1package deployment
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "net"
9 "net/url"
10 "strings"
11 "time"
12
13 "github.com/bnema/gordon/internal/domain"
14)
15
16// ProbeDeps abstracts the runtime hooks readiness needs. The deployment
17// service wires these to the ContainerRuntime adapter; tests inject the
18// mockery-generated ContainerRuntime mock through NewProbeDeps.
19type ProbeDeps struct {
20 networkInfo func(ctx context.Context, containerID string) (string, int, error)
21 containerStart func(ctx context.Context, containerID string) (time.Time, error)
22 logStream func(ctx context.Context, containerID string, since time.Time) (io.ReadCloser, error)
23 httpGet func(ctx context.Context, url, hostAuthority string) (int, error)
24 tcpDial func(ctx context.Context, addr string) error
25 // networkProbe runs one bounded session against an internal port that
26 // has no host publication. Nil disables the internal path: it must
27 // never fall back to a loopback bind that does not exist.
28 networkProbe func(ctx context.Context, request domain.ContainerNetworkProbeRequest) (domain.ContainerNetworkProbeResult, error)
29}
30
31// ProbeRuntime is the runtime subset readiness probing needs. It mirrors
32// out.ContainerRuntime signatures so both the live adapter and the
33// mockery-generated MockContainerRuntime satisfy it.
34type ProbeRuntime interface {
35 GetContainerNetworkInfo(ctx context.Context, containerID string) (string, int, error)
36 GetContainerLogsSince(ctx context.Context, containerID string, since time.Time, follow bool) (io.ReadCloser, error)
37 InspectContainer(ctx context.Context, containerID string) (*domain.Container, error)
38 ProbeContainerNetwork(ctx context.Context, request domain.ContainerNetworkProbeRequest) (domain.ContainerNetworkProbeResult, error)
39}
40
41// NewProbeDeps wires readiness probes to a runtime implementation.
42// Production passes the ContainerRuntime adapter; tests pass the mockery
43// MockContainerRuntime which satisfies ProbeRuntime.
44func NewProbeDeps(runtime ProbeRuntime) ProbeDeps {
45 return ProbeDeps{
46 networkInfo: func(ctx context.Context, containerID string) (string, int, error) {
47 return runtime.GetContainerNetworkInfo(ctx, containerID)
48 },
49 containerStart: func(ctx context.Context, containerID string) (time.Time, error) {
50 ctr, err := runtime.InspectContainer(ctx, containerID)
51 if err != nil {
52 return time.Time{}, err
53 }
54 return ctr.StartedAt, nil
55 },
56 logStream: func(ctx context.Context, containerID string, since time.Time) (io.ReadCloser, error) {
57 return runtime.GetContainerLogsSince(ctx, containerID, since, false)
58 },
59 httpGet: httpGetOnce,
60 tcpDial: tcpDialOnce,
61 networkProbe: func(ctx context.Context, request domain.ContainerNetworkProbeRequest) (domain.ContainerNetworkProbeResult, error) {
62 return runtime.ProbeContainerNetwork(ctx, request)
63 },
64 }
65}
66
67// NewTestProbeDeps builds ProbeDeps with injectable HTTP/TCP probers.
68// Tests control L4 outcomes without dialing; logStream still reads the
69// runtime mock.
70func NewTestProbeDeps(runtime ProbeRuntime, httpGet func(ctx context.Context, url, hostAuthority string) (int, error), tcpDial func(ctx context.Context, addr string) error) ProbeDeps {
71 deps := NewProbeDeps(runtime)
72 deps.httpGet = httpGet
73 deps.tcpDial = tcpDial
74 return deps
75}
76
77// defaultProbeDeps wires readiness to the live runtime adapter.
78func (s *Service) defaultProbeDeps() ProbeDeps {
79 return NewProbeDeps(s.deps.Runtime)
80}
81
82// probeDeps returns the injected test probes or the live adapter wiring.
83func (s *Service) probeDeps() ProbeDeps {
84 if s.probes != nil {
85 return *s.probes
86 }
87 return s.defaultProbeDeps()
88}
89
90// waitServiceReady dispatches readiness for one service. It resolves the
91// app private network lazily: only an internal interface needs it. A
92// service whose readiness port belongs to an internal HTTP interface
93// always probes over the private network; it never falls back to a
94// loopback bind, and a public service never takes the network path.
95func (s *Service) waitServiceReady(ctx context.Context, app, containerID string, spec domain.AppService, binds map[int]int) error {
96 switch spec.Readiness.Type {
97 case domain.AppReadinessHTTP, domain.AppReadinessTCP:
98 port := readinessContainerPort(spec)
99 if port > 0 && spec.InternallyOnlyPort(port) {
100 network, err := s.privateNetworkForApp(ctx, app)
101 if err != nil {
102 return err
103 }
104 return waitInternalReadyWithDeps(ctx, s.probeDeps(), containerID, spec, network, port)
105 }
106 }
107 return waitServiceReadyWithDeps(ctx, s.probeDeps(), containerID, spec, binds)
108}
109
110// privateNetworkForApp resolves the incarnation-owned private network, the
111// only network an internal readiness helper may join. It fails closed when
112// the app has no incarnation id.
113func (s *Service) privateNetworkForApp(ctx context.Context, app string) (string, error) {
114 ownership, err := s.deps.State.LoadOwnership(ctx, app)
115 if err != nil {
116 return "", fmt.Errorf("deployment: load ownership for private network: %w", err)
117 }
118 if ownership.ID == "" {
119 return "", fmt.Errorf("deployment: app %q has no incarnation id for internal readiness: %w", app, domain.ErrAppStateConflict)
120 }
121 return domain.AppPrivateNetworkName(s.deps.Networks.Prefix, ownership.ID), nil
122}
123
124// waitInternalReadyWithDeps polls one bounded private-network session per
125// attempt until the target is ready or the readiness timeout expires. An
126// infrastructure error fails immediately and is never retried as an
127// unhealthy attempt: a helper that cannot run says nothing about the
128// target.
129func waitInternalReadyWithDeps(ctx context.Context, deps ProbeDeps, containerID string, spec domain.AppService, network string, port int) error {
130 timeout := readinessTimeout(spec)
131 protocol, path := internalProbeShape(spec)
132 if deps.networkProbe == nil || deps.containerStart == nil {
133 return fmt.Errorf("deployment: internal readiness probe is not wired")
134 }
135 startedAt, err := deps.containerStart(ctx, containerID)
136 if err != nil {
137 return fmt.Errorf("deployment: internal readiness execution start for %s: %w", containerID, err)
138 }
139 if startedAt.IsZero() {
140 // Without an execution boundary the probe could report readiness
141 // for a generation that already restarted. Fail closed.
142 return fmt.Errorf("deployment: internal readiness needs the execution start of %s, runtime reported none", containerID)
143 }
144 deadline := time.Now().Add(timeout)
145 lastAttempt := "no attempt completed"
146 for {
147 if err := ctx.Err(); err != nil {
148 return err
149 }
150 remaining := time.Until(deadline)
151 if remaining <= 0 {
152 return fmt.Errorf("deployment: internal readiness timeout after %s on container port %d (last attempt: %s)", timeout, port, lastAttempt)
153 }
154 attemptTimeout := min(5*time.Second, remaining)
155 result, probeErr := runInternalProbeAttempt(ctx, deps, domain.ContainerNetworkProbeRequest{
156 TargetContainerID: containerID,
157 ExpectedStartedAt: startedAt,
158 Network: network,
159 Protocol: protocol,
160 Port: port,
161 Path: path,
162 Timeout: attemptTimeout,
163 })
164 ready, retryStale, diagnostic, classifyErr := classifyInternalProbe(result, probeErr)
165 if classifyErr != nil {
166 return classifyErr
167 }
168 if ready {
169 return nil
170 }
171 if diagnostic != "" {
172 lastAttempt = diagnostic
173 }
174 if retryStale {
175 // The candidate restarted or briefly vanished: re-read its
176 // execution boundary and keep polling until the deadline. This
177 // is "not ready yet", not a broken helper, so it must never
178 // abort the wait as an infrastructure error.
179 next, startErr := rereadInternalProbeStart(ctx, deps, containerID, startedAt)
180 if startErr != nil {
181 return startErr
182 }
183 startedAt = next
184 }
185 select {
186 case <-ctx.Done():
187 return ctx.Err()
188 case <-time.After(500 * time.Millisecond):
189 }
190 }
191}
192
193func runInternalProbeAttempt(ctx context.Context, deps ProbeDeps, request domain.ContainerNetworkProbeRequest) (domain.ContainerNetworkProbeResult, error) {
194 attemptCtx, cancel := context.WithTimeout(ctx, request.Timeout)
195 result, err := deps.networkProbe(attemptCtx, request)
196 attemptErr := attemptCtx.Err()
197 cancel()
198 if err != nil && !errors.Is(err, domain.ErrNetworkProbeCleanup) && errors.Is(attemptErr, context.DeadlineExceeded) && ctx.Err() == nil {
199 return domain.ContainerNetworkProbeResult{Diagnostic: "probe attempt timed out"}, nil
200 }
201 return result, err
202}
203
204// rereadInternalProbeStart refreshes the execution boundary after a stale
205// or vanished candidate. A candidate that briefly vanished (not found)
206// keeps its previous boundary so the caller keeps polling until the
207// deadline; any other runtime failure is infrastructure and aborts. It
208// keeps the previous value when the runtime reports none so the next
209// attempt still fails closed.
210func rereadInternalProbeStart(ctx context.Context, deps ProbeDeps, containerID string, current time.Time) (time.Time, error) {
211 next, err := deps.containerStart(ctx, containerID)
212 if err != nil {
213 // ErrContainerNotFound is the same "not ready yet" condition the
214 // probe reported: preserve the boundary and keep polling instead of
215 // aborting the wait as an infrastructure error.
216 if errors.Is(err, domain.ErrContainerNotFound) {
217 return current, nil
218 }
219 return current, fmt.Errorf("deployment: internal readiness execution start for %s: %w", containerID, err)
220 }
221 if next.IsZero() {
222 return current, nil
223 }
224 return next, nil
225}
226
227// classifyInternalProbe maps one attempt's raw outcome to the wait loop's
228// next step: ready, not-ready (diagnostic), a stale or vanished candidate
229// that should be re-read and retried, or an infrastructure failure that
230// must abort the wait immediately.
231func classifyInternalProbe(result domain.ContainerNetworkProbeResult, probeErr error) (ready, retryStale bool, diagnostic string, err error) {
232 switch {
233 case probeErr == nil:
234 if result.Ready {
235 return true, false, "", nil
236 }
237 if result.Diagnostic != "" {
238 return false, false, result.Diagnostic, nil
239 }
240 return false, false, "not ready", nil
241 case errors.Is(probeErr, domain.ErrNetworkProbeCleanup):
242 // A helper that could not be removed is a broken helper, not a
243 // stale candidate: an error wrapping both this and a stale or
244 // vanished sentinel must abort the wait, never be retried.
245 return false, false, "", fmt.Errorf("deployment: internal readiness probe helper cleanup failed: %w", probeErr)
246 case errors.Is(probeErr, domain.ErrAppStateConflict), errors.Is(probeErr, domain.ErrContainerNotFound):
247 return false, true, "target execution changed; retrying", nil
248 default:
249 return false, false, "", fmt.Errorf("deployment: internal readiness probe infrastructure error: %w", probeErr)
250 }
251}
252
253// internalProbeShape selects the probe transport for an internal port:
254// http readiness maps to a bounded HTTP probe, tcp readiness to a TCP
255// connection probe.
256func internalProbeShape(spec domain.AppService) (domain.ProbeProtocol, string) {
257 if spec.Readiness.Type == domain.AppReadinessHTTP {
258 path := spec.Readiness.Path
259 if path == "" {
260 path = "/"
261 }
262 return domain.ProbeProtocolHTTP, path
263 }
264 return domain.ProbeProtocolTCP, ""
265}
266
267// waitServiceReadyWithDeps dispatches readiness by manifest type.
268// binds carries the container's 127.0.0.1 backend publishes; L4 probes
269// dial them rootless-first and never container IPs (plan D3: loopback-
270// only generated backends). Restart/start reuse the recorded binds from
271// the active record; deploy passes the freshly read binds.
272func waitServiceReadyWithDeps(ctx context.Context, deps ProbeDeps, containerID string, spec domain.AppService, binds map[int]int) error {
273 switch spec.Readiness.Type {
274 case "", domain.AppReadinessNone:
275 return nil
276 case domain.AppReadinessLog:
277 return waitLogReadyWithDeps(ctx, deps, containerID, spec)
278 case domain.AppReadinessHTTP:
279 return waitHTTPReadyWithDeps(ctx, deps, spec, binds)
280 case domain.AppReadinessTCP:
281 return waitTCPReadyWithDeps(ctx, deps, spec, binds)
282 default:
283 return fmt.Errorf("deployment: unknown readiness type %q", spec.Readiness.Type)
284 }
285}
286
287// waitHTTPReadyWithDeps polls GET path until 2xx/3xx or timeout.
288func waitHTTPReadyWithDeps(ctx context.Context, deps ProbeDeps, spec domain.AppService, binds map[int]int) error {
289 timeout := readinessTimeout(spec)
290 containerPort := readinessContainerPort(spec)
291 port, err := dialPort(spec, binds)
292 if err != nil {
293 return err
294 }
295 path := spec.Readiness.Path
296 if path == "" {
297 path = "/"
298 }
299 pathPart, rawQuery, _ := strings.Cut(path, "?")
300 probeURL := (&url.URL{
301 Scheme: "http",
302 Host: net.JoinHostPort("127.0.0.1", itoa(port)),
303 Path: pathPart,
304 RawQuery: rawQuery,
305 }).String()
306 // The probe dials the random loopback publication but must address the
307 // workload at its declared container port: a workload that validates the
308 // Host port (e.g. qBittorrent) rejects a published-port authority with
309 // 401 "Invalid Host header, port mismatch".
310 hostAuthority := net.JoinHostPort("127.0.0.1", itoa(containerPort))
311 deadline := time.Now().Add(timeout)
312 var lastAttempt string
313 for {
314 if err := ctx.Err(); err != nil {
315 return err
316 }
317 attemptCtx, cancel := context.WithTimeout(ctx, min(2*time.Second, time.Until(deadline)))
318 status, err := deps.httpGet(attemptCtx, probeURL, hostAuthority)
319 cancel()
320 if err == nil && status >= 200 && status < 400 {
321 return nil
322 }
323 if err != nil {
324 lastAttempt = err.Error()
325 } else {
326 lastAttempt = fmt.Sprintf("HTTP status %d", status)
327 }
328 if time.Now().After(deadline) {
329 return fmt.Errorf("deployment: HTTP readiness timeout after %s: no 2xx/3xx from %s (last attempt: %s)", timeout, probeURL, lastAttempt)
330 }
331 select {
332 case <-ctx.Done():
333 return ctx.Err()
334 case <-time.After(time.Second):
335 }
336 }
337}
338
339// waitTCPReadyWithDeps dials the container port until accepted.
340func waitTCPReadyWithDeps(ctx context.Context, deps ProbeDeps, spec domain.AppService, binds map[int]int) error {
341 timeout := readinessTimeout(spec)
342 port, err := dialPort(spec, binds)
343 if err != nil {
344 return err
345 }
346 addr := net.JoinHostPort("127.0.0.1", itoa(port))
347 // Every attempt and retry wait runs under the readiness budget, so a
348 // connection that only settles after the budget expired is never accepted.
349 attemptCtx, cancel := context.WithTimeout(ctx, timeout)
350 defer cancel()
351 timedOut := func() error {
352 if err := ctx.Err(); err != nil {
353 return err
354 }
355 return fmt.Errorf("deployment: TCP readiness timeout after %s: %s not reachable", timeout, addr)
356 }
357 for {
358 if attemptCtx.Err() != nil {
359 return timedOut()
360 }
361 if err := deps.tcpDial(attemptCtx, addr); err == nil && attemptCtx.Err() == nil {
362 return nil
363 }
364 select {
365 case <-attemptCtx.Done():
366 return timedOut()
367 case <-time.After(500 * time.Millisecond):
368 }
369 }
370}
371
372// waitLogReadyWithDeps tails the logs of one exact container ID until the
373// marker appears, bounded by the manifest readiness timeout and by the
374// caller's deadline context. Logs are read from the observed start of the
375// current execution onward, so a marker emitted by a previous execution
376// of the same container ID can never satisfy the probe.
377func waitLogReadyWithDeps(ctx context.Context, deps ProbeDeps, containerID string, spec domain.AppService) error {
378 timeout := readinessTimeout(spec)
379 deadlineCtx, cancel := context.WithTimeout(ctx, timeout)
380 defer cancel()
381 marker := spec.Readiness.Contains
382 if containerID == "" {
383 return fmt.Errorf("deployment: log readiness requires a container ID")
384 }
385 since, err := deps.containerStart(deadlineCtx, containerID)
386 if err != nil {
387 return fmt.Errorf("deployment: log readiness execution start for %s: %w", containerID, err)
388 }
389 if since.IsZero() {
390 // Without an execution boundary the probe would scan the
391 // container's whole history and could match a marker from a
392 // previous execution. Fail closed instead.
393 return fmt.Errorf("deployment: log readiness needs the execution start of %s, runtime reported none", containerID)
394 }
395 for {
396 if err := deadlineCtx.Err(); err != nil {
397 return fmt.Errorf("deployment: log readiness timeout waiting for %q: %w", marker, err)
398 }
399 stream, err := deps.logStream(deadlineCtx, containerID, since)
400 if err != nil {
401 return fmt.Errorf("deployment: log readiness stream for %s: %w", containerID, err)
402 }
403 found, readErr := scanForMarker(stream, marker)
404 _ = stream.Close()
405 if readErr != nil {
406 return readErr
407 }
408 if found {
409 return nil
410 }
411 select {
412 case <-deadlineCtx.Done():
413 return fmt.Errorf("deployment: log readiness timeout waiting for %q: %w", marker, deadlineCtx.Err())
414 case <-time.After(time.Second):
415 }
416 }
417}
418
419// dialPort resolves the 127.0.0.1 host port for the readiness probe:
420// the manifest's readiness port when set, else the single TCP-capable
421// interface port (validation guarantees unambiguous selection).
422func dialPort(spec domain.AppService, binds map[int]int) (int, error) {
423 containerPort := readinessContainerPort(spec)
424 hostPort, ok := binds[containerPort]
425 if !ok || hostPort <= 0 {
426 return 0, fmt.Errorf("deployment: no 127.0.0.1 backend bind for container port %d", containerPort)
427 }
428 return hostPort, nil
429}
430
431// readinessContainerPort resolves the container port a readiness probe
432// targets: the manifest's readiness port when set, else the single
433// effective-public HTTP port (the backend the proxy dials), else the
434// single TCP-capable interface port. Validation guarantees unambiguous
435// selection for the latter two; preferring the public HTTP port keeps a
436// public+internal service probeable when it declares no explicit
437// readiness port.
438func readinessContainerPort(spec domain.AppService) int {
439 if spec.Readiness.Port > 0 {
440 return spec.Readiness.Port
441 }
442 if port := singlePublicHTTPPort(spec); port > 0 {
443 return port
444 }
445 return singleTCPPort(spec)
446}
447
448// singlePublicHTTPPort returns the container port when the service
449// declares exactly one effective-public HTTP interface and no TCP
450// interface, else 0.
451func singlePublicHTTPPort(spec domain.AppService) int {
452 if len(spec.TCP) > 0 {
453 return 0
454 }
455 port := 0
456 for _, h := range spec.HTTP {
457 if !h.IsPublic() {
458 continue
459 }
460 if port != 0 && port != h.Port {
461 return 0
462 }
463 port = h.Port
464 }
465 return port
466}
467
468// singleTCPPort returns the container port when exactly one TCP-capable
469// interface exists; validation rejects ambiguity (readiness.port
470// required) before the engine runs.
471func singleTCPPort(spec domain.AppService) int {
472 port := 0
473 count := 0
474 for _, h := range spec.HTTP {
475 port, count = h.Port, count+1
476 }
477 for _, t := range spec.TCP {
478 port, count = t.Port, count+1
479 }
480 if count == 1 {
481 return port
482 }
483 return 0
484}
485
486// readinessTimeout applies the manifest default.
487func readinessTimeout(spec domain.AppService) time.Duration {
488 if spec.Readiness.Timeout > 0 {
489 return spec.Readiness.Timeout
490 }
491 return domain.AppDefaultReadinessTimeout
492}