Merge pull request #97 from bnema/fix/default-http-readiness-probe
fix: improve default HTTP readiness probes
4 files · +109 -35
M internal/adapters/out/logwriter/logwriter.go +7 -9
| @@ -8,6 +8,7 @@ import ( | |||
| 8 | 8 | "path/filepath" | |
| 9 | 9 | "strings" | |
| 10 | 10 | "sync" | |
| 11 | + | "sync/atomic" | |
| 11 | 12 | ||
| 12 | 13 | "github.com/bnema/zerowrap" | |
| 13 | 14 | "github.com/docker/docker/pkg/stdcopy" | |
| @@ -39,7 +40,7 @@ type streamInfo struct { | |||
| 39 | 40 | domain string | |
| 40 | 41 | logStream io.ReadCloser | |
| 41 | 42 | logger *lumberjack.Logger | |
| 42 | - | cancel context.CancelFunc | |
| 43 | + | stopped atomic.Bool | |
| 43 | 44 | done chan struct{} | |
| 44 | 45 | } | |
| 45 | 46 | ||
| @@ -89,8 +90,6 @@ func (w *LogWriter) StartLogging(ctx context.Context, containerID string, domain | |||
| 89 | 90 | Compress: true, | |
| 90 | 91 | } | |
| 91 | 92 | ||
| 92 | - | // Create cancellable context for the streaming goroutine | |
| 93 | - | streamCtx, cancel := context.WithCancel(context.Background()) | |
| 94 | 93 | done := make(chan struct{}) | |
| 95 | 94 | ||
| 96 | 95 | info := &streamInfo{ | |
| @@ -98,14 +97,13 @@ func (w *LogWriter) StartLogging(ctx context.Context, containerID string, domain | |||
| 98 | 97 | domain: domain, | |
| 99 | 98 | logStream: logStream, | |
| 100 | 99 | logger: logger, | |
| 101 | - | cancel: cancel, | |
| 102 | 100 | done: done, | |
| 103 | 101 | } | |
| 104 | 102 | ||
| 105 | 103 | w.streams[containerID] = info | |
| 106 | 104 | ||
| 107 | 105 | // Start streaming goroutine | |
| 108 | - | go w.streamLogs(streamCtx, info, log) | |
| 106 | + | go w.streamLogs(info, log) | |
| 109 | 107 | ||
| 110 | 108 | log.Info().Str("path", logPath).Msg("started container log collection") | |
| 111 | 109 | return nil | |
| @@ -138,7 +136,7 @@ func (w *LogWriter) Close() error { | |||
| 138 | 136 | // stopStreamLocked stops a stream (must be called with lock held). | |
| 139 | 137 | func (w *LogWriter) stopStreamLocked(info *streamInfo) { | |
| 140 | 138 | // Signal the goroutine to stop | |
| 141 | - | info.cancel() | |
| 139 | + | info.stopped.Store(true) | |
| 142 | 140 | ||
| 143 | 141 | // Close the log stream to unblock reads | |
| 144 | 142 | info.logStream.Close() | |
| @@ -151,7 +149,7 @@ func (w *LogWriter) stopStreamLocked(info *streamInfo) { | |||
| 151 | 149 | } | |
| 152 | 150 | ||
| 153 | 151 | // streamLogs copies from the Docker log stream to the file. | |
| 154 | - | func (w *LogWriter) streamLogs(ctx context.Context, info *streamInfo, log zerowrap.Logger) { | |
| 152 | + | func (w *LogWriter) streamLogs(info *streamInfo, log zerowrap.Logger) { | |
| 155 | 153 | defer close(info.done) | |
| 156 | 154 | ||
| 157 | 155 | // Docker container logs are multiplexed with 8-byte headers. | |
| @@ -159,8 +157,8 @@ func (w *LogWriter) streamLogs(ctx context.Context, info *streamInfo, log zerowr | |||
| 159 | 157 | // We combine both to a single file. | |
| 160 | 158 | _, err := stdcopy.StdCopy(info.logger, info.logger, info.logStream) | |
| 161 | 159 | ||
| 162 | - | if err != nil && ctx.Err() == nil { | |
| 163 | - | // Only log error if we weren't cancelled | |
| 160 | + | if err != nil && !info.stopped.Load() { | |
| 161 | + | // Only log error if we weren't intentionally stopped | |
| 164 | 162 | log.Warn().Err(err). | |
| 165 | 163 | Str("container_id", info.containerID). | |
| 166 | 164 | Str("domain", info.domain). | |
| @@ -8,6 +8,7 @@ import ( | |||||
| 8 | "path/filepath" | 8 | "path/filepath" | ||
| 9 | "strings" | 9 | "strings" | ||
| 10 | "sync" | 10 | "sync" | ||
| 11 | + | "sync/atomic" | |||
| 11 | 12 | ||||
| 12 | "github.com/bnema/zerowrap" | 13 | "github.com/bnema/zerowrap" | ||
| 13 | "github.com/docker/docker/pkg/stdcopy" | 14 | "github.com/docker/docker/pkg/stdcopy" | ||
| @@ -39,7 +40,7 @@ type streamInfo struct { | |||||
| 39 | domain string | 40 | domain string | ||
| 40 | logStream io.ReadCloser | 41 | logStream io.ReadCloser | ||
| 41 | logger *lumberjack.Logger | 42 | logger *lumberjack.Logger | ||
| 42 | - | cancel context.CancelFunc | 43 | + | stopped atomic.Bool |
| 43 | done chan struct{} | 44 | done chan struct{} | ||
| 44 | } | 45 | } | ||
| 45 | 46 | ||||
| @@ -89,8 +90,6 @@ func (w *LogWriter) StartLogging(ctx context.Context, containerID string, domain | |||||
| 89 | Compress: true, | 90 | Compress: true, | ||
| 90 | } | 91 | } | ||
| 91 | 92 | ||||
| 92 | - | // Create cancellable context for the streaming goroutine | |||
| 93 | - | streamCtx, cancel := context.WithCancel(context.Background()) | |||
| 94 | done := make(chan struct{}) | 93 | done := make(chan struct{}) | ||
| 95 | 94 | ||||
| 96 | info := &streamInfo{ | 95 | info := &streamInfo{ | ||
| @@ -98,14 +97,13 @@ func (w *LogWriter) StartLogging(ctx context.Context, containerID string, domain | |||||
| 98 | domain: domain, | 97 | domain: domain, | ||
| 99 | logStream: logStream, | 98 | logStream: logStream, | ||
| 100 | logger: logger, | 99 | logger: logger, | ||
| 101 | - | cancel: cancel, | |||
| 102 | done: done, | 100 | done: done, | ||
| 103 | } | 101 | } | ||
| 104 | 102 | ||||
| 105 | w.streams[containerID] = info | 103 | w.streams[containerID] = info | ||
| 106 | 104 | ||||
| 107 | // Start streaming goroutine | 105 | // Start streaming goroutine | ||
| 108 | - | go w.streamLogs(streamCtx, info, log) | 106 | + | go w.streamLogs(info, log) |
| 109 | 107 | ||||
| 110 | log.Info().Str("path", logPath).Msg("started container log collection") | 108 | log.Info().Str("path", logPath).Msg("started container log collection") | ||
| 111 | return nil | 109 | return nil | ||
| @@ -138,7 +136,7 @@ func (w *LogWriter) Close() error { | |||||
| 138 | // stopStreamLocked stops a stream (must be called with lock held). | 136 | // stopStreamLocked stops a stream (must be called with lock held). | ||
| 139 | func (w *LogWriter) stopStreamLocked(info *streamInfo) { | 137 | func (w *LogWriter) stopStreamLocked(info *streamInfo) { | ||
| 140 | // Signal the goroutine to stop | 138 | // Signal the goroutine to stop | ||
| 141 | - | info.cancel() | 139 | + | info.stopped.Store(true) |
| 142 | 140 | ||||
| 143 | // Close the log stream to unblock reads | 141 | // Close the log stream to unblock reads | ||
| 144 | info.logStream.Close() | 142 | info.logStream.Close() | ||
| @@ -151,7 +149,7 @@ func (w *LogWriter) stopStreamLocked(info *streamInfo) { | |||||
| 151 | } | 149 | } | ||
| 152 | 150 | ||||
| 153 | // streamLogs copies from the Docker log stream to the file. | 151 | // streamLogs copies from the Docker log stream to the file. | ||
| 154 | - | func (w *LogWriter) streamLogs(ctx context.Context, info *streamInfo, log zerowrap.Logger) { | 152 | + | func (w *LogWriter) streamLogs(info *streamInfo, log zerowrap.Logger) { |
| 155 | defer close(info.done) | 153 | defer close(info.done) | ||
| 156 | 154 | ||||
| 157 | // Docker container logs are multiplexed with 8-byte headers. | 155 | // Docker container logs are multiplexed with 8-byte headers. | ||
| @@ -159,8 +157,8 @@ func (w *LogWriter) streamLogs(ctx context.Context, info *streamInfo, log zerowr | |||||
| 159 | // We combine both to a single file. | 157 | // We combine both to a single file. | ||
| 160 | _, err := stdcopy.StdCopy(info.logger, info.logger, info.logStream) | 158 | _, err := stdcopy.StdCopy(info.logger, info.logger, info.logStream) | ||
| 161 | 159 | ||||
| 162 | - | if err != nil && ctx.Err() == nil { | 160 | + | if err != nil && !info.stopped.Load() { |
| 163 | - | // Only log error if we weren't cancelled | 161 | + | // Only log error if we weren't intentionally stopped |
| 164 | log.Warn().Err(err). | 162 | log.Warn().Err(err). | ||
| 165 | Str("container_id", info.containerID). | 163 | Str("container_id", info.containerID). | ||
| 166 | Str("domain", info.domain). | 164 | Str("domain", info.domain). | ||
M internal/usecase/container/readiness.go +33 -0
| @@ -135,3 +135,36 @@ func httpProbe(ctx context.Context, url string, timeout time.Duration) error { | |||
| 135 | 135 | } | |
| 136 | 136 | return err | |
| 137 | 137 | } | |
| 138 | + | ||
| 139 | + | // httpAliveProbe performs HTTP GET requests to url, retrying every 1s until | |
| 140 | + | // any HTTP response is received or timeout. Unlike httpProbe, ANY status code | |
| 141 | + | // counts as success -- the goal is to verify the HTTP server is accepting and | |
| 142 | + | // processing requests, not that a specific endpoint is healthy. | |
| 143 | + | // Used as the default readiness check for web containers. | |
| 144 | + | func httpAliveProbe(ctx context.Context, url string, timeout time.Duration) error { | |
| 145 | + | client := &http.Client{ | |
| 146 | + | CheckRedirect: func(req *http.Request, via []*http.Request) error { | |
| 147 | + | return http.ErrUseLastResponse | |
| 148 | + | }, | |
| 149 | + | } | |
| 150 | + | deadline := time.Now().Add(timeout) | |
| 151 | + | err := probeLoop(ctx, deadline, time.Second, 2*time.Second, func(attemptCtx context.Context) (bool, error) { | |
| 152 | + | req, reqErr := http.NewRequestWithContext(attemptCtx, http.MethodGet, url, nil) | |
| 153 | + | if reqErr != nil { | |
| 154 | + | return false, reqErr | |
| 155 | + | } | |
| 156 | + | ||
| 157 | + | resp, err := client.Do(req) | |
| 158 | + | if err != nil { | |
| 159 | + | // Connection refused, timeout, etc. -- not ready yet. | |
| 160 | + | return false, nil | |
| 161 | + | } | |
| 162 | + | resp.Body.Close() | |
| 163 | + | // Any HTTP response means the server is alive and processing. | |
| 164 | + | return true, nil | |
| 165 | + | }) | |
| 166 | + | if errors.Is(err, errProbeTimeout) { | |
| 167 | + | return fmt.Errorf("HTTP alive probe timeout after %s: no HTTP response from %s", timeout, url) | |
| 168 | + | } | |
| 169 | + | return err | |
| 170 | + | } | |
| @@ -135,3 +135,36 @@ func httpProbe(ctx context.Context, url string, timeout time.Duration) error { | |||||
| 135 | } | 135 | } | ||
| 136 | return err | 136 | return err | ||
| 137 | } | 137 | } | ||
| 138 | + | ||||
| 139 | + | // httpAliveProbe performs HTTP GET requests to url, retrying every 1s until | |||
| 140 | + | // any HTTP response is received or timeout. Unlike httpProbe, ANY status code | |||
| 141 | + | // counts as success -- the goal is to verify the HTTP server is accepting and | |||
| 142 | + | // processing requests, not that a specific endpoint is healthy. | |||
| 143 | + | // Used as the default readiness check for web containers. | |||
| 144 | + | func httpAliveProbe(ctx context.Context, url string, timeout time.Duration) error { | |||
| 145 | + | client := &http.Client{ | |||
| 146 | + | CheckRedirect: func(req *http.Request, via []*http.Request) error { | |||
| 147 | + | return http.ErrUseLastResponse | |||
| 148 | + | }, | |||
| 149 | + | } | |||
| 150 | + | deadline := time.Now().Add(timeout) | |||
| 151 | + | err := probeLoop(ctx, deadline, time.Second, 2*time.Second, func(attemptCtx context.Context) (bool, error) { | |||
| 152 | + | req, reqErr := http.NewRequestWithContext(attemptCtx, http.MethodGet, url, nil) | |||
| 153 | + | if reqErr != nil { | |||
| 154 | + | return false, reqErr | |||
| 155 | + | } | |||
| 156 | + | ||||
| 157 | + | resp, err := client.Do(req) | |||
| 158 | + | if err != nil { | |||
| 159 | + | // Connection refused, timeout, etc. -- not ready yet. | |||
| 160 | + | return false, nil | |||
| 161 | + | } | |||
| 162 | + | resp.Body.Close() | |||
| 163 | + | // Any HTTP response means the server is alive and processing. | |||
| 164 | + | return true, nil | |||
| 165 | + | }) | |||
| 166 | + | if errors.Is(err, errProbeTimeout) { | |||
| 167 | + | return fmt.Errorf("HTTP alive probe timeout after %s: no HTTP response from %s", timeout, url) | |||
| 168 | + | } | |||
| 169 | + | return err | |||
| 170 | + | } | |||
M internal/usecase/container/service.go +60 -14
| @@ -7,6 +7,7 @@ import ( | |||
| 7 | 7 | "fmt" | |
| 8 | 8 | "maps" | |
| 9 | 9 | "slices" | |
| 10 | + | "strconv" | |
| 10 | 11 | "strings" | |
| 11 | 12 | "sync" | |
| 12 | 13 | "time" | |
| @@ -1330,7 +1331,7 @@ func (s *Service) startLogCollection(ctx context.Context, containerID, domainNam | |||
| 1330 | 1331 | return | |
| 1331 | 1332 | } | |
| 1332 | 1333 | ||
| 1333 | - | if err := s.logWriter.StartLogging(ctx, containerID, domainName, logStream); err != nil { | |
| 1334 | + | if err := s.logWriter.StartLogging(context.WithoutCancel(ctx), containerID, domainName, logStream); err != nil { | |
| 1334 | 1335 | log.Warn().Err(err).Msg("failed to start container log collection") | |
| 1335 | 1336 | logStream.Close() | |
| 1336 | 1337 | } | |
| @@ -2297,8 +2298,9 @@ func (s *Service) waitForReady(ctx context.Context, containerID string, containe | |||
| 2297 | 2298 | // readinessCascade auto-detects the strongest available readiness signal: | |
| 2298 | 2299 | // 1. Docker healthcheck (if present) → wait for healthy status | |
| 2299 | 2300 | // 2. HTTP probe (if gordon.health label set) → GET until 2xx/3xx | |
| 2300 | - | // 3. TCP probe (if port info available) → connect until accepted | |
| 2301 | - | // 4. Delay fallback (last resort) → waitForReadyByDelay | |
| 2301 | + | // 3. Default HTTP probe (GET / on exposed port) → wait for 2xx/3xx | |
| 2302 | + | // 4. TCP probe (if port info available) → connect until accepted | |
| 2303 | + | // 5. Delay fallback (last resort) → waitForReadyByDelay | |
| 2302 | 2304 | func (s *Service) readinessCascade(ctx context.Context, containerID string, containerConfig *domain.ContainerConfig, cfg Config) error { | |
| 2303 | 2305 | log := zerowrap.FromCtx(ctx) | |
| 2304 | 2306 | ||
| @@ -2317,12 +2319,17 @@ func (s *Service) readinessCascade(ctx context.Context, containerID string, cont | |||
| 2317 | 2319 | return probeErr | |
| 2318 | 2320 | } | |
| 2319 | 2321 | ||
| 2320 | - | // 3. TCP probe (if port info available) | |
| 2322 | + | // 3. Default HTTP probe on root path (if port info available) | |
| 2323 | + | if probed, probeErr := s.tryDefaultHTTPProbe(ctx, containerID, containerConfig, cfg); probed { | |
| 2324 | + | return probeErr | |
| 2325 | + | } | |
| 2326 | + | ||
| 2327 | + | // 4. TCP probe (if port info available) | |
| 2321 | 2328 | if probed, probeErr := s.tryTCPProbe(ctx, containerID, containerConfig, cfg); probed { | |
| 2322 | 2329 | return probeErr | |
| 2323 | 2330 | } | |
| 2324 | 2331 | ||
| 2325 | - | // 4. Delay fallback | |
| 2332 | + | // 5. Delay fallback | |
| 2326 | 2333 | log.Info().Msg("readiness cascade: using delay fallback") | |
| 2327 | 2334 | return s.waitForReadyByDelay(ctx, containerID) | |
| 2328 | 2335 | } | |
| @@ -2338,7 +2345,7 @@ func (s *Service) tryHTTPProbe(ctx context.Context, containerID string, containe | |||
| 2338 | 2345 | if !ok || healthPath == "" { | |
| 2339 | 2346 | return false, nil | |
| 2340 | 2347 | } | |
| 2341 | - | ip, port, probeErr := s.resolveContainerEndpoint(ctx, containerID) | |
| 2348 | + | ip, port, probeErr := s.resolveProbeEndpoint(ctx, containerID, containerConfig) | |
| 2342 | 2349 | if probeErr != nil || ip == "" || port <= 0 { | |
| 2343 | 2350 | log.Debug().Err(probeErr).Msg("readiness cascade: HTTP probe skipped, could not resolve container endpoint") | |
| 2344 | 2351 | return false, nil | |
| @@ -2352,6 +2359,29 @@ func (s *Service) tryHTTPProbe(ctx context.Context, containerID string, containe | |||
| 2352 | 2359 | return true, httpProbe(ctx, url, timeout) | |
| 2353 | 2360 | } | |
| 2354 | 2361 | ||
| 2362 | + | // tryDefaultHTTPProbe attempts an HTTP GET on "/" using the container's | |
| 2363 | + | // exposed port. This is the default readiness check for web containers — | |
| 2364 | + | // it verifies the app is actually serving HTTP, not just accepting TCP. | |
| 2365 | + | // Returns (true, err) if the probe was attempted, (false, nil) if skipped. | |
| 2366 | + | func (s *Service) tryDefaultHTTPProbe(ctx context.Context, containerID string, containerConfig *domain.ContainerConfig, cfg Config) (bool, error) { | |
| 2367 | + | log := zerowrap.FromCtx(ctx) | |
| 2368 | + | if containerConfig == nil || len(containerConfig.Ports) == 0 { | |
| 2369 | + | return false, nil | |
| 2370 | + | } | |
| 2371 | + | ip, port, probeErr := s.resolveProbeEndpoint(ctx, containerID, containerConfig) | |
| 2372 | + | if probeErr != nil || ip == "" || port <= 0 { | |
| 2373 | + | log.Debug().Err(probeErr).Msg("readiness cascade: default HTTP probe skipped, could not resolve container endpoint") | |
| 2374 | + | return false, nil | |
| 2375 | + | } | |
| 2376 | + | url := fmt.Sprintf("http://%s:%d/", ip, port) | |
| 2377 | + | timeout := cfg.HTTPProbeTimeout | |
| 2378 | + | if timeout == 0 { | |
| 2379 | + | timeout = 60 * time.Second | |
| 2380 | + | } | |
| 2381 | + | log.Info().Str("url", url).Dur("timeout", timeout).Msg("readiness cascade: using default HTTP alive probe") | |
| 2382 | + | return true, httpAliveProbe(ctx, url, timeout) | |
| 2383 | + | } | |
| 2384 | + | ||
| 2355 | 2385 | // tryTCPProbe attempts a TCP probe if port info is available. | |
| 2356 | 2386 | // Returns (true, err) if the probe was attempted, (false, nil) if skipped. | |
| 2357 | 2387 | func (s *Service) tryTCPProbe(ctx context.Context, containerID string, containerConfig *domain.ContainerConfig, cfg Config) (bool, error) { | |
| @@ -2359,7 +2389,7 @@ func (s *Service) tryTCPProbe(ctx context.Context, containerID string, container | |||
| 2359 | 2389 | if containerConfig == nil || len(containerConfig.Ports) == 0 { | |
| 2360 | 2390 | return false, nil | |
| 2361 | 2391 | } | |
| 2362 | - | ip, port, probeErr := s.resolveContainerEndpoint(ctx, containerID) | |
| 2392 | + | ip, port, probeErr := s.resolveProbeEndpoint(ctx, containerID, containerConfig) | |
| 2363 | 2393 | if probeErr != nil || ip == "" || port <= 0 { | |
| 2364 | 2394 | log.Debug().Err(probeErr).Msg("readiness cascade: TCP probe skipped, could not resolve container endpoint") | |
| 2365 | 2395 | return false, nil | |
| @@ -2378,16 +2408,33 @@ func (s *Service) tryTCPProbe(ctx context.Context, containerID string, container | |||
| 2378 | 2408 | // from the host. We use the host port binding (127.0.0.1:<mapped_port>) | |
| 2379 | 2409 | // which is always reachable. Falls back to the container's internal IP | |
| 2380 | 2410 | // only if no host port mapping exists (e.g. host-network mode). | |
| 2381 | - | func (s *Service) resolveContainerEndpoint(ctx context.Context, containerID string) (string, int, error) { | |
| 2411 | + | func (s *Service) resolveProbeEndpoint(ctx context.Context, containerID string, containerConfig *domain.ContainerConfig) (string, int, error) { | |
| 2382 | 2412 | log := zerowrap.FromCtx(ctx) | |
| 2383 | 2413 | ||
| 2384 | - | // First, get the container's internal port from network info | |
| 2385 | - | _, internalPort, err := s.runtime.GetContainerNetworkInfo(ctx, containerID) | |
| 2386 | - | if err != nil { | |
| 2387 | - | return "", 0, err | |
| 2414 | + | internalPort := 0 | |
| 2415 | + | if containerConfig != nil { | |
| 2416 | + | if labels := containerConfig.Labels; labels != nil { | |
| 2417 | + | for _, key := range []string{domain.LabelProxyPort, domain.LabelPort} { | |
| 2418 | + | if portStr, ok := labels[key]; ok && portStr != "" { | |
| 2419 | + | port, err := strconv.Atoi(portStr) | |
| 2420 | + | if err == nil && port > 0 { | |
| 2421 | + | internalPort = port | |
| 2422 | + | break | |
| 2423 | + | } | |
| 2424 | + | log.Warn().Str("label", key).Str("port_value", portStr).Msg("invalid probe port label value") | |
| 2425 | + | } | |
| 2426 | + | } | |
| 2427 | + | } | |
| 2428 | + | } | |
| 2429 | + | ||
| 2430 | + | if internalPort == 0 { | |
| 2431 | + | _, resolvedPort, err := s.runtime.GetContainerNetworkInfo(ctx, containerID) | |
| 2432 | + | if err != nil { | |
| 2433 | + | return "", 0, err | |
| 2434 | + | } | |
| 2435 | + | internalPort = resolvedPort | |
| 2388 | 2436 | } | |
| 2389 | 2437 | ||
| 2390 | - | // Try to resolve via host port binding (works in rootless podman/Docker) | |
| 2391 | 2438 | hostPort, hostErr := s.runtime.GetContainerPort(ctx, containerID, internalPort) | |
| 2392 | 2439 | if hostErr == nil && hostPort > 0 { | |
| 2393 | 2440 | log.Debug(). | |
| @@ -2397,7 +2444,6 @@ func (s *Service) resolveContainerEndpoint(ctx context.Context, containerID stri | |||
| 2397 | 2444 | return "127.0.0.1", hostPort, nil | |
| 2398 | 2445 | } | |
| 2399 | 2446 | ||
| 2400 | - | // Fallback: use internal IP (works in Docker rootful / host network) | |
| 2401 | 2447 | ip, _, fallbackErr := s.runtime.GetContainerNetworkInfo(ctx, containerID) | |
| 2402 | 2448 | if fallbackErr != nil { | |
| 2403 | 2449 | return "", 0, fallbackErr | |
| @@ -7,6 +7,7 @@ import ( | |||||
| 7 | "fmt" | 7 | "fmt" | ||
| 8 | "maps" | 8 | "maps" | ||
| 9 | "slices" | 9 | "slices" | ||
| 10 | + | "strconv" | |||
| 10 | "strings" | 11 | "strings" | ||
| 11 | "sync" | 12 | "sync" | ||
| 12 | "time" | 13 | "time" | ||
| @@ -1330,7 +1331,7 @@ func (s *Service) startLogCollection(ctx context.Context, containerID, domainNam | |||||
| 1330 | return | 1331 | return | ||
| 1331 | } | 1332 | } | ||
| 1332 | 1333 | ||||
| 1333 | - | if err := s.logWriter.StartLogging(ctx, containerID, domainName, logStream); err != nil { | 1334 | + | if err := s.logWriter.StartLogging(context.WithoutCancel(ctx), containerID, domainName, logStream); err != nil { |
| 1334 | log.Warn().Err(err).Msg("failed to start container log collection") | 1335 | log.Warn().Err(err).Msg("failed to start container log collection") | ||
| 1335 | logStream.Close() | 1336 | logStream.Close() | ||
| 1336 | } | 1337 | } | ||
| @@ -2297,8 +2298,9 @@ func (s *Service) waitForReady(ctx context.Context, containerID string, containe | |||||
| 2297 | // readinessCascade auto-detects the strongest available readiness signal: | 2298 | // readinessCascade auto-detects the strongest available readiness signal: | ||
| 2298 | // 1. Docker healthcheck (if present) → wait for healthy status | 2299 | // 1. Docker healthcheck (if present) → wait for healthy status | ||
| 2299 | // 2. HTTP probe (if gordon.health label set) → GET until 2xx/3xx | 2300 | // 2. HTTP probe (if gordon.health label set) → GET until 2xx/3xx | ||
| 2300 | - | // 3. TCP probe (if port info available) → connect until accepted | 2301 | + | // 3. Default HTTP probe (GET / on exposed port) → wait for 2xx/3xx |
| 2301 | - | // 4. Delay fallback (last resort) → waitForReadyByDelay | 2302 | + | // 4. TCP probe (if port info available) → connect until accepted |
| 2303 | + | // 5. Delay fallback (last resort) → waitForReadyByDelay | |||
| 2302 | func (s *Service) readinessCascade(ctx context.Context, containerID string, containerConfig *domain.ContainerConfig, cfg Config) error { | 2304 | func (s *Service) readinessCascade(ctx context.Context, containerID string, containerConfig *domain.ContainerConfig, cfg Config) error { | ||
| 2303 | log := zerowrap.FromCtx(ctx) | 2305 | log := zerowrap.FromCtx(ctx) | ||
| 2304 | 2306 | ||||
| @@ -2317,12 +2319,17 @@ func (s *Service) readinessCascade(ctx context.Context, containerID string, cont | |||||
| 2317 | return probeErr | 2319 | return probeErr | ||
| 2318 | } | 2320 | } | ||
| 2319 | 2321 | ||||
| 2320 | - | // 3. TCP probe (if port info available) | 2322 | + | // 3. Default HTTP probe on root path (if port info available) |
| 2323 | + | if probed, probeErr := s.tryDefaultHTTPProbe(ctx, containerID, containerConfig, cfg); probed { | |||
| 2324 | + | return probeErr | |||
| 2325 | + | } | |||
| 2326 | + | ||||
| 2327 | + | // 4. TCP probe (if port info available) | |||
| 2321 | if probed, probeErr := s.tryTCPProbe(ctx, containerID, containerConfig, cfg); probed { | 2328 | if probed, probeErr := s.tryTCPProbe(ctx, containerID, containerConfig, cfg); probed { | ||
| 2322 | return probeErr | 2329 | return probeErr | ||
| 2323 | } | 2330 | } | ||
| 2324 | 2331 | ||||
| 2325 | - | // 4. Delay fallback | 2332 | + | // 5. Delay fallback |
| 2326 | log.Info().Msg("readiness cascade: using delay fallback") | 2333 | log.Info().Msg("readiness cascade: using delay fallback") | ||
| 2327 | return s.waitForReadyByDelay(ctx, containerID) | 2334 | return s.waitForReadyByDelay(ctx, containerID) | ||
| 2328 | } | 2335 | } | ||
| @@ -2338,7 +2345,7 @@ func (s *Service) tryHTTPProbe(ctx context.Context, containerID string, containe | |||||
| 2338 | if !ok || healthPath == "" { | 2345 | if !ok || healthPath == "" { | ||
| 2339 | return false, nil | 2346 | return false, nil | ||
| 2340 | } | 2347 | } | ||
| 2341 | - | ip, port, probeErr := s.resolveContainerEndpoint(ctx, containerID) | 2348 | + | ip, port, probeErr := s.resolveProbeEndpoint(ctx, containerID, containerConfig) |
| 2342 | if probeErr != nil || ip == "" || port <= 0 { | 2349 | if probeErr != nil || ip == "" || port <= 0 { | ||
| 2343 | log.Debug().Err(probeErr).Msg("readiness cascade: HTTP probe skipped, could not resolve container endpoint") | 2350 | log.Debug().Err(probeErr).Msg("readiness cascade: HTTP probe skipped, could not resolve container endpoint") | ||
| 2344 | return false, nil | 2351 | return false, nil | ||
| @@ -2352,6 +2359,29 @@ func (s *Service) tryHTTPProbe(ctx context.Context, containerID string, containe | |||||
| 2352 | return true, httpProbe(ctx, url, timeout) | 2359 | return true, httpProbe(ctx, url, timeout) | ||
| 2353 | } | 2360 | } | ||
| 2354 | 2361 | ||||
| 2362 | + | // tryDefaultHTTPProbe attempts an HTTP GET on "/" using the container's | |||
| 2363 | + | // exposed port. This is the default readiness check for web containers — | |||
| 2364 | + | // it verifies the app is actually serving HTTP, not just accepting TCP. | |||
| 2365 | + | // Returns (true, err) if the probe was attempted, (false, nil) if skipped. | |||
| 2366 | + | func (s *Service) tryDefaultHTTPProbe(ctx context.Context, containerID string, containerConfig *domain.ContainerConfig, cfg Config) (bool, error) { | |||
| 2367 | + | log := zerowrap.FromCtx(ctx) | |||
| 2368 | + | if containerConfig == nil || len(containerConfig.Ports) == 0 { | |||
| 2369 | + | return false, nil | |||
| 2370 | + | } | |||
| 2371 | + | ip, port, probeErr := s.resolveProbeEndpoint(ctx, containerID, containerConfig) | |||
| 2372 | + | if probeErr != nil || ip == "" || port <= 0 { | |||
| 2373 | + | log.Debug().Err(probeErr).Msg("readiness cascade: default HTTP probe skipped, could not resolve container endpoint") | |||
| 2374 | + | return false, nil | |||
| 2375 | + | } | |||
| 2376 | + | url := fmt.Sprintf("http://%s:%d/", ip, port) | |||
| 2377 | + | timeout := cfg.HTTPProbeTimeout | |||
| 2378 | + | if timeout == 0 { | |||
| 2379 | + | timeout = 60 * time.Second | |||
| 2380 | + | } | |||
| 2381 | + | log.Info().Str("url", url).Dur("timeout", timeout).Msg("readiness cascade: using default HTTP alive probe") | |||
| 2382 | + | return true, httpAliveProbe(ctx, url, timeout) | |||
| 2383 | + | } | |||
| 2384 | + | ||||
| 2355 | // tryTCPProbe attempts a TCP probe if port info is available. | 2385 | // tryTCPProbe attempts a TCP probe if port info is available. | ||
| 2356 | // Returns (true, err) if the probe was attempted, (false, nil) if skipped. | 2386 | // Returns (true, err) if the probe was attempted, (false, nil) if skipped. | ||
| 2357 | func (s *Service) tryTCPProbe(ctx context.Context, containerID string, containerConfig *domain.ContainerConfig, cfg Config) (bool, error) { | 2387 | func (s *Service) tryTCPProbe(ctx context.Context, containerID string, containerConfig *domain.ContainerConfig, cfg Config) (bool, error) { | ||
| @@ -2359,7 +2389,7 @@ func (s *Service) tryTCPProbe(ctx context.Context, containerID string, container | |||||
| 2359 | if containerConfig == nil || len(containerConfig.Ports) == 0 { | 2389 | if containerConfig == nil || len(containerConfig.Ports) == 0 { | ||
| 2360 | return false, nil | 2390 | return false, nil | ||
| 2361 | } | 2391 | } | ||
| 2362 | - | ip, port, probeErr := s.resolveContainerEndpoint(ctx, containerID) | 2392 | + | ip, port, probeErr := s.resolveProbeEndpoint(ctx, containerID, containerConfig) |
| 2363 | if probeErr != nil || ip == "" || port <= 0 { | 2393 | if probeErr != nil || ip == "" || port <= 0 { | ||
| 2364 | log.Debug().Err(probeErr).Msg("readiness cascade: TCP probe skipped, could not resolve container endpoint") | 2394 | log.Debug().Err(probeErr).Msg("readiness cascade: TCP probe skipped, could not resolve container endpoint") | ||
| 2365 | return false, nil | 2395 | return false, nil | ||
| @@ -2378,16 +2408,33 @@ func (s *Service) tryTCPProbe(ctx context.Context, containerID string, container | |||||
| 2378 | // from the host. We use the host port binding (127.0.0.1:<mapped_port>) | 2408 | // from the host. We use the host port binding (127.0.0.1:<mapped_port>) | ||
| 2379 | // which is always reachable. Falls back to the container's internal IP | 2409 | // which is always reachable. Falls back to the container's internal IP | ||
| 2380 | // only if no host port mapping exists (e.g. host-network mode). | 2410 | // only if no host port mapping exists (e.g. host-network mode). | ||
| 2381 | - | func (s *Service) resolveContainerEndpoint(ctx context.Context, containerID string) (string, int, error) { | 2411 | + | func (s *Service) resolveProbeEndpoint(ctx context.Context, containerID string, containerConfig *domain.ContainerConfig) (string, int, error) { |
| 2382 | log := zerowrap.FromCtx(ctx) | 2412 | log := zerowrap.FromCtx(ctx) | ||
| 2383 | 2413 | ||||
| 2384 | - | // First, get the container's internal port from network info | 2414 | + | internalPort := 0 |
| 2385 | - | _, internalPort, err := s.runtime.GetContainerNetworkInfo(ctx, containerID) | 2415 | + | if containerConfig != nil { |
| 2386 | - | if err != nil { | 2416 | + | if labels := containerConfig.Labels; labels != nil { |
| 2387 | - | return "", 0, err | 2417 | + | for _, key := range []string{domain.LabelProxyPort, domain.LabelPort} { |
| 2418 | + | if portStr, ok := labels[key]; ok && portStr != "" { | |||
| 2419 | + | port, err := strconv.Atoi(portStr) | |||
| 2420 | + | if err == nil && port > 0 { | |||
| 2421 | + | internalPort = port | |||
| 2422 | + | break | |||
| 2423 | + | } | |||
| 2424 | + | log.Warn().Str("label", key).Str("port_value", portStr).Msg("invalid probe port label value") | |||
| 2425 | + | } | |||
| 2426 | + | } | |||
| 2427 | + | } | |||
| 2428 | + | } | |||
| 2429 | + | ||||
| 2430 | + | if internalPort == 0 { | |||
| 2431 | + | _, resolvedPort, err := s.runtime.GetContainerNetworkInfo(ctx, containerID) | |||
| 2432 | + | if err != nil { | |||
| 2433 | + | return "", 0, err | |||
| 2434 | + | } | |||
| 2435 | + | internalPort = resolvedPort | |||
| 2388 | } | 2436 | } | ||
| 2389 | 2437 | ||||
| 2390 | - | // Try to resolve via host port binding (works in rootless podman/Docker) | |||
| 2391 | hostPort, hostErr := s.runtime.GetContainerPort(ctx, containerID, internalPort) | 2438 | hostPort, hostErr := s.runtime.GetContainerPort(ctx, containerID, internalPort) | ||
| 2392 | if hostErr == nil && hostPort > 0 { | 2439 | if hostErr == nil && hostPort > 0 { | ||
| 2393 | log.Debug(). | 2440 | log.Debug(). | ||
| @@ -2397,7 +2444,6 @@ func (s *Service) resolveContainerEndpoint(ctx context.Context, containerID stri | |||||
| 2397 | return "127.0.0.1", hostPort, nil | 2444 | return "127.0.0.1", hostPort, nil | ||
| 2398 | } | 2445 | } | ||
| 2399 | 2446 | ||||
| 2400 | - | // Fallback: use internal IP (works in Docker rootful / host network) | |||
| 2401 | ip, _, fallbackErr := s.runtime.GetContainerNetworkInfo(ctx, containerID) | 2447 | ip, _, fallbackErr := s.runtime.GetContainerNetworkInfo(ctx, containerID) | ||
| 2402 | if fallbackErr != nil { | 2448 | if fallbackErr != nil { | ||
| 2403 | return "", 0, fallbackErr | 2449 | return "", 0, fallbackErr | ||
M internal/usecase/container/service_readiness_test.go +9 -12
| @@ -47,10 +47,11 @@ func TestService_WaitForReady_AutoFallsBackToDelayWhenNoHealthcheck(t *testing.T | |||
| 47 | 47 | assert.NoError(t, err) | |
| 48 | 48 | } | |
| 49 | 49 | ||
| 50 | - | func TestService_WaitForReady_AutoCascadeUsesTCPWhenNoHealthcheckAndNoHealthLabel(t *testing.T) { | |
| 50 | + | func TestService_WaitForReady_AutoCascadeUsesDefaultHTTPAliveProbeWhenNoHealthcheckAndNoHealthLabel(t *testing.T) { | |
| 51 | 51 | runtime := mocks.NewMockContainerRuntime(t) | |
| 52 | 52 | svc := NewService(runtime, nil, nil, nil, Config{ | |
| 53 | - | ReadinessMode: "auto", | |
| 53 | + | ReadinessMode: "auto", | |
| 54 | + | HTTPProbeTimeout: 50 * time.Millisecond, | |
| 54 | 55 | }) | |
| 55 | 56 | ||
| 56 | 57 | ctx := testContext() | |
| @@ -68,21 +69,17 @@ func TestService_WaitForReady_AutoCascadeUsesTCPWhenNoHealthcheckAndNoHealthLabe | |||
| 68 | 69 | runtime.EXPECT().IsContainerRunning(mock.Anything, containerID).Return(true, nil).Once() | |
| 69 | 70 | // No Docker healthcheck | |
| 70 | 71 | runtime.EXPECT().GetContainerHealthStatus(mock.Anything, containerID).Return("", false, nil).Once() | |
| 71 | - | // Cascade resolves container endpoint for TCP probe | |
| 72 | + | // Cascade resolves container endpoint for default HTTP alive probe | |
| 72 | 73 | runtime.EXPECT().GetContainerNetworkInfo(mock.Anything, containerID).Return(probeIP, probePort, nil).Once() | |
| 73 | 74 | // Host port binding resolution — return the same loopback addr so probe hits it | |
| 74 | 75 | runtime.EXPECT().GetContainerPort(mock.Anything, containerID, probePort).Return(probePort, nil).Once() | |
| 75 | 76 | ||
| 76 | - | // TCP probe will try to connect — which will fail since the port is closed. | |
| 77 | - | // Use a short health timeout so the test doesn't hang. | |
| 78 | - | svc.mu.Lock() | |
| 79 | - | svc.config.TCPProbeTimeout = 50 * time.Millisecond | |
| 80 | - | svc.mu.Unlock() | |
| 77 | + | // Default HTTP alive probe will try to connect — which will fail since the port is closed. | |
| 81 | 78 | ||
| 82 | 79 | err := svc.waitForReady(ctx, containerID, containerConfig) | |
| 83 | - | // Expect TCP probe timeout (no server listening) | |
| 80 | + | // Expect HTTP alive probe timeout (no server listening) | |
| 84 | 81 | assert.Error(t, err) | |
| 85 | - | assert.Contains(t, err.Error(), "TCP probe timeout") | |
| 82 | + | assert.Contains(t, err.Error(), "HTTP alive probe timeout") | |
| 86 | 83 | } | |
| 87 | 84 | ||
| 88 | 85 | func TestService_WaitForReady_AutoCascadeUsesHTTPProbeWhenHealthLabelSet(t *testing.T) { | |
| @@ -169,8 +166,8 @@ func TestService_WaitForReady_AutoCascadeFallsToDelayWhenNoEndpoint(t *testing.T | |||
| 169 | 166 | runtime.EXPECT().IsContainerRunning(mock.Anything, containerID).Return(true, nil).Once() | |
| 170 | 167 | // No Docker healthcheck | |
| 171 | 168 | runtime.EXPECT().GetContainerHealthStatus(mock.Anything, containerID).Return("", false, nil).Once() | |
| 172 | - | // Network info fails — can't do TCP probe, fall through to delay | |
| 173 | - | runtime.EXPECT().GetContainerNetworkInfo(mock.Anything, containerID).Return("", 0, assert.AnError).Once() | |
| 169 | + | // Network info fails for default HTTP alive probe, then again for TCP fallback before delay | |
| 170 | + | runtime.EXPECT().GetContainerNetworkInfo(mock.Anything, containerID).Return("", 0, assert.AnError).Twice() | |
| 174 | 171 | // Delay-mode fallback verification | |
| 175 | 172 | runtime.EXPECT().IsContainerRunning(mock.Anything, containerID).Return(true, nil).Once() | |
| 176 | 173 | ||
| @@ -47,10 +47,11 @@ func TestService_WaitForReady_AutoFallsBackToDelayWhenNoHealthcheck(t *testing.T | |||||
| 47 | assert.NoError(t, err) | 47 | assert.NoError(t, err) | ||
| 48 | } | 48 | } | ||
| 49 | 49 | ||||
| 50 | - | func TestService_WaitForReady_AutoCascadeUsesTCPWhenNoHealthcheckAndNoHealthLabel(t *testing.T) { | 50 | + | func TestService_WaitForReady_AutoCascadeUsesDefaultHTTPAliveProbeWhenNoHealthcheckAndNoHealthLabel(t *testing.T) { |
| 51 | runtime := mocks.NewMockContainerRuntime(t) | 51 | runtime := mocks.NewMockContainerRuntime(t) | ||
| 52 | svc := NewService(runtime, nil, nil, nil, Config{ | 52 | svc := NewService(runtime, nil, nil, nil, Config{ | ||
| 53 | - | ReadinessMode: "auto", | 53 | + | ReadinessMode: "auto", |
| 54 | + | HTTPProbeTimeout: 50 * time.Millisecond, | |||
| 54 | }) | 55 | }) | ||
| 55 | 56 | ||||
| 56 | ctx := testContext() | 57 | ctx := testContext() | ||
| @@ -68,21 +69,17 @@ func TestService_WaitForReady_AutoCascadeUsesTCPWhenNoHealthcheckAndNoHealthLabe | |||||
| 68 | runtime.EXPECT().IsContainerRunning(mock.Anything, containerID).Return(true, nil).Once() | 69 | runtime.EXPECT().IsContainerRunning(mock.Anything, containerID).Return(true, nil).Once() | ||
| 69 | // No Docker healthcheck | 70 | // No Docker healthcheck | ||
| 70 | runtime.EXPECT().GetContainerHealthStatus(mock.Anything, containerID).Return("", false, nil).Once() | 71 | runtime.EXPECT().GetContainerHealthStatus(mock.Anything, containerID).Return("", false, nil).Once() | ||
| 71 | - | // Cascade resolves container endpoint for TCP probe | 72 | + | // Cascade resolves container endpoint for default HTTP alive probe |
| 72 | runtime.EXPECT().GetContainerNetworkInfo(mock.Anything, containerID).Return(probeIP, probePort, nil).Once() | 73 | runtime.EXPECT().GetContainerNetworkInfo(mock.Anything, containerID).Return(probeIP, probePort, nil).Once() | ||
| 73 | // Host port binding resolution — return the same loopback addr so probe hits it | 74 | // Host port binding resolution — return the same loopback addr so probe hits it | ||
| 74 | runtime.EXPECT().GetContainerPort(mock.Anything, containerID, probePort).Return(probePort, nil).Once() | 75 | runtime.EXPECT().GetContainerPort(mock.Anything, containerID, probePort).Return(probePort, nil).Once() | ||
| 75 | 76 | ||||
| 76 | - | // TCP probe will try to connect — which will fail since the port is closed. | 77 | + | // Default HTTP alive probe will try to connect — which will fail since the port is closed. |
| 77 | - | // Use a short health timeout so the test doesn't hang. | |||
| 78 | - | svc.mu.Lock() | |||
| 79 | - | svc.config.TCPProbeTimeout = 50 * time.Millisecond | |||
| 80 | - | svc.mu.Unlock() | |||
| 81 | 78 | ||||
| 82 | err := svc.waitForReady(ctx, containerID, containerConfig) | 79 | err := svc.waitForReady(ctx, containerID, containerConfig) | ||
| 83 | - | // Expect TCP probe timeout (no server listening) | 80 | + | // Expect HTTP alive probe timeout (no server listening) |
| 84 | assert.Error(t, err) | 81 | assert.Error(t, err) | ||
| 85 | - | assert.Contains(t, err.Error(), "TCP probe timeout") | 82 | + | assert.Contains(t, err.Error(), "HTTP alive probe timeout") |
| 86 | } | 83 | } | ||
| 87 | 84 | ||||
| 88 | func TestService_WaitForReady_AutoCascadeUsesHTTPProbeWhenHealthLabelSet(t *testing.T) { | 85 | func TestService_WaitForReady_AutoCascadeUsesHTTPProbeWhenHealthLabelSet(t *testing.T) { | ||
| @@ -169,8 +166,8 @@ func TestService_WaitForReady_AutoCascadeFallsToDelayWhenNoEndpoint(t *testing.T | |||||
| 169 | runtime.EXPECT().IsContainerRunning(mock.Anything, containerID).Return(true, nil).Once() | 166 | runtime.EXPECT().IsContainerRunning(mock.Anything, containerID).Return(true, nil).Once() | ||
| 170 | // No Docker healthcheck | 167 | // No Docker healthcheck | ||
| 171 | runtime.EXPECT().GetContainerHealthStatus(mock.Anything, containerID).Return("", false, nil).Once() | 168 | runtime.EXPECT().GetContainerHealthStatus(mock.Anything, containerID).Return("", false, nil).Once() | ||
| 172 | - | // Network info fails — can't do TCP probe, fall through to delay | 169 | + | // Network info fails for default HTTP alive probe, then again for TCP fallback before delay |
| 173 | - | runtime.EXPECT().GetContainerNetworkInfo(mock.Anything, containerID).Return("", 0, assert.AnError).Once() | 170 | + | runtime.EXPECT().GetContainerNetworkInfo(mock.Anything, containerID).Return("", 0, assert.AnError).Twice() |
| 174 | // Delay-mode fallback verification | 171 | // Delay-mode fallback verification | ||
| 175 | runtime.EXPECT().IsContainerRunning(mock.Anything, containerID).Return(true, nil).Once() | 172 | runtime.EXPECT().IsContainerRunning(mock.Anything, containerID).Return(true, nil).Once() | ||
| 176 | 173 | ||||