real init
This commit is contained in:
@@ -0,0 +1,117 @@
|
||||
package containers
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"git.dcntrlized.org/Dcntrlized/Maestro/lib/docker"
|
||||
"git.dcntrlized.org/Dcntrlized/Maestro/lib/types"
|
||||
"github.com/docker/docker/api/types/events"
|
||||
"github.com/docker/docker/api/types/filters"
|
||||
)
|
||||
|
||||
func (cache *ContainerCache) WatchContainers(ctx context.Context, dc *docker.DockerClient) {
|
||||
filter := filters.NewArgs()
|
||||
filter.Add("type", "container")
|
||||
filter.Add("event", "start")
|
||||
filter.Add("event", "die")
|
||||
filter.Add("event", "destroy")
|
||||
|
||||
messages, errs := dc.Client.Events(ctx, events.ListOptions{Filters: filter})
|
||||
|
||||
log.Println("Subscribed to Docker container events...")
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case err := <-errs:
|
||||
log.Printf("Docker event stream error: %v", err)
|
||||
os.Exit(1)
|
||||
return
|
||||
case event := <-messages:
|
||||
serviceLabel, ok := event.Actor.Attributes["org.dcntrlized.service"]
|
||||
if !ok {
|
||||
log.Default().Printf("Found unrelated service %s\n", serviceLabel)
|
||||
for attribute := range event.Actor.Attributes {
|
||||
log.Default().Printf("\tKey: %s\tValue: %s\n", attribute, event.Actor.Attributes[attribute])
|
||||
}
|
||||
continue
|
||||
}
|
||||
_ = serviceLabel
|
||||
containerID := event.Actor.ID
|
||||
containerName := event.Actor.Attributes["name"]
|
||||
|
||||
cache.Lock()
|
||||
|
||||
switch event.Action {
|
||||
case "start":
|
||||
// Inspect Container
|
||||
inspectData, inspectErr := dc.Client.ContainerInspect(ctx, containerID)
|
||||
|
||||
// Get Network information
|
||||
var containerNetwork types.Network
|
||||
|
||||
if inspectErr != nil {
|
||||
log.Printf("Failed to inspect container %s: %v", containerID, inspectErr)
|
||||
} else {
|
||||
netInfo, shouldCache, err := docker.ExtractNetworkInfo(ctx, dc, inspectData)
|
||||
if err != nil {
|
||||
log.Printf("Error extracting network details: %v", err)
|
||||
}
|
||||
if !shouldCache {
|
||||
fmt.Printf("Bypassing special network mode: %s\n", inspectData.HostConfig.NetworkMode)
|
||||
cache.Unlock()
|
||||
return
|
||||
}
|
||||
containerNetwork = netInfo
|
||||
}
|
||||
|
||||
// Get Versions
|
||||
ver, err := docker.GetContainerVersion(ctx, dc, containerID)
|
||||
|
||||
// Fetch Manifest
|
||||
host := containerID[:12]
|
||||
port := event.Actor.Attributes["org.dcntrlized.port"]
|
||||
manifest, err := docker.FetchManifestWithRetry(ctx, host, port, 20, 50*time.Millisecond)
|
||||
if err != nil {
|
||||
log.Printf("Error fetching manifest for container %s: %v", containerID, err)
|
||||
cache.Unlock()
|
||||
continue
|
||||
}
|
||||
|
||||
// Save to Cache
|
||||
log.Printf("Service started. Adding container %s - %s to cache", host, containerName)
|
||||
cache.services[containerID] = types.Container{
|
||||
ID: containerID,
|
||||
Name: manifest.Name,
|
||||
DockerName: containerName,
|
||||
Image: event.Actor.Attributes["image"],
|
||||
CreatedAt: fmt.Sprint(event.TimeNano),
|
||||
Network: containerNetwork,
|
||||
Status: "running",
|
||||
Version: ver,
|
||||
}
|
||||
|
||||
case "die":
|
||||
if container, exists := cache.services[containerID]; exists {
|
||||
log.Printf("Service stopping: %s", containerName)
|
||||
container.Status = "stopping"
|
||||
cache.services[containerID] = container
|
||||
}
|
||||
|
||||
case "destroy":
|
||||
if container, exists := cache.services[containerID]; exists {
|
||||
log.Printf("Service destroyed: %s", containerName)
|
||||
cache.RemoveService(container)
|
||||
}
|
||||
}
|
||||
|
||||
cache.Unlock()
|
||||
cache.TriggerUpdate()
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user