Skip to content

Commit f41ab3c

Browse files
steveiliop56steveiliop56
authored
refactor: cache docker labels and monitor for changes (#1189)
1 parent d3e7e92 commit f41ab3c

2 files changed

Lines changed: 190 additions & 67 deletions

File tree

‎Makefile‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@ PROD_COMPOSE := $(shell test -f "docker-compose.test.prod.yml" && echo "docker-c
1919

2020
.DEFAULT_GOAL := binary
2121

22-
.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
22+
.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
2323

2424
# Tools
2525
tools:

‎internal/service/docker_service.go‎

Lines changed: 189 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,12 @@ import (
55
"errors"
66
"fmt"
77
"os"
8+
"sync"
89
"time"
910

1011
"github.com/cenkalti/backoff/v5"
12+
"github.com/docker/docker/api/types/events"
13+
"github.com/docker/docker/api/types/filters"
1114
"github.com/steveiliop56/ding"
1215
"github.com/tinyauthapp/tinyauth/internal/model"
1316
"github.com/tinyauthapp/tinyauth/internal/utils/decoders"
@@ -23,11 +26,12 @@ var (
2326
)
2427

2528
type DockerService struct {
26-
log *logger.Logger
27-
client *client.Client
28-
context context.Context
29+
log *logger.Logger
30+
client *client.Client
31+
ctx context.Context
2932

30-
isConnected bool
33+
apps map[string]model.App
34+
mu sync.RWMutex
3135
}
3236

3337
type DockerServiceInput struct {
@@ -40,124 +44,243 @@ type DockerServiceInput struct {
4044

4145
func NewDockerService(i DockerServiceInput) (*DockerService, error) {
4246
service := &DockerService{
43-
log: i.Log,
44-
context: i.Ctx,
47+
log: i.Log,
48+
ctx: i.Ctx,
49+
apps: make(map[string]model.App),
50+
mu: sync.RWMutex{},
4551
}
4652

4753
service.log.App.Debug().Msg("Attempting to connect to Docker")
4854

49-
if os.Getenv("DOCKER_HOST") == "" {
50-
cli, err := service.connect()
51-
if err != nil {
52-
if errors.Is(err, ErrPingFailed) {
53-
service.log.App.Debug().Msg("Docker not connected")
54-
return nil, nil
55-
}
56-
return nil, fmt.Errorf("failed to connect to docker: %w", err)
57-
}
58-
service.client = cli
55+
var cli *client.Client
56+
var err error
57+
58+
if os.Getenv("DOCKER_HOST") != "" {
59+
cli, err = service.connectWithBackoff()
5960
} else {
60-
exp := backoff.NewExponentialBackOff()
61-
exp.InitialInterval = 3 * time.Second
62-
exp.RandomizationFactor = 0.1
63-
exp.Multiplier = 1.5
64-
exp.Reset()
65-
66-
operation := func() (*client.Client, error) {
67-
if service.client != nil {
68-
service.client.Close()
69-
}
70-
cli, err := service.connect()
71-
if err != nil {
72-
return nil, err
73-
}
74-
return cli, nil
61+
cli, err = service.connect()
62+
}
63+
64+
if err != nil {
65+
if errors.Is(err, ErrPingFailed) {
66+
service.log.App.Debug().Msg("Docker not connected, skipping")
67+
return nil, nil
7568
}
69+
return nil, fmt.Errorf("failed to connect to docker: %w", err)
70+
}
7671

77-
cli, err := backoff.Retry(service.context, operation, backoff.WithBackOff(exp), backoff.WithMaxTries(3))
72+
if cli == nil {
73+
return nil, nil
74+
}
7875

79-
if err != nil {
80-
if errors.Is(err, ErrPingFailed) {
81-
service.log.App.Debug().Msg("Docker not connected after retrying")
82-
return nil, nil
83-
}
84-
return nil, fmt.Errorf("failed to connect to docker after retrying: %w", err)
85-
}
76+
service.client = cli
77+
78+
// initial sync
79+
err = service.sync()
8680

87-
service.client = cli
81+
if err != nil {
82+
return nil, fmt.Errorf("failed to do initial sync: %w", err)
8883
}
8984

90-
service.isConnected = true
9185
service.log.App.Debug().Msg("Docker connected successfully")
9286

93-
i.Ding.Go(service.watchAndClose, ding.RingMajor)
87+
i.Ding.Go(service.monitor, ding.RingMajor)
9488

9589
return service, nil
9690
}
9791

92+
func (docker *DockerService) connectWithBackoff() (*client.Client, error) {
93+
exp := backoff.NewExponentialBackOff()
94+
exp.InitialInterval = 3 * time.Second
95+
exp.RandomizationFactor = 0.1
96+
exp.Multiplier = 1.5
97+
exp.Reset()
98+
99+
operation := func() (*client.Client, error) {
100+
if docker.client != nil {
101+
docker.client.Close()
102+
}
103+
return docker.connect()
104+
}
105+
106+
return backoff.Retry(docker.ctx, operation, backoff.WithBackOff(exp), backoff.WithMaxTries(3))
107+
}
108+
98109
func (docker *DockerService) connect() (*client.Client, error) {
99110
cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
100111

101112
if err != nil {
102113
return nil, err
103114
}
104115

105-
_, err = cli.Ping(docker.context)
116+
_, err = cli.Ping(docker.ctx)
106117

107118
if err != nil {
119+
cli.Close()
108120
return nil, ErrPingFailed
109121
}
110122

111123
return cli, nil
112124
}
113125

114126
func (docker *DockerService) getContainers() ([]container.Summary, error) {
115-
return docker.client.ContainerList(docker.context, container.ListOptions{})
127+
return docker.client.ContainerList(docker.ctx, container.ListOptions{})
116128
}
117129

118130
func (docker *DockerService) inspectContainer(containerId string) (container.InspectResponse, error) {
119-
return docker.client.ContainerInspect(docker.context, containerId)
131+
return docker.client.ContainerInspect(docker.ctx, containerId)
120132
}
121133

122-
func (docker *DockerService) Lookup(locator func(name string, app *model.App)) error {
123-
if !docker.isConnected {
124-
docker.log.App.Debug().Msg("Docker service not connected, returning empty labels")
125-
return nil
126-
}
127-
134+
func (docker *DockerService) sync() error {
128135
containers, err := docker.getContainers()
129136
if err != nil {
130137
return fmt.Errorf("failed to get containers: %w", err)
131138
}
132139

140+
apps := make(map[string]model.App)
141+
133142
for _, ctr := range containers {
134-
inspect, err := docker.inspectContainer(ctr.ID)
143+
labels, err := decoders.DecodeLabels[model.Apps](ctr.Labels, "apps")
135144
if err != nil {
136-
docker.log.App.Error().Err(err).Msgf("Failed to inspect container %s", ctr.ID)
145+
docker.log.App.Warn().Err(err).Str("container_id", ctr.ID).Msg("Failed to decode container labels, skipping")
137146
continue
138147
}
139148

140-
labels, err := decoders.DecodeLabels[model.Apps](inspect.Config.Labels, "apps")
141-
if err != nil {
142-
docker.log.App.Warn().Err(err).Msgf("Failed to decode labels for container %s", ctr.ID)
149+
for app, config := range labels.Apps {
150+
apps[app] = config
151+
}
152+
}
153+
154+
docker.mu.Lock()
155+
docker.apps = apps
156+
docker.mu.Unlock()
157+
158+
return nil
159+
}
160+
161+
func (docker *DockerService) syncWithContainer(containerId string) error {
162+
ctr, err := docker.inspectContainer(containerId)
163+
if err != nil {
164+
return fmt.Errorf("failed to inspect container: %w", err)
165+
}
166+
labels, err := decoders.DecodeLabels[model.Apps](ctr.Config.Labels, "apps")
167+
if err != nil {
168+
return fmt.Errorf("failed to decode container labels: %w", err)
169+
}
170+
docker.mu.Lock()
171+
for app, config := range labels.Apps {
172+
if !ctr.State.Running {
173+
delete(docker.apps, app)
143174
continue
144175
}
176+
docker.apps[app] = config
177+
}
178+
docker.mu.Unlock()
179+
return nil
180+
}
145181

146-
for app, config := range labels.Apps {
147-
locator(app, &config)
182+
func (docker *DockerService) syncFromEvent(event events.Message) {
183+
switch event.Action {
184+
case events.ActionStart, events.ActionRestart,
185+
events.ActionStop, events.ActionKill, events.ActionDie:
186+
187+
if err := docker.syncWithContainer(event.Actor.ID); err != nil {
188+
docker.log.App.Warn().Err(err).
189+
Msg("Failed to sync container, performing full sync")
190+
191+
if err := docker.sync(); err != nil {
192+
docker.log.App.Error().Err(err).
193+
Msg("Failed to resync Docker apps")
194+
}
148195
}
149196
}
197+
}
150198

151-
return nil
199+
func (docker *DockerService) buildEventsKV() filters.Args {
200+
vals := []string{"start", "restart", "stop", "kill", "die"}
201+
kv := make([]filters.KeyValuePair, len(vals))
202+
for i, v := range vals {
203+
kv[i] = filters.KeyValuePair{
204+
Key: "event",
205+
Value: v,
206+
}
207+
}
208+
return filters.NewArgs(kv...)
152209
}
153210

154-
func (docker *DockerService) watchAndClose(ctx context.Context) {
155-
<-ctx.Done()
156-
docker.log.App.Debug().Msg("Closing Docker client")
157-
if docker.client != nil {
158-
err := docker.client.Close()
159-
if err != nil {
160-
docker.log.App.Error().Err(err).Msg("Error closing Docker client")
211+
func (docker *DockerService) monitor(ctx context.Context) {
212+
defer func() {
213+
if docker.client != nil {
214+
_ = docker.client.Close()
161215
}
216+
}()
217+
218+
for ctx.Err() == nil {
219+
streamCtx, cancel := context.WithCancel(ctx)
220+
221+
stream, chErr := docker.client.Events(streamCtx, events.ListOptions{
222+
Filters: docker.buildEventsKV(),
223+
})
224+
225+
if err := docker.sync(); err != nil {
226+
docker.log.App.Error().Err(err).Msg("Failed to sync docker apps")
227+
} else {
228+
watch:
229+
for {
230+
select {
231+
case <-ctx.Done():
232+
cancel()
233+
docker.log.App.Debug().Msg("Shutting down docker monitor")
234+
return
235+
236+
case event, ok := <-stream:
237+
if !ok {
238+
docker.log.App.Warn().Msg("Docker event stream closed")
239+
break watch
240+
}
241+
docker.syncFromEvent(event)
242+
243+
case err, ok := <-chErr:
244+
if ok && err != nil {
245+
docker.log.App.Error().Err(err).Msg("Docker event stream failed")
246+
}
247+
break watch
248+
}
249+
}
250+
}
251+
252+
cancel()
253+
254+
for ctx.Err() == nil {
255+
docker.log.App.Warn().Msg("Attempting to reconnect to docker")
256+
257+
cli, err := docker.connect()
258+
259+
if err == nil && cli != nil {
260+
docker.client.Close()
261+
docker.client = cli
262+
docker.log.App.Info().Msg("Docker connection recovered")
263+
break
264+
}
265+
266+
if err != nil {
267+
docker.log.App.Error().Err(err).Msg("Failed to reconnect to docker")
268+
}
269+
270+
select {
271+
case <-ctx.Done():
272+
return
273+
case <-time.After(5 * time.Second):
274+
}
275+
}
276+
}
277+
}
278+
279+
func (docker *DockerService) Lookup(locator func(name string, app *model.App)) error {
280+
docker.mu.RLock()
281+
for app, config := range docker.apps {
282+
locator(app, &config)
162283
}
284+
docker.mu.RUnlock()
285+
return nil
163286
}

0 commit comments

Comments
 (0)