Fix await annotation precedence

CHANGELOG.md

@@ -5,6 +5,14 @@
 - [nodejs] Resolves `punycode` deprecation warnings by using native `fetch` instead of `node-fetch`.
   (https://github.com/pulumi/pulumi-kubernetes/issues/3301)
 
+### Fixed
+
+- `pulumi.com/waitFor` and other await annotations now correctly take precedence over default await logic.
+  (https://github.com/pulumi/pulumi-kubernetes/issues/3329)
+
+- JSONPath expressions used with the `pulumi.com/waitFor` annotation will no longer hang indefinitely if they match non-primitive fields.
+  (https://github.com/pulumi/pulumi-kubernetes/issues/3345)
+
 ## 4.18.3 (October 31, 2024)
 
 ### Fixed

provider/pkg/await/await.go

@@ -272,7 +272,7 @@ func Creation(c CreateConfig) (*unstructured.Unstructured, error) {
 	defer cancel()
 
 	source := condition.NewDynamicSource(ctx, c.ClientSet, outputs.GetNamespace())
-	ready, err := metadata.ReadyCondition(ctx, source, c.ClientSet, c.DedupLogger, c.Inputs, outputs)
+	ready, custom, err := metadata.ReadyCondition(ctx, source, c.ClientSet, c.DedupLogger, c.Inputs, outputs)
 	if err != nil {
 		return outputs, err
 	}
@@ -281,7 +281,9 @@ func Creation(c CreateConfig) (*unstructured.Unstructured, error) {
 	if c.awaiters != nil {
 		a = c.awaiters
 	}
-	if spec, ok := a[id]; ok && spec.await != nil {
+	// Use our built-in await logic only if the user hasn't specified any await
+	// overrides.
+	if spec, ok := a[id]; ok && spec.await != nil && !custom {
 		conf := awaitConfig{
 			ctx:               c.Context,
 			urn:               c.URN,
@@ -432,7 +434,7 @@ func Update(c UpdateConfig) (*unstructured.Unstructured, error) {
 	defer cancel()
 
 	source := condition.NewDynamicSource(ctx, c.ClientSet, currentOutputs.GetNamespace())
-	ready, err := metadata.ReadyCondition(ctx, source, c.ClientSet, c.DedupLogger, c.Inputs, currentOutputs)
+	ready, custom, err := metadata.ReadyCondition(ctx, source, c.ClientSet, c.DedupLogger, c.Inputs, currentOutputs)
 	if err != nil {
 		return currentOutputs, err
 	}
@@ -441,7 +443,9 @@ func Update(c UpdateConfig) (*unstructured.Unstructured, error) {
 	if c.awaiters != nil {
 		a = c.awaiters
 	}
-	if spec, ok := a[id]; ok && spec.await != nil {
+	// Use our built-in await logic only if the user hasn't specified any await
+	// overrides.
+	if spec, ok := a[id]; ok && spec.await != nil && !custom {
 		conf := awaitConfig{
 			ctx:               c.Context,
 			urn:               c.URN,

provider/pkg/await/await_test.go

@@ -350,6 +350,15 @@ func TestCreation(t *testing.T) {
 			awaiter: awaitUnexpected,
 			expect:  []expectF{previewed("default", "foo")},
 		},
+		{
+			name: "WaitFor",
+			args: args{
+				resType: tokens.Type("kubernetes:core/v1:Pod"),
+				inputs:  withWaitFor(validPodUnstructured),
+			},
+			awaiter: awaitUnexpected, // waitFor annotation takes precedence.
+			expect:  []expectF{created("default", "foo")},
+		},
 		// FUTURE: test server-side apply (depends on https://github.com/kubernetes/kubernetes/issues/115598)
 	}
 
@@ -625,6 +634,15 @@ func TestUpdate(t *testing.T) {
 			awaiter: awaitUnexpected,
 			expect:  []expectF{previewed("default", "foo")},
 		},
+		{
+			name: "WaitFor",
+			args: args{
+				resType: tokens.Type("kubernetes:core/v1:Pod"),
+				inputs:  withWaitFor(validPodUnstructured),
+			},
+			awaiter: awaitUnexpected, // waitFor annotation takes precedence.
+			expect:  []expectF{updated("default", "foo"), logged()},
+		},
 		// FUTURE: test server-side apply (depends on https://github.com/kubernetes/kubernetes/issues/115598)
 	}
 
@@ -1054,6 +1072,14 @@ func withSkipAwait(obj *unstructured.Unstructured) *unstructured.Unstructured {
 	return copy
 }
 
+func withWaitFor(obj *unstructured.Unstructured) *unstructured.Unstructured {
+	copy := obj.DeepCopy()
+	copy.SetAnnotations(map[string]string{
+		"pulumi.com/waitFor": "jsonpath={.metadata}", // Succeeds immediately.
+	})
+	return copy
+}
+
 func withGenerateName(obj *unstructured.Unstructured) *unstructured.Unstructured {
 	copy := obj.DeepCopy()
 	copy.SetGenerateName(fmt.Sprintf("%s-", obj.GetName()))

provider/pkg/jsonpath/jsonpath.go

@@ -38,6 +38,10 @@ func (i *Parsed) Matches(uns *unstructured.Unstructured) (MatchResult, error) {
 	value := results[0][0]
 	switch value.Interface().(type) {
 	case []any, map[string]any:
+		if i.Value == "" {
+			// We don't care about complex types if we're matching anything.
+			return MatchResult{Matched: true}, nil
+		}
 		return MatchResult{}, fmt.Errorf("%q has a non-primitive value (%v)", i.Path, value.String())
 	}
 	found := fmt.Sprint(value.Interface())

provider/pkg/jsonpath/jsonpath_test.go

@@ -108,6 +108,16 @@ func TestMatches(t *testing.T) {
 			}},
 			want: MatchResult{Matched: true, Found: "<nil>"},
 		},
+		{
+			name: "object exists",
+			expr: "jsonpath={ .foo }",
+			uns: &unstructured.Unstructured{Object: map[string]any{
+				"foo": map[string]any{
+					"bar": "baz",
+				},
+			}},
+			want: MatchResult{Matched: true},
+		},
 		{
 			name: "key exists with non-primitive value",
 			expr: "jsonpath={.foo}",

provider/pkg/metadata/overrides.go

@@ -103,17 +103,17 @@ func ReadyCondition(
 	logger *logging.DedupLogger,
 	inputs *unstructured.Unstructured,
 	obj *unstructured.Unstructured,
-) (condition.Satisfier, error) {
+) (condition.Satisfier, bool, error) {
 	if SkipAwaitLogic(inputs) {
-		return condition.NewImmediate(logger, obj), nil
+		return condition.NewImmediate(logger, obj), true, nil
 	}
 
 	val := GetAnnotationValue(obj, AnnotationWaitFor)
 	if val == "" {
 		if os.Getenv("PULUMI_K8S_AWAIT_ALL") != "true" {
-			return condition.NewImmediate(nil, obj), nil
+			return condition.NewImmediate(nil, obj), false, nil
 		}
-		return condition.NewReady(ctx, source, logger, obj), nil
+		return condition.NewReady(ctx, source, logger, obj), false, nil
 	}
 
 	// Attempt to interpret the annotation as a JSON string array, and if that
@@ -124,7 +124,7 @@ func ReadyCondition(
 		values = append(values, val)
 	}
 	if len(values) == 0 {
-		return nil, fmt.Errorf("at least one condition must be specified")
+		return nil, false, fmt.Errorf("at least one condition must be specified")
 	}
 
 	conditions := make([]condition.Satisfier, 0, len(values))
@@ -133,28 +133,29 @@ func ReadyCondition(
 		case strings.HasPrefix(expr, "jsonpath="):
 			jsp, err := jsonpath.Parse(expr)
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
 			cond, err := condition.NewJSONPath(ctx, source, logger, obj, jsp)
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
 			conditions = append(conditions, cond)
 		case strings.HasPrefix(expr, "condition="):
 			cond, err := condition.NewCustom(ctx, source, logger, obj, expr)
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
 			conditions = append(conditions, cond)
 		default:
-			return nil, fmt.Errorf(`expected a "jsonpath=" or "condition=" prefix, got %q`, expr)
+			return nil, false, fmt.Errorf(`expected a "jsonpath=" or "condition=" prefix, got %q`, expr)
 		}
 	}
 
 	if len(conditions) == 1 {
-		return conditions[0], nil
+		return conditions[0], true, nil
 	}
-	return condition.NewAll(conditions...)
+	all, err := condition.NewAll(conditions...)
+	return all, true, err
 }
 
 // DeletedCondition inspects the object's annotations and returns a

provider/pkg/metadata/overrides_test.go

@@ -140,6 +140,7 @@ func TestReadyCondition(t *testing.T) {
 		inputs         *unstructured.Unstructured
 		genericEnabled bool
 		want           any
+		wantCustom     bool
 		wantErr        string
 	}{
 		{
@@ -164,7 +165,8 @@ func TestReadyCondition(t *testing.T) {
 					},
 				},
 			},
-			want: condition.Immediate{},
+			want:       condition.Immediate{},
+			wantCustom: true,
 		},
 		{
 			name: "skipAwait=true, generic await enabled",
@@ -179,6 +181,7 @@ func TestReadyCondition(t *testing.T) {
 			},
 			genericEnabled: true,
 			want:           condition.Immediate{},
+			wantCustom:     true,
 		},
 		{
 			name: "skipAwait=true with custom ready condition",
@@ -190,7 +193,8 @@ func TestReadyCondition(t *testing.T) {
 					},
 				},
 			}},
-			want: condition.Immediate{},
+			want:       condition.Immediate{},
+			wantCustom: true,
 		},
 		{
 			name: "skipAwait=false with custom ready condition",
@@ -202,7 +206,8 @@ func TestReadyCondition(t *testing.T) {
 					},
 				},
 			}},
-			want: &condition.JSONPath{},
+			want:       &condition.JSONPath{},
+			wantCustom: true,
 		},
 		{
 			name: "parse JSON array",
@@ -213,7 +218,8 @@ func TestReadyCondition(t *testing.T) {
 					},
 				},
 			}},
-			want: &condition.All{},
+			want:       &condition.All{},
+			wantCustom: true,
 		},
 		{
 			name: "parse empty array",
@@ -235,7 +241,8 @@ func TestReadyCondition(t *testing.T) {
 					},
 				},
 			}},
-			want: &condition.JSONPath{},
+			want:       &condition.JSONPath{},
+			wantCustom: true,
 		},
 		{
 			name: "invalid expression",
@@ -259,12 +266,13 @@ func TestReadyCondition(t *testing.T) {
 			if obj == nil {
 				obj = tt.inputs
 			}
-			cond, err := ReadyCondition(context.Background(), nil, nil, nil, tt.inputs, obj)
+			cond, custom, err := ReadyCondition(context.Background(), nil, nil, nil, tt.inputs, obj)
 			if tt.wantErr != "" {
 				assert.ErrorContains(t, err, tt.wantErr)
 				return
 			}
 			assert.IsType(t, tt.want, cond)
+			assert.Equal(t, tt.wantCustom, custom)
 		})
 	}
 }