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