Delete nimbus inbox jars with a simple scheduled delete jars job

trevorsummerssmith · trevorsummerssmith · commit 5046c24eef63 · 2012-01-08T12:42:27.000-05:00
This is meant to close issue 34 (Nimbus should clean inbox)

Adds two new config variables to control how often the cleaner jobs
runs and how old a jar needs to be before it will be deleted.
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
@@ -18,6 +18,8 @@ nimbus.childopts: "-Xmx1024m"
 nimbus.task.timeout.secs: 30
 nimbus.supervisor.timeout.secs: 60
 nimbus.monitor.freq.secs: 10
+nimbus.cleanup.freq.secs: 1800
+nimbus.inbox.jar.expiration.secs: 3600
 nimbus.task.launch.secs: 120
 nimbus.reassign: true
 nimbus.file.copy.expiration.secs: 600
diff --git a/src/clj/backtype/storm/daemon/nimbus.clj b/src/clj/backtype/storm/daemon/nimbus.clj
@@ -194,6 +194,7 @@
 
 
 (defmulti setup-jar cluster-mode)
+(defmulti clean-inbox cluster-mode)
 
 ;; status types
 ;; -- killed (:kill-time-secs)
@@ -640,6 +641,13 @@
                             (transition! nimbus storm-id :monitor))
                           (do-cleanup nimbus)
                           ))
+    ;; Schedule Nimbus inbox cleaner
+    (schedule-recurring (:timer nimbus)
+                        0
+                        (conf NIMBUS-CLEANUP-FREQ-SECS)
+                        (fn []
+                          (clean-inbox conf (inbox nimbus) (conf NIMBUS-INBOX-JAR-EXPIRATION-SECS))
+                          ))
     (reify Nimbus$Iface
       (^void submitTopology
         [this ^String storm-name ^String uploadedJarLocation ^String serializedConf ^StormTopology topology]
@@ -857,12 +865,34 @@
              (FileUtils/copyFile src-file (File. (master-stormjar-path stormroot)))
              ))
 
+(defn- file-older-than? [now seconds file]
+  (<= (+ (.lastModified file) (* seconds 1000)) now))
+
+(defn- is-jar? [f]
+  (-> f .getName (.endsWith ".jar")))
+
+(defmethod clean-inbox :distributed [conf dir-location seconds]
+  "Deletes jar files in dir older than seconds."
+  (let [now (Time/currentTimeMillis)
+        pred #(and (is-jar? %) (file-older-than? now seconds %))
+        files (filter pred (file-seq (File. dir-location)))]
+    (doseq [f files]
+      (if (.delete f)
+        (log-message "Cleaning inbox ... deleted: " (.getName f))
+        ;; This should never happen
+        (log-error "Cleaning inbox ... error deleting: " (.getName f))
+        ))))
+
 ;; local implementation
 
 (defmethod setup-jar :local [conf & args]
   nil
   )
 
+(defmethod clean-inbox :local [conf & args]
+  nil
+  )
+
 
 (defn -main []
   (launch-server! (read-storm-config)))
diff --git a/src/jvm/backtype/storm/Config.java b/src/jvm/backtype/storm/Config.java
@@ -105,6 +105,22 @@ public class Config extends HashMap<String, Object> {
      */
     public static String NIMBUS_MONITOR_FREQ_SECS = "nimbus.monitor.freq.secs";
 
+    /**
+     * How often nimbus should wake the cleanup thread to clean the inbox.
+     * @see NIMBUS_INBOX_JAR_EXPIRATION_SECS
+     */
+    public static String NIMBUS_CLEANUP_FREQ_SECS = "nimbus.cleanup.freq.secs";
+
+    /**
+     * The length of time a jar file lives in the inbox before being deleted by the cleanup thread.
+     *
+     * Probably keep this value greater than or equal to NIMBUS_CLEANUP_INBOX_JAR_EXPIRATION_SECS.
+     * Note that the time it takes to delete an inbox jar file is going to be somewhat more than
+     * NIMBUS_CLEANUP_INBOX_JAR_EXPIRATION_SECS (depending on how often NIMBUS_CLEANUP_FREQ_SECS
+     * is set to).
+     * @see NIMBUS_CLEANUP_FREQ_SECS
+     */
+    public static String NIMBUS_INBOX_JAR_EXPIRATION_SECS = "nimbus.inbox.jar.expiration.secs";
 
     /**
      * How long before a supervisor can go without heartbeating before nimbus considers it dead
diff --git a/test/clj/backtype/storm/nimbus_test.clj b/test/clj/backtype/storm/nimbus_test.clj
@@ -430,3 +430,37 @@
   ;; test that nimbus can die and restart without any problems
   )
 
+(deftest test-clean-inbox
+  "Tests that the inbox correctly cleans jar files."
+  (with-simulated-time
+   (with-local-tmp [dir-location]
+     (let [dir (File. dir-location)
+           mk-file (fn [name seconds-ago]
+                     (let [f (File. (str dir-location "/" name))
+                           t (- (Time/currentTimeMillis) (* seconds-ago 1000))]
+                       (FileUtils/touch f)
+                       (.setLastModified f t)))
+           assert-files-in-dir (fn [compare-file-names]
+                                 (let [file-names (map #(.getName %) (file-seq dir))]
+                                   (is (= (sort compare-file-names)
+                                          (sort (filter #(.endsWith % ".jar") file-names))
+                                          ))))]
+       ;; Make three files a.jar, b.jar, c.jar.
+       ;; a and b are older than c and should be deleted first.
+       (advance-time-secs! 100)
+       (doseq [fs [["a.jar" 20] ["b.jar" 20] ["c.jar" 0]]]
+         (apply mk-file fs))
+       (assert-files-in-dir ["a.jar" "b.jar" "c.jar"])
+       (nimbus/clean-inbox {STORM-CLUSTER-MODE "distributed"} dir-location 10)
+       (assert-files-in-dir ["c.jar"])
+       ;; Cleanit again, c.jar should stay
+       (advance-time-secs! 5)
+       (nimbus/clean-inbox {STORM-CLUSTER-MODE "distributed"} dir-location 10)
+       (assert-files-in-dir ["c.jar"])
+       ;; Advance time, clean again, c.jar should be deleted.
+       (advance-time-secs! 5)
+       (nimbus/clean-inbox {STORM-CLUSTER-MODE "distributed"} dir-location 10)
+       (assert-files-in-dir [])
+       )))
+
+  )
