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 deadline := time.Now().Add(timeout)
348 for {
349 if err := ctx.Err(); err != nil {
350 return err
351 }
352 if err := deps.tcpDial(ctx, addr); err == nil {
353 return nil
354 }
355 if time.Now().After(deadline) {
356 return fmt.Errorf("deployment: TCP readiness timeout after %s: %s not reachable", timeout, addr)
357 }
358 select {
359 case <-ctx.Done():
360 return ctx.Err()
361 case <-time.After(500 * time.Millisecond):
362 }
363 }
364}
365
366// waitLogReadyWithDeps tails the logs of one exact container ID until the
367// marker appears, bounded by the manifest readiness timeout and by the
368// caller's deadline context. Logs are read from the observed start of the
369// current execution onward, so a marker emitted by a previous execution
370// of the same container ID can never satisfy the probe.
371func waitLogReadyWithDeps(ctx context.Context, deps ProbeDeps, containerID string, spec domain.AppService) error {
372 timeout := readinessTimeout(spec)
373 deadlineCtx, cancel := context.WithTimeout(ctx, timeout)
374 defer cancel()
375 marker := spec.Readiness.Contains
376 if containerID == "" {
377 return fmt.Errorf("deployment: log readiness requires a container ID")
378 }
379 since, err := deps.containerStart(deadlineCtx, containerID)
380 if err != nil {
381 return fmt.Errorf("deployment: log readiness execution start for %s: %w", containerID, err)
382 }
383 if since.IsZero() {
384 // Without an execution boundary the probe would scan the
385 // container's whole history and could match a marker from a
386 // previous execution. Fail closed instead.
387 return fmt.Errorf("deployment: log readiness needs the execution start of %s, runtime reported none", containerID)
388 }
389 for {
390 if err := deadlineCtx.Err(); err != nil {
391 return fmt.Errorf("deployment: log readiness timeout waiting for %q: %w", marker, err)
392 }
393 stream, err := deps.logStream(deadlineCtx, containerID, since)
394 if err != nil {
395 return fmt.Errorf("deployment: log readiness stream for %s: %w", containerID, err)
396 }
397 found, readErr := scanForMarker(stream, marker)
398 _ = stream.Close()
399 if readErr != nil {
400 return readErr
401 }
402 if found {
403 return nil
404 }
405 select {
406 case <-deadlineCtx.Done():
407 return fmt.Errorf("deployment: log readiness timeout waiting for %q: %w", marker, deadlineCtx.Err())
408 case <-time.After(time.Second):
409 }
410 }
411}
412
413// dialPort resolves the 127.0.0.1 host port for the readiness probe:
414// the manifest's readiness port when set, else the single TCP-capable
415// interface port (validation guarantees unambiguous selection).
416func dialPort(spec domain.AppService, binds map[int]int) (int, error) {
417 containerPort := readinessContainerPort(spec)
418 hostPort, ok := binds[containerPort]
419 if !ok || hostPort <= 0 {
420 return 0, fmt.Errorf("deployment: no 127.0.0.1 backend bind for container port %d", containerPort)
421 }
422 return hostPort, nil
423}
424
425// readinessContainerPort resolves the container port a readiness probe
426// targets: the manifest's readiness port when set, else the single
427// effective-public HTTP port (the backend the proxy dials), else the
428// single TCP-capable interface port. Validation guarantees unambiguous
429// selection for the latter two; preferring the public HTTP port keeps a
430// public+internal service probeable when it declares no explicit
431// readiness port.
432func readinessContainerPort(spec domain.AppService) int {
433 if spec.Readiness.Port > 0 {
434 return spec.Readiness.Port
435 }
436 if port := singlePublicHTTPPort(spec); port > 0 {
437 return port
438 }
439 return singleTCPPort(spec)
440}
441
442// singlePublicHTTPPort returns the container port when the service
443// declares exactly one effective-public HTTP interface and no TCP
444// interface, else 0.
445func singlePublicHTTPPort(spec domain.AppService) int {
446 if len(spec.TCP) > 0 {
447 return 0
448 }
449 port := 0
450 for _, h := range spec.HTTP {
451 if !h.IsPublic() {
452 continue
453 }
454 if port != 0 && port != h.Port {
455 return 0
456 }
457 port = h.Port
458 }
459 return port
460}
461
462// singleTCPPort returns the container port when exactly one TCP-capable
463// interface exists; validation rejects ambiguity (readiness.port
464// required) before the engine runs.
465func singleTCPPort(spec domain.AppService) int {
466 port := 0
467 count := 0
468 for _, h := range spec.HTTP {
469 port, count = h.Port, count+1
470 }
471 for _, t := range spec.TCP {
472 port, count = t.Port, count+1
473 }
474 if count == 1 {
475 return port
476 }
477 return 0
478}
479
480// readinessTimeout applies the manifest default.
481func readinessTimeout(spec domain.AppService) time.Duration {
482 if spec.Readiness.Timeout > 0 {
483 return spec.Readiness.Timeout
484 }
485 return domain.AppDefaultReadinessTimeout
486}