fix: resolve readiness probe lint issues
Replace the log stream cancellation marker with explicit stopped state so lint no longer flags an uncancelled context, and drop the unused readiness endpoint wrapper left behind by the probe refactor.
- commit
- fced65178c906feadc58c401bfa83407a43f1386
- author
- brice · · 2026-03-12 19:09 UTC
- parent
- 50a9fab
- signature
- unsigned
2 files · +7 -13
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/service.go +0 -4
| @@ -2455,10 +2455,6 @@ func (s *Service) resolveProbeEndpoint(ctx context.Context, containerID string, | |||
| 2455 | 2455 | return ip, internalPort, nil | |
| 2456 | 2456 | } | |
| 2457 | 2457 | ||
| 2458 | - | func (s *Service) resolveContainerEndpoint(ctx context.Context, containerID string) (string, int, error) { | |
| 2459 | - | return s.resolveProbeEndpoint(ctx, containerID, nil) | |
| 2460 | - | } | |
| 2461 | - | ||
| 2462 | 2458 | // waitForReadyByDelay waits using the legacy running+delay strategy. | |
| 2463 | 2459 | func (s *Service) waitForReadyByDelay(ctx context.Context, containerID string) error { | |
| 2464 | 2460 | log := zerowrap.FromCtx(ctx) | |
| @@ -2455,10 +2455,6 @@ func (s *Service) resolveProbeEndpoint(ctx context.Context, containerID string, | |||||
| 2455 | return ip, internalPort, nil | 2455 | return ip, internalPort, nil | ||
| 2456 | } | 2456 | } | ||
| 2457 | 2457 | ||||
| 2458 | - | func (s *Service) resolveContainerEndpoint(ctx context.Context, containerID string) (string, int, error) { | |||
| 2459 | - | return s.resolveProbeEndpoint(ctx, containerID, nil) | |||
| 2460 | - | } | |||
| 2461 | - | ||||
| 2462 | // waitForReadyByDelay waits using the legacy running+delay strategy. | 2458 | // waitForReadyByDelay waits using the legacy running+delay strategy. | ||
| 2463 | func (s *Service) waitForReadyByDelay(ctx context.Context, containerID string) error { | 2459 | func (s *Service) waitForReadyByDelay(ctx context.Context, containerID string) error { | ||
| 2464 | log := zerowrap.FromCtx(ctx) | 2460 | log := zerowrap.FromCtx(ctx) | ||