diff --git a/cmd/crossplane/render/render.go b/cmd/crossplane/render/render.go index bed3f635..f09ea681 100644 --- a/cmd/crossplane/render/render.go +++ b/cmd/crossplane/render/render.go @@ -95,8 +95,18 @@ type OperationOutputs struct { // FunctionAddresses maps function names to their gRPC target addresses. type FunctionAddresses struct { - addrs map[string]string - contexts map[string]RuntimeContext + addrs map[string]string + // runtimes holds every runtime started, in start order. It's a slice + // rather than a map keyed by name so that Functions sharing a name all + // get stopped, even though only the last one's address is used. + runtimes []startedRuntime +} + +// startedRuntime is a started runtime and the name of its Function. +type startedRuntime struct { + RuntimeContext + + name string } // Addresses returns the function name to gRPC address map. @@ -108,9 +118,9 @@ func (fa *FunctionAddresses) Addresses() map[string]string { // returned error joins all failures. Each runtime bounds its own cleanup. func (fa *FunctionAddresses) Stop(ctx context.Context) error { var errs []error - for name, rctx := range fa.contexts { - if err := rctx.Stop(ctx); err != nil { - errs = append(errs, errors.Wrapf(err, "cannot stop function %q runtime (target %q)", name, rctx.Target)) + for _, rt := range fa.runtimes { + if err := rt.Stop(ctx); err != nil { + errs = append(errs, errors.Wrapf(err, "cannot stop function %q runtime (target %q)", rt.name, rt.Target)) } } return errors.Join(errs...) @@ -120,25 +130,41 @@ func (fa *FunctionAddresses) Stop(ctx context.Context) error { // gRPC addresses. The caller must call Stop on the returned FunctionAddresses // when done. func StartFunctionRuntimes(ctx context.Context, log logging.Logger, fns []pkgv1.Function) (*FunctionAddresses, error) { - addrs := make(map[string]string, len(fns)) - contexts := make(map[string]RuntimeContext, len(fns)) + return startFunctionRuntimes(ctx, log, fns, GetRuntime) +} + +// startFunctionRuntimes implements StartFunctionRuntimes, using getRuntime to +// get each Function's runtime. +func startFunctionRuntimes(ctx context.Context, log logging.Logger, fns []pkgv1.Function, getRuntime func(pkgv1.Function, logging.Logger) (Runtime, error)) (*FunctionAddresses, error) { + fa := &FunctionAddresses{addrs: make(map[string]string, len(fns))} for _, fn := range fns { - rt, err := GetRuntime(fn, log) + rt, err := getRuntime(fn, log) if err != nil { - return nil, errors.Wrapf(err, "cannot get runtime for Function %q", fn.GetName()) + return nil, stopStarted(ctx, fa, errors.Wrapf(err, "cannot get runtime for Function %q", fn.GetName())) } rctx, err := rt.Start(ctx) if err != nil { - return nil, errors.Wrapf(err, "cannot start Function %q", fn.GetName()) + return nil, stopStarted(ctx, fa, errors.Wrapf(err, "cannot start Function %q", fn.GetName())) } - addrs[fn.GetName()] = rctx.Target - contexts[fn.GetName()] = rctx + fa.addrs[fn.GetName()] = rctx.Target + fa.runtimes = append(fa.runtimes, startedRuntime{name: fn.GetName(), RuntimeContext: rctx}) } - return &FunctionAddresses{addrs: addrs, contexts: contexts}, nil + return fa, nil +} + +// stopStarted stops the runtimes started before a later Function failed to +// start. It returns startErr, joined with any stop errors. StopFunctionRuntimes +// detaches the stop from ctx's cancellation, since ctx may be why the start +// failed. +func stopStarted(ctx context.Context, fa *FunctionAddresses, startErr error) error { + if err := StopFunctionRuntimes(ctx, fa); err != nil { + return errors.Join(startErr, err) + } + return startErr } // RewriteAddressesForDocker rewrites function addresses so they are reachable diff --git a/cmd/crossplane/render/render_test.go b/cmd/crossplane/render/render_test.go index c54f44a8..96490396 100644 --- a/cmd/crossplane/render/render_test.go +++ b/cmd/crossplane/render/render_test.go @@ -9,6 +9,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "github.com/crossplane/crossplane-runtime/v2/pkg/errors" + "github.com/crossplane/crossplane-runtime/v2/pkg/logging" pkgv1 "github.com/crossplane/crossplane/apis/v2/pkg/v1" ) @@ -99,12 +100,12 @@ func TestFunctionAddressesStop(t *testing.T) { errA := errors.New("boom a") errB := errors.New("boom b") - stopReturns := func(err error) RuntimeContext { - return RuntimeContext{Target: "fn:9443", Stop: func(context.Context) error { return err }} + stopReturns := func(name string, err error) startedRuntime { + return startedRuntime{name: name, RuntimeContext: RuntimeContext{Target: name + ":9443", Stop: func(context.Context) error { return err }}} } type args struct { - contexts map[string]RuntimeContext + runtimes []startedRuntime } type want struct { // errs are the errors the returned error must wrap. When empty, Stop @@ -120,20 +121,20 @@ func TestFunctionAddressesStop(t *testing.T) { "AllSucceed": { reason: "Stop should return nil when every runtime stops.", args: args{ - contexts: map[string]RuntimeContext{ - "ok-a": stopReturns(nil), - "ok-b": stopReturns(nil), + runtimes: []startedRuntime{ + stopReturns("ok-a", nil), + stopReturns("ok-b", nil), }, }, }, "SomeFail": { reason: "Stop should stop every runtime even if some fail, and return every failure.", args: args{ - contexts: map[string]RuntimeContext{ - "ok-a": stopReturns(nil), - "fail-a": stopReturns(errA), - "ok-b": stopReturns(nil), - "fail-b": stopReturns(errB), + runtimes: []startedRuntime{ + stopReturns("fail-a", errA), + stopReturns("ok-a", nil), + stopReturns("fail-b", errB), + stopReturns("ok-b", nil), }, }, want: want{ @@ -144,21 +145,17 @@ func TestFunctionAddressesStop(t *testing.T) { for name, tc := range cases { t.Run(name, func(t *testing.T) { - // Repeat so Go's randomized map iteration can't hide a regression - // to returning on the first error. - for range 20 { - fa := &FunctionAddresses{contexts: tc.args.contexts} + fa := &FunctionAddresses{runtimes: tc.args.runtimes} - err := fa.Stop(t.Context()) + err := fa.Stop(t.Context()) - wantErrs := tc.want.errs - if len(wantErrs) == 0 { - wantErrs = []error{nil} - } - for _, want := range wantErrs { - if diff := cmp.Diff(want, err, cmpopts.EquateErrors()); diff != "" { - t.Fatalf("\n%s\nStop(...): -want error, +got error:\n%s", tc.reason, diff) - } + wantErrs := tc.want.errs + if len(wantErrs) == 0 { + wantErrs = []error{nil} + } + for _, want := range wantErrs { + if diff := cmp.Diff(want, err, cmpopts.EquateErrors()); diff != "" { + t.Errorf("\n%s\nStop(...): -want error, +got error:\n%s", tc.reason, diff) } } }) @@ -217,11 +214,11 @@ func TestStopFunctionRuntimes(t *testing.T) { var got want var fa *FunctionAddresses if !tc.args.noRuntimes { - fa = &FunctionAddresses{contexts: map[string]RuntimeContext{ - "fn": {Target: "fn:9443", Stop: func(ctx context.Context) error { + fa = &FunctionAddresses{runtimes: []startedRuntime{ + {name: "fn", RuntimeContext: RuntimeContext{Target: "fn:9443", Stop: func(ctx context.Context) error { got.ctxErr = ctx.Err() return tc.args.stopErr - }}, + }}}, }} } @@ -239,3 +236,197 @@ func TestStopFunctionRuntimes(t *testing.T) { }) } } + +type mockRuntime struct { + MockStart func(ctx context.Context) (RuntimeContext, error) +} + +func (m *mockRuntime) Start(ctx context.Context) (RuntimeContext, error) { return m.MockStart(ctx) } + +var _ Runtime = &mockRuntime{} + +func TestStartFunctionRuntimes(t *testing.T) { + errGet := errors.New("get boom") + errStart := errors.New("start boom") + errStopA := errors.New("stop a boom") + errStopB := errors.New("stop b boom") + + // started returns a runtime that starts at the supplied target, and whose + // Stop returns the supplied error. + started := func(target string, stopErr error) Runtime { + return &mockRuntime{MockStart: func(context.Context) (RuntimeContext, error) { + return RuntimeContext{Target: target, Stop: func(context.Context) error { return stopErr }}, nil + }} + } + // startFails returns a runtime that fails to start with the supplied + // error. + startFails := func(err error) Runtime { + return &mockRuntime{MockStart: func(context.Context) (RuntimeContext, error) { + return RuntimeContext{}, err + }} + } + // getRuntimes returns a runtime getter that gets each Function's runtime + // from the supplied map, and returns errGet for a Function not in it. + getRuntimes := func(rts map[string]Runtime) func(pkgv1.Function, logging.Logger) (Runtime, error) { + return func(fn pkgv1.Function, _ logging.Logger) (Runtime, error) { + rt, ok := rts[fn.GetName()] + if !ok { + return nil, errGet + } + return rt, nil + } + } + // getRuntimesInOrder returns a runtime getter that returns the supplied + // runtimes in turn, for Functions that share a name. + getRuntimesInOrder := func(rts ...Runtime) func(pkgv1.Function, logging.Logger) (Runtime, error) { + return func(pkgv1.Function, logging.Logger) (Runtime, error) { + rt := rts[0] + rts = rts[1:] + return rt, nil + } + } + fn := func(name string) pkgv1.Function { + return pkgv1.Function{ObjectMeta: metav1.ObjectMeta{Name: name}} + } + + type args struct { + fns []pkgv1.Function + getRuntime func(pkgv1.Function, logging.Logger) (Runtime, error) + } + type want struct { + // addrs are the addresses of the returned FunctionAddresses, or nil + // if none are returned. + addrs map[string]string + // errs are the errors the returned error must wrap. When empty, + // startFunctionRuntimes must return nil. + errs []error + // stopErrs are the errors stopping the returned FunctionAddresses + // must return, proving which runtimes it stops. When empty, it must + // stop without error. + stopErrs []error + } + + cases := map[string]struct { + reason string + args args + want want + }{ + "StartsAll": { + reason: "startFunctionRuntimes should start every Function's runtime and return their addresses.", + args: args{ + fns: []pkgv1.Function{fn("fn-a"), fn("fn-b")}, + getRuntime: getRuntimes(map[string]Runtime{ + "fn-a": started("fn-a:9443", nil), + "fn-b": started("fn-b:9443", nil), + }), + }, + want: want{ + addrs: map[string]string{"fn-a": "fn-a:9443", "fn-b": "fn-b:9443"}, + }, + }, + "StopsStartedOnStartFailure": { + reason: "startFunctionRuntimes should stop the runtimes it already started when a later one fails to start, and return the start error joined with any stop errors.", + args: args{ + fns: []pkgv1.Function{fn("fn-a"), fn("fn-b"), fn("fn-c")}, + getRuntime: getRuntimes(map[string]Runtime{ + "fn-a": started("fn-a:9443", errStopA), + "fn-b": started("fn-b:9443", errStopB), + "fn-c": startFails(errStart), + }), + }, + want: want{ + errs: []error{errStart, errStopA, errStopB}, + }, + }, + "StopsStartedOnGetRuntimeFailure": { + reason: "startFunctionRuntimes should stop the runtimes it already started when it cannot get a later Function's runtime.", + args: args{ + fns: []pkgv1.Function{fn("fn-a"), fn("fn-c")}, + getRuntime: getRuntimes(map[string]Runtime{ + "fn-a": started("fn-a:9443", errStopA), + }), + }, + want: want{ + errs: []error{errGet, errStopA}, + }, + }, + "StartsDuplicateNames": { + reason: "startFunctionRuntimes should use the last address for Functions that share a name, but keep every runtime so stopping them stops all of them.", + args: args{ + fns: []pkgv1.Function{fn("fn-a"), fn("fn-a")}, + getRuntime: getRuntimesInOrder( + started("fn-a-1:9443", errStopA), + started("fn-a-2:9443", errStopB), + ), + }, + want: want{ + addrs: map[string]string{"fn-a": "fn-a-2:9443"}, + stopErrs: []error{errStopA, errStopB}, + }, + }, + "StopsDuplicateNamesOnStartFailure": { + reason: "startFunctionRuntimes should stop every runtime it already started, including Functions that share a name, when a later one fails to start.", + args: args{ + fns: []pkgv1.Function{fn("fn-a"), fn("fn-a"), fn("fn-c")}, + getRuntime: getRuntimesInOrder( + started("fn-a-1:9443", errStopA), + started("fn-a-2:9443", errStopB), + startFails(errStart), + ), + }, + want: want{ + errs: []error{errStart, errStopA, errStopB}, + }, + }, + "FirstStartFailure": { + reason: "startFunctionRuntimes should return the start error when the first Function fails to start.", + args: args{ + fns: []pkgv1.Function{fn("fn-c")}, + getRuntime: getRuntimes(map[string]Runtime{ + "fn-c": startFails(errStart), + }), + }, + want: want{ + errs: []error{errStart}, + }, + }, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + fa, err := startFunctionRuntimes(t.Context(), logging.NewNopLogger(), tc.args.fns, tc.args.getRuntime) + + var addrs map[string]string + if fa != nil { + addrs = fa.Addresses() + } + if diff := cmp.Diff(tc.want.addrs, addrs); diff != "" { + t.Errorf("\n%s\nstartFunctionRuntimes(...): -want addresses, +got addresses:\n%s", tc.reason, diff) + } + + wantErrs := tc.want.errs + if len(wantErrs) == 0 { + wantErrs = []error{nil} + } + for _, want := range wantErrs { + if diff := cmp.Diff(want, err, cmpopts.EquateErrors()); diff != "" { + t.Errorf("\n%s\nstartFunctionRuntimes(...): -want error, +got error:\n%s", tc.reason, diff) + } + } + + if fa == nil { + return + } + stopErr := fa.Stop(t.Context()) + wantStopErrs := tc.want.stopErrs + if len(wantStopErrs) == 0 { + wantStopErrs = []error{nil} + } + for _, want := range wantStopErrs { + if diff := cmp.Diff(want, stopErr, cmpopts.EquateErrors()); diff != "" { + t.Errorf("\n%s\nfa.Stop(...): -want error, +got error:\n%s", tc.reason, diff) + } + } + }) + } +}