refactor: cache docker labels and monitor for changes

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:

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,124 +44,243 @@ 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())
 
 	if err != nil {
 		return nil, err
 	}
 
-	_, err = cli.Ping(docker.context)
+	_, err = cli.Ping(docker.ctx)
 
 	if err != nil {
+		cli.Close()
 		return nil, ErrPingFailed
 	}
 
 	return cli, nil
 }
 
 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
 }