Skip to content

Commit bf81d58

Browse files
dejanzeledejanzele
committed
remove duplicated code and fix minor bugs
Signed-off-by: Dejan Zele Pejchev <pejcev.dejan@gmail.com>
1 parent e3bcf44 commit bf81d58

20 files changed

+807
-670
lines changed

‎go.mod

+1
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ require (
2323
github.com/cespare/xxhash/v2 v2.3.0 // indirect
2424
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
2525
github.com/emicklei/go-restful/v3 v3.12.1 // indirect
26+
github.com/evanphx/json-patch v4.12.0+incompatible // indirect
2627
github.com/evanphx/json-patch/v5 v5.9.0 // indirect
2728
github.com/fsnotify/fsnotify v1.7.0 // indirect
2829
github.com/go-logr/zapr v1.3.0 // indirect

‎internal/controller/install/armadaserver_controller.go

+53-114
Original file line numberDiff line numberDiff line change
@@ -64,67 +64,34 @@ type ArmadaServerReconciler struct {
6464
// move the current state of the cluster closer to the desired state.
6565
func (r *ArmadaServerReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
6666
logger := log.FromContext(ctx).WithValues("namespace", req.Namespace, "name", req.Name)
67+
6768
started := time.Now()
68-
logger.Info("Reconciling ArmadaServer object")
69-
70-
logger.Info("Fetching ArmadaServer object from cache")
71-
var as installv1alpha1.ArmadaServer
72-
if err := r.Client.Get(ctx, req.NamespacedName, &as); err != nil {
73-
if k8serrors.IsNotFound(err) {
74-
logger.Info("ArmadaServer not found in cache, ending reconcile...", "namespace", req.Namespace, "name", req.Name)
75-
return ctrl.Result{}, nil
76-
}
69+
70+
logger.Info("Reconciling object")
71+
72+
var server installv1alpha1.ArmadaServer
73+
if miss, err := getObject(ctx, r.Client, &server, req.NamespacedName, logger); err != nil || miss {
7774
return ctrl.Result{}, err
7875
}
7976

80-
pc, err := installv1alpha1.BuildPortConfig(as.Spec.ApplicationConfig)
77+
pc, err := installv1alpha1.BuildPortConfig(server.Spec.ApplicationConfig)
8178
if err != nil {
8279
return ctrl.Result{}, err
8380
}
84-
as.Spec.PortConfig = pc
81+
server.Spec.PortConfig = pc
8582

8683
var components *CommonComponents
87-
components, err = generateArmadaServerInstallComponents(&as, r.Scheme)
84+
components, err = generateArmadaServerInstallComponents(&server, r.Scheme)
8885
if err != nil {
8986
return ctrl.Result{}, err
9087
}
9188

92-
deletionTimestamp := as.ObjectMeta.DeletionTimestamp
93-
// examine DeletionTimestamp to determine if object is under deletion
94-
if deletionTimestamp.IsZero() {
95-
// The object is not being deleted, so if it does not have our finalizer,
96-
// then lets add the finalizer and update the object. This is equivalent
97-
// registering our finalizer.
98-
if !controllerutil.ContainsFinalizer(&as, operatorFinalizer) {
99-
logger.Info("Attaching finalizer to As object", "finalizer", operatorFinalizer)
100-
controllerutil.AddFinalizer(&as, operatorFinalizer)
101-
if err := r.Update(ctx, &as); err != nil {
102-
return ctrl.Result{}, err
103-
}
104-
}
105-
} else {
106-
logger.Info("ArmadaServer object is being deleted", "finalizer", operatorFinalizer)
107-
logger.Info("Namespace-scoped resources will be deleted by Kubernetes based on their OwnerReference")
108-
// The object is being deleted
109-
if controllerutil.ContainsFinalizer(&as, operatorFinalizer) {
110-
// our finalizer is present, so lets handle any external dependency
111-
logger.Info("Running cleanup function for ArmadaServer cluster-scoped components", "finalizer", operatorFinalizer)
112-
if err := r.deleteExternalResources(ctx, components, logger); err != nil {
113-
// if fail to delete the external dependency here, return with error
114-
// so that it can be retried
115-
return ctrl.Result{}, err
116-
}
117-
118-
// remove our finalizer from the list and update it.
119-
logger.Info("Removing finalizer from ArmadaServer object", "finalizer", operatorFinalizer)
120-
controllerutil.RemoveFinalizer(&as, operatorFinalizer)
121-
if err := r.Update(ctx, &as); err != nil {
122-
return ctrl.Result{}, err
123-
}
124-
}
125-
126-
// Stop reconciliation as the item is being deleted
127-
return ctrl.Result{}, nil
89+
cleanupF := func(ctx context.Context) error {
90+
return r.deleteExternalResources(ctx, components, logger)
91+
}
92+
finish, err := checkAndHandleObjectDeletion(ctx, r.Client, &server, operatorFinalizer, cleanupF, logger)
93+
if err != nil || finish {
94+
return ctrl.Result{}, err
12895
}
12996

13097
componentsCopy := components.DeepCopy()
@@ -134,90 +101,58 @@ func (r *ArmadaServerReconciler) Reconcile(ctx context.Context, req ctrl.Request
134101
return nil
135102
}
136103

137-
if components.ServiceAccount != nil {
138-
logger.Info("Upserting ArmadaServer ServiceAccount object")
139-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.ServiceAccount, mutateFn); err != nil {
140-
return ctrl.Result{}, err
141-
}
104+
if err := upsertObjectIfNeeded(ctx, r.Client, components.ServiceAccount, server.Kind, mutateFn, logger); err != nil {
105+
return ctrl.Result{}, err
142106
}
143107

144-
if components.Secret != nil {
145-
logger.Info("Upserting ArmadaServer Secret object")
146-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.Secret, mutateFn); err != nil {
147-
return ctrl.Result{}, err
148-
}
108+
if err := upsertObjectIfNeeded(ctx, r.Client, components.Secret, server.Kind, mutateFn, logger); err != nil {
109+
return ctrl.Result{}, err
149110
}
150111

151-
if as.Spec.PulsarInit {
152-
for idx := range components.Jobs {
153-
err = func() error {
154-
if components.Jobs[idx] != nil {
155-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.Jobs[idx], mutateFn); err != nil {
156-
return err
157-
}
158-
ctxTimeout, cancel := context.WithTimeout(ctx, migrationTimeout)
159-
defer cancel()
160-
161-
err := waitForJob(ctxTimeout, r.Client, components.Jobs[idx], migrationPollSleep)
162-
if err != nil {
163-
return err
164-
}
112+
if server.Spec.PulsarInit {
113+
for _, job := range components.Jobs {
114+
err = func(job *batchv1.Job) error {
115+
if err := upsertObjectIfNeeded(ctx, r.Client, job, server.Kind, mutateFn, logger); err != nil {
116+
return err
117+
}
118+
119+
if err := waitForJob(ctx, r.Client, job, jobPollInterval, jobTimeout); err != nil {
120+
return err
165121
}
166122
return nil
167-
}()
123+
}(job)
168124
if err != nil {
169125
return ctrl.Result{}, err
170126
}
171127
}
172128
}
173129

174-
if components.Deployment != nil {
175-
logger.Info("Upserting ArmadaServer Deployment object")
176-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.Deployment, mutateFn); err != nil {
177-
return ctrl.Result{}, err
178-
}
130+
if err := upsertObjectIfNeeded(ctx, r.Client, components.Deployment, server.Kind, mutateFn, logger); err != nil {
131+
return ctrl.Result{}, err
179132
}
180133

181-
if components.Service != nil {
182-
logger.Info("Upserting ArmadaServer Service object")
183-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.Service, mutateFn); err != nil {
184-
return ctrl.Result{}, err
185-
}
134+
if err := upsertObjectIfNeeded(ctx, r.Client, components.Service, server.Kind, mutateFn, logger); err != nil {
135+
return ctrl.Result{}, err
186136
}
187137

188-
if components.IngressGrpc != nil {
189-
logger.Info("Upserting ArmadaServer GRPC Ingress object")
190-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.IngressGrpc, mutateFn); err != nil {
191-
return ctrl.Result{}, err
192-
}
138+
if err := upsertObjectIfNeeded(ctx, r.Client, components.IngressGrpc, server.Kind, mutateFn, logger); err != nil {
139+
return ctrl.Result{}, err
193140
}
194141

195-
if components.IngressHttp != nil {
196-
logger.Info("Upserting ArmadaServer REST Ingress object")
197-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.IngressHttp, mutateFn); err != nil {
198-
return ctrl.Result{}, err
199-
}
142+
if err := upsertObjectIfNeeded(ctx, r.Client, components.IngressHttp, server.Kind, mutateFn, logger); err != nil {
143+
return ctrl.Result{}, err
200144
}
201145

202-
if components.PodDisruptionBudget != nil {
203-
logger.Info("Upserting ArmadaServer PodDisruptionBudget object")
204-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.PodDisruptionBudget, mutateFn); err != nil {
205-
return ctrl.Result{}, err
206-
}
146+
if err := upsertObjectIfNeeded(ctx, r.Client, components.PodDisruptionBudget, server.Kind, mutateFn, logger); err != nil {
147+
return ctrl.Result{}, err
207148
}
208149

209-
if components.PrometheusRule != nil {
210-
logger.Info("Upserting ArmadaServer PrometheusRule object")
211-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.PrometheusRule, mutateFn); err != nil {
212-
return ctrl.Result{}, err
213-
}
150+
if err := upsertObjectIfNeeded(ctx, r.Client, components.PrometheusRule, server.Kind, mutateFn, logger); err != nil {
151+
return ctrl.Result{}, err
214152
}
215153

216-
if components.ServiceMonitor != nil {
217-
logger.Info("Upserting ArmadaServer ServiceMonitor object")
218-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.ServiceMonitor, mutateFn); err != nil {
219-
return ctrl.Result{}, err
220-
}
154+
if err := upsertObjectIfNeeded(ctx, r.Client, components.ServiceMonitor, server.Kind, mutateFn, logger); err != nil {
155+
return ctrl.Result{}, err
221156
}
222157

223158
logger.Info("Successfully reconciled ArmadaServer object", "durationMillis", time.Since(started).Milliseconds())
@@ -283,13 +218,17 @@ func generateArmadaServerInstallComponents(as *installv1alpha1.ArmadaServer, sch
283218
}
284219

285220
var pr *monitoringv1.PrometheusRule
221+
var sm *monitoringv1.ServiceMonitor
286222
if as.Spec.Prometheus != nil && as.Spec.Prometheus.Enabled {
287223
pr = createServerPrometheusRule(as.Name, as.Namespace, as.Spec.Prometheus.ScrapeInterval, as.Spec.Labels, as.Spec.Prometheus.Labels)
288-
}
224+
if err := controllerutil.SetOwnerReference(as, pr, scheme); err != nil {
225+
return nil, err
226+
}
289227

290-
sm := createServiceMonitor(as)
291-
if err := controllerutil.SetOwnerReference(as, sm, scheme); err != nil {
292-
return nil, err
228+
sm = createServiceMonitor(as)
229+
if err := controllerutil.SetOwnerReference(as, sm, scheme); err != nil {
230+
return nil, err
231+
}
293232
}
294233

295234
jobs := []*batchv1.Job{{}}
@@ -331,12 +270,12 @@ func createArmadaServerMigrationJobs(as *installv1alpha1.ArmadaServer) ([]*batch
331270

332271
appConfig, err := builders.ConvertRawExtensionToYaml(as.Spec.ApplicationConfig)
333272
if err != nil {
334-
return []*batchv1.Job{}, err
273+
return nil, err
335274
}
336275
var asConfig AppConfig
337276
err = yaml.Unmarshal([]byte(appConfig), &asConfig)
338277
if err != nil {
339-
return []*batchv1.Job{}, err
278+
return nil, err
340279
}
341280

342281
// First job is to poll/wait for Pulsar to be fully started

‎internal/controller/install/binoculars_controller.go

+30-81
Original file line numberDiff line numberDiff line change
@@ -61,17 +61,13 @@ type BinocularsReconciler struct {
6161
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.13.0/pkg/reconcile
6262
func (r *BinocularsReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
6363
logger := log.FromContext(ctx).WithValues("namespace", req.Namespace, "name", req.Name)
64+
6465
started := time.Now()
65-
logger.Info("Reconciling Binoculars object")
6666

67-
logger.Info("Fetching Binoculars object from cache")
67+
logger.Info("Reconciling Binoculars object")
6868

6969
var binoculars installv1alpha1.Binoculars
70-
if err := r.Client.Get(ctx, req.NamespacedName, &binoculars); err != nil {
71-
if k8serrors.IsNotFound(err) {
72-
logger.Info("Binoculars not found in cache, ending reconcile...", "namespace", req.Namespace, "name", req.Name)
73-
return ctrl.Result{}, nil
74-
}
70+
if miss, err := getObject(ctx, r.Client, &binoculars, req.NamespacedName, logger); err != nil || miss {
7571
return ctrl.Result{}, err
7672
}
7773

@@ -87,41 +83,12 @@ func (r *BinocularsReconciler) Reconcile(ctx context.Context, req ctrl.Request)
8783
return ctrl.Result{}, err
8884
}
8985

90-
deletionTimestamp := binoculars.ObjectMeta.DeletionTimestamp
91-
// examine DeletionTimestamp to determine if object is under deletion
92-
if deletionTimestamp.IsZero() {
93-
// The object is not being deleted, so if it does not have our finalizer,
94-
// then lets add the finalizer and update the object. This is equivalent
95-
// registering our finalizer.
96-
if !controllerutil.ContainsFinalizer(&binoculars, operatorFinalizer) {
97-
logger.Info("Attaching finalizer to Binoculars object", "finalizer", operatorFinalizer)
98-
controllerutil.AddFinalizer(&binoculars, operatorFinalizer)
99-
if err := r.Update(ctx, &binoculars); err != nil {
100-
return ctrl.Result{}, err
101-
}
102-
}
103-
} else {
104-
logger.Info("Binoculars object is being deleted", "finalizer", operatorFinalizer)
105-
// The object is being deleted
106-
if controllerutil.ContainsFinalizer(&binoculars, operatorFinalizer) {
107-
// our finalizer is present, so lets handle any external dependency
108-
logger.Info("Running cleanup function for Binoculars object", "finalizer", operatorFinalizer)
109-
if err := r.deleteExternalResources(ctx, components); err != nil {
110-
// if fail to delete the external dependency here, return with error
111-
// so that it can be retried
112-
return ctrl.Result{}, err
113-
}
114-
115-
// remove our finalizer from the list and update it.
116-
logger.Info("Removing finalizer from Binoculars object", "finalizer", operatorFinalizer)
117-
controllerutil.RemoveFinalizer(&binoculars, operatorFinalizer)
118-
if err := r.Update(ctx, &binoculars); err != nil {
119-
return ctrl.Result{}, err
120-
}
121-
}
122-
123-
// Stop reconciliation as the item is being deleted
124-
return ctrl.Result{}, nil
86+
cleanupF := func(ctx context.Context) error {
87+
return r.deleteExternalResources(ctx, components)
88+
}
89+
finish, err := checkAndHandleObjectDeletion(ctx, r.Client, &binoculars, operatorFinalizer, cleanupF, logger)
90+
if err != nil || finish {
91+
return ctrl.Result{}, err
12592
}
12693

12794
componentsCopy := components.DeepCopy()
@@ -131,58 +98,40 @@ func (r *BinocularsReconciler) Reconcile(ctx context.Context, req ctrl.Request)
13198
return nil
13299
}
133100

134-
if components.ServiceAccount != nil {
135-
logger.Info("Upserting Binoculars ServiceAccount object")
136-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.ServiceAccount, mutateFn); err != nil {
137-
return ctrl.Result{}, err
138-
}
101+
if err := upsertObjectIfNeeded(ctx, r.Client, components.ServiceAccount, binoculars.Kind, mutateFn, logger); err != nil {
102+
return ctrl.Result{}, err
139103
}
140104

141-
if components.ClusterRole != nil {
142-
logger.Info("Upserting Binoculars ClusterRole object")
143-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.ClusterRole, mutateFn); err != nil {
144-
return ctrl.Result{}, err
145-
}
105+
if err := upsertObjectIfNeeded(ctx, r.Client, components.ClusterRole, binoculars.Kind, mutateFn, logger); err != nil {
106+
return ctrl.Result{}, err
146107
}
147108

148-
if components.ClusterRoleBindings != nil && len(components.ClusterRoleBindings) > 0 {
149-
logger.Info("Upserting Binoculars ClusterRoleBinding object")
150-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.ClusterRoleBindings[0], mutateFn); err != nil {
151-
return ctrl.Result{}, err
109+
if len(components.ClusterRoleBindings) > 0 {
110+
for _, crb := range components.ClusterRoleBindings {
111+
if err := upsertObjectIfNeeded(ctx, r.Client, crb, binoculars.Kind, mutateFn, logger); err != nil {
112+
return ctrl.Result{}, err
113+
}
152114
}
153115
}
154116

155-
if components.Secret != nil {
156-
logger.Info("Upserting Binoculars Secret object")
157-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.Secret, mutateFn); err != nil {
158-
return ctrl.Result{}, err
159-
}
117+
if err := upsertObjectIfNeeded(ctx, r.Client, components.Secret, binoculars.Kind, mutateFn, logger); err != nil {
118+
return ctrl.Result{}, err
160119
}
161120

162-
if components.Deployment != nil {
163-
logger.Info("Upserting Binoculars Deployment object")
164-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.Deployment, mutateFn); err != nil {
165-
return ctrl.Result{}, err
166-
}
121+
if err := upsertObjectIfNeeded(ctx, r.Client, components.Deployment, binoculars.Kind, mutateFn, logger); err != nil {
122+
return ctrl.Result{}, err
167123
}
168124

169-
if components.Service != nil {
170-
logger.Info("Upserting Binoculars Service object")
171-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.Service, mutateFn); err != nil {
172-
return ctrl.Result{}, err
173-
}
125+
if err := upsertObjectIfNeeded(ctx, r.Client, components.Service, binoculars.Kind, mutateFn, logger); err != nil {
126+
return ctrl.Result{}, err
174127
}
175-
if components.IngressGrpc != nil {
176-
logger.Info("Upserting GRPC Ingress object")
177-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.IngressGrpc, mutateFn); err != nil {
178-
return ctrl.Result{}, err
179-
}
128+
129+
if err := upsertObjectIfNeeded(ctx, r.Client, components.IngressGrpc, binoculars.Kind, mutateFn, logger); err != nil {
130+
return ctrl.Result{}, err
180131
}
181-
if components.IngressHttp != nil {
182-
logger.Info("Upserting REST Ingress object")
183-
if _, err := controllerutil.CreateOrUpdate(ctx, r.Client, components.IngressHttp, mutateFn); err != nil {
184-
return ctrl.Result{}, err
185-
}
132+
133+
if err := upsertObjectIfNeeded(ctx, r.Client, components.IngressHttp, binoculars.Kind, mutateFn, logger); err != nil {
134+
return ctrl.Result{}, err
186135
}
187136

188137
logger.Info("Successfully reconciled Binoculars object", "durationMillis", time.Since(started).Milliseconds())

0 commit comments

Comments
 (0)