diff --git a/Makefile b/Makefile index 2a628c65..ddb78a4b 100644 --- a/Makefile +++ b/Makefile @@ -19,7 +19,7 @@ PROD_COMPOSE := $(shell test -f "docker-compose.test.prod.yml" && echo "docker-c .DEFAULT_GOAL := binary -.PHONY: deps clean-data clean-webui webui binary binary-linux-amd64 binary-linux-arm64 test vet test-race dev dev-infisical prod prod-infisical sql generate docker docker-distroless tools +.PHONY: deps clean-data clean-webui webui binary binary-linux-amd64 binary-linux-arm64 test vet test-race dev prod sql generate docker docker-distroless tools # Tools tools: diff --git a/internal/service/docker_service.go b/internal/service/docker_service.go index 800e8d17..6ea8a968 100644 --- a/internal/service/docker_service.go +++ b/internal/service/docker_service.go @@ -5,9 +5,12 @@ import ( "errors" "fmt" "os" + "sync" "time" "github.com/cenkalti/backoff/v5" + "github.com/docker/docker/api/types/events" + "github.com/docker/docker/api/types/filters" "github.com/steveiliop56/ding" "github.com/tinyauthapp/tinyauth/internal/model" "github.com/tinyauthapp/tinyauth/internal/utils/decoders" @@ -23,11 +26,12 @@ var ( ) type DockerService struct { - log *logger.Logger - client *client.Client - context context.Context + log *logger.Logger + client *client.Client + ctx context.Context - isConnected bool + apps map[string]model.App + mu sync.RWMutex } type DockerServiceInput struct { @@ -40,61 +44,68 @@ type DockerServiceInput struct { func NewDockerService(i DockerServiceInput) (*DockerService, error) { service := &DockerService{ - log: i.Log, - context: i.Ctx, + log: i.Log, + ctx: i.Ctx, + apps: make(map[string]model.App), + mu: sync.RWMutex{}, } service.log.App.Debug().Msg("Attempting to connect to Docker") - if os.Getenv("DOCKER_HOST") == "" { - cli, err := service.connect() - if err != nil { - if errors.Is(err, ErrPingFailed) { - service.log.App.Debug().Msg("Docker not connected") - return nil, nil - } - return nil, fmt.Errorf("failed to connect to docker: %w", err) - } - service.client = cli + var cli *client.Client + var err error + + if os.Getenv("DOCKER_HOST") != "" { + cli, err = service.connectWithBackoff() } else { - exp := backoff.NewExponentialBackOff() - exp.InitialInterval = 3 * time.Second - exp.RandomizationFactor = 0.1 - exp.Multiplier = 1.5 - exp.Reset() - - operation := func() (*client.Client, error) { - if service.client != nil { - service.client.Close() - } - cli, err := service.connect() - if err != nil { - return nil, err - } - return cli, nil + cli, err = service.connect() + } + + if err != nil { + if errors.Is(err, ErrPingFailed) { + service.log.App.Debug().Msg("Docker not connected, skipping") + return nil, nil } + return nil, fmt.Errorf("failed to connect to docker: %w", err) + } - cli, err := backoff.Retry(service.context, operation, backoff.WithBackOff(exp), backoff.WithMaxTries(3)) + if cli == nil { + return nil, nil + } - if err != nil { - if errors.Is(err, ErrPingFailed) { - service.log.App.Debug().Msg("Docker not connected after retrying") - return nil, nil - } - return nil, fmt.Errorf("failed to connect to docker after retrying: %w", err) - } + service.client = cli + + // initial sync + err = service.sync() - service.client = cli + if err != nil { + return nil, fmt.Errorf("failed to do initial sync: %w", err) } - service.isConnected = true service.log.App.Debug().Msg("Docker connected successfully") - i.Ding.Go(service.watchAndClose, ding.RingMajor) + i.Ding.Go(service.monitor, ding.RingMajor) return service, nil } +func (docker *DockerService) connectWithBackoff() (*client.Client, error) { + exp := backoff.NewExponentialBackOff() + exp.InitialInterval = 3 * time.Second + exp.RandomizationFactor = 0.1 + exp.Multiplier = 1.5 + exp.Reset() + + operation := func() (*client.Client, error) { + if docker.client != nil { + docker.client.Close() + } + return docker.connect() + } + + return backoff.Retry(docker.ctx, operation, backoff.WithBackOff(exp), backoff.WithMaxTries(3)) +} + func (docker *DockerService) connect() (*client.Client, error) { cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) @@ -102,9 +113,10 @@ func (docker *DockerService) connect() (*client.Client, error) { return nil, err } - _, err = cli.Ping(docker.context) + _, err = cli.Ping(docker.ctx) if err != nil { + cli.Close() return nil, ErrPingFailed } @@ -112,52 +124,163 @@ func (docker *DockerService) connect() (*client.Client, error) { } func (docker *DockerService) getContainers() ([]container.Summary, error) { - return docker.client.ContainerList(docker.context, container.ListOptions{}) + return docker.client.ContainerList(docker.ctx, container.ListOptions{}) } func (docker *DockerService) inspectContainer(containerId string) (container.InspectResponse, error) { - return docker.client.ContainerInspect(docker.context, containerId) + return docker.client.ContainerInspect(docker.ctx, containerId) } -func (docker *DockerService) Lookup(locator func(name string, app *model.App)) error { - if !docker.isConnected { - docker.log.App.Debug().Msg("Docker service not connected, returning empty labels") - return nil - } - +func (docker *DockerService) sync() error { containers, err := docker.getContainers() if err != nil { return fmt.Errorf("failed to get containers: %w", err) } + apps := make(map[string]model.App) + for _, ctr := range containers { - inspect, err := docker.inspectContainer(ctr.ID) + labels, err := decoders.DecodeLabels[model.Apps](ctr.Labels, "apps") if err != nil { - docker.log.App.Error().Err(err).Msgf("Failed to inspect container %s", ctr.ID) + docker.log.App.Warn().Err(err).Str("container_id", ctr.ID).Msg("Failed to decode container labels, skipping") continue } - labels, err := decoders.DecodeLabels[model.Apps](inspect.Config.Labels, "apps") - if err != nil { - docker.log.App.Warn().Err(err).Msgf("Failed to decode labels for container %s", ctr.ID) + for app, config := range labels.Apps { + apps[app] = config + } + } + + docker.mu.Lock() + docker.apps = apps + docker.mu.Unlock() + + return nil +} + +func (docker *DockerService) syncWithContainer(containerId string) error { + ctr, err := docker.inspectContainer(containerId) + if err != nil { + return fmt.Errorf("failed to inspect container: %w", err) + } + labels, err := decoders.DecodeLabels[model.Apps](ctr.Config.Labels, "apps") + if err != nil { + return fmt.Errorf("failed to decode container labels: %w", err) + } + docker.mu.Lock() + for app, config := range labels.Apps { + if !ctr.State.Running { + delete(docker.apps, app) continue } + docker.apps[app] = config + } + docker.mu.Unlock() + return nil +} - for app, config := range labels.Apps { - locator(app, &config) +func (docker *DockerService) syncFromEvent(event events.Message) { + switch event.Action { + case events.ActionStart, events.ActionRestart, + events.ActionStop, events.ActionKill, events.ActionDie: + + if err := docker.syncWithContainer(event.Actor.ID); err != nil { + docker.log.App.Warn().Err(err). + Msg("Failed to sync container, performing full sync") + + if err := docker.sync(); err != nil { + docker.log.App.Error().Err(err). + Msg("Failed to resync Docker apps") + } } } +} - return nil +func (docker *DockerService) buildEventsKV() filters.Args { + vals := []string{"start", "restart", "stop", "kill", "die"} + kv := make([]filters.KeyValuePair, len(vals)) + for i, v := range vals { + kv[i] = filters.KeyValuePair{ + Key: "event", + Value: v, + } + } + return filters.NewArgs(kv...) } -func (docker *DockerService) watchAndClose(ctx context.Context) { - <-ctx.Done() - docker.log.App.Debug().Msg("Closing Docker client") - if docker.client != nil { - err := docker.client.Close() - if err != nil { - docker.log.App.Error().Err(err).Msg("Error closing Docker client") +func (docker *DockerService) monitor(ctx context.Context) { + defer func() { + if docker.client != nil { + _ = docker.client.Close() } + }() + + for ctx.Err() == nil { + streamCtx, cancel := context.WithCancel(ctx) + + stream, chErr := docker.client.Events(streamCtx, events.ListOptions{ + Filters: docker.buildEventsKV(), + }) + + if err := docker.sync(); err != nil { + docker.log.App.Error().Err(err).Msg("Failed to sync docker apps") + } else { + watch: + for { + select { + case <-ctx.Done(): + cancel() + docker.log.App.Debug().Msg("Shutting down docker monitor") + return + + case event, ok := <-stream: + if !ok { + docker.log.App.Warn().Msg("Docker event stream closed") + break watch + } + docker.syncFromEvent(event) + + case err, ok := <-chErr: + if ok && err != nil { + docker.log.App.Error().Err(err).Msg("Docker event stream failed") + } + break watch + } + } + } + + cancel() + + for ctx.Err() == nil { + docker.log.App.Warn().Msg("Attempting to reconnect to docker") + + cli, err := docker.connect() + + if err == nil && cli != nil { + docker.client.Close() + docker.client = cli + docker.log.App.Info().Msg("Docker connection recovered") + break + } + + if err != nil { + docker.log.App.Error().Err(err).Msg("Failed to reconnect to docker") + } + + select { + case <-ctx.Done(): + return + case <-time.After(5 * time.Second): + } + } + } +} + +func (docker *DockerService) Lookup(locator func(name string, app *model.App)) error { + docker.mu.RLock() + for app, config := range docker.apps { + locator(app, &config) } + docker.mu.RUnlock() + return nil }