Add configurable timeouts for complex awaiters (#457)

Allow the default timeout for each complex awaiter to be
overridden by setting the `pulumi.com/timeoutSeconds`
annotation.

CHANGELOG.md

@@ -6,7 +6,7 @@
 
 ### Improvements
 
--   None
+-   Allow the default timeout for awaiters to be overridden (https://github.com/pulumi/pulumi-kubernetes/pull/457)
 
 ### Bug fixes
 

pkg/await/apps_deployment.go

@@ -10,6 +10,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/pulumi/pulumi-kubernetes/pkg/clients"
 	"github.com/pulumi/pulumi-kubernetes/pkg/kinds"
+	"github.com/pulumi/pulumi-kubernetes/pkg/metadata"
 	"github.com/pulumi/pulumi-kubernetes/pkg/openapi"
 	"github.com/pulumi/pulumi/pkg/diag"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -180,7 +181,8 @@ func (dia *deploymentInitAwaiter) Await() error {
 	period := time.NewTicker(10 * time.Second)
 	defer period.Stop()
 
-	return dia.await(deploymentWatcher, replicaSetWatcher, podWatcher, pvcWatcher, time.After(5*time.Minute), period.C)
+	timeout := time.Duration(metadata.TimeoutSeconds(dia.config.currentInputs, 5*60)) * time.Second
+	return dia.await(deploymentWatcher, replicaSetWatcher, podWatcher, pvcWatcher, time.After(timeout), period.C)
 }
 
 func (dia *deploymentInitAwaiter) Read() error {

pkg/await/apps_statefulset.go

@@ -9,6 +9,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/pulumi/pulumi-kubernetes/pkg/clients"
 	"github.com/pulumi/pulumi-kubernetes/pkg/kinds"
+	"github.com/pulumi/pulumi-kubernetes/pkg/metadata"
 	"github.com/pulumi/pulumi-kubernetes/pkg/openapi"
 	"github.com/pulumi/pulumi/pkg/diag"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -155,7 +156,8 @@ func (sia *statefulsetInitAwaiter) Await() error {
 	period := time.NewTicker(10 * time.Second)
 	defer period.Stop()
 
-	return sia.await(statefulSetWatcher, podWatcher, time.After(5*time.Minute), period.C)
+	timeout := time.Duration(metadata.TimeoutSeconds(sia.config.currentInputs, 5*60)) * time.Second
+	return sia.await(statefulSetWatcher, podWatcher, time.After(timeout), period.C)
 }
 
 func (sia *statefulsetInitAwaiter) Read() error {

pkg/await/core_pod.go

@@ -10,6 +10,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/pulumi/pulumi-kubernetes/pkg/clients"
 	"github.com/pulumi/pulumi-kubernetes/pkg/kinds"
+	"github.com/pulumi/pulumi-kubernetes/pkg/metadata"
 	"github.com/pulumi/pulumi-kubernetes/pkg/openapi"
 	"github.com/pulumi/pulumi/pkg/diag"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -360,7 +361,8 @@ func (pia *podInitAwaiter) Await() error {
 	}
 	defer podWatcher.Stop()
 
-	return pia.await(podWatcher, time.After(5*time.Minute))
+	timeout := time.Duration(metadata.TimeoutSeconds(pia.config.currentInputs, 5*60)) * time.Second
+	return pia.await(podWatcher, time.After(timeout))
 }
 
 func (pia *podInitAwaiter) Read() error {

pkg/await/core_service.go

@@ -8,6 +8,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/pulumi/pulumi-kubernetes/pkg/clients"
 	"github.com/pulumi/pulumi-kubernetes/pkg/kinds"
+	"github.com/pulumi/pulumi-kubernetes/pkg/metadata"
 	"github.com/pulumi/pulumi-kubernetes/pkg/openapi"
 	"github.com/pulumi/pulumi/pkg/diag"
 	"k8s.io/api/core/v1"
@@ -134,7 +135,8 @@ func (sia *serviceInitAwaiter) Await() error {
 
 	version := ServerVersion(sia.config.clientSet.DiscoveryClientCached)
 
-	return sia.await(serviceWatcher, endpointWatcher, time.After(10*time.Minute), make(chan struct{}), version)
+	timeout := time.Duration(metadata.TimeoutSeconds(sia.config.currentInputs, 10*60)) * time.Second
+	return sia.await(serviceWatcher, endpointWatcher, time.After(timeout), make(chan struct{}), version)
 }
 
 func (sia *serviceInitAwaiter) Read() error {

pkg/await/extensions_ingress.go

@@ -10,6 +10,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/pulumi/pulumi-kubernetes/pkg/clients"
 	"github.com/pulumi/pulumi-kubernetes/pkg/kinds"
+	"github.com/pulumi/pulumi-kubernetes/pkg/metadata"
 	"github.com/pulumi/pulumi-kubernetes/pkg/openapi"
 	"github.com/pulumi/pulumi/pkg/diag"
 	"k8s.io/api/extensions/v1beta1"
@@ -116,7 +117,8 @@ func (iia *ingressInitAwaiter) Await() error {
 			iia.config.currentInputs.GetName())
 	}
 
-	return iia.await(ingressWatcher, serviceWatcher, endpointWatcher, make(chan struct{}), time.After(10*time.Minute))
+	timeout := time.Duration(metadata.TimeoutSeconds(iia.config.currentInputs, 10*60)) * time.Second
+	return iia.await(ingressWatcher, serviceWatcher, endpointWatcher, make(chan struct{}), time.After(timeout))
 }
 
 func (iia *ingressInitAwaiter) Read() error {

pkg/metadata/annotations.go

@@ -26,13 +26,15 @@ const (
 
 	AnnotationPrefix = "pulumi.com/"
 
-	AnnotationAutonamed = AnnotationPrefix + "autonamed"
-	AnnotationSkipAwait = AnnotationPrefix + "skipAwait"
+	AnnotationAutonamed      = AnnotationPrefix + "autonamed"
+	AnnotationSkipAwait      = AnnotationPrefix + "skipAwait"
+	AnnotationTimeoutSeconds = AnnotationPrefix + "timeoutSeconds"
 )
 
 // Annotations for internal Pulumi use only.
 var internalAnnotationPrefixes = []string{AnnotationAutonamed}
 
+// IsInternalAnnotation returns true if the specified annotation has the `pulumi.com/` prefix, false otherwise.
 func IsInternalAnnotation(key string) bool {
 	for _, annotationPrefix := range internalAnnotationPrefixes {
 		if strings.HasPrefix(key, annotationPrefix) {
@@ -43,6 +45,7 @@ func IsInternalAnnotation(key string) bool {
 	return false
 }
 
+// SetAnnotation sets the specified key, value annotation on the provided Unstructured object.
 func SetAnnotation(obj *unstructured.Unstructured, key, value string) {
 	annotations := obj.GetAnnotations()
 	if annotations == nil {
@@ -52,12 +55,20 @@ func SetAnnotation(obj *unstructured.Unstructured, key, value string) {
 	obj.SetAnnotations(annotations)
 }
 
+// SetAnnotationTrue sets the specified annotation key to "true" on the provided Unstructured object.
 func SetAnnotationTrue(obj *unstructured.Unstructured, key string) {
 	SetAnnotation(obj, key, AnnotationTrue)
 }
 
+// IsAnnotationTrue returns true if the specified annotation has the value "true", false otherwise.
 func IsAnnotationTrue(obj *unstructured.Unstructured, key string) bool {
 	annotations := obj.GetAnnotations()
 	value := annotations[key]
 	return value == AnnotationTrue
 }
+
+// GetAnnotationValue returns the value of the specified annotation on the provided Unstructured object.
+func GetAnnotationValue(obj *unstructured.Unstructured, key string) string {
+	annotations := obj.GetAnnotations()
+	return annotations[key]
+}

pkg/metadata/overrides.go

@@ -15,9 +15,25 @@
 package metadata
 
 import (
+	"strconv"
+
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 )
 
+// SkipAwaitLogic returns true if the `pulumi.com/skipAwait` annotation is "true", false otherwise.
 func SkipAwaitLogic(obj *unstructured.Unstructured) bool {
 	return IsAnnotationTrue(obj, AnnotationSkipAwait)
 }
+
+// TimeoutSeconds returns the int value of the `pulumi.com/timeoutSeconds` annotation, or the defaultSeconds value
+// if the annotation is unset/invalid.
+func TimeoutSeconds(obj *unstructured.Unstructured, defaultSeconds int) int {
+	if s := GetAnnotationValue(obj, AnnotationTimeoutSeconds); s != "" {
+		val, err := strconv.Atoi(s)
+		if err == nil {
+			return val
+		}
+	}
+
+	return defaultSeconds
+}

pkg/metadata/overrides_test.go

@@ -49,3 +49,34 @@ func TestSkipAwaitLogic(t *testing.T) {
 		})
 	}
 }
+
+func TestTimeoutSeconds(t *testing.T) {
+	resource := &unstructured.Unstructured{}
+
+	annotatedResource15 := &unstructured.Unstructured{}
+	annotatedResource15.SetAnnotations(map[string]string{AnnotationTimeoutSeconds: "15"})
+
+	annotatedResourceInvalid := &unstructured.Unstructured{}
+	annotatedResourceInvalid.SetAnnotations(map[string]string{AnnotationTimeoutSeconds: "foo"})
+
+	type args struct {
+		obj            *unstructured.Unstructured
+		defaultSeconds int
+	}
+	tests := []struct {
+		name string
+		args args
+		want int
+	}{
+		{"Timeout annotation unset", args{obj: resource, defaultSeconds: 300}, 300},
+		{"Timeout annotation set", args{obj: annotatedResource15, defaultSeconds: 300}, 15},
+		{"Timeout annotation invalid", args{obj: annotatedResourceInvalid, defaultSeconds: 300}, 300},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := TimeoutSeconds(tt.args.obj, tt.args.defaultSeconds); got != tt.want {
+				t.Errorf("TimeoutSeconds() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}