fix: hashing job spec and storing in annotation

Author: Amr Asem

controllers/djob_controller.go

@@ -96,9 +96,12 @@ func (r *DjobReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
 				r.Recorder.Event(instance, corev1.EventTypeWarning, "Resetting object", fmt.Sprintf("Failed to reset object: %s", err))
 				return ctrl.Result{}, fmt.Errorf("error when resetting job: %v", err)
 			}
+			if err := r.UpdateHash(instance); err != nil {
+				return ctrl.Result{}, fmt.Errorf("Failed to update the hash key: %v", err)
+			}
 			r.Recorder.Event(instance, corev1.EventTypeNormal, "Reset", "Object is reset")
 		} else {
-			r.Log.Info(fmt.Sprintf("Refreshesss for %v", req.NamespacedName))
+			r.Log.Info(fmt.Sprintf("Refresh for %v", req.NamespacedName))
 			if err := r.refresh(instance); err != nil {
 				r.Recorder.Event(instance, corev1.EventTypeWarning, "Refreshing object", fmt.Sprintf("Failed to refresh object: %s", err))
 				return ctrl.Result{}, fmt.Errorf("error when refreshing job: %v", err)

controllers/djob_controller_databricks.go

@@ -18,16 +18,16 @@ package controllers
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
-	"strings"
-
 	databricksv1alpha1 "github.com/microsoft/azure-databricks-operator/api/v1alpha1"
+	"github.com/mitchellh/hashstructure"
 	dbmodels "github.com/xinsnake/databricks-sdk-golang/azure/models"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"reflect"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"strconv"
+	"strings"
 )
 
 func (r *DjobReconciler) submit(instance *databricksv1alpha1.Djob) error {
@@ -71,6 +71,13 @@ func (r *DjobReconciler) submit(instance *databricksv1alpha1.Djob) error {
 		}
 		instance.ObjectMeta.SetOwnerReferences(references)
 	}
+
+	if hash, err := hashstructure.Hash(instance.Spec, nil); err == nil {
+		instance.ObjectMeta.SetAnnotations(map[string]string{instance.GetName(): strconv.FormatUint(hash, 10)})
+	} else {
+		r.Log.Info(fmt.Sprintf("Failed to hash the Spec for job %s", instance.GetName()))
+	}
+
 	jobSettings := databricksv1alpha1.ToDatabricksJobSettings(instance.Spec)
 	job, err := r.createJob(jobSettings)
 
@@ -124,86 +131,36 @@ func (r *DjobReconciler) refresh(instance *databricksv1alpha1.Djob) error {
 }
 
 /*
-IsDJobUpdated  is a method to check if the cluster has the latest version of a certain Djob
+IsDJobUpdated checks if the cluster has the latest version of a certain Djob
 */
 func (r *DjobReconciler) IsDJobUpdated(instance *databricksv1alpha1.Djob) bool {
-	//	func (r *DjobReconciler) checkIdentity(oldDJob, newDJob *[]byte) bool {
-	jobID := instance.Status.JobStatus.JobID
-	jobExisting, err := r.APIClient.Jobs().Get(jobID)
-	if err != nil {
-		if strings.Contains(err.Error(), "does not exist") {
-			return true
-		}
-		return true
-	}
-	newDJob := []interface{}{}
-	ExistingDJob := []interface{}{}
 
-	if instance.Spec.NotebookTask != nil {
-		v, _ := json.Marshal(instance.Spec.NotebookTask)
-		newDJob = append(newDJob, v)
-	}
-	if instance.Spec.SparkJarTask != nil {
-		v, _ := json.Marshal(instance.Spec.SparkJarTask)
-		newDJob = append(newDJob, v)
+	currentAnnotation := instance.ObjectMeta.GetAnnotations()[instance.GetName()]
+	var updatedHash uint64
+	if returnUpdatedHash, err := hashstructure.Hash(instance.Spec, nil); err != nil {
+		r.Log.Info(fmt.Sprintf("Deleting job %s", instance.GetName()))
+	} else {
+		updatedHash = returnUpdatedHash
 	}
 
-	if instance.Spec.SparkPythonTask != nil {
-		v, _ := json.Marshal(instance.Spec.SparkPythonTask)
-		newDJob = append(newDJob, v)
-	}
+	// jobID := instance.Status.JobStatus.JobID
+	// jobExisting, err := r.APIClient.Jobs().Get(jobID)
+	// if err != nil {
+	// 	if strings.Contains(err.Error(), "does not exist") {
+	// 		return true
+	// 	}
+	// 	return true
+	// }
 
-	if instance.Spec.SparkSubmitTask != nil {
-		v, _ := json.Marshal(instance.Spec.SparkSubmitTask)
-		newDJob = append(newDJob, v)
-	}
+	// hash2, err2 := hashstructure.Hash(jobExisting.Settings, nil)
+	// if err2 != nil {
+	// 	panic(err2)
+	// }
+	// r.Log.Info(fmt.Sprintf("Hashs %v", hash2))
 
-	if instance.Spec.NewCluster != nil {
-		v, _ := json.Marshal(instance.Spec.NewCluster)
-		newDJob = append(newDJob, v)
-	}
+	// r.Log.Info(fmt.Sprintf("2 object old %v and new %v", jobExisting.Settings, instance.Spec))
 
-	if instance.Spec.Schedule != nil {
-		v, _ := json.Marshal(instance.Spec.Schedule)
-		newDJob = append(newDJob, v)
-	}
-	newDJob = append(newDJob, instance.Spec.TimeoutSeconds)
-	newDJob = append(newDJob, instance.Spec.MaxRetries)
-
-	////////////////////////////////////////////////
-
-	if jobExisting.Settings.NotebookTask != nil {
-		v, _ := json.Marshal(jobExisting.Settings.NotebookTask)
-		ExistingDJob = append(ExistingDJob, v)
-	}
-	if jobExisting.Settings.SparkJarTask != nil {
-		v, _ := json.Marshal(jobExisting.Settings.SparkJarTask)
-		ExistingDJob = append(ExistingDJob, v)
-	}
-
-	if jobExisting.Settings.SparkPythonTask != nil {
-		v, _ := json.Marshal(jobExisting.Settings.SparkPythonTask)
-		ExistingDJob = append(ExistingDJob, v)
-	}
-
-	if jobExisting.Settings.SparkSubmitTask != nil {
-		v, _ := json.Marshal(jobExisting.Settings.SparkSubmitTask)
-		ExistingDJob = append(ExistingDJob, v)
-	}
-
-	if jobExisting.Settings.NewCluster != nil {
-		v, _ := json.Marshal(jobExisting.Settings.NewCluster)
-		ExistingDJob = append(ExistingDJob, v)
-	}
-
-	if jobExisting.Settings.Schedule != nil {
-		v, _ := json.Marshal(jobExisting.Settings.Schedule)
-		ExistingDJob = append(ExistingDJob, v)
-	}
-
-	ExistingDJob = append(ExistingDJob, jobExisting.Settings.TimeoutSeconds)
-	ExistingDJob = append(ExistingDJob, jobExisting.Settings.MaxRetries)
-	return reflect.DeepEqual(ExistingDJob, newDJob)
+	return currentAnnotation == strconv.FormatUint(updatedHash, 10)
 }
 
 func (r *DjobReconciler) delete(instance *databricksv1alpha1.Djob) error {
@@ -243,6 +200,19 @@ func (r *DjobReconciler) createJob(jobSettings dbmodels.JobSettings) (job dbmode
 	return job, err
 }
 
+// UpdateHash updates the current job with a new annotation key
+func (r *DjobReconciler) UpdateHash(instance *databricksv1alpha1.Djob) error {
+	hash, err := hashstructure.Hash(instance.Spec, nil)
+	if err != nil {
+		return err
+	}
+
+	delete(instance.GetAnnotations(), instance.GetName())
+	instance.ObjectMeta.SetAnnotations(map[string]string{instance.GetName(): strconv.FormatUint(hash, 10)})
+
+	return r.Update(context.Background(), instance)
+}
+
 func (r *DjobReconciler) reset(instance *databricksv1alpha1.Djob) error {
 	r.Log.Info(fmt.Sprintf("Reset job %s", instance.GetName()))
 	jobSettings := databricksv1alpha1.ToDatabricksJobSettings(instance.Spec)

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/go-logr/logr v0.1.0
 	github.com/jstemmer/go-junit-report v0.9.1 // indirect
 	github.com/matm/gocov-html v0.0.0-20191111163307-9ee104d84c82 // indirect
+	github.com/mitchellh/hashstructure v1.0.0
 	github.com/onsi/ginkgo v1.10.3
 	github.com/onsi/gomega v1.7.0
 	github.com/prometheus/client_golang v0.9.2

go.sum

@@ -196,6 +196,8 @@ github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hd
 github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU=
 github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
+github.com/mitchellh/hashstructure v1.0.0 h1:ZkRJX1CyOoTkar7p/mLS5TZU4nJ1Rn/F8u9dGS02Q3Y=
+github.com/mitchellh/hashstructure v1.0.0/go.mod h1:QjSHrPWS+BGUVBYkbTZWEnOh3G1DutKwClXU/ABz6AQ=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=