feat(agent): collect compose, swarm and container inventory from docker

internal/agent/inventory/collector.go

@@ -0,0 +1,193 @@
+// Package inventory turns what the local Docker daemon runs into the
+// services an agent reports to the master.
+package inventory
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/image"
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/docker/docker/api/types/system"
+
+	"github.com/shcizo/package-updater/internal/imageref"
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+const (
+	labelProject    = "com.docker.compose.project"
+	labelService    = "com.docker.compose.service"
+	labelWorkingDir = "com.docker.compose.project.working_dir"
+	labelSwarmTask  = "com.docker.swarm.service.id"
+)
+
+// DockerClient is the subset of the Docker SDK the collector needs.
+// *client.Client satisfies it.
+type DockerClient interface {
+	ContainerList(ctx context.Context, opts container.ListOptions) ([]container.Summary, error)
+	ImageList(ctx context.Context, opts image.ListOptions) ([]image.Summary, error)
+	ServiceList(ctx context.Context, opts swarm.ServiceListOptions) ([]swarm.Service, error)
+	Info(ctx context.Context) (system.Info, error)
+}
+
+// Collector reads the local daemon's inventory.
+type Collector struct {
+	cli    DockerClient
+	selfID string
+}
+
+// New returns a Collector. selfID is a prefix of the agent's own container
+// ID (inside a container, the hostname); that container is marked Self.
+func New(cli DockerClient, selfID string) *Collector {
+	return &Collector{cli: cli, selfID: selfID}
+}
+
+// Hello returns the machine facts sent on connect. AgentVersion is left for
+// the caller to fill in.
+func (c *Collector) Hello(ctx context.Context) (protocol.Hello, error) {
+	info, err := c.cli.Info(ctx)
+	if err != nil {
+		return protocol.Hello{}, fmt.Errorf("docker info: %w", err)
+	}
+	h := protocol.Hello{Hostname: info.Name, SwarmManager: info.Swarm.ControlAvailable}
+	if info.Swarm.Cluster != nil {
+		h.SwarmClusterID = info.Swarm.Cluster.ID
+	}
+	return h, nil
+}
+
+// Snapshot returns every service on this machine, sorted by Key.
+func (c *Collector) Snapshot(ctx context.Context) ([]protocol.Service, error) {
+	info, err := c.cli.Info(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("docker info: %w", err)
+	}
+	containers, err := c.cli.ContainerList(ctx, container.ListOptions{All: true})
+	if err != nil {
+		return nil, fmt.Errorf("docker container list: %w", err)
+	}
+	images, err := c.cli.ImageList(ctx, image.ListOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("docker image list: %w", err)
+	}
+	repoDigests := make(map[string][]string, len(images))
+	for _, img := range images {
+		repoDigests[img.ID] = img.RepoDigests
+	}
+
+	byKey := make(map[string]protocol.Service)
+	for _, ct := range containers {
+		if _, isTask := ct.Labels[labelSwarmTask]; isTask {
+			continue
+		}
+		svc := c.fromContainer(ct, repoDigests[ct.ImageID])
+		if prev, ok := byKey[svc.Key()]; ok {
+			svc = mergeReplicas(prev, svc)
+		}
+		byKey[svc.Key()] = svc
+	}
+
+	if info.Swarm.ControlAvailable {
+		services, err := c.cli.ServiceList(ctx, swarm.ServiceListOptions{Status: true})
+		if err != nil {
+			return nil, fmt.Errorf("docker service list: %w", err)
+		}
+		for _, s := range services {
+			if s.Spec.TaskTemplate.ContainerSpec == nil {
+				continue
+			}
+			svc := fromSwarmService(s)
+			byKey[svc.Key()] = svc
+		}
+	}
+
+	out := make([]protocol.Service, 0, len(byKey))
+	for _, s := range byKey {
+		out = append(out, s)
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Key() < out[j].Key() })
+	return out, nil
+}
+
+func (c *Collector) fromContainer(ct container.Summary, repoDigests []string) protocol.Service {
+	svc := protocol.Service{
+		Image:  ct.Image,
+		Digest: digestFor(ct.Image, repoDigests),
+		Status: containerStatus(ct.State),
+		Self:   c.selfID != "" && strings.HasPrefix(ct.ID, c.selfID),
+	}
+	project, service := ct.Labels[labelProject], ct.Labels[labelService]
+	if project != "" && service != "" {
+		svc.Kind = protocol.KindCompose
+		svc.ComposeProject = project
+		svc.ComposeService = service
+		svc.WorkingDir = ct.Labels[labelWorkingDir]
+		return svc
+	}
+	svc.Kind = protocol.KindContainer
+	svc.ContainerName = ct.ID
+	if len(ct.Names) > 0 {
+		svc.ContainerName = strings.TrimPrefix(ct.Names[0], "/")
+	}
+	return svc
+}
+
+func fromSwarmService(s swarm.Service) protocol.Service {
+	ref := s.Spec.TaskTemplate.ContainerSpec.Image
+	return protocol.Service{
+		Kind:         protocol.KindSwarm,
+		SwarmService: s.Spec.Name,
+		Image:        ref,
+		Digest:       imageref.Parse(ref).Digest,
+		Status:       swarmStatus(s.ServiceStatus),
+	}
+}
+
+// mergeReplicas folds a second container of the same Compose service into
+// the first: the service counts as running if any replica runs.
+func mergeReplicas(a, b protocol.Service) protocol.Service {
+	if b.Status == "running" {
+		a.Status = "running"
+	}
+	a.Self = a.Self || b.Self
+	return a
+}
+
+// digestFor returns the registry digest the container's image was pulled
+// as. A reference that already pins a digest wins; otherwise the image's
+// RepoDigests entry for the same repository is used.
+func digestFor(ref string, repoDigests []string) string {
+	r := imageref.Parse(ref)
+	if r.Digest != "" {
+		return r.Digest
+	}
+	for _, rd := range repoDigests {
+		if p := imageref.Parse(rd); p.Repo == r.Repo {
+			return p.Digest
+		}
+	}
+	return ""
+}
+
+func containerStatus(state string) string {
+	if state == "running" {
+		return "running"
+	}
+	return "stopped"
+}
+
+func swarmStatus(st *swarm.ServiceStatus) string {
+	switch {
+	case st == nil:
+		return "unknown"
+	case st.DesiredTasks == 0:
+		return "stopped"
+	case st.RunningTasks >= st.DesiredTasks:
+		return "running"
+	default:
+		return "degraded"
+	}
+}

