From 6d148a0c5768ff9dddae9f0f5330ab71f4bbc577 Mon Sep 17 00:00:00 2001 From: AIWithShrey Date: Fri, 30 Oct 2026 18:07:35 +0530 Subject: [PATCH] fix(cel): rebuild the resource binding per evaluation and surface cancelled prometheusMetrics queries A reused variables map kept the first evaluation's context in its resource binding, so a later evaluation could fail on an earlier cancellation or ignore its own deadline. The binding is now rebuilt on every evaluation. A prometheusMetrics query cancelled by its context returned an error map; it now returns an error wrapping the context error. Other query errors keep the error map. Co-Authored-By: Claude Opus 5.5 --- internal/workflow/executor/cel.go | 15 +-- .../workflow/executor/cel_ctx_binding_test.go | 106 ++++++++++++++++++ internal/workflow/executor/resource_macros.go | 8 +- 3 files changed, 119 insertions(+), 10 deletions(-) create mode 100644 internal/workflow/executor/cel_ctx_binding_test.go diff --git a/internal/workflow/executor/cel.go b/internal/workflow/executor/cel.go index aa50abc..1888e99 100644 --- a/internal/workflow/executor/cel.go +++ b/internal/workflow/executor/cel.go @@ -469,16 +469,13 @@ func (e *CELEvaluator) EvaluateExpression(ctx context.Context, expr string, vars vars["image"] = kyvernoimagedata.Context{ContextInterface: imageDataCtx} // Add resource context with controller-runtime client implementation (for CEL expressions that use resource.Get/List). - // If vars["resource"] or vars["items"] are already set (e.g. by resourceQuery step output evaluation), do not overwrite. - if _, set := vars["resource"]; !set { - resCtx := &resourceContext{ - client: e.client, - namespace: namespace, - pageSize: e.resourceListPageSize, - ctx: ctx, - } - vars["resource"] = kyvernoresource.Context{ContextInterface: resCtx} + resCtx := &resourceContext{ + client: e.client, + namespace: namespace, + pageSize: e.resourceListPageSize, + ctx: ctx, } + vars["resource"] = kyvernoresource.Context{ContextInterface: resCtx} // Evaluate with variables. When macro functions are registered (resourceLogs, resourceEvents, // etc.), their closures read context from the shared macroContextHolder. Serialise the diff --git a/internal/workflow/executor/cel_ctx_binding_test.go b/internal/workflow/executor/cel_ctx_binding_test.go new file mode 100644 index 0000000..8f1ad93 --- /dev/null +++ b/internal/workflow/executor/cel_ctx_binding_test.go @@ -0,0 +1,106 @@ +/* +Copyright 2026 Nirmata, Inc. + +Use of this source code is governed by the Business Source License 1.1 +that can be found in the LICENSE.md file. +*/ + +package executor + +import ( + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" +) + +const ctxBindingListExpr = `dyn(resource.List("v1","Pod","a",{})).items.size()` + +func TestCELReusedVarsHonourLaterDeadline(t *testing.T) { + k8s := fake.NewClientBuilder().WithScheme(lockScopeScheme()).WithInterceptorFuncs(interceptor.Funcs{ + List: func(ctx context.Context, c client.WithWatch, l client.ObjectList, o ...client.ListOption) error { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(time.Second): + return c.List(ctx, l, o...) + } + }, + }).Build() + ev, err := NewCELEvaluator(k8s, nil) + if err != nil { + t.Fatal(err) + } + vars := map[string]interface{}{} + if _, err := ev.EvaluateExpression(context.Background(), "1 + 1", vars); err != nil { + t.Fatal(err) + } + ctx2, cancel2 := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel2() + start := time.Now() + _, err = ev.EvaluateExpression(ctx2, ctxBindingListExpr, vars) + if el := time.Since(start); el > 700*time.Millisecond { + t.Errorf("evaluation ran %v past its 200ms deadline (err=%v)", el, err) + } +} + +func TestCELPrometheusMetricsCancelledQueryReturnsError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + select { + case <-r.Context().Done(): + case <-time.After(400 * time.Millisecond): + } + })) + defer srv.Close() + pc, err := NewHTTPPrometheusClient(srv.URL) + if err != nil { + t.Fatal(err) + } + ev, err := NewCELEvaluatorWithMetrics(nil, nil, nil, pc, nil, nil, 0, nil) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + res, err := ev.EvaluateExpression(ctx, `prometheusMetrics("up", "5m")`, map[string]interface{}{}) + if err == nil { + t.Fatalf("expected an error for a cancelled query, got result %v", res) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("error does not wrap context.DeadlineExceeded: %v", err) + } +} + +func TestCELPrometheusMetricsUnrelatedErrorOnCancelledCtxKeepsErrorMap(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + requirePrometheusErrorMap(t, ctx, errors.New("upstream: context canceled")) +} + +func TestCELPrometheusMetricsDeadlineErrorOnLiveCtxKeepsErrorMap(t *testing.T) { + requirePrometheusErrorMap(t, context.Background(), fmt.Errorf("dial tcp: %w", context.DeadlineExceeded)) +} + +func requirePrometheusErrorMap(t *testing.T, ctx context.Context, queryErr error) { + t.Helper() + ev, err := NewCELEvaluatorWithMetrics(nil, nil, nil, &mockPrometheusClient{err: queryErr}, nil, nil, 0, nil) + if err != nil { + t.Fatal(err) + } + res, err := ev.EvaluateExpression(ctx, `prometheusMetrics("up", "5m")`, map[string]interface{}{}) + if err != nil { + t.Fatalf("expected the error map, got error: %v", err) + } + if m, ok := res.(map[string]interface{}); !ok || m["error"] == nil { + t.Errorf("expected a result map with an error field, got %v", res) + } +} diff --git a/internal/workflow/executor/resource_macros.go b/internal/workflow/executor/resource_macros.go index 4d0e360..ef70f17 100644 --- a/internal/workflow/executor/resource_macros.go +++ b/internal/workflow/executor/resource_macros.go @@ -9,6 +9,7 @@ package executor import ( "context" + "errors" "fmt" "io" "time" @@ -523,6 +524,7 @@ func GetResourceMacroOptionsWithMetrics( // Queries Prometheus for metrics using PromQL. Returns query results as a structured map // (type, samples, value). On query failure returns the same shape with empty data and // an "error" field set so workflows can check has(result.error) and fail or continue. + // A query failing with ctx.Err() returns a CEL error instead. prometheusMetricsFunc := celapi.Function("prometheusMetrics", celapi.Overload("prometheusMetrics_string_string", []*celapi.Type{celapi.StringType, celapi.StringType}, @@ -545,7 +547,11 @@ func GetResourceMacroOptionsWithMetrics( } // Query Prometheus; on failure return empty vector with error field so workflows can fail or continue - result, err := prometheusClient.Query(macroCtx.get(), query, time.Now().Add(-duration)) + qctx := macroCtx.get() + result, err := prometheusClient.Query(qctx, query, time.Now().Add(-duration)) + if ctxErr := qctx.Err(); ctxErr != nil && errors.Is(err, ctxErr) { + return types.NewErr("prometheus query cancelled: %w", ctxErr) + } if err != nil { if _, isNoOp := prometheusClient.(*NoOpPrometheusClient); !isNoOp { klog.Warningf("prometheusMetrics query failed: query=%q timeRange=%q: %v", query, timeRange, err)