53 lines
1.5 KiB
Go
53 lines
1.5 KiB
Go
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()
|
|
}
|
|
}
|
|
}
|