automatically upgrade json jobs

Author: milt

src/cli/com/yetanalytics/xapipe/cli/options.clj

@@ -149,15 +149,15 @@
     [nil "--delete-job ID" "Delete the job specified and exit."]
     ["-f" "--force-resume" "If resuming a job, clear any errors and force it to resume."
      :default false]
-    [nil "--json JSON" "Take a job specification as a JSON string"
+    [nil "--json JSON" "Take a job specification as a JSON string (automatically upgrades old jobs)"
      :parse-fn job-json/json->job]
-    [nil "--json-file FILE" "Take a job specification from a JSON file"
+    [nil "--json-file FILE" "Take a job specification from a JSON file (automatically upgrades old jobs)"
      :parse-fn (fn [filepath]
                  (-> filepath
                      slurp
                      job-json/json->job))]
     [nil "--json-out FILE" "Write JOB to a JSON file"]
-    [nil "--[no-]upgrade" "Automatically upgrade jobs to the latest version"
+    [nil "--[no-]upgrade" "Automatically upgrade jobs started from storage to the latest version (jobs started from JSON will always be upgraded)"
      :id :upgrade-jobs
      :default true]]
    (concat storage-options

src/lib/com/yetanalytics/xapipe/job.clj

@@ -13,7 +13,7 @@
 
 (s/def ::config config/config-spec)
 (s/def ::state state/state-spec)
-(s/def ::version #{0 current-version})
+(s/def ::version #{current-version})
 (s/def ::id (s/and string? not-empty))
 
 (defn valid-get-params-vs-state?

src/lib/com/yetanalytics/xapipe/job/config.clj

@@ -69,6 +69,7 @@
      poll-interval    :poll-interval
      {?since :since
       ?until :until}  :get-params
+     get-req-config   :request-config
      :as              source-config
      :or              {get-batch-size   50
                        get-backoff-opts {:budget      10000
@@ -77,6 +78,7 @@
     :source
     {post-batch-size   :batch-size
      post-backoff-opts :backoff-opts
+     post-req-config   :request-config
      :as               target-config
      :or               {post-backoff-opts {:budget      10000
                                            :max-attempt 10}}}
@@ -108,27 +110,32 @@
         cleanup-buffer-size
         (or cleanup-buffer-size
             get-batch-size
-            0)]
-    {:get-buffer-size       get-buffer-size
-     :statement-buffer-size statement-buffer-size
-     :batch-buffer-size     batch-buffer-size
-     :batch-timeout         batch-timeout
-     :cleanup-buffer-size   cleanup-buffer-size
-     :source
-     (-> source-config
-         (assoc :batch-size get-batch-size
-                :backoff-opts get-backoff-opts
-                :poll-interval poll-interval)
-         (assoc-in [:get-params :limit] get-batch-size)
-         (cond->
-          ?since (update-in [:get-params :since] t/normalize-stamp)
-          ?until (update-in [:get-params :until] t/normalize-stamp)))
-     :target
-     (assoc target-config
-            :batch-size post-batch-size
-            :backoff-opts post-backoff-opts)
-     :filter
-     (or filter-config {})}))
+            0)
+        source-version (get get-req-config :xapi-version "1.0.3")
+        target-version (get post-req-config :xapi-version "1.0.3")]
+    (->
+     {:get-buffer-size       get-buffer-size
+      :statement-buffer-size statement-buffer-size
+      :batch-buffer-size     batch-buffer-size
+      :batch-timeout         batch-timeout
+      :cleanup-buffer-size   cleanup-buffer-size
+      :source
+      (-> source-config
+          (assoc :batch-size get-batch-size
+                 :backoff-opts get-backoff-opts
+                 :poll-interval poll-interval)
+          (assoc-in [:get-params :limit] get-batch-size)
+          (cond->
+              ?since (update-in [:get-params :since] t/normalize-stamp)
+              ?until (update-in [:get-params :until] t/normalize-stamp)))
+      :target
+      (assoc target-config
+             :batch-size post-batch-size
+             :backoff-opts post-backoff-opts)
+      :filter
+      (or filter-config {})}
+     (assoc-in [:source :request-config :xapi-version] source-version)
+     (assoc-in [:target :request-config :xapi-version] target-version))))
 
 (s/fdef sanitize-req-cfg
   :args (s/cat :rcfg ::client/request-config)

src/lib/com/yetanalytics/xapipe/job/json.clj

@@ -95,11 +95,14 @@
 
 (defn json->job
   "Parse a job from JSON, possibly updating it to the current version."
-  [^String json-str]
+  [^String json-str & {:keys [upgrade]
+                       :or {upgrade true}}]
   (-> (json/parse-string json-str (partial keyword nil))
       keywordize-status
       keywordize-job-error-types
       (unpack-paths [[:state :filter :pattern]])
+      (cond->
+          upgrade job/upgrade-job)
       (update :config config/ensure-defaults)))
 
 (s/def ::pretty boolean?)