1package services
2
3import (
4 "bufio"
5 "context"
6 "crypto/sha256"
7 "encoding/hex"
8 "encoding/json"
9 "fmt"
10 "io"
11 "maps"
12 "net"
13 "os"
14 "sort"
15 "strconv"
16 "strings"
17 "time"
18
19 "github.com/bnema/gordon/internal/boundaries/out"
20 "github.com/bnema/gordon/internal/domain"
21)
22
23const defaultVolumePrefix = "gordon-service"
24
25type Service struct {
26 runtime out.ContainerRuntime
27 secretProvider out.SecretProvider
28 volumePrefix string
29}
30
31func NewService(runtime out.ContainerRuntime) *Service {
32 return &Service{runtime: runtime, volumePrefix: defaultVolumePrefix}
33}
34
35func NewServiceWithSecretProvider(runtime out.ContainerRuntime, secretProvider out.SecretProvider) *Service {
36 return &Service{runtime: runtime, secretProvider: secretProvider, volumePrefix: defaultVolumePrefix}
37}
38
39func NewServiceWithVolumePrefix(runtime out.ContainerRuntime, volumePrefix string) *Service {
40 if volumePrefix == "" {
41 volumePrefix = defaultVolumePrefix
42 }
43 return &Service{runtime: runtime, volumePrefix: volumePrefix}
44}
45
46func (s *Service) Reconcile(ctx context.Context, configured []domain.StandaloneService) error {
47 containers, err := s.runtime.ListContainers(ctx, true)
48 if err != nil {
49 return fmt.Errorf("list standalone service containers: %w", err)
50 }
51 existing := managedServiceContainers(containers)
52 configuredNames := make(map[string]struct{}, len(configured))
53 for _, svc := range configured {
54 configuredNames[svc.Name] = struct{}{}
55 if err := s.reconcileOne(ctx, svc, existing[svc.Name]); err != nil {
56 return err
57 }
58 }
59 for name, containers := range existing {
60 if _, ok := configuredNames[name]; ok {
61 continue
62 }
63 if err := s.stopRemoved(ctx, name, containers); err != nil {
64 return err
65 }
66 }
67 return nil
68}
69
70func (s *Service) Status(ctx context.Context) ([]domain.StandaloneServiceStatus, error) {
71 containers, err := s.runtime.ListContainers(ctx, true)
72 if err != nil {
73 return nil, fmt.Errorf("list standalone service containers: %w", err)
74 }
75 statuses := make([]domain.StandaloneServiceStatus, 0)
76 for _, container := range containers {
77 if container.Labels[domain.LabelService] != "true" {
78 continue
79 }
80 statuses = append(statuses, domain.StandaloneServiceStatus{
81 Name: container.Labels[domain.LabelServiceName],
82 ContainerID: container.ID,
83 ContainerName: container.Name,
84 Status: containerStatus(container),
85 ConfigHash: container.Labels[domain.LabelServiceConfigHash],
86 })
87 }
88 sort.Slice(statuses, func(i, j int) bool { return statuses[i].Name < statuses[j].Name })
89 return statuses, nil
90}
91
92func (s *Service) reconcileOne(ctx context.Context, svc domain.StandaloneService, existing []*domain.Container) error {
93 cleanup := normalizeCleanup(svc.Cleanup)
94 if !svc.Enabled {
95 return s.stopDisabled(ctx, svc.Name, cleanup, existing)
96 }
97 env, err := s.serviceEnv(ctx, svc)
98 if err != nil {
99 return fmt.Errorf("resolve standalone service %q environment: %w", svc.Name, err)
100 }
101 hash, err := serviceConfigHashWithEnv(svc, env)
102 if err != nil {
103 return fmt.Errorf("hash standalone service %q config: %w", svc.Name, err)
104 }
105 if len(existing) == 0 {
106 return s.createAndStart(ctx, svc, hash, env)
107 }
108 sort.SliceStable(existing, func(i, j int) bool {
109 leftRunning := containerStatus(existing[i]) == domain.ContainerStatusRunning
110 rightRunning := containerStatus(existing[j]) == domain.ContainerStatusRunning
111 if leftRunning != rightRunning {
112 return leftRunning
113 }
114 return existing[i].ID < existing[j].ID
115 })
116 current := existing[0]
117 if current.Labels[domain.LabelServiceConfigHash] != hash {
118 if err := s.recreate(ctx, svc, cleanup, existing, hash, env); err != nil {
119 return err
120 }
121 return nil
122 }
123 for _, duplicate := range existing[1:] {
124 if err := s.cleanupContainer(ctx, svc.Name, "duplicate", cleanup, duplicate); err != nil {
125 return err
126 }
127 }
128 if containerStatus(current) != domain.ContainerStatusRunning {
129 if err := s.runtime.StartContainer(ctx, current.ID); err != nil {
130 return fmt.Errorf("start standalone service %q container: %w", svc.Name, err)
131 }
132 }
133 if err := s.waitReadiness(ctx, current.ID, svc); err != nil {
134 return err
135 }
136 return nil
137}
138
139func (s *Service) recreate(ctx context.Context, svc domain.StandaloneService, cleanup domain.StandaloneServiceCleanup, existing []*domain.Container, hash string, env []string) error {
140 for _, container := range existing {
141 if containerStatus(container) == domain.ContainerStatusRunning {
142 if err := s.runtime.StopContainer(ctx, container.ID); err != nil {
143 return fmt.Errorf("stop stale standalone service %q container: %w", svc.Name, err)
144 }
145 }
146 if cleanup.RemoveContainer {
147 if err := s.runtime.RemoveContainer(ctx, container.ID, true); err != nil {
148 return fmt.Errorf("remove stale standalone service %q container: %w", svc.Name, err)
149 }
150 }
151 }
152 return s.createAndStart(ctx, svc, hash, env)
153}
154
155func (s *Service) stopDisabled(ctx context.Context, name string, cleanup domain.StandaloneServiceCleanup, existing []*domain.Container) error {
156 for _, container := range existing {
157 if err := s.cleanupContainer(ctx, name, "disabled", cleanup, container); err != nil {
158 return err
159 }
160 }
161 return nil
162}
163
164func (s *Service) stopRemoved(ctx context.Context, name string, existing []*domain.Container) error {
165 for _, container := range existing {
166 cleanup := cleanupFromLabels(container.Labels)
167 if err := s.cleanupContainer(ctx, name, "removed", cleanup, container); err != nil {
168 return err
169 }
170 }
171 return nil
172}
173
174func (s *Service) cleanupContainer(ctx context.Context, name, reason string, cleanup domain.StandaloneServiceCleanup, container *domain.Container) error {
175 if containerStatus(container) == domain.ContainerStatusRunning {
176 if err := s.runtime.StopContainer(ctx, container.ID); err != nil {
177 return fmt.Errorf("stop %s standalone service %q container: %w", reason, name, err)
178 }
179 }
180 if cleanup.RemoveContainer {
181 if err := s.runtime.RemoveContainer(ctx, container.ID, true); err != nil {
182 return fmt.Errorf("remove %s standalone service %q container: %w", reason, name, err)
183 }
184 }
185 if err := s.removeManagedVolumes(ctx, name, cleanup, container); err != nil {
186 return err
187 }
188 return nil
189}
190
191func (s *Service) removeManagedVolumes(ctx context.Context, name string, cleanup domain.StandaloneServiceCleanup, container *domain.Container) error {
192 if cleanup.PreserveVolumes {
193 return nil
194 }
195 managed := managedVolumeSet(container.Labels)
196 for _, mount := range container.VolumeMounts {
197 if mount.Type != "volume" || mount.Name == "" {
198 continue
199 }
200 if _, ok := managed[mount.Name]; !ok {
201 continue
202 }
203 if err := s.runtime.RemoveVolume(ctx, mount.Name, true); err != nil {
204 return fmt.Errorf("remove standalone service %q volume %q: %w", name, mount.Name, err)
205 }
206 }
207 return nil
208}
209
210func managedVolumeSet(labels map[string]string) map[string]struct{} {
211 values := strings.Split(labels[domain.LabelServiceManagedVolumes], ",")
212 managed := make(map[string]struct{}, len(values))
213 for _, value := range values {
214 name := strings.TrimSpace(value)
215 if name == "" {
216 continue
217 }
218 managed[name] = struct{}{}
219 }
220 return managed
221}
222
223func cleanupFromLabels(labels map[string]string) domain.StandaloneServiceCleanup {
224 cleanup := domain.StandaloneServiceCleanup{PreserveVolumes: true, RemoveContainer: true}
225 if value, ok := labels[domain.LabelServiceCleanupPreserveVolumes]; ok {
226 if parsed, err := strconv.ParseBool(value); err == nil {
227 cleanup.PreserveVolumes = parsed
228 }
229 }
230 if value, ok := labels[domain.LabelServiceCleanupRemoveContainer]; ok {
231 if parsed, err := strconv.ParseBool(value); err == nil {
232 cleanup.RemoveContainer = parsed
233 }
234 }
235 return cleanup
236}
237
238func (s *Service) createAndStart(ctx context.Context, svc domain.StandaloneService, hash string, env []string) error {
239 config, err := s.containerConfig(ctx, svc, hash, env)
240 if err != nil {
241 return err
242 }
243 container, err := s.runtime.CreateContainer(ctx, config)
244 if err != nil {
245 return fmt.Errorf("create standalone service %q container: %w", svc.Name, err)
246 }
247 if err := s.runtime.StartContainer(ctx, container.ID); err != nil {
248 return fmt.Errorf("start standalone service %q container: %w", svc.Name, err)
249 }
250 if err := s.waitReadiness(ctx, container.ID, svc); err != nil {
251 return err
252 }
253 return nil
254}
255
256func (s *Service) containerConfig(ctx context.Context, svc domain.StandaloneService, hash string, env []string) (*domain.ContainerConfig, error) {
257 var imageVolumes []string
258 if len(svc.Volumes) == 0 {
259 var err error
260 imageVolumes, err = s.runtime.InspectImageVolumes(ctx, svc.Image)
261 if err != nil {
262 return nil, fmt.Errorf("inspect standalone service %q image volumes: %w", svc.Name, err)
263 }
264 }
265 mounts := ResolveVolumeMounts(s.volumePrefix, svc.Name, svc.Volumes, imageVolumes)
266 volumes := make(map[string]string)
267 readOnlyVolumes := make(map[string]string)
268 managedVolumes := make([]string, 0)
269 for _, mount := range mounts {
270 if mount.Managed {
271 managedVolumes = append(managedVolumes, mount.Source)
272 }
273 if mount.ReadOnly {
274 readOnlyVolumes[mount.Target] = mount.Source
275 continue
276 }
277 volumes[mount.Target] = mount.Source
278 }
279 publishes, err := portPublishes(svc)
280 if err != nil {
281 return nil, err
282 }
283 return &domain.ContainerConfig{
284 Image: svc.Image,
285 Name: serviceContainerName(svc.Name),
286 Env: env,
287 PortPublishes: publishes,
288 Labels: serviceLabels(svc.Name, hash, normalizeCleanup(svc.Cleanup), managedVolumes),
289 AutoRemove: false,
290 RestartPolicy: domain.RestartPolicyAlways,
291 Volumes: emptyToNil(volumes),
292 ReadOnlyVolumes: emptyToNil(readOnlyVolumes),
293 }, nil
294}
295
296func managedServiceContainers(containers []*domain.Container) map[string][]*domain.Container {
297 managed := make(map[string][]*domain.Container)
298 for _, container := range containers {
299 if container == nil || container.Labels[domain.LabelService] != "true" {
300 continue
301 }
302 name := container.Labels[domain.LabelServiceName]
303 if name == "" {
304 continue
305 }
306 managed[name] = append(managed[name], container)
307 }
308 return managed
309}
310
311func serviceLabels(name, hash string, cleanup domain.StandaloneServiceCleanup, managedVolumes []string) map[string]string {
312 sort.Strings(managedVolumes)
313 return map[string]string{
314 domain.LabelManaged: "true",
315 domain.LabelService: "true",
316 domain.LabelServiceName: name,
317 domain.LabelServiceConfigHash: hash,
318 domain.LabelServiceManagedVolumes: strings.Join(managedVolumes, ","),
319 domain.LabelServiceCleanupPreserveVolumes: strconv.FormatBool(cleanup.PreserveVolumes),
320 domain.LabelServiceCleanupRemoveContainer: strconv.FormatBool(cleanup.RemoveContainer),
321 }
322}
323
324func portPublishes(svc domain.StandaloneService) ([]domain.ContainerPortPublish, error) {
325 publishes := make([]domain.ContainerPortPublish, 0, len(svc.Ports))
326 for _, port := range svc.Ports {
327 if port.Publish == "" {
328 continue
329 }
330 host, hostPort, err := net.SplitHostPort(port.Publish)
331 if err != nil {
332 return nil, fmt.Errorf("parse standalone service %q port %q publish: %w", svc.Name, port.Name, err)
333 }
334 hostPortNumber, err := strconv.Atoi(hostPort)
335 if err != nil {
336 return nil, fmt.Errorf("parse standalone service %q port %q publish port: %w", svc.Name, port.Name, err)
337 }
338 publishes = append(publishes, domain.ContainerPortPublish{
339 HostIP: host, HostPort: hostPortNumber, ContainerPort: port.Container, Protocol: port.Protocol,
340 })
341 }
342 return publishes, nil
343}
344
345func serviceConfigHash(svc domain.StandaloneService) (string, error) {
346 return NewService(nil).serviceConfigHash(context.Background(), svc)
347}
348
349func (s *Service) serviceConfigHash(ctx context.Context, svc domain.StandaloneService) (string, error) {
350 resolvedEnv, err := s.serviceEnv(ctx, svc)
351 if err != nil {
352 return "", err
353 }
354 return serviceConfigHashWithEnv(svc, resolvedEnv)
355}
356
357func serviceConfigHashWithEnv(svc domain.StandaloneService, resolvedEnv []string) (string, error) {
358 payload := struct {
359 Image string
360 ResolvedEnv []string
361 Readiness domain.StandaloneServiceReadiness
362 Ports []domain.StandaloneServicePort
363 Volumes []domain.StandaloneServiceVolume
364 Cleanup domain.StandaloneServiceCleanup
365 }{svc.Image, append([]string(nil), resolvedEnv...), svc.Readiness, append([]domain.StandaloneServicePort(nil), svc.Ports...), append([]domain.StandaloneServiceVolume(nil), svc.Volumes...), normalizeCleanup(svc.Cleanup)}
366 bytes, err := json.Marshal(payload)
367 if err != nil {
368 return "", err
369 }
370 sum := sha256.Sum256(bytes)
371 return hex.EncodeToString(sum[:]), nil
372}
373
374func (s *Service) serviceEnv(ctx context.Context, svc domain.StandaloneService) ([]string, error) {
375 envMap := make(map[string]string)
376 if svc.EnvFile != "" {
377 fileEnv, err := loadEnvFile(ctx, svc.EnvFile)
378 if err != nil {
379 return nil, fmt.Errorf("load standalone service %q env file: %w", svc.Name, err)
380 }
381 maps.Copy(envMap, fileEnv)
382 }
383 for _, entry := range svc.Env {
384 key, value, ok := strings.Cut(entry, "=")
385 if !ok || key == "" {
386 return nil, fmt.Errorf("parse standalone service %q env entry", svc.Name)
387 }
388 envMap[key] = value
389 }
390 if len(svc.Secrets) > 0 && s.secretProvider == nil {
391 return nil, fmt.Errorf("resolve standalone service %q secrets: secret provider is not configured", svc.Name)
392 }
393 for _, secret := range svc.Secrets {
394 value, err := s.secretProvider.GetSecret(ctx, serviceSecretPath(svc.Name, secret.Name))
395 if err != nil {
396 return nil, fmt.Errorf("resolve standalone service %q secret %q: %w", svc.Name, secret.Name, err)
397 }
398 envMap[secret.Key] = value
399 }
400 return envMapToList(envMap), nil
401}
402
403func loadEnvFile(ctx context.Context, path string) (map[string]string, error) {
404 if err := ctx.Err(); err != nil {
405 return nil, err
406 }
407 file, err := os.Open(path)
408 if err != nil {
409 return nil, err
410 }
411 defer file.Close()
412 values := make(map[string]string)
413 scanner := bufio.NewScanner(file)
414 for scanner.Scan() {
415 if err := ctx.Err(); err != nil {
416 return nil, err
417 }
418 line := strings.TrimSpace(scanner.Text())
419 if line == "" || strings.HasPrefix(line, "#") {
420 continue
421 }
422 key, value, ok := strings.Cut(line, "=")
423 if !ok || strings.TrimSpace(key) == "" {
424 return nil, fmt.Errorf("invalid env file line")
425 }
426 values[strings.TrimSpace(key)] = strings.Trim(strings.TrimSpace(value), `"'`)
427 }
428 if err := scanner.Err(); err != nil {
429 return nil, err
430 }
431 return values, nil
432}
433
434func serviceSecretPath(serviceName, secretName string) string {
435 if strings.Contains(secretName, ":") {
436 return secretName
437 }
438 return "service/" + serviceName + "/" + secretName
439}
440
441func envMapToList(values map[string]string) []string {
442 if len(values) == 0 {
443 return nil
444 }
445 keys := make([]string, 0, len(values))
446 for key := range values {
447 keys = append(keys, key)
448 }
449 sort.Strings(keys)
450 env := make([]string, 0, len(keys))
451 for _, key := range keys {
452 env = append(env, key+"="+values[key])
453 }
454 return env
455}
456
457func (s *Service) waitReadiness(ctx context.Context, containerID string, svc domain.StandaloneService) error {
458 readiness := svc.Readiness
459 if readiness.Type == "" || readiness.Type == domain.StandaloneServiceReadinessNone {
460 return nil
461 }
462 readyCtx := ctx
463 cancel := func() {}
464 if readiness.Timeout > 0 {
465 readyCtx, cancel = context.WithTimeout(ctx, readiness.Timeout)
466 }
467 defer cancel()
468 switch readiness.Type {
469 case domain.StandaloneServiceReadinessTCP:
470 return waitTCPReadiness(readyCtx, svc)
471 case domain.StandaloneServiceReadinessLog:
472 return s.waitLogReadiness(readyCtx, containerID, svc)
473 default:
474 return fmt.Errorf("unsupported standalone service %q readiness type %q", svc.Name, readiness.Type)
475 }
476}
477
478func waitTCPReadiness(ctx context.Context, svc domain.StandaloneService) error {
479 address, err := tcpReadinessAddress(svc)
480 if err != nil {
481 return err
482 }
483 ticker := time.NewTicker(10 * time.Millisecond)
484 defer ticker.Stop()
485 for {
486 if err := ctx.Err(); err != nil {
487 return fmt.Errorf("standalone service %q tcp readiness timed out: %w", svc.Name, err)
488 }
489 dialer := net.Dialer{Timeout: 100 * time.Millisecond}
490 conn, err := dialer.DialContext(ctx, "tcp", address)
491 if err == nil {
492 _ = conn.Close()
493 return nil
494 }
495 select {
496 case <-ctx.Done():
497 return fmt.Errorf("standalone service %q tcp readiness timed out: %w", svc.Name, ctx.Err())
498 case <-ticker.C:
499 }
500 }
501}
502
503func tcpReadinessAddress(svc domain.StandaloneService) (string, error) {
504 for _, port := range svc.Ports {
505 if port.Protocol != domain.NetworkProtocolTCP || port.Publish == "" {
506 continue
507 }
508 host, hostPort, err := net.SplitHostPort(port.Publish)
509 if err != nil {
510 return "", fmt.Errorf("parse standalone service %q tcp readiness publish: %w", svc.Name, err)
511 }
512 switch host {
513 case "::":
514 host = "::1"
515 case "0.0.0.0", "":
516 host = "127.0.0.1"
517 }
518 return net.JoinHostPort(host, hostPort), nil
519 }
520 return "", fmt.Errorf("standalone service %q tcp readiness requires a published tcp port", svc.Name)
521}
522
523func (s *Service) waitLogReadiness(ctx context.Context, containerID string, svc domain.StandaloneService) error {
524 ticker := time.NewTicker(10 * time.Millisecond)
525 defer ticker.Stop()
526 var lastErr error
527 for {
528 if err := ctx.Err(); err != nil {
529 return logReadinessTimeoutError(svc.Name, err, lastErr)
530 }
531 found, err := s.logContains(ctx, containerID, svc.Readiness.Path, svc.Readiness.Contains)
532 if err == nil && found {
533 return nil
534 }
535 if err != nil {
536 lastErr = err
537 }
538 // Continue polling until timeout; many services create the log after start.
539 select {
540 case <-ctx.Done():
541 return logReadinessTimeoutError(svc.Name, ctx.Err(), lastErr)
542 case <-ticker.C:
543 }
544 }
545}
546
547func logReadinessTimeoutError(serviceName string, timeoutErr, lastErr error) error {
548 if lastErr != nil {
549 return fmt.Errorf("standalone service %q log readiness timed out: %w; last read error: %w", serviceName, timeoutErr, lastErr)
550 }
551 return fmt.Errorf("standalone service %q log readiness timed out: %w", serviceName, timeoutErr)
552}
553
554func (s *Service) logContains(ctx context.Context, containerID, path, contains string) (bool, error) {
555 reader, err := s.runtime.CopyFromContainer(ctx, containerID, path)
556 if err != nil {
557 return false, err
558 }
559 defer reader.Close()
560 content, err := io.ReadAll(reader)
561 if err != nil {
562 return false, err
563 }
564 return strings.Contains(string(content), contains), nil
565}
566
567func normalizeCleanup(cleanup domain.StandaloneServiceCleanup) domain.StandaloneServiceCleanup {
568 if !cleanup.PreserveVolumes && !cleanup.RemoveContainer {
569 return domain.StandaloneServiceCleanup{PreserveVolumes: true, RemoveContainer: true}
570 }
571 return cleanup
572}
573
574func containerStatus(container *domain.Container) domain.ContainerStatus {
575 status := strings.ToLower(container.Status)
576 if strings.Contains(status, string(domain.ContainerStatusRunning)) {
577 return domain.ContainerStatusRunning
578 }
579 if strings.Contains(status, string(domain.ContainerStatusExited)) {
580 return domain.ContainerStatusExited
581 }
582 if strings.Contains(status, string(domain.ContainerStatusCreated)) {
583 return domain.ContainerStatusCreated
584 }
585 if strings.Contains(status, string(domain.ContainerStatusPaused)) {
586 return domain.ContainerStatusPaused
587 }
588 if strings.Contains(status, string(domain.ContainerStatusStopped)) {
589 return domain.ContainerStatusStopped
590 }
591 return domain.ContainerStatusUnknown
592}
593
594func serviceContainerName(name string) string {
595 return "gordon-service-" + strings.NewReplacer(".", "-", "_", "-", "/", "-").Replace(name)
596}
597
598func emptyToNil(values map[string]string) map[string]string {
599 if len(values) == 0 {
600 return nil
601 }
602 return values
603}