internal/agent/inventory/collector_test.go

@@ -0,0 +1,170 @@
+package inventory
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/image"
+	"github.com/docker/docker/api/types/swarm"
+	"github.com/docker/docker/api/types/system"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+type fakeDocker struct {
+	containers       []container.Summary
+	images           []image.Summary
+	services         []swarm.Service
+	info             system.Info
+	containerErr     error
+	serviceListCalls int
+}
+
+func (f *fakeDocker) ContainerList(context.Context, container.ListOptions) ([]container.Summary, error) {
+	return f.containers, f.containerErr
+}
+
+func (f *fakeDocker) ImageList(context.Context, image.ListOptions) ([]image.Summary, error) {
+	return f.images, nil
+}
+
+func (f *fakeDocker) ServiceList(context.Context, swarm.ServiceListOptions) ([]swarm.Service, error) {
+	f.serviceListCalls++
+	return f.services, nil
+}
+
+func (f *fakeDocker) Info(context.Context) (system.Info, error) { return f.info, nil }
+
+func composeContainer(id, project, service, img, imageID, state string) container.Summary {
+	return container.Summary{
+		ID: id, Image: img, ImageID: imageID, State: state, Names: []string{"/" + project + "-" + service + "-1"},
+		Labels: map[string]string{
+			labelProject:    project,
+			labelService:    service,
+			labelWorkingDir: "/home/shcizo/self-hosted/" + project,
+		},
+	}
+}
+
+func managerInfo() system.Info {
+	var info system.Info
+	info.Name = "swarm-mgr"
+	info.Swarm.ControlAvailable = true
+	info.Swarm.Cluster = &swarm.ClusterInfo{ID: "cluster-1"}
+	return info
+}
+
+func TestSnapshot_ComposeContainerWithRepoDigest(t *testing.T) {
+	cli := &fakeDocker{
+		containers: []container.Summary{composeContainer("abc123", "slusk", "web", "gitea.shcizo.se/shcizo/slusk:1.4.2", "sha256:img1", "running")},
+		images:     []image.Summary{{ID: "sha256:img1", RepoDigests: []string{"gitea.shcizo.se/shcizo/slusk@sha256:d1"}}},
+	}
+	got, err := New(cli, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, []protocol.Service{{
+		Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web",
+		WorkingDir: "/home/shcizo/self-hosted/slusk",
+		Image:      "gitea.shcizo.se/shcizo/slusk:1.4.2", Digest: "sha256:d1", Status: "running",
+	}}, got)
+}
+
+func TestSnapshot_ReplicasMergeAndRunningWins(t *testing.T) {
+	cli := &fakeDocker{containers: []container.Summary{
+		composeContainer("a1", "slusk", "worker", "slusk:1", "sha256:i", "exited"),
+		composeContainer("a2", "slusk", "worker", "slusk:1", "sha256:i", "running"),
+	}}
+	got, err := New(cli, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.Equal(t, "running", got[0].Status)
+}
+
+func TestSnapshot_PlainContainerAndSwarmTaskSkipped(t *testing.T) {
+	cli := &fakeDocker{containers: []container.Summary{
+		{ID: "p1", Names: []string{"/adhoc"}, Image: "redis:alpine", State: "exited"},
+		{ID: "t1", Names: []string{"/gitea_gitea.1.x"}, Image: "gitea/gitea:1.27.3", State: "running",
+			Labels: map[string]string{labelSwarmTask: "svc1"}},
+	}}
+	got, err := New(cli, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, []protocol.Service{{
+		Kind: protocol.KindContainer, ContainerName: "adhoc", Image: "redis:alpine", Status: "stopped",
+	}}, got)
+}
+
+func TestSnapshot_SelfFlag(t *testing.T) {
+	cli := &fakeDocker{containers: []container.Summary{
+		composeContainer("0123456789abcdef", "updater-agent", "agent", "pu:1", "sha256:i", "running"),
+	}}
+	got, err := New(cli, "0123456789ab").Snapshot(context.Background())
+	require.NoError(t, err)
+	require.Len(t, got, 1)
+	assert.True(t, got[0].Self)
+}
+
+func TestSnapshot_SwarmServicesOnlyOnManager(t *testing.T) {
+	svc := swarm.Service{
+		Spec: swarm.ServiceSpec{
+			Annotations:  swarm.Annotations{Name: "gitea_gitea"},
+			TaskTemplate: swarm.TaskSpec{ContainerSpec: &swarm.ContainerSpec{Image: "gitea/gitea:1.27.3@sha256:87a6"}},
+		},
+		ServiceStatus: &swarm.ServiceStatus{RunningTasks: 1, DesiredTasks: 1},
+	}
+
+	worker := &fakeDocker{services: []swarm.Service{svc}}
+	got, err := New(worker, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	assert.Empty(t, got)
+	assert.Equal(t, 0, worker.serviceListCalls, "workers must not call the manager-only API")
+
+	manager := &fakeDocker{services: []swarm.Service{svc}, info: managerInfo()}
+	got, err = New(manager, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, []protocol.Service{{
+		Kind: protocol.KindSwarm, SwarmService: "gitea_gitea",
+		Image: "gitea/gitea:1.27.3@sha256:87a6", Digest: "sha256:87a6", Status: "running",
+	}}, got)
+}
+
+func TestSwarmStatus(t *testing.T) {
+	cases := []struct {
+		name string
+		in   *swarm.ServiceStatus
+		want string
+	}{
+		{"no status", nil, "unknown"},
+		{"scaled to zero", &swarm.ServiceStatus{DesiredTasks: 0}, "stopped"},
+		{"all running", &swarm.ServiceStatus{RunningTasks: 2, DesiredTasks: 2}, "running"},
+		{"some running", &swarm.ServiceStatus{RunningTasks: 1, DesiredTasks: 2}, "degraded"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) { assert.Equal(t, c.want, swarmStatus(c.in)) })
+	}
+}
+
+func TestSnapshot_SortedByKey(t *testing.T) {
+	cli := &fakeDocker{containers: []container.Summary{
+		{ID: "2", Names: []string{"/zeta"}, Image: "z", State: "running"},
+		{ID: "1", Names: []string{"/alpha"}, Image: "a", State: "running"},
+	}}
+	got, err := New(cli, "").Snapshot(context.Background())
+	require.NoError(t, err)
+	require.Len(t, got, 2)
+	assert.Equal(t, "alpha", got[0].ContainerName)
+}
+
+func TestSnapshot_PropagatesDockerErrors(t *testing.T) {
+	cli := &fakeDocker{containerErr: errors.New("daemon down")}
+	_, err := New(cli, "").Snapshot(context.Background())
+	assert.ErrorContains(t, err, "daemon down")
+}
+
+func TestHello(t *testing.T) {
+	got, err := New(&fakeDocker{info: managerInfo()}, "").Hello(context.Background())
+	require.NoError(t, err)
+	assert.Equal(t, protocol.Hello{Hostname: "swarm-mgr", SwarmManager: true, SwarmClusterID: "cluster-1"}, got)
+}