Delete resources' cr if pulsar cluster doesn't exist (#236)

* Delete resources cr if pulsar cluster doesn't exist * Declare an error function and deal with topic * Lint code
streamnative · Sep 4, 2024 · baaba51 · baaba51
1 parent 0a10ea2
commit baaba51
Show file tree

Hide file tree

Showing 5 changed files with 42 additions and 13 deletions.
diff --git a/pkg/admin/errors.go b/pkg/admin/errors.go
@@ -15,6 +15,9 @@
 package admin
 
 import (
+	"errors"
+	"net"
+
 	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/rest"
 )
 
@@ -70,3 +73,9 @@ func IsAlreadyExist(err error) bool {
 func IsInternalServerError(err error) bool {
 	return ErrorReason(err) == ReasonInternalServerError
 }
+
+// IsNoSuchHostError returns true if operator cannot connect the resource host
+func IsNoSuchHostError(err error) bool {
+	var dnsErr *net.DNSError
+	return errors.As(err, &dnsErr) && dnsErr.Err == "no such host"
+}
diff --git a/pkg/connection/reconcile_geo_replication.go b/pkg/connection/reconcile_geo_replication.go
@@ -156,8 +156,12 @@ func (r *PulsarGeoReplicationReconciler) ReconcileGeoReplication(ctx context.Con
 				// Delete the cluster that created with destination cluster info.
 				// TODO it can only be deleted after the cluster has been removed from the tenant, namespace, and topic
 				if err := pulsarAdmin.DeleteCluster(destClusterName); err != nil && !admin.IsNotFound(err) {
-					log.Error(err, "Failed to delete geo replication cluster")
-					return err
+					if admin.IsNoSuchHostError(err) {
+						log.Info("Pulsar cluster has been deleted")
+					} else {
+						log.Error(err, "Failed to delete geo replication cluster")
+						return err
+					}
 				}
 			}
 			controllerutil.RemoveFinalizer(geoReplication, resourcev1alpha1.FinalizerName)

diff --git a/pkg/connection/reconcile_namespace.go b/pkg/connection/reconcile_namespace.go
@@ -98,20 +98,28 @@ func (r *PulsarNamespaceReconciler) ReconcileNamespace(ctx context.Context, puls
 		if namespace.Status.GeoReplicationEnabled {
 			log.Info("GeoReplication is enabled. Reset namespace cluster", "LifecyclePolicy", namespace.Spec.LifecyclePolicy, "ClusterName", r.conn.connection.Spec.ClusterName)
 			if err := pulsarAdmin.SetNamespaceClusters(namespace.Spec.Name, []string{r.conn.connection.Spec.ClusterName}); err != nil {
-				log.Error(err, "Failed to reset the cluster for namespace")
-				return err
+				if admin.IsNoSuchHostError(err) {
+					log.Info("Pulsar cluster has been deleted")
+				} else {
+					log.Error(err, "Failed to reset the cluster for namespace")
+					return err
+				}
 			}
 		}
 
 		if namespace.Spec.LifecyclePolicy != resourcev1alpha1.KeepAfterDeletion {
 			if err := pulsarAdmin.DeleteNamespace(namespace.Spec.Name); err != nil && !admin.IsNotFound(err) {
-				log.Error(err, "Failed to delete namespace")
-				meta.SetStatusCondition(&namespace.Status.Conditions, *NewErrorCondition(namespace.Generation, err.Error()))
-				if err := r.conn.client.Status().Update(ctx, namespace); err != nil {
-					log.Error(err, "Failed to update the geo replication status")
+				if admin.IsNoSuchHostError(err) {
+					log.Info("Pulsar cluster has been deleted")
+				} else {
+					log.Error(err, "Failed to delete namespace")
+					meta.SetStatusCondition(&namespace.Status.Conditions, *NewErrorCondition(namespace.Generation, err.Error()))
+					if err := r.conn.client.Status().Update(ctx, namespace); err != nil {
+						log.Error(err, "Failed to update the geo replication status")
+						return err
+					}
 					return err
 				}
-				return err
 			}
 		}
 

diff --git a/pkg/connection/reconcile_tenant.go b/pkg/connection/reconcile_tenant.go
@@ -88,8 +88,12 @@ func (r *PulsarTenantReconciler) ReconcileTenant(ctx context.Context, pulsarAdmi
 		log.Info("Deleting tenant", "LifecyclePolicy", tenant.Spec.LifecyclePolicy)
 		if tenant.Spec.LifecyclePolicy != resourcev1alpha1.KeepAfterDeletion {
 			if err := pulsarAdmin.DeleteTenant(tenant.Spec.Name); err != nil && !admin.IsNotFound(err) {
-				log.Error(err, "Failed to delete tenant")
-				return err
+				if admin.IsNoSuchHostError(err) {
+					log.Info("Pulsar cluster has been deleted")
+				} else {
+					log.Error(err, "Failed to delete tenant")
+					return err
+				}
 			}
 		}
 

diff --git a/pkg/connection/reconcile_topic.go b/pkg/connection/reconcile_topic.go
@@ -101,8 +101,12 @@ func (r *PulsarTopicReconciler) ReconcileTopic(ctx context.Context, pulsarAdmin
 			if topic.Status.GeoReplicationEnabled {
 				log.Info("GeoReplication is enabled. Reset topic cluster first", "LifecyclePolicy", topic.Spec.LifecyclePolicy, "ClusterName", r.conn.connection.Spec.ClusterName)
 				if err := pulsarAdmin.SetTopicClusters(topic.Spec.Name, topic.Spec.Persistent, []string{r.conn.connection.Spec.ClusterName}); err != nil {
-					log.Error(err, "Failed to reset the cluster for topic")
-					return err
+					if admin.IsNoSuchHostError(err) {
+						log.Info("Pulsar cluster has been deleted")
+					} else {
+						log.Error(err, "Failed to reset the cluster for topic")
+						return err
+					}
 				}
 			}