Reconcilers delete old Topics and PullSubscriptions

pkg/duck/v1alpha1/identifiable.go

@@ -17,13 +17,17 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"k8s.io/apimachinery/pkg/runtime"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/kmeta"
 
 	duckv1alpha1 "github.com/google/knative-gcp/pkg/apis/duck/v1alpha1"
 )
 
 type Identifiable interface {
+	// runtime.Object can be removed once the old Topic and PullSubscription are removed.
+	runtime.Object
+
 	kmeta.OwnerRefable
 	// IdentitySpec returns the IdentitySpec portion of the Spec.
 	IdentitySpec() *duckv1alpha1.IdentitySpec

pkg/reconciler/intevents/reconciler.go

@@ -22,6 +22,7 @@ import (
 
 	duckv1alpha1 "github.com/google/knative-gcp/pkg/apis/duck/v1alpha1"
 	inteventsv1alpha1 "github.com/google/knative-gcp/pkg/apis/intevents/v1alpha1"
+	pubsubv1alpha1 "github.com/google/knative-gcp/pkg/apis/pubsub/v1alpha1"
 	clientset "github.com/google/knative-gcp/pkg/client/clientset/versioned"
 	duck "github.com/google/knative-gcp/pkg/duck/v1alpha1"
 	"github.com/google/knative-gcp/pkg/reconciler"
@@ -62,39 +63,56 @@ type PubSubBase struct {
 // Also sets the following fields in the pubsubable.Status upon success
 // TopicID, ProjectID, and SinkURI
 func (psb *PubSubBase) ReconcilePubSub(ctx context.Context, pubsubable duck.PubSubable, topic, resourceGroup string) (*inteventsv1alpha1.Topic, *inteventsv1alpha1.PullSubscription, error) {
+	t, err := psb.reconcileTopic(ctx, pubsubable, topic)
+	if err != nil {
+		return t, nil, err
+	}
+
+	ps, err := psb.ReconcilePullSubscription(ctx, pubsubable, topic, resourceGroup, false)
+	if err != nil {
+		return t, ps, err
+	}
+	return t, ps, nil
+}
+
+func (psb *PubSubBase) reconcileTopic(ctx context.Context, pubsubable duck.PubSubable, topic string) (*inteventsv1alpha1.Topic, pkgreconciler.Event) {
 	if pubsubable == nil {
-		return nil, nil, fmt.Errorf("nil pubsubable passed in")
+		return nil, fmt.Errorf("nil pubsubable passed in")
 	}
-	namespace := pubsubable.GetObjectMeta().GetNamespace()
-	name := pubsubable.GetObjectMeta().GetName()
-	spec := pubsubable.PubSubSpec()
-	annotations := pubsubable.GetObjectMeta().GetAnnotations()
-	status := pubsubable.PubSubStatus()
 
+	name := pubsubable.GetObjectMeta().GetName()
 	args := &resources.TopicArgs{
-		Namespace:   namespace,
+		Namespace:   pubsubable.GetObjectMeta().GetNamespace(),
 		Name:        name,
-		Spec:        spec,
+		Spec:        pubsubable.PubSubSpec(),
 		Owner:       pubsubable,
 		Topic:       topic,
 		Labels:      resources.GetLabels(psb.receiveAdapterName, name),
-		Annotations: annotations,
+		Annotations: pubsubable.GetObjectMeta().GetAnnotations(),
 	}
 	newTopic := resources.MakeTopic(args)
 
-	topics := psb.pubsubClient.InternalV1alpha1().Topics(namespace)
-	t, err := topics.Get(name, v1.GetOptions{})
+	// The old and new Topics use the same, deterministic names. So delete the old one before
+	// creating the new one. They cannot both be Ready=true at the same time, so by deleting the old
+	// Topic, we allow the new Topic to become ready.
+	err := psb.deleteOldPubSubTopic(ctx, pubsubable, newTopic)
 	if err != nil {
-		if !apierrs.IsNotFound(err) {
-			logging.FromContext(ctx).Desugar().Error("Failed to get Topics", zap.Error(err))
-			return nil, nil, fmt.Errorf("failed to get Topics: %w", err)
-		}
+		logging.FromContext(ctx).Desugar().Info("Unable to delete old Topic", zap.Error(err))
+		return nil, err
+	}
+
+	topics := psb.pubsubClient.InternalV1alpha1().Topics(newTopic.Namespace)
+	t, err := topics.Get(newTopic.Name, v1.GetOptions{})
+	if apierrs.IsNotFound(err) {
 		logging.FromContext(ctx).Desugar().Debug("Creating Topic", zap.Any("topic", newTopic))
 		t, err = topics.Create(newTopic)
 		if err != nil {
 			logging.FromContext(ctx).Desugar().Error("Failed to create Topic", zap.Any("topic", newTopic), zap.Error(err))
-			return nil, nil, fmt.Errorf("failed to create Topic: %w", err)
+			return nil, fmt.Errorf("failed to create Topic: %w", err)
 		}
+	} else if err != nil {
+		logging.FromContext(ctx).Desugar().Error("Failed to get Topic", zap.Error(err))
+		return nil, fmt.Errorf("failed to get Topic: %w", err)
 		// Check whether the specs differ and update the Topic if so.
 	} else if !equality.Semantic.DeepDerivative(newTopic.Spec, t.Spec) {
 		// Don't modify the informers copy.
@@ -104,21 +122,17 @@ func (psb *PubSubBase) ReconcilePubSub(ctx context.Context, pubsubable duck.PubS
 		t, err = topics.Update(desired)
 		if err != nil {
 			logging.FromContext(ctx).Desugar().Error("Failed to update Topic", zap.Any("topic", t), zap.Error(err))
-			return nil, nil, fmt.Errorf("failed to update Topic: %w", err)
+			return nil, fmt.Errorf("failed to update Topic: %w", err)
 		}
 	}
 
+	status := pubsubable.PubSubStatus()
 	cs := pubsubable.ConditionSet()
-
 	if err := propagateTopicStatus(t, status, cs, topic); err != nil {
-		return t, nil, err
+		return t, err
 	}
 
-	ps, err := psb.ReconcilePullSubscription(ctx, pubsubable, topic, resourceGroup, false)
-	if err != nil {
-		return t, ps, err
-	}
-	return t, ps, nil
+	return t, nil
 }
 
 func (psb *PubSubBase) ReconcilePullSubscription(ctx context.Context, pubsubable duck.PubSubable, topic, resourceGroup string, isPushCompatible bool) (*inteventsv1alpha1.PullSubscription, pkgreconciler.Event) {
@@ -181,9 +195,97 @@ func (psb *PubSubBase) ReconcilePullSubscription(ctx context.Context, pubsubable
 	}
 
 	status.SinkURI = ps.Status.SinkURI
+
+	// The old and new PullSubscriptions can co-exist without any problems. So to bias in favor of
+	// double event delivery over dropped events, don't delete the old one until the new one is
+	// ready.
+	if ps.Status.IsReady() {
+		err = psb.deleteOldPubSubPullSubscription(ctx, pubsubable, ps)
+		if err != nil {
+			return ps, err
+		}
+	}
+
 	return ps, nil
 }
 
+func (psb *PubSubBase) deleteOldPubSubTopic(_ context.Context, pubsubable duck.PubSubable, t *inteventsv1alpha1.Topic) pkgreconciler.Event {
+	// TODO This will be deleted at the same time as the old pubsub.cloud.google.com CRDs. That is
+	// expected to happen after 0.15, before 0.16.
+	oldT, err := psb.pubsubClient.PubsubV1alpha1().Topics(t.Namespace).Get(t.Name, v1.GetOptions{})
+	if apierrs.IsNotFound(err) {
+		// It doesn't exist, so there is nothing to delete.
+		return nil
+	} else if err != nil {
+		return pkgreconciler.NewEvent(corev1.EventTypeWarning, "OldTopicGetFailed", "unable to get old Topic in the `pubsub.events.cloud.google.com` API group: %w", err)
+	}
+	if !v1.IsControlledBy(oldT, pubsubable.GetObjectMeta()) {
+		// If this pubsubable doesn't own it, then just ignore it. Generate an event in case users
+		// are interested, but do not stop reconciliation of pubsubable, nor give it a Ready=false
+		// status.
+		psb.Recorder.Eventf(pubsubable,
+			corev1.EventTypeWarning,
+			"OldTopicNotControlled",
+			"old Topic '%s/%s' in the `pubsub.events.cloud.google.com` API group is not controlled by this pubsubable, so won't be deleted. Actual owners: %v",
+			oldT.Namespace, oldT.Name, oldT.OwnerReferences)
+		return nil
+	}
+
+	// First, to make sure the Topic is not deleted in GCP, update the Topic with a new deletion
+	// policy.
+	switch pp := oldT.Spec.PropagationPolicy; pp {
+	case pubsubv1alpha1.TopicPolicyCreateDelete:
+		c := oldT.DeepCopy()
+		c.Spec.PropagationPolicy = pubsubv1alpha1.TopicPolicyCreateNoDelete
+		oldT, err = psb.pubsubClient.PubsubV1alpha1().Topics(oldT.Namespace).Update(c)
+		if err != nil {
+			return pkgreconciler.NewEvent(corev1.EventTypeWarning, "OldTopicUpdateFailed", "unable to update propagation policy on old Topic: %w", err)
+		}
+	case pubsubv1alpha1.TopicPolicyCreateNoDelete:
+		// Already marked for non-deletion.
+		break
+	case pubsubv1alpha1.TopicPolicyNoCreateNoDelete:
+		// Already marked for non-deletion.
+		break
+	default:
+		return pkgreconciler.NewEvent(corev1.EventTypeWarning, "OldTopicUnknownPropagationPolicy", "unknown propagation policy on old Topic: %v", pp)
+	}
+
+	err = psb.pubsubClient.PubsubV1alpha1().Topics(oldT.Namespace).Delete(oldT.Name, nil)
+	if err != nil {
+		return pkgreconciler.NewEvent(corev1.EventTypeWarning, "OldTopicDeletionFailed", "unable to delete old Topic in the `pubsub.events.cloud.google.com` API group: %w", err)
+	}
+	return nil
+}
+
+func (psb *PubSubBase) deleteOldPubSubPullSubscription(_ context.Context, pubsubable duck.PubSubable, ps *inteventsv1alpha1.PullSubscription) pkgreconciler.Event {
+	// TODO This will be deleted at the same time as the old pubsub.cloud.google.com CRDs. That is
+	// expected to happen after 0.15, before 0.16.
+	oldPS, err := psb.pubsubClient.PubsubV1alpha1().PullSubscriptions(ps.Namespace).Get(ps.Name, v1.GetOptions{})
+	if apierrs.IsNotFound(err) {
+		// It doesn't exist, so there is nothing to delete.
+		return nil
+	} else if err != nil {
+		return pkgreconciler.NewEvent(corev1.EventTypeWarning, "OldPullSubscriptionGetFailed", "unable to get old PullSubscription in the `pubsub.events.cloud.google.com` API group: %w", err)
+	}
+	if !v1.IsControlledBy(oldPS, pubsubable.GetObjectMeta()) {
+		// If this pubsubable doesn't own it, then just ignore it. Generate an event in case users
+		// are interested, but do not stop reconciliation of pubsubable, nor give it a Ready=false
+		// status.
+		psb.Recorder.Eventf(pubsubable,
+			corev1.EventTypeWarning,
+			"oldPullSubscriptionNotControlled",
+			"old PullSubscription '%s/%s' in the `pubsub.events.cloud.google.com` API group is not controlled by this pubsubable, so won't be deleted. Actual owners: %v",
+			oldPS.Namespace, oldPS.Name, oldPS.OwnerReferences)
+		return nil
+	}
+	err = psb.pubsubClient.PubsubV1alpha1().PullSubscriptions(oldPS.Namespace).Delete(oldPS.Name, nil)
+	if err != nil {
+		return pkgreconciler.NewEvent(corev1.EventTypeWarning, "OldPullSubscriptionDeletionFailed", "unable to delete old PullSubscription in the `pubsub.events.cloud.google.com` API group: %w", err)
+	}
+	return nil
+}
+
 func propagatePullSubscriptionStatus(ps *inteventsv1alpha1.PullSubscription, status *duckv1alpha1.PubSubStatus, cs *apis.ConditionSet) error {
 	pc := ps.Status.GetTopLevelCondition()
 	if pc == nil {