Skip to content

Commit da563de

Browse files
itamar-maromitamar-marom
authored
(finalizers): refactor (#124)
* (finalizers): refactor * (finalizers): added tests * (finalizers): convert finalizer logs to events
1 parent a0beeb3 commit da563de

File tree

8 files changed

+341
-86
lines changed

8 files changed

+341
-86
lines changed

‎controllers/druid/druid_controller.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -69,7 +69,7 @@ func (r *DruidReconciler) Reconcile(ctx context.Context, request reconcile.Reque
6969
return ctrl.Result{}, err
7070
}
7171

72-
// Intialize Emit Events
72+
// Initialize Emit Events
7373
var emitEvent EventEmitter = EmitEventFuncs{r.Recorder}
7474

7575
if err := deployDruidCluster(ctx, r.Client, instance, emitEvent); err != nil {

‎controllers/druid/finalizers.go

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,116 @@
1+
package druid
2+
3+
import (
4+
"context"
5+
"fmt"
6+
7+
"github.com/datainfrahq/druid-operator/apis/druid/v1alpha1"
8+
appsv1 "k8s.io/api/apps/v1"
9+
v1 "k8s.io/api/core/v1"
10+
"k8s.io/apimachinery/pkg/runtime"
11+
"sigs.k8s.io/controller-runtime/pkg/client"
12+
)
13+
14+
const (
15+
deletePVCFinalizerName = "deletepvc.finalizers.druid.apache.org"
16+
)
17+
18+
var (
19+
defaultFinalizers []string
20+
)
21+
22+
func updateFinalizers(ctx context.Context, sdk client.Client, m *v1alpha1.Druid, emitEvents EventEmitter) error {
23+
desiredFinalizers := m.GetFinalizers()
24+
additionFinalizers := defaultFinalizers
25+
26+
desiredFinalizers = RemoveString(desiredFinalizers, deletePVCFinalizerName)
27+
if !m.Spec.DisablePVCDeletionFinalizer {
28+
additionFinalizers = append(additionFinalizers, deletePVCFinalizerName)
29+
}
30+
31+
for _, finalizer := range additionFinalizers {
32+
if !ContainsString(desiredFinalizers, finalizer) {
33+
desiredFinalizers = append(desiredFinalizers, finalizer)
34+
}
35+
}
36+
37+
m.SetFinalizers(desiredFinalizers)
38+
_, err := writers.Update(ctx, sdk, m, m, emitEvents)
39+
if err != nil {
40+
return err
41+
}
42+
43+
return nil
44+
}
45+
46+
func executeFinalizers(ctx context.Context, sdk client.Client, m *v1alpha1.Druid, emitEvents EventEmitter) error {
47+
if m.Spec.DisablePVCDeletionFinalizer == false {
48+
if err := executePVCFinalizer(ctx, sdk, m, emitEvents); err != nil {
49+
return err
50+
}
51+
}
52+
return nil
53+
}
54+
55+
/*
56+
executePVCFinalizer will execute a PVC deletion of all Druid's PVCs.
57+
Flow:
58+
1. Get sts List and PVC List
59+
2. Range and Delete sts first and then delete pvc. PVC must be deleted after sts termination has been executed
60+
else pvc finalizer shall block deletion since a pod/sts is referencing it.
61+
3. Once delete is executed we block program and return.
62+
*/
63+
func executePVCFinalizer(ctx context.Context, sdk client.Client, druid *v1alpha1.Druid, eventEmitter EventEmitter) error {
64+
if ContainsString(druid.ObjectMeta.Finalizers, deletePVCFinalizerName) {
65+
pvcLabels := map[string]string{
66+
"druid_cr": druid.Name,
67+
}
68+
69+
pvcList, err := readers.List(ctx, sdk, druid, pvcLabels, eventEmitter, func() objectList { return &v1.PersistentVolumeClaimList{} }, func(listObj runtime.Object) []object {
70+
items := listObj.(*v1.PersistentVolumeClaimList).Items
71+
result := make([]object, len(items))
72+
for i := 0; i < len(items); i++ {
73+
result[i] = &items[i]
74+
}
75+
return result
76+
})
77+
if err != nil {
78+
return err
79+
}
80+
81+
stsList, err := readers.List(ctx, sdk, druid, makeLabelsForDruid(druid.Name), eventEmitter, func() objectList { return &appsv1.StatefulSetList{} }, func(listObj runtime.Object) []object {
82+
items := listObj.(*appsv1.StatefulSetList).Items
83+
result := make([]object, len(items))
84+
for i := 0; i < len(items); i++ {
85+
result[i] = &items[i]
86+
}
87+
return result
88+
})
89+
if err != nil {
90+
return err
91+
}
92+
93+
eventEmitter.EmitEventGeneric(druid, string(druidFinalizerTriggered),
94+
fmt.Sprintf("Trigerring finalizer [%s] for CR [%s] in namespace [%s]", deletePVCFinalizerName, druid.Name, druid.Namespace), nil)
95+
96+
if err = deleteSTSAndPVC(ctx, sdk, druid, stsList, pvcList, eventEmitter); err != nil {
97+
eventEmitter.EmitEventGeneric(druid, string(druidFinalizerFailed),
98+
fmt.Sprintf("Finalizer [%s] failed for CR [%s] in namespace [%s]", deletePVCFinalizerName, druid.Name, druid.Namespace), err)
99+
100+
return err
101+
}
102+
103+
eventEmitter.EmitEventGeneric(druid, string(druidFinalizerSuccess),
104+
fmt.Sprintf("Finalizer [%s] success for CR [%s] in namespace [%s]", deletePVCFinalizerName, druid.Name, druid.Namespace), nil)
105+
106+
// remove our finalizer from the list and update it.
107+
druid.ObjectMeta.Finalizers = RemoveString(druid.ObjectMeta.Finalizers, deletePVCFinalizerName)
108+
109+
_, err = writers.Update(ctx, sdk, druid, druid, eventEmitter)
110+
if err != nil {
111+
return err
112+
}
113+
114+
}
115+
return nil
116+
}

‎controllers/druid/finalizers_test.go

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
package druid
2+
3+
import (
4+
"time"
5+
6+
druidv1alpha1 "github.com/datainfrahq/druid-operator/apis/druid/v1alpha1"
7+
. "github.com/onsi/ginkgo/v2"
8+
. "github.com/onsi/gomega"
9+
"k8s.io/apimachinery/pkg/types"
10+
)
11+
12+
// +kubebuilder:docs-gen:collapse=Imports
13+
14+
/*
15+
finalizers_test
16+
*/
17+
var _ = Describe("Test finalizers logic", func() {
18+
const (
19+
filePath = "testdata/finalizers.yaml"
20+
timeout = time.Second * 45
21+
interval = time.Millisecond * 250
22+
)
23+
24+
var (
25+
druid = &druidv1alpha1.Druid{}
26+
)
27+
28+
Context("When creating a druid cluster", func() {
29+
It("Should create the druid object", func() {
30+
By("Creating a new druid")
31+
druidCR, err := readDruidClusterSpecFromFile(filePath)
32+
Expect(err).Should(BeNil())
33+
Expect(k8sClient.Create(ctx, druidCR)).To(Succeed())
34+
35+
By("Getting a newly created druid")
36+
Eventually(func() bool {
37+
err := k8sClient.Get(ctx, types.NamespacedName{Name: druidCR.Name, Namespace: druidCR.Namespace}, druid)
38+
return err == nil
39+
}, timeout, interval).Should(BeTrue())
40+
})
41+
It("Should add the delete PVC finalizer", func() {
42+
By("Waiting for the finalizer to be created")
43+
Eventually(func() bool {
44+
err := k8sClient.Get(ctx, types.NamespacedName{Name: druid.Name, Namespace: druid.Namespace}, druid)
45+
if err == nil && ContainsString(druid.GetFinalizers(), deletePVCFinalizerName) {
46+
return true
47+
}
48+
return false
49+
}, timeout, interval).Should(BeTrue())
50+
})
51+
It("Should delete druid successfully", func() {
52+
By("Waiting for the druid cluster to be deleted")
53+
Expect(k8sClient.Delete(ctx, druid)).To(Succeed())
54+
Eventually(func() bool {
55+
err := k8sClient.Get(ctx, types.NamespacedName{Name: druid.Name, Namespace: druid.Namespace}, druid)
56+
return err != nil
57+
}, timeout, interval).Should(BeTrue())
58+
})
59+
})
60+
61+
Context("When creating a druid cluster with disablePVCDeletion", func() {
62+
It("Should create the druid object", func() {
63+
By("Creating a new druid")
64+
druidCR, err := readDruidClusterSpecFromFile(filePath)
65+
druidCR.Spec.DisablePVCDeletionFinalizer = true
66+
Expect(err).Should(BeNil())
67+
Expect(k8sClient.Create(ctx, druidCR)).To(Succeed())
68+
69+
By("Getting a newly created druid")
70+
Eventually(func() bool {
71+
err := k8sClient.Get(ctx, types.NamespacedName{Name: druidCR.Name, Namespace: druidCR.Namespace}, druid)
72+
return err == nil
73+
}, timeout, interval).Should(BeTrue())
74+
})
75+
It("Should not add the delete PVC finalizer", func() {
76+
By("Call for the update finalizer function")
77+
Expect(updateFinalizers(ctx, k8sClient, druid, emitEvent)).Should(BeNil())
78+
79+
By("Getting a updated druid")
80+
Eventually(func() bool {
81+
err := k8sClient.Get(ctx, types.NamespacedName{Name: druid.Name, Namespace: druid.Namespace}, druid)
82+
return err == nil
83+
}, timeout, interval).Should(BeTrue())
84+
85+
By("Checking the absence of the finalizer")
86+
Expect(ContainsString(druid.GetFinalizers(), deletePVCFinalizerName)).Should(BeFalse())
87+
})
88+
It("Should delete druid successfully", func() {
89+
Expect(k8sClient.Delete(ctx, druid)).To(Succeed())
90+
Eventually(func() bool {
91+
err := k8sClient.Get(ctx, types.NamespacedName{Name: druid.Name, Namespace: druid.Namespace}, druid)
92+
return err != nil
93+
}, timeout, interval).Should(BeTrue())
94+
})
95+
})
96+
97+
Context("When creating a druid cluster", func() {
98+
It("Should create the druid object", func() {
99+
By("Creating a new druid")
100+
druidCR, err := readDruidClusterSpecFromFile(filePath)
101+
Expect(err).Should(BeNil())
102+
Expect(k8sClient.Create(ctx, druidCR)).To(Succeed())
103+
104+
By("Getting the CR")
105+
Eventually(func() bool {
106+
err := k8sClient.Get(ctx, types.NamespacedName{Name: druidCR.Name, Namespace: druidCR.Namespace}, druid)
107+
return err == nil
108+
}, timeout, interval).Should(BeTrue())
109+
})
110+
It("Should add the delete PVC finalizer", func() {
111+
By("Waiting for the finalizer to be created")
112+
Eventually(func() bool {
113+
err := k8sClient.Get(ctx, types.NamespacedName{Name: druid.Name, Namespace: druid.Namespace}, druid)
114+
if err == nil && ContainsString(druid.GetFinalizers(), deletePVCFinalizerName) {
115+
return true
116+
}
117+
return false
118+
}, timeout, interval).Should(BeTrue())
119+
})
120+
It("Should remove the delete PVC finalizer", func() {
121+
By("Disabling the deletePVC finalizer")
122+
druid.Spec.DisablePVCDeletionFinalizer = true
123+
Expect(k8sClient.Update(ctx, druid)).To(BeNil())
124+
By("Waiting for the finalizer to be deleted")
125+
Eventually(func() bool {
126+
err := k8sClient.Get(ctx, types.NamespacedName{Name: druid.Name, Namespace: druid.Namespace}, druid)
127+
if err == nil && !ContainsString(druid.GetFinalizers(), deletePVCFinalizerName) {
128+
return true
129+
}
130+
return false
131+
}, timeout, interval).Should(BeTrue())
132+
})
133+
It("Should delete druid successfully", func() {
134+
Expect(k8sClient.Delete(ctx, druid)).To(Succeed())
135+
Eventually(func() bool {
136+
err := k8sClient.Get(ctx, types.NamespacedName{Name: druid.Name, Namespace: druid.Namespace}, druid)
137+
return err != nil
138+
}, timeout, interval).Should(BeTrue())
139+
})
140+
})
141+
})

‎controllers/druid/handler.go

Lines changed: 6 additions & 84 deletions
Original file line numberDiff line numberDiff line change
@@ -80,35 +80,12 @@ func deployDruidCluster(ctx context.Context, sdk client.Client, m *v1alpha1.Drui
8080
return err
8181
}
8282

83-
/*
84-
Default Behavior: Finalizer shall be always executed resulting in deletion of pvc post deletion of Druid CR
85-
When the object (druid CR) has for deletion time stamp set, execute the finalizer.
86-
Finalizer shall execute the following flow :
87-
1. Get sts List and PVC List
88-
2. Range and Delete sts first and then delete pvc. PVC must be deleted after sts termination has been executed
89-
else pvc finalizer shall block deletion since a pod/sts is referencing it.
90-
3. Once delete is executed we block program and return.
91-
*/
92-
93-
if m.Spec.DisablePVCDeletionFinalizer == false {
94-
md := m.GetDeletionTimestamp() != nil
95-
if md {
96-
return executeFinalizers(ctx, sdk, m, emitEvents)
97-
}
98-
/*
99-
If finalizer isn't present add it to object meta.
100-
In case cr is already deleted do not call this function
101-
*/
102-
cr := checkIfCRExists(ctx, sdk, m, emitEvents)
103-
if cr {
104-
if !ContainsString(m.ObjectMeta.Finalizers, finalizerName) {
105-
m.SetFinalizers(append(m.GetFinalizers(), finalizerName))
106-
_, err := writers.Update(context.Background(), sdk, m, m, emitEvents)
107-
if err != nil {
108-
return err
109-
}
110-
}
111-
}
83+
if m.GetDeletionTimestamp() != nil {
84+
return executeFinalizers(ctx, sdk, m, emitEvents)
85+
}
86+
87+
if err := updateFinalizers(ctx, sdk, m, emitEvents); err != nil {
88+
return err
11289
}
11390

11491
for _, elem := range allNodeSpecs {
@@ -574,61 +551,6 @@ func setPVCLabels(ctx context.Context, sdk client.Client, drd *v1alpha1.Druid, e
574551
return nil
575552
}
576553

577-
func executeFinalizers(ctx context.Context, sdk client.Client, m *v1alpha1.Druid, emitEvents EventEmitter) error {
578-
579-
if ContainsString(m.ObjectMeta.Finalizers, finalizerName) {
580-
pvcLabels := map[string]string{
581-
"druid_cr": m.Name,
582-
}
583-
584-
pvcList, err := readers.List(ctx, sdk, m, pvcLabels, emitEvents, func() objectList { return &v1.PersistentVolumeClaimList{} }, func(listObj runtime.Object) []object {
585-
items := listObj.(*v1.PersistentVolumeClaimList).Items
586-
result := make([]object, len(items))
587-
for i := 0; i < len(items); i++ {
588-
result[i] = &items[i]
589-
}
590-
return result
591-
})
592-
if err != nil {
593-
return err
594-
}
595-
596-
stsList, err := readers.List(ctx, sdk, m, makeLabelsForDruid(m.Name), emitEvents, func() objectList { return &appsv1.StatefulSetList{} }, func(listObj runtime.Object) []object {
597-
items := listObj.(*appsv1.StatefulSetList).Items
598-
result := make([]object, len(items))
599-
for i := 0; i < len(items); i++ {
600-
result[i] = &items[i]
601-
}
602-
return result
603-
})
604-
if err != nil {
605-
return err
606-
}
607-
608-
msg := fmt.Sprintf("Trigerring finalizer for CR [%s] in namespace [%s]", m.Name, m.Namespace)
609-
// sendEvent(sdk, m, v1.EventTypeNormal, DruidFinalizer, msg)
610-
logger.Info(msg)
611-
if err := deleteSTSAndPVC(ctx, sdk, m, stsList, pvcList, emitEvents); err != nil {
612-
return err
613-
} else {
614-
msg := fmt.Sprintf("Finalizer success for CR [%s] in namespace [%s]", m.Name, m.Namespace)
615-
// sendEvent(sdk, m, v1.EventTypeNormal, DruidFinalizerSuccess, msg)
616-
logger.Info(msg)
617-
}
618-
619-
// remove our finalizer from the list and update it.
620-
m.ObjectMeta.Finalizers = RemoveString(m.ObjectMeta.Finalizers, finalizerName)
621-
622-
_, err = writers.Update(ctx, sdk, m, m, emitEvents)
623-
if err != nil {
624-
return err
625-
}
626-
627-
}
628-
return nil
629-
630-
}
631-
632554
func execCheckCrashStatus(ctx context.Context, sdk client.Client, nodeSpec *v1alpha1.DruidNodeSpec, m *v1alpha1.Druid, event EventEmitter) {
633555
if m.Spec.ForceDeleteStsPodOnError == false {
634556
return

‎controllers/druid/interface.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,10 @@ const (
3838
druidNodePatchFail druidEventReason = "DruidOperatorPatchFail"
3939
druidNodePatchSucess druidEventReason = "DruidOperatorPatchSuccess"
4040
druidObjectListFail druidEventReason = "DruidOperatorListFail"
41+
42+
druidFinalizerTriggered druidEventReason = "DruidOperatorFinalizerTriggered"
43+
druidFinalizerFailed druidEventReason = "DruidFinalizerFailed"
44+
druidFinalizerSuccess druidEventReason = "DruidFinalizerSuccess"
4145
)
4246

4347
// Reader Interface

‎controllers/druid/suite_test.go

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ var (
3434
testEnv *envtest.Environment
3535
ctx context.Context
3636
cancel context.CancelFunc
37+
emitEvent EventEmitter
3738
)
3839

3940
func TestAPIs(t *testing.T) {
@@ -93,6 +94,8 @@ var _ = BeforeSuite(func() {
9394
}).SetupWithManager(k8sManager)
9495
Expect(err).ToNot(HaveOccurred())
9596

97+
emitEvent = EmitEventFuncs{k8sManager.GetEventRecorderFor("druid-operator")}
98+
9699
go func() {
97100
defer GinkgoRecover()
98101
err = k8sManager.Start(ctx)

0 commit comments

Comments
 (0)