// Copyright (C) 2026 Dcntrlized // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published // by the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Affero General Public License for more details. // // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . package containers import ( "context" "fmt" "log" "os" "strings" "time" "git.dcntrlized.org/Dcntrlized/Maestro/lib/docker" "git.dcntrlized.org/Dcntrlized/Maestro/lib/types" "github.com/docker/docker/api/types/container" "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 } key := normalizeName(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) } else if !shouldCache { log.Printf("Bypassing special network mode: %s", inspectData.HostConfig.NetworkMode) } else { containerNetwork = netInfo } } // Get Versions ver, _ := docker.GetContainerVersion(ctx, dc, containerID) // Fetch Manifest (best effort; fall back to basic info if it fails) // Infrastructure containers (e.g. the shared database) have no HTTP // manifest endpoint; keep their role marker and skip the fetch. role := event.Actor.Attributes["org.dcntrlized.role"] var manifest *types.Manifest if role != "db" { host := containerID[:12] port := event.Actor.Attributes["org.dcntrlized.port"] var fetchErr error manifest, fetchErr = docker.FetchManifestWithRetry(ctx, host, port, 20, 50*time.Millisecond) if fetchErr != nil { log.Printf("Error fetching manifest for container %s: %v", containerID, fetchErr) } } // Save to Cache log.Printf("Service started. Adding container %s - %s to cache", containerID[:12], containerName) existing, _ := cache.services[key] if existing.Version != "" && existing.Version != "latest" && (ver == "" || ver == "latest") { ver = existing.Version } capabilities := existing.Capabilities hasAuth := existing.HasAuth publicKey := existing.PublicKey if manifest != nil { capabilities = manifest.Capabilities hasAuth = manifest.HasAuth publicKey = manifest.PublicKey updateInstalledMetadataFromManifest(serviceLabel, manifest) } container := types.Container{ ID: containerID, ServiceName: serviceLabel, DockerName: containerName, Image: event.Actor.Attributes["image"], CreatedAt: fmt.Sprint(event.TimeNano), Network: containerNetwork, Status: "running", Version: ver, Role: role, Hidden: existing.Hidden || event.Actor.Attributes["org.dcntrlized.hidden"] == "true", Capabilities: capabilities, HasAuth: hasAuth, PublicKey: publicKey, Title: existing.Title, Vendor: existing.Vendor, Author: existing.Author, Description: existing.Description, Source: existing.Source, } if manifest != nil { container.Name = manifest.Name } else { container.Name = containerName } cache.services[key] = container case "die": if container, exists := cache.services[key]; exists { log.Printf("Service stopped: %s", containerName) container.Status = "stopped" cache.services[key] = container } case "destroy": if container, exists := cache.services[key]; exists { log.Printf("Service destroyed: %s", containerName) if IsInstalled(serviceLabel) || IsInstalled(container.ServiceName) || IsInstalled(container.Name) { container.Status = "installed" container.ID = "" container.DockerName = "" container.Network = types.Network{} cache.services[key] = container } else { // We already hold the lock; delete directly instead of // calling RemoveService (which re-locks and deadlocks). delete(cache.services, key) } } } cache.Unlock() cache.TriggerUpdate() } } } // imageTag returns the tag portion of an image reference, defaulting to latest. func imageTag(ref string) string { base := ref if i := strings.LastIndex(base, "/"); i >= 0 { base = base[i+1:] } if i := strings.LastIndex(base, ":"); i >= 0 { return base[i+1:] } return "latest" } // normalizeDockerState maps raw Docker container states onto the smaller status // vocabulary used by the frontend (running / starting / stopped / installed). func normalizeDockerState(state string) string { switch state { case "running": return "running" case "restarting", "starting": return "starting" case "exited", "stopped", "paused", "dead", "removing": return "stopped" case "created": return "installed" default: return "installed" } } // ReconcileContainers lists all dcntrlized-labeled containers (running or not) and // syncs their status into the cache, so state survives a Maestro restart. func (cache *ContainerCache) ReconcileContainers(ctx context.Context, dc *docker.DockerClient) { filter := filters.NewArgs() filter.Add("label", "org.dcntrlized.service") options := container.ListOptions{ All: true, Filters: filter, } existing, err := dc.Client.ContainerList(ctx, options) if err != nil { log.Printf("Failed to list existing containers during reconciliation: %v", err) return } for _, c := range existing { serviceLabel := c.Labels["org.dcntrlized.service"] if serviceLabel == "" || len(c.Names) == 0 { continue } key := normalizeName(serviceLabel) cache.Lock() entry, has := cache.services[key] if !has { entry = types.Container{ ServiceName: serviceLabel, Name: strings.TrimPrefix(c.Names[0], "/"), Image: c.Image, Version: imageTag(c.Image), Status: "installed", } } entry.Status = normalizeDockerState(c.State) entry.ID = c.ID entry.DockerName = strings.TrimPrefix(c.Names[0], "/") if role := c.Labels["org.dcntrlized.role"]; role != "" { entry.Role = role } if c.Labels["org.dcntrlized.hidden"] == "true" { entry.Hidden = true } // Running containers get their manifest fields (capabilities, has_auth, // public_key) fetched best-effort so a Maestro restart surfaces them even // when the container was already up before the watcher subscribed. if c.State == "running" && c.Labels["org.dcntrlized.role"] != "db" { if manifest, err := docker.FetchManifestWithRetry(ctx, c.ID[:12], c.Labels["org.dcntrlized.port"], 20, 50*time.Millisecond); err != nil { log.Printf("Error fetching manifest for container %s: %v", c.ID[:12], err) } else if manifest != nil { entry.Capabilities = manifest.Capabilities entry.HasAuth = manifest.HasAuth entry.PublicKey = manifest.PublicKey updateInstalledMetadataFromManifest(serviceLabel, manifest) } } cache.services[key] = entry cache.Unlock() } cache.TriggerUpdate() log.Printf("Reconciled %d existing containers with Docker", len(existing)) }