From d3f825a3663a2f1d8d46d506e3e242d84d0e34bd Mon Sep 17 00:00:00 2001 From: Stavros Date: Tue, 13 Oct 2026 20:44:04 +0300 Subject: [PATCH 1/4] refactor: cache docker labels and monitor for changes --- Makefile | 2 +- internal/service/docker_service.go | 120 +++++++++++++++++++++++------ 2 files changed, 98 insertions(+), 24 deletions(-) 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..78cecab1 100644 --- a/internal/service/docker_service.go +++ b/internal/service/docker_service.go @@ -8,6 +8,8 @@ import ( "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" @@ -27,7 +29,7 @@ type DockerService struct { client *client.Client context context.Context - isConnected bool + apps map[string]model.App } type DockerServiceInput struct { @@ -42,6 +44,7 @@ func NewDockerService(i DockerServiceInput) (*DockerService, error) { service := &DockerService{ log: i.Log, context: i.Ctx, + apps: make(map[string]model.App), } service.log.App.Debug().Msg("Attempting to connect to Docker") @@ -87,10 +90,16 @@ func NewDockerService(i DockerServiceInput) (*DockerService, error) { service.client = cli } - service.isConnected = true + // initial sync + err := service.initialSync() + + if err != nil { + return nil, fmt.Errorf("failed to do initial sync: %w", err) + } + 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 } @@ -120,44 +129,109 @@ func (docker *DockerService) inspectContainer(containerId string) (container.Ins } 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 + for app, config := range docker.apps { + locator(app, &config) } + return nil +} + +func (docker *DockerService) initialSync() error { containers, err := docker.getContainers() if err != nil { return fmt.Errorf("failed to get containers: %w", err) } for _, ctr := range containers { - inspect, err := docker.inspectContainer(ctr.ID) + err = docker.updateFromContainerID(ctr.ID, false) if err != nil { - docker.log.App.Error().Err(err).Msgf("Failed to inspect container %s", ctr.ID) - continue + return fmt.Errorf("failed to update apps from container: %w", err) } + } - 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) - continue - } + return nil +} + +func (docker *DockerService) updateFromContainerID(containerId string, shouldDelete bool) error { + if containerId == "" { + return nil + } - for app, config := range labels.Apps { - locator(app, &config) + inspect, err := docker.inspectContainer(containerId) + if err != nil { + return fmt.Errorf("failed to inspect container: %w", err) + } + + labels, err := decoders.DecodeLabels[model.Apps](inspect.Config.Labels, "apps") + if err != nil { + return fmt.Errorf("failed to decode labels: %w", err) + } + + for app, config := range labels.Apps { + if shouldDelete { + delete(docker.apps, app) + continue } + docker.apps[app] = config } return nil } -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) updateFromEvent(event events.Message) { + docker.log.App.Debug().Msgf("Received event: %s", event.Action) + + containerId := event.Actor.ID + + var err error + + switch event.Action { + case events.ActionStart, events.ActionRestart: + err = docker.updateFromContainerID(containerId, false) + case events.ActionStop, events.ActionKill: + err = docker.updateFromContainerID(containerId, true) + default: + docker.log.App.Debug().Msgf("Ignoring event: %s", event.Action) + } + + if err != nil { + docker.log.App.Error().Err(err).Msg("Failed to update apps from event") + } +} + +func (docker *DockerService) buildEventsKV() filters.Args { + vals := []string{"start", "restart", "stop", "kill"} + 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) monitor(ctx context.Context) { + defer docker.client.Close() + + stream, chErr := docker.client.Events(ctx, events.ListOptions{ + Filters: docker.buildEventsKV(), + }) + + for { + select { + case <-ctx.Done(): + docker.log.App.Debug().Msg("Shutting down docker monitor") + return + + case event := <-stream: + docker.updateFromEvent(event) + + case err := <-chErr: + if err != nil { + docker.log.App.Error().Err(err).Msg("Docker event stream failed") + return + } } } } From 142590dd640b5050c5c2144005239081317d7a4d Mon Sep 17 00:00:00 2001 From: Stavros Date: Tue, 13 Oct 2026 21:01:35 +0300 Subject: [PATCH 2/4] refactor: no need to get and inspect on initial sync --- internal/service/docker_service.go | 67 +++++++++++++----------------- 1 file changed, 28 insertions(+), 39 deletions(-) diff --git a/internal/service/docker_service.go b/internal/service/docker_service.go index 78cecab1..ae00dd71 100644 --- a/internal/service/docker_service.go +++ b/internal/service/docker_service.go @@ -91,7 +91,7 @@ func NewDockerService(i DockerServiceInput) (*DockerService, error) { } // initial sync - err := service.initialSync() + err := service.sync() if err != nil { return nil, fmt.Errorf("failed to do initial sync: %w", err) @@ -128,75 +128,56 @@ func (docker *DockerService) inspectContainer(containerId string) (container.Ins return docker.client.ContainerInspect(docker.context, containerId) } -func (docker *DockerService) Lookup(locator func(name string, app *model.App)) error { - for app, config := range docker.apps { - locator(app, &config) - } - - return nil -} - -func (docker *DockerService) initialSync() error { +func (docker *DockerService) sync() error { containers, err := docker.getContainers() if err != nil { return fmt.Errorf("failed to get containers: %w", err) } for _, ctr := range containers { - err = docker.updateFromContainerID(ctr.ID, false) + labels, err := decoders.DecodeLabels[model.Apps](ctr.Labels, "apps") if err != nil { - return fmt.Errorf("failed to update apps from container: %w", err) + docker.log.App.Warn().Err(err).Str("container_id", ctr.ID).Msg("Failed to decode container labels, skipping") + continue + } + for app, config := range labels.Apps { + docker.apps[app] = config } } return nil } -func (docker *DockerService) updateFromContainerID(containerId string, shouldDelete bool) error { - if containerId == "" { - return nil - } - - inspect, err := docker.inspectContainer(containerId) +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](inspect.Config.Labels, "apps") + labels, err := decoders.DecodeLabels[model.Apps](ctr.Config.Labels, "apps") if err != nil { - return fmt.Errorf("failed to decode labels: %w", err) + return fmt.Errorf("failed to decode container labels: %w", err) } - for app, config := range labels.Apps { - if shouldDelete { + if !ctr.State.Running { delete(docker.apps, app) continue } docker.apps[app] = config } - return nil } - -func (docker *DockerService) updateFromEvent(event events.Message) { +func (docker *DockerService) syncFromEvent(event events.Message) { docker.log.App.Debug().Msgf("Received event: %s", event.Action) - containerId := event.Actor.ID - - var err error - switch event.Action { - case events.ActionStart, events.ActionRestart: - err = docker.updateFromContainerID(containerId, false) - case events.ActionStop, events.ActionKill: - err = docker.updateFromContainerID(containerId, true) + case events.ActionStart, events.ActionRestart, events.ActionStop, events.ActionKill: + err := docker.syncWithContainer(event.Actor.ID) + if err != nil { + docker.log.App.Error().Err(err).Msg("Failed to update apps from event") + } default: docker.log.App.Debug().Msgf("Ignoring event: %s", event.Action) } - - if err != nil { - docker.log.App.Error().Err(err).Msg("Failed to update apps from event") - } } func (docker *DockerService) buildEventsKV() filters.Args { @@ -225,7 +206,7 @@ func (docker *DockerService) monitor(ctx context.Context) { return case event := <-stream: - docker.updateFromEvent(event) + docker.syncFromEvent(event) case err := <-chErr: if err != nil { @@ -235,3 +216,11 @@ func (docker *DockerService) monitor(ctx context.Context) { } } } + +func (docker *DockerService) Lookup(locator func(name string, app *model.App)) error { + for app, config := range docker.apps { + locator(app, &config) + } + + return nil +} From 952b50d344c9ea63511ed3eb8eb03ebca9df821c Mon Sep 17 00:00:00 2001 From: Stavros Date: Wed, 14 Oct 2026 16:22:51 +0300 Subject: [PATCH 3/4] fix: rabbit comments --- internal/service/docker_service.go | 202 +++++++++++++++++++---------- 1 file changed, 130 insertions(+), 72 deletions(-) diff --git a/internal/service/docker_service.go b/internal/service/docker_service.go index ae00dd71..5488d0fe 100644 --- a/internal/service/docker_service.go +++ b/internal/service/docker_service.go @@ -4,7 +4,7 @@ import ( "context" "errors" "fmt" - "os" + "sync" "time" "github.com/cenkalti/backoff/v5" @@ -25,11 +25,12 @@ var ( ) type DockerService struct { - log *logger.Logger - client *client.Client - context context.Context + log *logger.Logger + client *client.Client + ctx context.Context apps map[string]model.App + mu sync.RWMutex } type DockerServiceInput struct { @@ -42,56 +43,28 @@ type DockerServiceInput struct { func NewDockerService(i DockerServiceInput) (*DockerService, error) { service := &DockerService{ - log: i.Log, - context: i.Ctx, - apps: make(map[string]model.App), + 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 - } 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 := backoff.Retry(service.context, operation, backoff.WithBackOff(exp), backoff.WithMaxTries(3)) + cli, err := service.connectWithBackoff() - 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) - } + if err != nil { + return nil, fmt.Errorf("failed to connect to docker: %w", err) + } - service.client = cli + if cli == nil { + return nil, nil } + service.client = cli + // initial sync - err := service.sync() + err = service.sync() if err != nil { return nil, fmt.Errorf("failed to do initial sync: %w", err) @@ -104,6 +77,33 @@ func NewDockerService(i DockerServiceInput) (*DockerService, error) { 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() + } + + cli, err := backoff.Retry(docker.ctx, operation, backoff.WithBackOff(exp), backoff.WithMaxTries(3)) + + if err != nil { + if errors.Is(err, ErrPingFailed) { + docker.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) + } + + return cli, nil +} + func (docker *DockerService) connect() (*client.Client, error) { cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) @@ -111,9 +111,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 } @@ -121,11 +122,11 @@ 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) sync() error { @@ -134,17 +135,24 @@ func (docker *DockerService) sync() error { return fmt.Errorf("failed to get containers: %w", err) } + apps := make(map[string]model.App) + for _, ctr := range containers { labels, err := decoders.DecodeLabels[model.Apps](ctr.Labels, "apps") if err != nil { docker.log.App.Warn().Err(err).Str("container_id", ctr.ID).Msg("Failed to decode container labels, skipping") continue } + for app, config := range labels.Apps { - docker.apps[app] = config + apps[app] = config } } + docker.mu.Lock() + docker.apps = apps + docker.mu.Unlock() + return nil } @@ -157,6 +165,7 @@ func (docker *DockerService) syncWithContainer(containerId string) error { 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) @@ -164,24 +173,29 @@ func (docker *DockerService) syncWithContainer(containerId string) error { } docker.apps[app] = config } + docker.mu.Unlock() return nil } -func (docker *DockerService) syncFromEvent(event events.Message) { - docker.log.App.Debug().Msgf("Received event: %s", event.Action) +func (docker *DockerService) syncFromEvent(event events.Message) { switch event.Action { - case events.ActionStart, events.ActionRestart, events.ActionStop, events.ActionKill: - err := docker.syncWithContainer(event.Actor.ID) - if err != nil { - docker.log.App.Error().Err(err).Msg("Failed to update apps from event") + 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") + } } - default: - docker.log.App.Debug().Msgf("Ignoring event: %s", event.Action) } } func (docker *DockerService) buildEventsKV() filters.Args { - vals := []string{"start", "restart", "stop", "kill"} + vals := []string{"start", "restart", "stop", "kill", "die"} kv := make([]filters.KeyValuePair, len(vals)) for i, v := range vals { kv[i] = filters.KeyValuePair{ @@ -193,34 +207,78 @@ func (docker *DockerService) buildEventsKV() filters.Args { } func (docker *DockerService) monitor(ctx context.Context) { - defer docker.client.Close() + 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 + } + } + } - stream, chErr := docker.client.Events(ctx, events.ListOptions{ - Filters: docker.buildEventsKV(), - }) + cancel() - for { - select { - case <-ctx.Done(): - docker.log.App.Debug().Msg("Shutting down docker monitor") - return + for ctx.Err() == nil { + docker.log.App.Warn().Msg("Attempting to reconnect to docker") - case event := <-stream: - docker.syncFromEvent(event) + cli, err := docker.connect() + + if err == nil && cli != nil { + docker.client.Close() + docker.client = cli + docker.log.App.Info().Msg("Docker connection recovered") + break + } - case err := <-chErr: if err != nil { - docker.log.App.Error().Err(err).Msg("Docker event stream failed") + 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 } From a934d2c85196c53af559146cc2536a794d743c44 Mon Sep 17 00:00:00 2001 From: Stavros Date: Thu, 15 Oct 2026 22:34:18 +0300 Subject: [PATCH 4/4] fix: only use backoff for network based docker connections --- internal/service/docker_service.go | 26 ++++++++++++++------------ 1 file changed, 14 insertions(+), 12 deletions(-) diff --git a/internal/service/docker_service.go b/internal/service/docker_service.go index 5488d0fe..6ea8a968 100644 --- a/internal/service/docker_service.go +++ b/internal/service/docker_service.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "os" "sync" "time" @@ -51,9 +52,20 @@ func NewDockerService(i DockerServiceInput) (*DockerService, error) { service.log.App.Debug().Msg("Attempting to connect to Docker") - cli, err := service.connectWithBackoff() + var cli *client.Client + var err error + + if os.Getenv("DOCKER_HOST") != "" { + cli, err = service.connectWithBackoff() + } else { + 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) } @@ -91,17 +103,7 @@ func (docker *DockerService) connectWithBackoff() (*client.Client, error) { return docker.connect() } - cli, err := backoff.Retry(docker.ctx, operation, backoff.WithBackOff(exp), backoff.WithMaxTries(3)) - - if err != nil { - if errors.Is(err, ErrPingFailed) { - docker.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) - } - - return cli, nil + return backoff.Retry(docker.ctx, operation, backoff.WithBackOff(exp), backoff.WithMaxTries(3)) } func (docker *DockerService) connect() (*client.Client, error) {