Skip to content

Commit

Permalink
issue-572, handling of the resource external deletion was added
Browse files Browse the repository at this point in the history
  • Loading branch information
Bohdan Siryk authored and Bohdan Siryk committed Sep 27, 2023
1 parent 3723b60 commit ac411f4
Show file tree
Hide file tree
Showing 13 changed files with 287 additions and 296 deletions.
78 changes: 39 additions & 39 deletions controllers/clusters/cadence_controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -790,45 +790,7 @@ func (r *CadenceReconciler) newWatchStatusJob(cadence *v1beta1.Cadence) schedule
iData, err := r.API.GetCadence(cadence.Status.ID)
if err != nil {
if errors.Is(err, instaclustr.NotFound) {
activeClusters, err := r.API.ListClusters()
if err != nil {
l.Error(err, "Cannot list account active clusters")
return err
}

if !isClusterActive(cadence.Status.ID, activeClusters) {
l.Info("Cluster is not found in Instaclustr. Deleting resource.",
"cluster ID", cadence.Status.ClusterStatus.ID,
"cluster name", cadence.Spec.Name,
)

patch := cadence.NewPatch()
cadence.Annotations[models.ClusterDeletionAnnotation] = ""
cadence.Annotations[models.ResourceStateAnnotation] = models.DeletingEvent
err = r.Patch(context.TODO(), cadence, patch)
if err != nil {
l.Error(err, "Cannot patch Cadence cluster resource",
"cluster ID", cadence.Status.ID,
"cluster name", cadence.Spec.Name,
"resource name", cadence.Name,
)

return err
}

err = r.Delete(context.TODO(), cadence)
if err != nil {
l.Error(err, "Cannot delete Cadence cluster resource",
"cluster ID", cadence.Status.ID,
"cluster name", cadence.Spec.Name,
"resource name", cadence.Name,
)

return err
}

return nil
}
return r.handleExternalDelete(context.Background(), cadence)
}

l.Error(err, "Cannot get Cadence cluster from the Instaclustr API",
Expand Down Expand Up @@ -1281,3 +1243,41 @@ func (r *CadenceReconciler) reconcileMaintenanceEvents(ctx context.Context, c *v

return nil
}

func (r *CadenceReconciler) handleExternalDelete(ctx context.Context, c *v1beta1.Cadence) error {
l := log.FromContext(ctx)

clusters, err := r.API.ListClusters()
if err != nil {
l.Error(err, "Cannot fetch active clusters for the account")
r.EventRecorder.Eventf(c, models.Warning, models.FetchFailed,
"Cannot fetch active clusters for the account. Reason: %w", err)

return err
}

if isClusterActive(c.Status.ID, clusters) {
l.Info("The resource is not found on Instaclustr but it exists in the list of active clusters for the account",
"cluster ID", c.Status.ID,
)
r.EventRecorder.Event(c, models.Warning, models.NotFound,
"The resource is not found on Instaclustr but it exists in the list of active clusters for the account")

return nil
}

patch := c.NewPatch()
c.Status.State = models.DeletedStatus
err = r.Status().Patch(ctx, c, patch)
if err != nil {
return err
}

l.Info(instaclustr.MsgInstaclustrResourceNotFound)
r.EventRecorder.Eventf(c, models.Warning, models.ExternalDeleted, instaclustr.MsgInstaclustrResourceNotFound)

r.Scheduler.RemoveJob(c.GetJobID(scheduler.BackupsChecker))
r.Scheduler.RemoveJob(c.GetJobID(scheduler.StatusChecker))

return nil
}
79 changes: 40 additions & 39 deletions controllers/clusters/cassandra_controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -885,45 +885,7 @@ func (r *CassandraReconciler) newWatchStatusJob(cassandra *v1beta1.Cassandra) sc
iData, err := r.API.GetCassandra(cassandra.Status.ID)
if err != nil {
if errors.Is(err, instaclustr.NotFound) {
activeClusters, err := r.API.ListClusters()
if err != nil {
l.Error(err, "Cannot list account active clusters")
return err
}

if !isClusterActive(cassandra.Status.ID, activeClusters) {
l.Info("Cluster is not found in Instaclustr. Deleting resource.",
"cluster ID", cassandra.Status.ClusterStatus.ID,
"cluster name", cassandra.Spec.Name,
)

patch := cassandra.NewPatch()
cassandra.Annotations[models.ClusterDeletionAnnotation] = ""
cassandra.Annotations[models.ResourceStateAnnotation] = models.DeletingEvent
err = r.Patch(context.TODO(), cassandra, patch)
if err != nil {
l.Error(err, "Cannot patch Cassandra cluster resource",
"cluster ID", cassandra.Status.ID,
"cluster name", cassandra.Spec.Name,
"resource name", cassandra.Name,
)

return err
}

err = r.Delete(context.TODO(), cassandra)
if err != nil {
l.Error(err, "Cannot delete Cassandra cluster resource",
"cluster ID", cassandra.Status.ID,
"cluster name", cassandra.Spec.Name,
"resource name", cassandra.Name,
)

return err
}

return nil
}
return r.handleExternalDelete(context.Background(), cassandra)
}

l.Error(err, "Cannot get cluster from the Instaclustr API",
Expand Down Expand Up @@ -1262,6 +1224,45 @@ func (r *CassandraReconciler) reconcileMaintenanceEvents(ctx context.Context, c
return nil
}

func (r *CassandraReconciler) handleExternalDelete(ctx context.Context, c *v1beta1.Cassandra) error {
l := log.FromContext(ctx)

clusters, err := r.API.ListClusters()
if err != nil {
l.Error(err, "Cannot fetch active clusters for the account")
r.EventRecorder.Eventf(c, models.Warning, models.FetchFailed,
"Cannot fetch active clusters for the account. Reason: %w", err)

return err
}

if isClusterActive(c.Status.ID, clusters) {
l.Info("The resource is not found on Instaclustr but it exists in the list of active clusters for the account",
"cluster ID", c.Status.ID,
)
r.EventRecorder.Event(c, models.Warning, models.NotFound,
"The resource is not found on Instaclustr but it exists in the list of active clusters for the account")

return nil
}

patch := c.NewPatch()
c.Status.State = models.DeletedStatus
err = r.Status().Patch(ctx, c, patch)
if err != nil {
return err
}

l.Info(instaclustr.MsgInstaclustrResourceNotFound)
r.EventRecorder.Eventf(c, models.Warning, models.ExternalDeleted, instaclustr.MsgInstaclustrResourceNotFound)

r.Scheduler.RemoveJob(c.GetJobID(scheduler.BackupsChecker))
r.Scheduler.RemoveJob(c.GetJobID(scheduler.UserCreator))
r.Scheduler.RemoveJob(c.GetJobID(scheduler.StatusChecker))

return nil
}

// SetupWithManager sets up the controller with the Manager.
func (r *CassandraReconciler) SetupWithManager(mgr ctrl.Manager) error {
return ctrl.NewControllerManagedBy(mgr).
Expand Down
5 changes: 2 additions & 3 deletions controllers/clusters/helpers.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,11 +22,10 @@ import (
"sort"

"github.com/hashicorp/go-version"
"k8s.io/utils/strings/slices"
"sigs.k8s.io/controller-runtime/pkg/client"

"github.com/instaclustr/operator/apis/clusters/v1beta1"
"github.com/instaclustr/operator/pkg/models"
"k8s.io/utils/strings/slices"
"sigs.k8s.io/controller-runtime/pkg/client"
)

// confirmDeletion confirms if resource is deleting and set appropriate annotation.
Expand Down
51 changes: 23 additions & 28 deletions controllers/clusters/kafka_controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -677,6 +677,7 @@ func (r *KafkaReconciler) newWatchStatusJob(kafka *v1beta1.Kafka) scheduler.Job
"namespaced name", namespacedName)
r.Scheduler.RemoveJob(kafka.GetJobID(scheduler.StatusChecker))
r.Scheduler.RemoveJob(kafka.GetJobID(scheduler.UserCreator))
r.Scheduler.RemoveJob(kafka.GetJobID(scheduler.BackupsChecker))
return nil
}
if err != nil {
Expand All @@ -688,7 +689,7 @@ func (r *KafkaReconciler) newWatchStatusJob(kafka *v1beta1.Kafka) scheduler.Job
iData, err := r.API.GetKafka(kafka.Status.ID)
if err != nil {
if errors.Is(err, instaclustr.NotFound) {
return r.handleDeleteFromInstaclustrUI(kafka, l)
return r.handleExternalDelete(context.Background(), kafka)
}

l.Error(err, "Cannot get cluster from the Instaclustr", "cluster ID", kafka.Status.ID)
Expand Down Expand Up @@ -868,47 +869,41 @@ func (r *KafkaReconciler) newUsersCreationJob(kafka *v1beta1.Kafka) scheduler.Jo
}
}

func (r *KafkaReconciler) handleDeleteFromInstaclustrUI(kafka *v1beta1.Kafka, l logr.Logger) error {
activeClusters, err := r.API.ListClusters()
func (r *KafkaReconciler) handleExternalDelete(ctx context.Context, kafka *v1beta1.Kafka) error {
l := log.FromContext(ctx)

clusters, err := r.API.ListClusters()
if err != nil {
l.Error(err, "Cannot list account active clusters")
l.Error(err, "Cannot fetch active clusters for the account")
r.EventRecorder.Eventf(kafka, models.Warning, models.FetchFailed,
"Cannot fetch active clusters for the account. Reason: %w", err)

return err
}

if isClusterActive(kafka.Status.ID, activeClusters) {
l.Info("Kafka is not found in the Instaclustr but still exist in the Instaclustr list of active cluster",
if isClusterActive(kafka.Status.ID, clusters) {
l.Info("The resource is not found on Instaclustr but it exists in the list of active clusters for the account",
"cluster ID", kafka.Status.ID,
"cluster name", kafka.Spec.Name,
"resource name", kafka.Name)
)
r.EventRecorder.Event(kafka, models.Warning, models.NotFound,
"The resource is not found on Instaclustr but it exists in the list of active clusters for the account")

return nil
}

l.Info("Cluster is not found in Instaclustr. Deleting resource.",
"cluster ID", kafka.Status.ClusterStatus.ID,
"cluster name", kafka.Spec.Name)

patch := kafka.NewPatch()

kafka.Annotations[models.ClusterDeletionAnnotation] = ""
kafka.Annotations[models.ResourceStateAnnotation] = models.DeletingEvent
err = r.Patch(context.TODO(), kafka, patch)
kafka.Status.State = models.DeletedStatus
err = r.Status().Patch(ctx, kafka, patch)
if err != nil {
l.Error(err, "Cannot patch Kafka cluster resource",
"cluster ID", kafka.Status.ID,
"cluster name", kafka.Spec.Name,
"resource name", kafka.Name)
return err
}

err = r.Delete(context.TODO(), kafka)
if err != nil {
l.Error(err, "Cannot delete Kafka cluster resource",
"cluster ID", kafka.Status.ID,
"cluster name", kafka.Spec.Name,
"resource name", kafka.Name)
return err
}
l.Info(instaclustr.MsgInstaclustrResourceNotFound)
r.EventRecorder.Eventf(kafka, models.Warning, models.ExternalDeleted, instaclustr.MsgInstaclustrResourceNotFound)

r.Scheduler.RemoveJob(kafka.GetJobID(scheduler.BackupsChecker))
r.Scheduler.RemoveJob(kafka.GetJobID(scheduler.UserCreator))
r.Scheduler.RemoveJob(kafka.GetJobID(scheduler.StatusChecker))

return nil
}
Expand Down
78 changes: 39 additions & 39 deletions controllers/clusters/kafkaconnect_controller.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -468,45 +468,7 @@ func (r *KafkaConnectReconciler) newWatchStatusJob(kc *v1beta1.KafkaConnect) sch
iData, err := r.API.GetKafkaConnect(kc.Status.ID)
if err != nil {
if errors.Is(err, instaclustr.NotFound) {
activeClusters, err := r.API.ListClusters()
if err != nil {
l.Error(err, "Cannot list account active clusters")
return err
}

if !isClusterActive(kc.Status.ID, activeClusters) {
l.Info("Cluster is not found in Instaclustr. Deleting resource.",
"cluster ID", kc.Status.ClusterStatus.ID,
"cluster name", kc.Spec.Name,
)

patch := kc.NewPatch()
kc.Annotations[models.ClusterDeletionAnnotation] = ""
kc.Annotations[models.ResourceStateAnnotation] = models.DeletingEvent
err = r.Patch(context.TODO(), kc, patch)
if err != nil {
l.Error(err, "Cannot patch KafkaConnect cluster resource",
"cluster ID", kc.Status.ID,
"cluster name", kc.Spec.Name,
"resource name", kc.Name,
)

return err
}

err = r.Delete(context.TODO(), kc)
if err != nil {
l.Error(err, "Cannot delete KafkaConnect cluster resource",
"cluster ID", kc.Status.ID,
"cluster name", kc.Spec.Name,
"resource name", kc.Name,
)

return err
}

return nil
}
return r.handleExternalDelete(context.Background(), kc)
}

l.Error(err, "Cannot get Kafka Connect from Instaclustr",
Expand Down Expand Up @@ -657,3 +619,41 @@ func (r *KafkaConnectReconciler) reconcileMaintenanceEvents(ctx context.Context,

return nil
}

func (r *KafkaConnectReconciler) handleExternalDelete(ctx context.Context, kc *v1beta1.KafkaConnect) error {
l := log.FromContext(ctx)

clusters, err := r.API.ListClusters()
if err != nil {
l.Error(err, "Cannot fetch active clusters for the account")
r.EventRecorder.Eventf(kc, models.Warning, models.FetchFailed,
"Cannot fetch active clusters for the account. Reason: %w", err)

return err
}

if isClusterActive(kc.Status.ID, clusters) {
l.Info("The resource is not found on Instaclustr but it exists in the list of active clusters for the account",
"cluster ID", kc.Status.ID,
)
r.EventRecorder.Event(kc, models.Warning, models.NotFound,
"The resource is not found on Instaclustr but it exists in the list of active clusters for the account")

return nil
}

patch := kc.NewPatch()
kc.Status.State = models.DeletedStatus
err = r.Status().Patch(ctx, kc, patch)
if err != nil {
return err
}

l.Info(instaclustr.MsgInstaclustrResourceNotFound)
r.EventRecorder.Eventf(kc, models.Warning, models.ExternalDeleted, instaclustr.MsgInstaclustrResourceNotFound)

r.Scheduler.RemoveJob(kc.GetJobID(scheduler.BackupsChecker))
r.Scheduler.RemoveJob(kc.GetJobID(scheduler.StatusChecker))

return nil
}
Loading

0 comments on commit ac411f4

Please sign in to comment.