From 52a4a8036eff9bd439620683b1f5a8e53e0b8d82 Mon Sep 17 00:00:00 2001 From: "elvandlie@gmail.com" Date: Thu, 21 May 2026 02:55:14 +0700 Subject: [PATCH 1/3] fix(knative): fix goroutine leak in Deploy The Deploy function (new-service creation path) starts a background goroutine that polls isImageInPrivateRegistry every 5 seconds and sends results on an unbuffered chprivate channel. If WaitForService completes first (via cherr), the main function returns and the poller goroutine blocks forever on the channel send with no receiver, leaking the goroutine and captured context. Fixed by adding a done channel that the main select loop closes when it exits. The poller goroutine now selects on done alongside its timer and channel send, so it always exits promptly. Fixes #3797 --- pkg/knative/deployer.go | 105 ++++++++++++++++++++++++----------- pkg/knative/deployer_test.go | 62 +++++++++++++++++++++ 2 files changed, 135 insertions(+), 32 deletions(-) diff --git a/pkg/knative/deployer.go b/pkg/knative/deployer.go index 8afe136362..15b2834646 100644 --- a/pkg/knative/deployer.go +++ b/pkg/knative/deployer.go @@ -46,6 +46,8 @@ const ( KnativeDeployerName = deployers.Knative ) +var errPrivateRegistry = stdErrors.New("your function image is unreachable. It is possible that your docker registry is private. If so, make sure you have set up pull secrets https://knative.dev/docs/developer/serving/deploying-from-private-registry") + type DeployerOpt func(*Deployer) type Deployer struct { @@ -254,44 +256,21 @@ consider using the --image-pull-secret flag, or setting up pull secrets manually if d.verbose { fmt.Println("Waiting for Knative Service to become ready") } - chprivate := make(chan bool) - cherr := make(chan error) - go func() { - private := false - for !private { - time.Sleep(5 * time.Second) - private = d.isImageInPrivateRegistry(ctx, client, f) - chprivate <- private - } - close(chprivate) - }() - go func() { + waitForService := func() error { err, _ := client.WaitForService(ctx, f.Name, clientservingv1.WaitConfig{Timeout: k8s.DefaultWaitingTimeout, ErrorWindow: k8s.DefaultErrorWindowTimeout}, wait.NoopMessageCallback()) - cherr <- err - close(cherr) - }() - - presumePrivate := false - main: - // Wait for either a timeout or a container condition signaling the image is unreachable - for { - select { - case private := <-chprivate: - if private { - presumePrivate = true - break main - } - case err = <-cherr: - break main - } + return err } - if presumePrivate { - err := fmt.Errorf("your function image is unreachable. It is possible that your docker registry is private. If so, make sure you have set up pull secrets https://knative.dev/docs/developer/serving/deploying-from-private-registry") - return fn.DeploymentResult{}, err + isPrivateRegistry := func() bool { + return d.isImageInPrivateRegistry(ctx, client, f) } + + err = waitForReadyOrPrivateRegistry(waitForService, isPrivateRegistry, 5*time.Second) if err != nil { + if stdErrors.Is(err, errPrivateRegistry) { + return fn.DeploymentResult{}, err + } err = fmt.Errorf("knative deployer failed to wait for the Knative Service to become ready: %v", err) if !d.verbose { fmt.Fprintln(os.Stderr, "\nService output:") @@ -890,3 +869,65 @@ func wrapK8sConnectionError(err error) error { return nil } + +// waitForReadyOrPrivateRegistry handles polling for a private registry image pull error +// or successfully waiting for the service to become ready. +func waitForReadyOrPrivateRegistry( + waitForService func() error, + isPrivateRegistry func() bool, + pollInterval time.Duration, +) error { + chprivate := make(chan bool) + cherr := make(chan error, 1) + done := make(chan struct{}) + + go func() { + defer close(chprivate) + for { + select { + case <-time.After(pollInterval): + case <-done: + return + } + private := isPrivateRegistry() + select { + case chprivate <- private: + if private { + return + } + case <-done: + return + } + } + }() + + go func() { + cherr <- waitForService() + close(cherr) + }() + + presumePrivate := false +main: + for { + select { + case private := <-chprivate: + if private { + presumePrivate = true + break main + } + case err := <-cherr: + if err != nil { + close(done) + return err + } + break main + } + } + close(done) + + if presumePrivate { + return errPrivateRegistry + } + + return nil +} diff --git a/pkg/knative/deployer_test.go b/pkg/knative/deployer_test.go index 523d2b3fa3..ceb7c059d3 100644 --- a/pkg/knative/deployer_test.go +++ b/pkg/knative/deployer_test.go @@ -13,6 +13,7 @@ import ( "strings" "sync/atomic" "testing" + "time" "github.com/docker/cli/cli/config/configfile" "github.com/docker/cli/cli/config/types" @@ -529,3 +530,64 @@ func assertAuth(uname, pwd string, w http.ResponseWriter, r *http.Request) bool _, _ = fmt.Fprintln(w, "Unauthorised.") return false } + +func TestWaitForReadyOrPrivateRegistry_Success(t *testing.T) { + waitForService := func() error { + return nil // returns immediately with success + } + isPrivateRegistry := func() bool { + return false + } + err := waitForReadyOrPrivateRegistry(waitForService, isPrivateRegistry, 10*time.Millisecond) + if err != nil { + t.Fatalf("expected nil error, got %v", err) + } +} + +func TestWaitForReadyOrPrivateRegistry_WaitError(t *testing.T) { + expectedErr := errors.New("timeout waiting for service") + waitForService := func() error { + return expectedErr // returns immediately with error + } + isPrivateRegistry := func() bool { + return false + } + err := waitForReadyOrPrivateRegistry(waitForService, isPrivateRegistry, 10*time.Millisecond) + if !errors.Is(err, expectedErr) { + t.Fatalf("expected error %v, got %v", expectedErr, err) + } +} + +func TestWaitForReadyOrPrivateRegistry_PrivateRegistry(t *testing.T) { + waitForService := func() error { + // block forever (or at least long enough for private registry check to fire) + time.Sleep(1 * time.Second) + return nil + } + isPrivate := false + isPrivateRegistry := func() bool { + isPrivate = true + return true // instantly return true + } + err := waitForReadyOrPrivateRegistry(waitForService, isPrivateRegistry, 10*time.Millisecond) + if !errors.Is(err, errPrivateRegistry) { + t.Fatalf("expected errPrivateRegistry, got %v", err) + } + if !isPrivate { + t.Fatalf("expected isPrivateRegistry to be called") + } +} + +func TestWaitForReadyOrPrivateRegistry_DelayedSuccess(t *testing.T) { + waitForService := func() error { + time.Sleep(50 * time.Millisecond) + return nil + } + isPrivateRegistry := func() bool { + return false // never private + } + err := waitForReadyOrPrivateRegistry(waitForService, isPrivateRegistry, 10*time.Millisecond) + if err != nil { + t.Fatalf("expected nil error, got %v", err) + } +} From b67aeb68041383450bbe8f04c16cf780c4ef5eb3 Mon Sep 17 00:00:00 2001 From: Tempris Admin Date: Fri, 26 Jun 2026 17:13:15 +0700 Subject: [PATCH 2/3] cancel knative wait on private registry detection --- pkg/knative/deployer.go | 72 ++++++++++++------------------------ pkg/knative/deployer_test.go | 36 ++++++++++-------- 2 files changed, 45 insertions(+), 63 deletions(-) diff --git a/pkg/knative/deployer.go b/pkg/knative/deployer.go index 15b2834646..d14a120be6 100644 --- a/pkg/knative/deployer.go +++ b/pkg/knative/deployer.go @@ -214,8 +214,10 @@ consider using the --image-pull-secret flag, or setting up pull secrets manually out = os.Stderr } since := time.Now() + deployCtx, cancelDeploy := context.WithCancel(ctx) + defer cancelDeploy() go func() { - _ = GetKServiceLogs(ctx, LogsOptions{ + _ = GetKServiceLogs(deployCtx, LogsOptions{ Name: f.Name, Namespace: namespace, Image: f.Deploy.Image, @@ -256,17 +258,17 @@ consider using the --image-pull-secret flag, or setting up pull secrets manually if d.verbose { fmt.Println("Waiting for Knative Service to become ready") } - waitForService := func() error { + waitForService := func(ctx context.Context) error { err, _ := client.WaitForService(ctx, f.Name, clientservingv1.WaitConfig{Timeout: k8s.DefaultWaitingTimeout, ErrorWindow: k8s.DefaultErrorWindowTimeout}, wait.NoopMessageCallback()) return err } - isPrivateRegistry := func() bool { + isPrivateRegistry := func(ctx context.Context) bool { return d.isImageInPrivateRegistry(ctx, client, f) } - err = waitForReadyOrPrivateRegistry(waitForService, isPrivateRegistry, 5*time.Second) + err = waitForReadyOrPrivateRegistry(deployCtx, waitForService, isPrivateRegistry, 5*time.Second) if err != nil { if stdErrors.Is(err, errPrivateRegistry) { return fn.DeploymentResult{}, err @@ -873,61 +875,35 @@ func wrapK8sConnectionError(err error) error { // waitForReadyOrPrivateRegistry handles polling for a private registry image pull error // or successfully waiting for the service to become ready. func waitForReadyOrPrivateRegistry( - waitForService func() error, - isPrivateRegistry func() bool, + ctx context.Context, + waitForService func(context.Context) error, + isPrivateRegistry func(context.Context) bool, pollInterval time.Duration, ) error { - chprivate := make(chan bool) - cherr := make(chan error, 1) - done := make(chan struct{}) + waitCtx, cancel := context.WithCancel(ctx) + defer cancel() - go func() { - defer close(chprivate) - for { - select { - case <-time.After(pollInterval): - case <-done: - return - } - private := isPrivateRegistry() - select { - case chprivate <- private: - if private { - return - } - case <-done: - return - } - } - }() + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() + + cherr := make(chan error, 1) go func() { - cherr <- waitForService() - close(cherr) + cherr <- waitForService(waitCtx) }() - presumePrivate := false -main: for { select { - case private := <-chprivate: - if private { - presumePrivate = true - break main - } + case <-ctx.Done(): + return ctx.Err() case err := <-cherr: - if err != nil { - close(done) - return err + cancel() + return err + case <-ticker.C: + if isPrivateRegistry(waitCtx) { + cancel() + return errPrivateRegistry } - break main } } - close(done) - - if presumePrivate { - return errPrivateRegistry - } - - return nil } diff --git a/pkg/knative/deployer_test.go b/pkg/knative/deployer_test.go index ceb7c059d3..7c7d9d3354 100644 --- a/pkg/knative/deployer_test.go +++ b/pkg/knative/deployer_test.go @@ -532,13 +532,13 @@ func assertAuth(uname, pwd string, w http.ResponseWriter, r *http.Request) bool } func TestWaitForReadyOrPrivateRegistry_Success(t *testing.T) { - waitForService := func() error { + waitForService := func(context.Context) error { return nil // returns immediately with success } - isPrivateRegistry := func() bool { + isPrivateRegistry := func(context.Context) bool { return false } - err := waitForReadyOrPrivateRegistry(waitForService, isPrivateRegistry, 10*time.Millisecond) + err := waitForReadyOrPrivateRegistry(t.Context(), waitForService, isPrivateRegistry, 10*time.Millisecond) if err != nil { t.Fatalf("expected nil error, got %v", err) } @@ -546,47 +546,53 @@ func TestWaitForReadyOrPrivateRegistry_Success(t *testing.T) { func TestWaitForReadyOrPrivateRegistry_WaitError(t *testing.T) { expectedErr := errors.New("timeout waiting for service") - waitForService := func() error { + waitForService := func(context.Context) error { return expectedErr // returns immediately with error } - isPrivateRegistry := func() bool { + isPrivateRegistry := func(context.Context) bool { return false } - err := waitForReadyOrPrivateRegistry(waitForService, isPrivateRegistry, 10*time.Millisecond) + err := waitForReadyOrPrivateRegistry(t.Context(), waitForService, isPrivateRegistry, 10*time.Millisecond) if !errors.Is(err, expectedErr) { t.Fatalf("expected error %v, got %v", expectedErr, err) } } func TestWaitForReadyOrPrivateRegistry_PrivateRegistry(t *testing.T) { - waitForService := func() error { - // block forever (or at least long enough for private registry check to fire) - time.Sleep(1 * time.Second) - return nil + waitCanceled := make(chan struct{}) + waitForService := func(ctx context.Context) error { + <-ctx.Done() + close(waitCanceled) + return ctx.Err() } isPrivate := false - isPrivateRegistry := func() bool { + isPrivateRegistry := func(context.Context) bool { isPrivate = true return true // instantly return true } - err := waitForReadyOrPrivateRegistry(waitForService, isPrivateRegistry, 10*time.Millisecond) + err := waitForReadyOrPrivateRegistry(t.Context(), waitForService, isPrivateRegistry, 10*time.Millisecond) if !errors.Is(err, errPrivateRegistry) { t.Fatalf("expected errPrivateRegistry, got %v", err) } if !isPrivate { t.Fatalf("expected isPrivateRegistry to be called") } + select { + case <-waitCanceled: + case <-time.After(time.Second): + t.Fatal("expected waitForService context to be canceled") + } } func TestWaitForReadyOrPrivateRegistry_DelayedSuccess(t *testing.T) { - waitForService := func() error { + waitForService := func(context.Context) error { time.Sleep(50 * time.Millisecond) return nil } - isPrivateRegistry := func() bool { + isPrivateRegistry := func(context.Context) bool { return false // never private } - err := waitForReadyOrPrivateRegistry(waitForService, isPrivateRegistry, 10*time.Millisecond) + err := waitForReadyOrPrivateRegistry(t.Context(), waitForService, isPrivateRegistry, 10*time.Millisecond) if err != nil { t.Fatalf("expected nil error, got %v", err) } From 8ffb67062c9dc64456492b3c44149c2d1c8886e4 Mon Sep 17 00:00:00 2001 From: Luke Kingland Date: Thu, 29 Oct 2026 21:39:55 +0900 Subject: [PATCH 3/3] fix: join deploy goroutines on return Cancel alone leaves WaitForService and the log follower running until they notice. Deploy returned in that window, so a private-registry exit could leave the waiter alive until its own timeout. Cancel, then wait for each goroutine to finish. Waiting without cancelling first deadlocks: both block until the context ends. --- pkg/knative/deployer.go | 31 +++++++++++++++++------ pkg/knative/deployer_test.go | 49 ++++++++++++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 8 deletions(-) diff --git a/pkg/knative/deployer.go b/pkg/knative/deployer.go index d14a120be6..7fa1f6f34a 100644 --- a/pkg/knative/deployer.go +++ b/pkg/knative/deployer.go @@ -215,8 +215,9 @@ consider using the --image-pull-secret flag, or setting up pull secrets manually } since := time.Now() deployCtx, cancelDeploy := context.WithCancel(ctx) - defer cancelDeploy() + logsDone := make(chan struct{}) go func() { + defer close(logsDone) _ = GetKServiceLogs(deployCtx, LogsOptions{ Name: f.Name, Namespace: namespace, @@ -225,6 +226,12 @@ consider using the --image-pull-secret flag, or setting up pull secrets manually Follow: true, }, out) }() + // Stop the log follower before Deploy returns. Cancel first: Follow blocks + // until the context is done, and waiting without cancelling would deadlock. + defer func() { + cancelDeploy() + <-logsDone + }() previousService, err := client.GetService(ctx, f.Name) if err != nil { @@ -887,23 +894,31 @@ func waitForReadyOrPrivateRegistry( defer ticker.Stop() cherr := make(chan error, 1) - + waitDone := make(chan struct{}) go func() { + defer close(waitDone) cherr <- waitForService(waitCtx) }() + var err error +loop: for { select { case <-ctx.Done(): - return ctx.Err() - case err := <-cherr: - cancel() - return err + err = ctx.Err() + break loop + case err = <-cherr: + break loop case <-ticker.C: if isPrivateRegistry(waitCtx) { - cancel() - return errPrivateRegistry + err = errPrivateRegistry + break loop } } } + // Cancel before joining. Otherwise a WaitForService that blocks on + // waitCtx never closes waitDone, and this wait deadlocks. + cancel() + <-waitDone + return err } diff --git a/pkg/knative/deployer_test.go b/pkg/knative/deployer_test.go index 7c7d9d3354..1f84f06b9d 100644 --- a/pkg/knative/deployer_test.go +++ b/pkg/knative/deployer_test.go @@ -597,3 +597,52 @@ func TestWaitForReadyOrPrivateRegistry_DelayedSuccess(t *testing.T) { t.Fatalf("expected nil error, got %v", err) } } + +// The waiter must have left waitForService before we return. Cancel alone is +// not enough: a goroutine can still be unwinding after Deploy has returned. +func TestWaitForReadyOrPrivateRegistry_PrivateRegistryJoinsWaiter(t *testing.T) { + var exited atomic.Bool + waitForService := func(ctx context.Context) error { + <-ctx.Done() + time.Sleep(150 * time.Millisecond) + exited.Store(true) + return ctx.Err() + } + isPrivateRegistry := func(context.Context) bool { + return true + } + err := waitForReadyOrPrivateRegistry(t.Context(), waitForService, isPrivateRegistry, time.Millisecond) + if !errors.Is(err, errPrivateRegistry) { + t.Fatalf("expected errPrivateRegistry, got %v", err) + } + if !exited.Load() { + t.Fatal("waitForService still running after return") + } +} + +func TestWaitForReadyOrPrivateRegistry_CallerCancelJoinsWaiter(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + started := make(chan struct{}) + var exited atomic.Bool + waitForService := func(ctx context.Context) error { + close(started) + <-ctx.Done() + time.Sleep(150 * time.Millisecond) + exited.Store(true) + return ctx.Err() + } + isPrivateRegistry := func(context.Context) bool { + return false + } + go func() { + <-started + cancel() + }() + err := waitForReadyOrPrivateRegistry(ctx, waitForService, isPrivateRegistry, time.Hour) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } + if !exited.Load() { + t.Fatal("waitForService still running after caller cancel") + } +}