From ee5188fce5bb32108b431f0c489191d0fc9281c0 Mon Sep 17 00:00:00 2001 From: Charles Wu Date: Tue, 25 Aug 2026 13:24:24 +1000 Subject: [PATCH 1/4] feat: make the worker handoff write path cancellable Follow-up to #1231, addressing both review threads left open there. Cancellation. WriteImagesPipe and WriteCompletionPipe now take a context. The collector and remover derive theirs from SIGTERM, so a terminating pod no longer leaves a worker blocked forever on a peer that is never going to arrive, and the scanner passes the context it already has. The Unix rendezvous is deliberately untouched. I had proposed O_NONBLOCK plus polling, but that changes the syscall every existing deployment depends on, and a non-blocking descriptor then has to handle EAGAIN on payloads larger than the pipe buffer. Instead the blocking open runs on a goroutine that hands the file back if the caller is still waiting and closes it if not. Linux keeps the exact open it has always used; only the waiting becomes interruptible. WriteCompletionPipe stats the path before opening, so an absent scanner is still reported as ENOENT even when the context is already done. Left to the select, that case would have been decided at random, which would have made "scanner disabled" indistinguishable from "we are shutting down". WriteScanErasePipe keeps its signature for out-of-tree scanners and waits indefinitely, as before. Endpoint safety. listen removed whatever sat at the endpoint path before binding. A socket left behind by an unclean exit does have to go, or a crashed worker would poison the endpoint for every retry, but anything else there is not ours to delete: the worker runs as NT AUTHORITY\SYSTEM and shares the volume with a scanner image we do not control. Lstat reports ModeSocket on Windows, so the two cases are separable. Signed-off-by: Charles Wu --- pkg/collector/collector.go | 10 ++++- pkg/remover/remover.go | 14 +++--- pkg/scanners/template/scanner_template.go | 2 +- pkg/utils/handoff_test.go | 31 +++++++++++-- pkg/utils/handoff_unix.go | 52 +++++++++++++++++++--- pkg/utils/handoff_windows.go | 53 +++++++++++++++-------- pkg/utils/platform_windows_test.go | 43 ++++++++++++++++++ pkg/utils/utils.go | 5 ++- 8 files changed, 173 insertions(+), 37 deletions(-) diff --git a/pkg/collector/collector.go b/pkg/collector/collector.go index 16646f84f6..8ea21ef59d 100644 --- a/pkg/collector/collector.go +++ b/pkg/collector/collector.go @@ -1,11 +1,14 @@ package main import ( + "context" "flag" "fmt" "net/http" _ "net/http/pprof" "os" + "os/signal" + "syscall" "time" "github.com/eraser-dev/eraser/pkg/cri" @@ -29,6 +32,11 @@ var ( func main() { flag.Parse() + // A terminating pod should not leave the worker blocked on a peer that is + // never going to arrive. The stop func is discarded rather than deferred + // because every exit path here is os.Exit, which would skip it anyway. + ctx, _ := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + if *enableProfile { go func() { server := &http.Server{ @@ -85,7 +93,7 @@ func main() { os.Exit(1) } - if err := util.WriteImagesPipe(path, finalImages); err != nil { + if err := util.WriteImagesPipe(ctx, path, finalImages); err != nil { log.Error(err, "failed to send images", "pipeFile", path) os.Exit(1) } diff --git a/pkg/remover/remover.go b/pkg/remover/remover.go index 5db5bb2dd5..f9f637547d 100644 --- a/pkg/remover/remover.go +++ b/pkg/remover/remover.go @@ -39,6 +39,11 @@ const ( func main() { flag.Parse() + // A terminating pod should not leave the worker blocked on a peer that is + // never going to arrive. The stop func is discarded rather than deferred + // because every exit path here is os.Exit, which would skip it anyway. + ctx, _ := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + if *enableProfile { go func() { server := &http.Server{ @@ -74,7 +79,7 @@ func main() { } if *imageListPtr == "" { - nonCompliantImages, err := util.ReadImagesPipe(context.Background(), util.ScanErasePath) + nonCompliantImages, err := util.ReadImagesPipe(ctx, util.ScanErasePath) if err != nil { log.Error(err, "error reading non-compliant images") os.Exit(generalErr) @@ -114,8 +119,6 @@ func main() { if os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT") != "" { // record metrics - ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - exporter, reader, provider := metrics.ConfigureMetrics(ctx, log, os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT")) otel.SetMeterProvider(provider) @@ -123,16 +126,15 @@ func main() { log.Error(err, "error recording metrics") } metrics.ExportMetrics(log, exporter, reader) - cancel() } if *imageListPtr == "" { - if err := util.WriteCompletionPipe(util.EraseCompleteCollectPath); err != nil { + if err := util.WriteCompletionPipe(ctx, util.EraseCompleteCollectPath); err != nil { log.Error(err, "unable to signal completion", "pipeFile", util.EraseCompleteCollectPath) os.Exit(generalErr) } - err := util.WriteCompletionPipe(util.EraseCompleteScanPath) + err := util.WriteCompletionPipe(ctx, util.EraseCompleteScanPath) // if the scanner is disabled if os.IsNotExist(err) { return diff --git a/pkg/scanners/template/scanner_template.go b/pkg/scanners/template/scanner_template.go index 9f12dcfa21..ba66200059 100644 --- a/pkg/scanners/template/scanner_template.go +++ b/pkg/scanners/template/scanner_template.go @@ -88,7 +88,7 @@ func (cfg *config) SendImages(nonCompliantImages, failedImages []unversioned.Ima nonCompliantImages = append(nonCompliantImages, failedImages...) } - if err := util.WriteScanErasePipe(nonCompliantImages); err != nil { + if err := util.WriteImagesPipe(cfg.ctx, util.ScanErasePath, nonCompliantImages); err != nil { cfg.log.Error(err, "unable to write non-compliant images to scan erase pipe") return err } diff --git a/pkg/utils/handoff_test.go b/pkg/utils/handoff_test.go index 0d1e45a227..b30476e603 100644 --- a/pkg/utils/handoff_test.go +++ b/pkg/utils/handoff_test.go @@ -2,9 +2,11 @@ package utils import ( "context" + "errors" "os" "path/filepath" "testing" + "time" "github.com/eraser-dev/eraser/api/unversioned" ) @@ -36,7 +38,7 @@ func TestImagesHandoffRoundTrip(t *testing.T) { } errCh := make(chan error, 1) - go func() { errCh <- WriteImagesPipe(path, want) }() + go func() { errCh <- WriteImagesPipe(context.Background(), path, want) }() got, err := ReadImagesPipe(context.Background(), path) if err != nil { @@ -66,7 +68,7 @@ func TestCompletionHandoffRoundTrip(t *testing.T) { defer func() { _ = pipe.Close() }() errCh := make(chan error, 1) - go func() { errCh <- WriteCompletionPipe(path) }() + go func() { errCh <- WriteCompletionPipe(context.Background(), path) }() data, err := pipe.Await() if err != nil { @@ -86,7 +88,7 @@ func TestCompletionHandoffRoundTrip(t *testing.T) { func TestWriteCompletionPipeAbsentPeerIsNotExist(t *testing.T) { path := filepath.Join(shortTempDir(t), "no-such-peer") - err := WriteCompletionPipe(path) + err := WriteCompletionPipe(context.Background(), path) if err == nil { t.Fatal("expected an error writing to an endpoint nobody published") } @@ -95,6 +97,29 @@ func TestWriteCompletionPipeAbsentPeerIsNotExist(t *testing.T) { } } +// The whole point of taking a context: a worker whose peer never arrives has to +// be able to give up, on either platform. +func TestWriteImagesPipeHonoursACanceledContext(t *testing.T) { + path := filepath.Join(shortTempDir(t), "never-read") + + ctx, cancel := context.WithCancel(context.Background()) + + errCh := make(chan error, 1) + go func() { errCh <- WriteImagesPipe(ctx, path, []unversioned.Image{{ImageID: "sha256:aaaa"}}) }() + + // nothing ever reads the endpoint, so the write is still waiting + cancel() + + select { + case err := <-errCh: + if !errors.Is(err, context.Canceled) { + t.Errorf("WriteImagesPipe = %v, want context.Canceled", err) + } + case <-time.After(30 * time.Second): + t.Fatal("WriteImagesPipe ignored the canceled context") + } +} + func TestCompletionPipeCloseIsIdempotentlySafe(t *testing.T) { path := filepath.Join(shortTempDir(t), "closed") diff --git a/pkg/utils/handoff_unix.go b/pkg/utils/handoff_unix.go index 49ae8dd35e..bd36cd4b01 100644 --- a/pkg/utils/handoff_unix.go +++ b/pkg/utils/handoff_unix.go @@ -68,8 +68,9 @@ func (p *CompletionPipe) Close() error { return nil } -// WriteImagesPipe publishes the endpoint and blocks until the reader connects. -func WriteImagesPipe(path string, images []unversioned.Image) error { +// WriteImagesPipe publishes the endpoint and blocks until the reader connects, +// or until ctx is done. +func WriteImagesPipe(ctx context.Context, path string, images []unversioned.Image) error { data, err := json.Marshal(images) if err != nil { return err @@ -79,8 +80,7 @@ func WriteImagesPipe(path string, images []unversioned.Image) error { return err } - //nolint:gosec // G304: Opening pipe file is intended functionality - file, err := os.OpenFile(path, os.O_WRONLY, 0) + file, err := openForWrite(ctx, path) if err != nil { return err } @@ -93,6 +93,38 @@ func WriteImagesPipe(path string, images []unversioned.Image) error { return err } +// openForWrite opens a FIFO for writing, which blocks in the kernel until a +// reader arrives. The open itself is left untouched -- the rendezvous, and the +// behavior every existing deployment depends on, is exactly as before. Only +// the waiting is made interruptible, by doing it on a goroutine that hands the +// file over if anyone is still listening and closes it if not. +func openForWrite(ctx context.Context, path string) (*os.File, error) { + type opened struct { + file *os.File + err error + } + + ch := make(chan opened, 1) + go func() { + //nolint:gosec // G304: Opening pipe file is intended functionality + file, err := os.OpenFile(path, os.O_WRONLY, 0) + select { + case ch <- opened{file: file, err: err}: + default: + if file != nil { + _ = file.Close() + } + } + }() + + select { + case <-ctx.Done(): + return nil, ctx.Err() + case o := <-ch: + return o.file, o.err + } +} + // ReadImagesPipe waits for the endpoint to appear, then reads until the writer // finishes. It returns ctx.Err() if the context is canceled while waiting. func ReadImagesPipe(ctx context.Context, path string) ([]unversioned.Image, error) { @@ -143,9 +175,15 @@ func ReadImagesPipe(ctx context.Context, path string) ([]unversioned.Image, erro // WriteCompletionPipe signals a peer that this stage is done. The returned error // satisfies os.IsNotExist when the peer never published the endpoint, which is // how an absent scanner is detected. -func WriteCompletionPipe(path string) error { - //nolint:gosec // G304: Opening pipe file is intended functionality - file, err := os.OpenFile(path, os.O_WRONLY, 0) +func WriteCompletionPipe(ctx context.Context, path string) error { + // Checked before the open so that an absent peer is reported as such even + // when ctx is already done; otherwise a terminating remover could mistake a + // disabled scanner for a cancellation, and vice versa. + if _, err := os.Stat(path); err != nil { + return err + } + + file, err := openForWrite(ctx, path) if err != nil { return err } diff --git a/pkg/utils/handoff_windows.go b/pkg/utils/handoff_windows.go index 0323da79dc..df8d3c1d83 100644 --- a/pkg/utils/handoff_windows.go +++ b/pkg/utils/handoff_windows.go @@ -73,16 +73,15 @@ func (p *CompletionPipe) Close() error { return l.Close() } -// WriteImagesPipe blocks until the reader is listening, then sends the list. -// The unbounded retry mirrors the Unix implementation, where opening a FIFO for -// writing blocks until a reader arrives. -func WriteImagesPipe(path string, images []unversioned.Image) error { +// WriteImagesPipe blocks until the reader is listening, then sends the list, or +// returns ctx.Err() if the context is done first. +func WriteImagesPipe(ctx context.Context, path string, images []unversioned.Image) error { data, err := json.Marshal(images) if err != nil { return err } - conn, err := dialForever(path) + conn, err := dial(ctx, path) if err != nil { return err } @@ -141,7 +140,7 @@ func ReadImagesPipe(ctx context.Context, path string) ([]unversioned.Image, erro // WriteCompletionPipe signals a peer that this stage is done. The returned error // satisfies os.IsNotExist when the peer never published the endpoint, which is // how an absent scanner is detected. -func WriteCompletionPipe(path string) error { +func WriteCompletionPipe(ctx context.Context, path string) error { // Dialing a socket that is not there reports connection-refused on Windows // rather than ENOENT, so the filesystem is the only reliable way to tell // "never published" from "published but gone". @@ -149,7 +148,8 @@ func WriteCompletionPipe(path string) error { return err } - conn, err := net.Dial("unix", path) + var d net.Dialer + conn, err := d.DialContext(ctx, "unix", path) if err != nil { return err } @@ -167,29 +167,48 @@ func listen(path string) (net.Listener, error) { return nil, fmt.Errorf("socket path %q is %d bytes, over the %d byte limit", path, len(path), maxSocketPath) } - // a socket left behind by a previous run would fail the bind - if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) { + // A socket left behind by an unclean exit would fail the bind, so it has to + // go. Anything else at this path is not ours to delete: the worker runs as + // SYSTEM and shares the volume with a scanner image we do not control. + switch fi, err := os.Lstat(path); { + case errors.Is(err, fs.ErrNotExist): + case err != nil: return nil, err + case fi.Mode()&os.ModeSocket == 0: + return nil, fmt.Errorf("refusing to replace %q: it exists and is not a socket", path) + default: + if err := os.Remove(path); err != nil && !errors.Is(err, fs.ErrNotExist) { + return nil, err + } } return net.Listen("unix", path) } -// dialForever waits for the reader to start listening. Errors are not -// classified: Windows reports a missing socket as connection-refused, so there -// is no reliable "not yet" error to match on. Retrying unconditionally mirrors -// the Unix implementation, where opening a FIFO for writing blocks until a -// reader arrives. -func dialForever(path string) (net.Conn, error) { +// dial waits for the reader to start listening. Errors are not classified: +// Windows reports a missing socket as connection-refused, so there is no +// reliable "not yet" error to match on. Retrying on a tick mirrors the Unix +// implementation, where opening a FIFO for writing blocks until a reader +// arrives. +func dial(ctx context.Context, path string) (net.Conn, error) { if len(path) > maxSocketPath { return nil, fmt.Errorf("socket path %q is %d bytes, over the %d byte limit", path, len(path), maxSocketPath) } + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + + var d net.Dialer for { - conn, err := net.Dial("unix", path) + conn, err := d.DialContext(ctx, "unix", path) if err == nil { return conn, nil } - time.Sleep(time.Second) + + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-ticker.C: + } } } diff --git a/pkg/utils/platform_windows_test.go b/pkg/utils/platform_windows_test.go index 4b4a911adb..11244a2b0a 100644 --- a/pkg/utils/platform_windows_test.go +++ b/pkg/utils/platform_windows_test.go @@ -6,6 +6,7 @@ import ( "context" "errors" "fmt" + "net" "os" "path/filepath" "strings" @@ -79,6 +80,48 @@ func TestSocketPathLimitBoundary(t *testing.T) { } } +// The worker runs as SYSTEM and shares the volume with a scanner image we do +// not control, so an occupied endpoint path is a reason to stop rather than to +// start deleting. +func TestListenRefusesToReplaceANonSocket(t *testing.T) { + dir := shortTempDir(t) + path := filepath.Join(dir, "occupied") + + if err := os.WriteFile(path, []byte("not a socket"), 0o600); err != nil { + t.Fatal(err) + } + + if l, err := listen(path); err == nil { + _ = l.Close() + t.Fatal("listen replaced a regular file, want an error") + } + + if _, err := os.Stat(path); err != nil { + t.Errorf("the file was removed anyway: %v", err) + } +} + +// A socket the previous run failed to clean up must still be reclaimable, +// otherwise a crashed worker would poison the endpoint for every retry. +func TestListenReclaimsAStaleSocket(t *testing.T) { + dir := shortTempDir(t) + path := filepath.Join(dir, "stale") + + stale, err := net.Listen("unix", path) + if err != nil { + t.Fatal(err) + } + // leaks the endpoint on purpose: Close would unlink it and remove the case + // under test + t.Cleanup(func() { _ = stale.Close() }) + + l, err := listen(path) + if err != nil { + t.Fatalf("listen over a stale socket: %v", err) + } + _ = l.Close() +} + func TestMkfifoUnsupported(t *testing.T) { if err := mkfifo("ignored", PipeMode); !errors.Is(err, ErrFifoUnsupported) { t.Errorf("mkfifo on windows = %v, want ErrFifoUnsupported", err) diff --git a/pkg/utils/utils.go b/pkg/utils/utils.go index f3d4521738..4de028ddef 100644 --- a/pkg/utils/utils.go +++ b/pkg/utils/utils.go @@ -334,9 +334,10 @@ func ReadCollectScanPipe(ctx context.Context) ([]unversioned.Image, error) { } // WriteScanErasePipe is the scanner-facing spelling of WriteImagesPipe, kept -// because custom scanners may call it directly. +// because custom scanners may call it directly. It waits indefinitely; reach +// for WriteImagesPipe when the wait needs to be cancellable. func WriteScanErasePipe(vulnerableImages []unversioned.Image) error { - return WriteImagesPipe(ScanErasePath, vulnerableImages) + return WriteImagesPipe(context.Background(), ScanErasePath, vulnerableImages) } func ProcessRepoDigests(repoDigests []string) ([]string, []error) { From 83eb67db82f6cffa1ab2aeb9ccd2e8c9f05622cd Mon Sep 17 00:00:00 2001 From: Charles Wu Date: Wed, 26 Aug 2026 11:31:56 +1000 Subject: [PATCH 2/4] fix: do not swallow SIGTERM, and do not orphan the opened FIFO All three found in review. The buffered channel in openForWrite defeated its own cleanup. With one slot free the send always succeeded, so the default case never ran: if the context won and a reader arrived later, the goroutine handed the file into a buffer nobody would ever read, leaking the descriptor and leaving the FIFO with a writer that never closes. Making the channel unbuffered is not enough on its own, because the open can win the race to that select before the caller reaches its own, and the default case would then close a file the caller was about to ask for. The channel is now unbuffered and paired with an explicit abandoned signal, so the goroutine blocks until the caller has either taken the file or given up on it. Registering signal notification also suppresses Go's default SIGTERM exit, and neither worker observed the context everywhere it mattered. removeImages built its five-minute timeout from context.Background, so a SIGTERM during deletion was ignored until the work finished or the kubelet escalated to SIGKILL; it now derives from the caller's context. In the collector the gap is after the write, where Await deliberately has no context, so notification is stopped before that wait and SIGTERM regains its default effect. Signed-off-by: Charles Wu --- pkg/collector/collector.go | 10 ++++++---- pkg/remover/helpers.go | 7 +++++-- pkg/remover/remover.go | 2 +- pkg/remover/remover_test.go | 3 ++- pkg/utils/handoff_unix.go | 11 +++++++++-- 5 files changed, 23 insertions(+), 10 deletions(-) diff --git a/pkg/collector/collector.go b/pkg/collector/collector.go index 8ea21ef59d..b02bce9306 100644 --- a/pkg/collector/collector.go +++ b/pkg/collector/collector.go @@ -32,10 +32,11 @@ var ( func main() { flag.Parse() - // A terminating pod should not leave the worker blocked on a peer that is - // never going to arrive. The stop func is discarded rather than deferred - // because every exit path here is os.Exit, which would skip it anyway. - ctx, _ := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + // Scoped to the cancellable write below and stopped before Await, which has + // no context: registering the handler suppresses the default SIGTERM exit, so + // holding it across an uncancellable wait would turn a terminating pod into a + // SIGKILL instead of a clean one. + ctx, stopSignals := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) if *enableProfile { go func() { @@ -97,6 +98,7 @@ func main() { log.Error(err, "failed to send images", "pipeFile", path) os.Exit(1) } + stopSignals() data, err := completion.Await() if err != nil { diff --git a/pkg/remover/helpers.go b/pkg/remover/helpers.go index 9d16e1d7d6..eb203e9e14 100644 --- a/pkg/remover/helpers.go +++ b/pkg/remover/helpers.go @@ -8,10 +8,13 @@ import ( util "github.com/eraser-dev/eraser/pkg/utils" ) -func removeImages(c cri.Remover, targetImages []string) (int, error) { +func removeImages(ctx context.Context, c cri.Remover, targetImages []string) (int, error) { removed := 0 - backgroundContext, cancel := context.WithTimeout(context.Background(), timeout) + // Derived from the caller's context, not Background: signal notification is + // registered for the whole process, so nothing would observe a SIGTERM during + // the deletion loop otherwise. + backgroundContext, cancel := context.WithTimeout(ctx, timeout) defer cancel() images, err := c.ListImages(backgroundContext) diff --git a/pkg/remover/remover.go b/pkg/remover/remover.go index f9f637547d..299eea9327 100644 --- a/pkg/remover/remover.go +++ b/pkg/remover/remover.go @@ -111,7 +111,7 @@ func main() { log.Info("no images to exclude") } - removed, err := removeImages(client, imagelist) + removed, err := removeImages(ctx, client, imagelist) if err != nil { log.Error(err, "failed to remove images") os.Exit(generalErr) diff --git a/pkg/remover/remover_test.go b/pkg/remover/remover_test.go index d6ca179578..4b3974a4ed 100644 --- a/pkg/remover/remover_test.go +++ b/pkg/remover/remover_test.go @@ -1,6 +1,7 @@ package main import ( + "context" "testing" v1 "k8s.io/cri-api/pkg/apis/runtime/v1" @@ -50,7 +51,7 @@ func TestRemoveImages(t *testing.T) { } } - _, err := removeImages(client, tc.remove) + _, err := removeImages(context.Background(), client, tc.remove) if tc.shouldErr && err == nil { t.Fatal("expected error, got none") } diff --git a/pkg/utils/handoff_unix.go b/pkg/utils/handoff_unix.go index bd36cd4b01..d43cc6fa62 100644 --- a/pkg/utils/handoff_unix.go +++ b/pkg/utils/handoff_unix.go @@ -104,13 +104,19 @@ func openForWrite(ctx context.Context, path string) (*os.File, error) { err error } - ch := make(chan opened, 1) + // Unbuffered, and paired with abandoned rather than a default case. A + // buffered channel would accept the file after the caller had already + // returned, orphaning the descriptor; a default case would close a file the + // caller was about to ask for, if the open won the race to this select. + ch := make(chan opened) + abandoned := make(chan struct{}) + go func() { //nolint:gosec // G304: Opening pipe file is intended functionality file, err := os.OpenFile(path, os.O_WRONLY, 0) select { case ch <- opened{file: file, err: err}: - default: + case <-abandoned: if file != nil { _ = file.Close() } @@ -119,6 +125,7 @@ func openForWrite(ctx context.Context, path string) (*os.File, error) { select { case <-ctx.Done(): + close(abandoned) return nil, ctx.Err() case o := <-ch: return o.file, o.err From 12469c307ec6e0856e38628a0303751337bdbd3a Mon Sep 17 00:00:00 2001 From: Charles Wu Date: Wed, 26 Aug 2026 16:03:47 +1000 Subject: [PATCH 3/4] fix: make the write itself cancellable, and stop consuming SIGTERM Four more from review. Only the rendezvous was cancellable, not the write. Once the socket or pipe buffer fills, a peer that connects and then stops draining blocks the worker indefinitely, so "the write path is cancellable" was not true for a large image list. Both platforms now watch the context and close the endpoint to unblock the write, and report ctx.Err() rather than the close-induced write error. The collector's signal handler covered far more than the one call that observes it. getImages builds its own timeout from context.Background, so registering the handler at the top of main meant a blocked CRI listing ignored SIGTERM for up to five minutes; the handler now starts immediately before the write. Stopping it afterwards also left a lost-signal window: a SIGTERM landing between the write returning and the handler stopping was consumed rather than killing the process, and the collector walked into Await and waited for SIGKILL. The context is checked once the handler is stopped. The absent-peer test only ever ran with a live context, so the precedence the pre-open Stat exists to guarantee was untested. A missing endpoint must report IsNotExist even when the context is already canceled, otherwise "the scanner is disabled" and "we are terminating" become indistinguishable. Signed-off-by: Charles Wu --- pkg/collector/collector.go | 20 +++++++++++++------ pkg/utils/handoff_test.go | 19 ++++++++++++++++++ pkg/utils/handoff_unix.go | 36 ++++++++++++++++++++++++++++------- pkg/utils/handoff_windows.go | 37 ++++++++++++++++++++++++++++-------- 4 files changed, 91 insertions(+), 21 deletions(-) diff --git a/pkg/collector/collector.go b/pkg/collector/collector.go index b02bce9306..8350d9c3ae 100644 --- a/pkg/collector/collector.go +++ b/pkg/collector/collector.go @@ -32,12 +32,6 @@ var ( func main() { flag.Parse() - // Scoped to the cancellable write below and stopped before Await, which has - // no context: registering the handler suppresses the default SIGTERM exit, so - // holding it across an uncancellable wait would turn a terminating pod into a - // SIGKILL instead of a clean one. - ctx, stopSignals := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - if *enableProfile { go func() { server := &http.Server{ @@ -94,12 +88,26 @@ func main() { os.Exit(1) } + // Registering the handler suppresses the default SIGTERM exit, so it covers + // exactly the one call that observes ctx. Everything above builds its own + // timeouts from Background, and Await below has no context at all; holding + // the handler across either would swallow the signal. + ctx, stopSignals := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + if err := util.WriteImagesPipe(ctx, path, finalImages); err != nil { + stopSignals() log.Error(err, "failed to send images", "pipeFile", path) os.Exit(1) } stopSignals() + // A signal that landed after the write completed was consumed by the handler + // rather than killing the process, so it has to be acted on here. + if err := ctx.Err(); err != nil { + log.Error(err, "terminating before waiting for completion") + os.Exit(1) + } + data, err := completion.Await() if err != nil { log.Error(err, "failed to read pipe", "pipeFile", util.EraseCompleteCollectPath) diff --git a/pkg/utils/handoff_test.go b/pkg/utils/handoff_test.go index b30476e603..a965370971 100644 --- a/pkg/utils/handoff_test.go +++ b/pkg/utils/handoff_test.go @@ -97,6 +97,25 @@ func TestWriteCompletionPipeAbsentPeerIsNotExist(t *testing.T) { } } +// The pre-open Stat exists so this precedence holds: an endpoint nobody +// published reports IsNotExist even when the caller is already shutting down. +// Left to a select, the two would race and "the scanner is disabled" would +// become indistinguishable from "we are terminating". +func TestWriteCompletionPipeAbsentPeerBeatsACanceledContext(t *testing.T) { + path := filepath.Join(shortTempDir(t), "no-such-peer") + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := WriteCompletionPipe(ctx, path) + if err == nil { + t.Fatal("expected an error writing to an endpoint nobody published") + } + if !os.IsNotExist(err) { + t.Errorf("os.IsNotExist(%v) = false, want true", err) + } +} + // The whole point of taking a context: a worker whose peer never arrives has to // be able to give up, on either platform. func TestWriteImagesPipeHonoursACanceledContext(t *testing.T) { diff --git a/pkg/utils/handoff_unix.go b/pkg/utils/handoff_unix.go index d43cc6fa62..da31a31d6a 100644 --- a/pkg/utils/handoff_unix.go +++ b/pkg/utils/handoff_unix.go @@ -85,7 +85,34 @@ func WriteImagesPipe(ctx context.Context, path string, images []unversioned.Imag return err } - _, err = file.Write(data) + return writeAndClose(ctx, file, data) +} + +// writeAndClose writes the payload and closes, which is what frames the message +// for the reader. The open is not the only place this can block: once the pipe +// buffer fills, a reader that stops draining blocks the write too, so the +// watcher closes the file to unblock it. +func writeAndClose(ctx context.Context, file *os.File, payload []byte) error { + done := make(chan struct{}) + defer close(done) + + go func() { + select { + case <-ctx.Done(): + _ = file.Close() + case <-done: + } + }() + + _, err := file.Write(payload) + + // The watcher may already have closed the file, which is what surfaced as the + // write error, so the context is checked before the error is trusted. + if ctxErr := ctx.Err(); ctxErr != nil { + _ = file.Close() + return ctxErr + } + if closeErr := file.Close(); closeErr != nil && err == nil { err = closeErr } @@ -195,10 +222,5 @@ func WriteCompletionPipe(ctx context.Context, path string) error { return err } - _, err = file.WriteString(EraseCompleteMessage) - if closeErr := file.Close(); closeErr != nil && err == nil { - err = closeErr - } - - return err + return writeAndClose(ctx, file, []byte(EraseCompleteMessage)) } diff --git a/pkg/utils/handoff_windows.go b/pkg/utils/handoff_windows.go index df8d3c1d83..24489dbb39 100644 --- a/pkg/utils/handoff_windows.go +++ b/pkg/utils/handoff_windows.go @@ -86,12 +86,38 @@ func WriteImagesPipe(ctx context.Context, path string, images []unversioned.Imag return err } - if _, err := conn.Write(data); err != nil { + return sendAndClose(ctx, conn, data) +} + +// sendAndClose writes the payload and closes, which is what frames the message +// for the reader. DialContext only makes connecting cancellable, so a peer that +// connects and then stops reading would block the write itself; closing the +// connection from the watcher is what unblocks it. +func sendAndClose(ctx context.Context, conn net.Conn, payload []byte) error { + done := make(chan struct{}) + defer close(done) + + go func() { + select { + case <-ctx.Done(): + _ = conn.Close() + case <-done: + } + }() + + _, err := conn.Write(payload) + + // The watcher may already have closed the connection, which is what surfaced + // as the write error, so the context is checked before the error is trusted. + if ctxErr := ctx.Err(); ctxErr != nil { + _ = conn.Close() + return ctxErr + } + if err != nil { _ = conn.Close() return err } - // closing is what signals end-of-message to the reader return conn.Close() } @@ -154,12 +180,7 @@ func WriteCompletionPipe(ctx context.Context, path string) error { return err } - if _, err := conn.Write([]byte(EraseCompleteMessage)); err != nil { - _ = conn.Close() - return err - } - - return conn.Close() + return sendAndClose(ctx, conn, []byte(EraseCompleteMessage)) } func listen(path string) (net.Listener, error) { From f18303cc6887c6d745e884ae0ef7f5e407b11e64 Mon Sep 17 00:00:00 2001 From: Charles Wu Date: Wed, 26 Aug 2026 16:38:59 +1000 Subject: [PATCH 4/4] fix: read ctx.Err before stopping signal delivery stopSignals cancels the context returned by NotifyContext, so checking ctx.Err afterwards always reported Canceled and the collector exited on every successful run instead of waiting for the erase to finish. The E2E suite caught it: collector_pipeline hung on all four Kubernetes versions while every other test passed, because the remover was left blocking on a completion endpoint whose reader had already exited. Signed-off-by: Charles Wu --- pkg/collector/collector.go | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/pkg/collector/collector.go b/pkg/collector/collector.go index 8350d9c3ae..17190fb7c2 100644 --- a/pkg/collector/collector.go +++ b/pkg/collector/collector.go @@ -99,12 +99,17 @@ func main() { log.Error(err, "failed to send images", "pipeFile", path) os.Exit(1) } + + // Read before stopping, because stopSignals cancels this context itself: + // checked afterwards it would always report Canceled, and the collector + // would exit on every successful run instead of waiting for the erase. + sigErr := ctx.Err() stopSignals() - // A signal that landed after the write completed was consumed by the handler - // rather than killing the process, so it has to be acted on here. - if err := ctx.Err(); err != nil { - log.Error(err, "terminating before waiting for completion") + // A signal that landed while the handler was registered was consumed rather + // than killing the process, so it has to be acted on here. + if sigErr != nil { + log.Error(sigErr, "terminating before waiting for completion") os.Exit(1) }