Notify deadline exceeded error on subscription creation (#180)

api/v1/subscription_types.go

@@ -23,15 +23,19 @@ import (
 // SubscriptionSpec defines the desired state of Subscription
 type SubscriptionSpec struct {
 	// subscription ID
+	//+kubebuilder:validation:XValidation:message="Immutable field",rule="self == oldSelf"
 	SubscriptionID string `json:"subscriptionID,omitempty"`
 
 	// project ID of subscription
+	//+kubebuilder:validation:XValidation:message="Immutable field",rule="self == oldSelf"
 	SubscriptionProjectID string `json:"subscriptionProjectID,omitempty"`
 
 	// topic ID
+	//+kubebuilder:validation:XValidation:message="Immutable field",rule="self == oldSelf"
 	TopicID string `json:"topicID,omitempty"`
 
 	// project ID of topic
+	//+kubebuilder:validation:XValidation:message="Immutable field",rule="self == oldSelf"
 	TopicProjectID string `json:"topicProjectID,omitempty"`
 }
 

api/v1/topic_types.go

@@ -23,9 +23,11 @@ import (
 // TopicSpec defines the desired state of Topic
 type TopicSpec struct {
 	// ID of project
+	//+kubebuilder:validation:XValidation:message="Immutable field",rule="self == oldSelf"
 	ProjectID string `json:"projectID,omitempty"`
 
 	// ID of topic
+	//+kubebuilder:validation:XValidation:message="Immutable field",rule="self == oldSelf"
 	TopicID string `json:"topicID,omitempty"`
 }
 

cmd/main.go

@@ -21,6 +21,8 @@ import (
 	"os"
 
 	"cloud.google.com/go/pubsub"
+	"k8s.io/utils/clock"
+
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
 	// to ensure that exec-entrypoint and run can make use of them.
 	_ "k8s.io/client-go/plugin/pkg/client/auth"
@@ -106,6 +108,7 @@ func main() {
 		Scheme:    mgr.GetScheme(),
 		NewClient: pubsub.NewClient,
 		Recorder:  mgr.GetEventRecorderFor("subscription-controller"),
+		Clock:     clock.RealClock{},
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Subscription")
 		os.Exit(1)

config/crd/bases/googlecloudpubsuboperator.quipper.github.io_subscriptions.yaml

@@ -37,15 +37,27 @@ spec:
               subscriptionID:
                 description: subscription ID
                 type: string
+                x-kubernetes-validations:
+                - message: Immutable field
+                  rule: self == oldSelf
               subscriptionProjectID:
                 description: project ID of subscription
                 type: string
+                x-kubernetes-validations:
+                - message: Immutable field
+                  rule: self == oldSelf
               topicID:
                 description: topic ID
                 type: string
+                x-kubernetes-validations:
+                - message: Immutable field
+                  rule: self == oldSelf
               topicProjectID:
                 description: project ID of topic
                 type: string
+                x-kubernetes-validations:
+                - message: Immutable field
+                  rule: self == oldSelf
             type: object
           status:
             description: SubscriptionStatus defines the observed state of Subscription

config/crd/bases/googlecloudpubsuboperator.quipper.github.io_topics.yaml

@@ -37,9 +37,15 @@ spec:
               projectID:
                 description: ID of project
                 type: string
+                x-kubernetes-validations:
+                - message: Immutable field
+                  rule: self == oldSelf
               topicID:
                 description: ID of topic
                 type: string
+                x-kubernetes-validations:
+                - message: Immutable field
+                  rule: self == oldSelf
             type: object
           status:
             description: TopicStatus defines the observed state of Topic

go.mod

@@ -11,6 +11,7 @@ require (
 	k8s.io/api v0.28.4
 	k8s.io/apimachinery v0.28.4
 	k8s.io/client-go v0.28.4
+	k8s.io/utils v0.0.0-20230406110748-d93618cff8a2
 	sigs.k8s.io/controller-runtime v0.15.1
 )
 
@@ -82,7 +83,6 @@ require (
 	k8s.io/component-base v0.27.2 // indirect
 	k8s.io/klog/v2 v2.100.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20230717233707-2695361300d9 // indirect
-	k8s.io/utils v0.0.0-20230406110748-d93618cff8a2 // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
 	sigs.k8s.io/yaml v1.3.0 // indirect

internal/controller/subscription_controller.go

@@ -25,6 +25,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/tools/record"
+	"k8s.io/utils/clock"
 	ctrl "sigs.k8s.io/controller-runtime"
 	crclient "sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -35,12 +36,15 @@ import (
 
 const subscriptionFinalizerName = "subscription.googlecloudpubsuboperator.quipper.github.io/finalizer"
 
+const subscriptionCreationWarningDeadline = 10 * time.Minute
+
 // SubscriptionReconciler reconciles a Subscription object
 type SubscriptionReconciler struct {
 	crclient.Client
 	Scheme    *runtime.Scheme
 	NewClient newPubSubClientFunc
 	Recorder  record.EventRecorder
+	Clock     clock.PassiveClock
 }
 
 //+kubebuilder:rbac:groups=googlecloudpubsuboperator.quipper.github.io,resources=subscriptions,verbs=get;list;watch;create;update;patch;delete
@@ -108,8 +112,13 @@ func (r *SubscriptionReconciler) Reconcile(ctx context.Context, req ctrl.Request
 			return ctrl.Result{}, nil
 		}
 
-		r.Recorder.Event(&subscription, corev1.EventTypeWarning, "SubscriptionCreateError",
-			fmt.Sprintf("Failed to create Subscription in Pub/Sub: %s", err))
+		if r.Clock.Since(subscription.CreationTimestamp.Time) > subscriptionCreationWarningDeadline {
+			r.Recorder.Event(&subscription, corev1.EventTypeWarning, "SubscriptionCreateErrorDeadlineExceeded",
+				fmt.Sprintf("Failed to create Subscription for %s in Pub/Sub: %s", subscriptionCreationWarningDeadline, err))
+		} else {
+			r.Recorder.Event(&subscription, corev1.EventTypeNormal, "SubscriptionCreateError",
+				fmt.Sprintf("Failed to create Subscription in Pub/Sub: %s", err))
+		}
 		subscriptionPatch := crclient.MergeFrom(subscription.DeepCopy())
 		subscription.Status.Phase = googlecloudpubsuboperatorv1.SubscriptionStatusPhaseError
 		if err := r.Client.Status().Patch(ctx, &subscription, subscriptionPatch); err != nil {

internal/controller/suite_test.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"cloud.google.com/go/pubsub"
 	"cloud.google.com/go/pubsub/pstest"
@@ -29,6 +30,7 @@ import (
 	"google.golang.org/api/option"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
+	clocktesting "k8s.io/utils/clock/testing"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/envtest"
@@ -113,6 +115,8 @@ var _ = BeforeSuite(func() {
 		Scheme:    k8sManager.GetScheme(),
 		NewClient: newClient,
 		Recorder:  k8sManager.GetEventRecorderFor("subscription-controller"),
+		// TODO: how to change the time in test code?
+		Clock: clocktesting.NewFakePassiveClock(time.Date(2021, 1, 1, 0, 0, 0, 0, time.UTC)),
 	}).SetupWithManager(k8sManager)
 	Expect(err).ToNot(HaveOccurred())