diff --git a/pkg/knative/deployer.go b/pkg/knative/deployer.go index 8afe136362..7fa1f6f34a 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 { @@ -212,8 +214,11 @@ consider using the --image-pull-secret flag, or setting up pull secrets manually out = os.Stderr } since := time.Now() + deployCtx, cancelDeploy := context.WithCancel(ctx) + logsDone := make(chan struct{}) go func() { - _ = GetKServiceLogs(ctx, LogsOptions{ + defer close(logsDone) + _ = GetKServiceLogs(deployCtx, LogsOptions{ Name: f.Name, Namespace: namespace, Image: f.Deploy.Image, @@ -221,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 { @@ -254,44 +265,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(ctx context.Context) 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(ctx context.Context) bool { + return d.isImageInPrivateRegistry(ctx, client, f) } + + err = waitForReadyOrPrivateRegistry(deployCtx, 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 +878,47 @@ 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( + ctx context.Context, + waitForService func(context.Context) error, + isPrivateRegistry func(context.Context) bool, + pollInterval time.Duration, +) error { + waitCtx, cancel := context.WithCancel(ctx) + defer cancel() + + ticker := time.NewTicker(pollInterval) + 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(): + err = ctx.Err() + break loop + case err = <-cherr: + break loop + case <-ticker.C: + if isPrivateRegistry(waitCtx) { + 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 523d2b3fa3..1f84f06b9d 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,119 @@ 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(context.Context) error { + return nil // returns immediately with success + } + isPrivateRegistry := func(context.Context) bool { + return false + } + err := waitForReadyOrPrivateRegistry(t.Context(), 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(context.Context) error { + return expectedErr // returns immediately with error + } + isPrivateRegistry := func(context.Context) bool { + return false + } + 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) { + waitCanceled := make(chan struct{}) + waitForService := func(ctx context.Context) error { + <-ctx.Done() + close(waitCanceled) + return ctx.Err() + } + isPrivate := false + isPrivateRegistry := func(context.Context) bool { + isPrivate = true + return true // instantly return true + } + 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(context.Context) error { + time.Sleep(50 * time.Millisecond) + return nil + } + isPrivateRegistry := func(context.Context) bool { + return false // never private + } + err := waitForReadyOrPrivateRegistry(t.Context(), waitForService, isPrivateRegistry, 10*time.Millisecond) + if err != nil { + 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") + } +}