diff --git a/internal/agent/inventory/diff.go b/internal/agent/inventory/diff.go new file mode 100644 index 0000000..d3fb08a --- /dev/null +++ b/internal/agent/inventory/diff.go @@ -0,0 +1,25 @@ +package inventory + +import "github.com/shcizo/package-updater/internal/protocol" + +// Diff compares two snapshots. upserts holds services that are new or +// changed in next; removes holds the keys of services missing from next. +func Diff(prev, next []protocol.Service) (upserts []protocol.Service, removes []string) { + old := make(map[string]protocol.Service, len(prev)) + for _, s := range prev { + old[s.Key()] = s + } + seen := make(map[string]struct{}, len(next)) + for _, s := range next { + seen[s.Key()] = struct{}{} + if o, ok := old[s.Key()]; !ok || o != s { + upserts = append(upserts, s) + } + } + for _, s := range prev { + if _, ok := seen[s.Key()]; !ok { + removes = append(removes, s.Key()) + } + } + return upserts, removes +} diff --git a/internal/agent/inventory/diff_test.go b/internal/agent/inventory/diff_test.go new file mode 100644 index 0000000..a71140c --- /dev/null +++ b/internal/agent/inventory/diff_test.go @@ -0,0 +1,37 @@ +package inventory + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/shcizo/package-updater/internal/protocol" +) + +func TestDiff(t *testing.T) { + a := protocol.Service{Kind: protocol.KindContainer, ContainerName: "a", Image: "a:1", Status: "running"} + b := protocol.Service{Kind: protocol.KindContainer, ContainerName: "b", Image: "b:1", Status: "running"} + bStopped := b + bStopped.Status = "stopped" + c := protocol.Service{Kind: protocol.KindContainer, ContainerName: "c", Image: "c:1", Status: "running"} + + cases := []struct { + name string + prev, next []protocol.Service + wantUpserts []protocol.Service + wantRemoves []string + }{ + {"unchanged", []protocol.Service{a, b}, []protocol.Service{a, b}, nil, nil}, + {"added", []protocol.Service{a}, []protocol.Service{a, c}, []protocol.Service{c}, nil}, + {"changed", []protocol.Service{a, b}, []protocol.Service{a, bStopped}, []protocol.Service{bStopped}, nil}, + {"removed", []protocol.Service{a, b}, []protocol.Service{a}, nil, []string{"container|b"}}, + {"from empty", nil, []protocol.Service{a}, []protocol.Service{a}, nil}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + ups, rms := Diff(tc.prev, tc.next) + assert.Equal(t, tc.wantUpserts, ups) + assert.Equal(t, tc.wantRemoves, rms) + }) + } +} diff --git a/internal/agent/inventory/watch.go b/internal/agent/inventory/watch.go new file mode 100644 index 0000000..4a4f5d4 --- /dev/null +++ b/internal/agent/inventory/watch.go @@ -0,0 +1,52 @@ +package inventory + +import ( + "context" + "fmt" + "time" + + "github.com/docker/docker/api/types/events" + "github.com/docker/docker/api/types/filters" +) + +// EventSource is the subset of the Docker SDK Watch needs. +// *client.Client satisfies it. +type EventSource interface { + Events(ctx context.Context, opts events.ListOptions) (<-chan events.Message, <-chan error) +} + +// Watch subscribes to container and service lifecycle events and calls +// onChange once after each burst has been quiet for the given duration. +// A Compose restart emits a dozen events; this turns them into one +// inventory refresh. Returns nil when ctx ends, or an error if the event +// stream breaks (for example when the daemon restarts) — the caller should +// resubscribe and refresh, since events may have been missed. +func Watch(ctx context.Context, src EventSource, quiet time.Duration, onChange func()) error { + args := filters.NewArgs( + filters.Arg("type", string(events.ContainerEventType)), + filters.Arg("type", string(events.ServiceEventType)), + ) + for _, action := range []string{"create", "start", "stop", "die", "destroy", "rename", "update", "remove"} { + args.Add("event", action) + } + msgs, errs := src.Events(ctx, events.ListOptions{Filters: args}) + + timer := time.NewTimer(quiet) + timer.Stop() + defer timer.Stop() + for { + select { + case <-ctx.Done(): + return nil + case err := <-errs: + if ctx.Err() != nil { + return nil + } + return fmt.Errorf("docker events: %w", err) + case <-msgs: + timer.Reset(quiet) + case <-timer.C: + onChange() + } + } +} diff --git a/internal/agent/inventory/watch_test.go b/internal/agent/inventory/watch_test.go new file mode 100644 index 0000000..df7c9f1 --- /dev/null +++ b/internal/agent/inventory/watch_test.go @@ -0,0 +1,52 @@ +package inventory + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/docker/docker/api/types/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type fakeEvents struct { + msgs chan events.Message + errs chan error +} + +func newFakeEvents() *fakeEvents { + return &fakeEvents{msgs: make(chan events.Message, 10), errs: make(chan error, 1)} +} + +func (f *fakeEvents) Events(context.Context, events.ListOptions) (<-chan events.Message, <-chan error) { + return f.msgs, f.errs +} + +func TestWatch_DebouncesBursts(t *testing.T) { + src := newFakeEvents() + var calls atomic.Int32 + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- Watch(ctx, src, 30*time.Millisecond, func() { calls.Add(1) }) }() + + for range 3 { + src.msgs <- events.Message{Type: events.ContainerEventType, Action: "start"} + } + require.Eventually(t, func() bool { return calls.Load() == 1 }, time.Second, 5*time.Millisecond) + + src.msgs <- events.Message{Type: events.ServiceEventType, Action: "update"} + require.Eventually(t, func() bool { return calls.Load() == 2 }, time.Second, 5*time.Millisecond) + + cancel() + assert.NoError(t, <-done) +} + +func TestWatch_ReturnsStreamError(t *testing.T) { + src := newFakeEvents() + src.errs <- errors.New("daemon restarted") + err := Watch(context.Background(), src, time.Millisecond, func() {}) + assert.ErrorContains(t, err, "daemon restarted") +}