Merge branch '120822/idempotency-token'

charts/databricks-kube-operator/Chart.yaml

@@ -15,7 +15,7 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.3.2
+version: 0.3.3
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to

charts/databricks-kube-operator/templates/sts.yaml

@@ -18,7 +18,7 @@ spec:
       terminationGracePeriodSeconds: 10
       containers:
       - name: dko
-        image: ghcr.io/mach-kernel/databricks-kube-operator:0.3.2
+        image: ghcr.io/mach-kernel/databricks-kube-operator:0.3.3
         imagePullPolicy: Always
         env:
           - name: DATABRICKS_KUBE_CONFIGMAP

databricks-kube/Cargo.toml

@@ -5,7 +5,7 @@ path = "src/crdgen.rs"
 [package]
 name = "databricks_kube"
 default-run = "databricks_kube"
-version = "0.3.2"
+version = "0.3.3"
 edition = "2021"
 
 [dependencies]

databricks-kube/src/crds/databricks_job.rs

@@ -1,16 +1,19 @@
 use std::collections::hash_map::DefaultHasher;
 use std::hash::Hasher;
+use std::pin::Pin;
 use std::sync::Arc;
 use std::time::SystemTime;
-use std::{hash::Hash, pin::Pin};
 
 use crate::traits::remote_api_status::RemoteAPIStatus;
+use crate::util::hash_json_value;
 use crate::{
     context::Context, error::DatabricksKubeError, traits::remote_api_resource::RemoteAPIResource,
     traits::rest_config::RestConfig,
 };
 
-use databricks_rust_jobs::models::{JobsRunsList200Response, RunLifeCycleState, RunState, JobSettings};
+use databricks_rust_jobs::models::{
+    JobSettings, JobsRunsList200Response, RunLifeCycleState, RunState,
+};
 use databricks_rust_jobs::{
     apis::default_api,
     models::{
@@ -155,17 +158,14 @@ impl DatabricksJob {
     fn hash_run_request(request: &JobsRunNowRequest, settings: Option<Box<JobSettings>>) -> u64 {
         let mut hasher = DefaultHasher::new();
 
-        let request_as_json = serde_json::to_string(&request).unwrap();
+        let request_as_value = serde_json::to_value(&request).unwrap();
+        hash_json_value(&mut hasher, &request_as_value);
 
         if let Some(settings) = settings {
-            let settings_as_json = serde_json::to_string(&settings).unwrap();
-            settings_as_json.hash(&mut hasher);
+            let settings_as_value = serde_json::to_value(&settings).unwrap();
+            hash_json_value(&mut hasher, &settings_as_value);
         }
 
-        request_as_json.hash(&mut hasher);
-
-        // Databricks docs state a 64 char limit for the idempotency token,
-        // so we can get away with coercing i64 to a string
         hasher.finish()
     }
 }
@@ -221,7 +221,8 @@ impl RemoteAPIResource<Job> for DatabricksJob {
                 job_id,
                 ..run_request.unwrap()
             };
-            run_request.idempotency_token = Some(Self::hash_run_request(&run_request, job_settings).to_string());
+            run_request.idempotency_token =
+                Some(Self::hash_run_request(&run_request, job_settings).to_string());
 
             let triggered = default_api::jobs_run_now(&config, Some(run_request)).await?;
 

databricks-kube/src/util.rs

@@ -1,5 +1,10 @@
 #![allow(dead_code)]
 
+use serde_json::{json, Value};
+use std::collections::hash_map::DefaultHasher;
+
+use std::hash::Hash;
+
 use std::{sync::Arc, time::Duration};
 
 use crate::context::CONFIGMAP_NAME;
@@ -23,7 +28,6 @@ use kube::{
 };
 
 use schemars::schema_for;
-use serde_json::json;
 use tokio::time::timeout;
 
 pub async fn watch_api_secret(
@@ -220,3 +224,25 @@ pub async fn ensure_crd(
         .flatten()
         .ok_or(DatabricksKubeError::CRDMissingError(name.to_string()))
 }
+
+pub fn hash_json_value(hash: &mut DefaultHasher, value: &Value) {
+    match value {
+        Value::Null => 0.hash(hash),
+        Value::Bool(b) => b.hash(hash),
+        Value::String(s) => s.hash(hash),
+        Value::Number(n) => n.hash(hash),
+        Value::Array(a) => {
+            for v in a {
+                hash_json_value(hash, v)
+            }
+        }
+        Value::Object(o) => {
+            let mut keys: Vec<String> = o.keys().cloned().collect();
+            keys.sort();
+
+            for k in keys {
+                hash_json_value(hash, o.get(&k).unwrap())
+            }
+        }
+    }
+}

databricks-kube/tests/fixtures/random-1.json

@@ -0,0 +1,57 @@
+{
+    "97": {
+      "<To": [
+        ":7oP!`",
+        [
+          [
+            "!dR%",
+            [
+              {
+                "03,~4JAg-": true,
+                "~%S#'": -1988141153,
+                "}": false,
+                "~4Q1tXi@Y": {
+                  "6": 1511003206,
+                  ":]Z`#{": 910201886.8581648,
+                  "V": false,
+                  "": "",
+                  "nI_He0}m": true,
+                  "lhHr?nT7y": true
+                },
+                "": false,
+                "&CmD'": 198916140.92531967
+              },
+              539682369.870666,
+              true,
+              -1296095952.9940119,
+              true,
+              202617390
+            ],
+            false,
+            true,
+            1177603502,
+            1055836135.0364754
+          ],
+          1236434144,
+          "[C1fjkHh",
+          -1029243481.1868985,
+          "!(<>p-fQ6",
+          true
+        ],
+        false,
+        ":2-K5J?e",
+        "RT~otYx=s!",
+        "Cfj\\"
+      ],
+      "": "[ kM$",
+      "!m o4Cn,@J": "&<kL.",
+      "2'm=zAgN$": " W0bh&SQ-",
+      "a'*": -998187725.4032295,
+      "=rW": true
+    },
+    "Kn": "8d|:q",
+    "w": 45134795,
+    "y)QTkW5q": "9|=1s4o",
+    "?lkQF": false,
+    ":mh$(B": "j9"
+  }