STORM-765: Thrift serialization for local state.

Author: Robert (Bobby) Evans

storm-core/src/clj/backtype/storm/daemon/common.clj

@@ -58,19 +58,6 @@
 (defprotocol DaemonCommon
   (waiting? [this]))
 
-(def LS-WORKER-HEARTBEAT "worker-heartbeat")
-
-;; LocalState constants
-(def LS-ID "supervisor-id")
-(def LS-LOCAL-ASSIGNMENTS "local-assignments")
-(def LS-APPROVED-WORKERS "approved-workers")
-
-(defn mk-local-worker-heartbeat [time-secs storm-id executors port]
-  {:time-secs time-secs
-   :storm-id storm-id
-   :executors executors
-   :port port})
-
 (defrecord ExecutorStats [^long processed
                           ^long acked
                           ^long emitted

storm-core/src/clj/backtype/storm/daemon/supervisor.clj

@@ -24,7 +24,7 @@
            [java.net URI]
            [org.apache.commons.io FileUtils]
            [java.io File])
-  (:use [backtype.storm config util log timer])
+  (:use [backtype.storm config util log timer local-state])
   (:use [backtype.storm.daemon common])
   (:require [backtype.storm.daemon [worker :as worker]]
             [backtype.storm [process-simulator :as psim] [cluster :as cluster] [event :as event]]
@@ -38,9 +38,6 @@
 (defmulti download-storm-code cluster-mode)
 (defmulti launch-worker (fn [supervisor & _] (cluster-mode (:conf supervisor))))
 
-;; used as part of a map from port to this
-(defrecord LocalAssignment [storm-id executors])
-
 (defprotocol SupervisorDaemon
   (get-id [this])
   (get-conf [this])
@@ -76,7 +73,7 @@
     (into {} (for [[port executors] port-executors]
                ;; need to cast to int b/c it might be a long (due to how yaml parses things)
                ;; doall is to avoid serialization/deserialization problems with lazy seqs
-               [(Integer. port) (LocalAssignment. storm-id (doall executors))]
+               [(Integer. port) (mk-local-assignment storm-id (doall executors))]
                ))))
 
 (defn- read-assignments
@@ -104,8 +101,8 @@
 (defn read-worker-heartbeat [conf id]
   (let [local-state (worker-state conf id)]
     (try
-      (.get local-state LS-WORKER-HEARTBEAT)
-      (catch IOException e
+      (ls-worker-heartbeat local-state)
+      (catch Exception e
         (log-warn e "Failed to read local heartbeat for workerId : " id ",Ignoring exception.")
         nil))))
 
@@ -148,7 +145,7 @@
   (let [conf (:conf supervisor)
         ^LocalState local-state (:local-state supervisor)
         id->heartbeat (read-worker-heartbeats conf)
-        approved-ids (set (keys (.get local-state LS-APPROVED-WORKERS)))]
+        approved-ids (set (keys (ls-approved-workers local-state)))]
     (into
      {}
      (dofor [[id hb] id->heartbeat]
@@ -174,7 +171,7 @@
 (defn- wait-for-worker-launch [conf id start-time]
   (let [state (worker-state conf id)]
     (loop []
-      (let [hb (.get state LS-WORKER-HEARTBEAT)]
+      (let [hb (ls-worker-heartbeat state)]
         (when (and
                (not hb)
                (<
@@ -185,7 +182,7 @@
           (Time/sleep 500)
           (recur)
           )))
-    (when-not (.get state LS-WORKER-HEARTBEAT)
+    (when-not (ls-worker-heartbeat state)
       (log-message "Worker " id " failed to start")
       )))
 
@@ -320,7 +317,7 @@
         download-lock (:download-lock supervisor)
         ^LocalState local-state (:local-state supervisor)
         storm-cluster-state (:storm-cluster-state supervisor)
-        assigned-executors (defaulted (.get local-state LS-LOCAL-ASSIGNMENTS) {})
+        assigned-executors (defaulted (ls-local-assignments local-state) {})
         now (current-time-secs)
         allocated (read-allocated-workers supervisor assigned-executors now)
         keepers (filter-val
@@ -358,9 +355,9 @@
     (doseq [id (vals new-worker-ids)]
       (local-mkdirs (worker-pids-root conf id))
       (local-mkdirs (worker-heartbeats-root conf id)))
-    (.put local-state LS-APPROVED-WORKERS
+    (ls-approved-workers! local-state
           (merge
-           (select-keys (.get local-state LS-APPROVED-WORKERS)
+           (select-keys (ls-approved-workers local-state)
                         (keys keepers))
            (zipmap (vals new-worker-ids) (keys new-worker-ids))
            ))
@@ -416,7 +413,7 @@
 (defn shutdown-disallowed-workers [supervisor]
   (let [conf (:conf supervisor)
         ^LocalState local-state (:local-state supervisor)
-        assigned-executors (defaulted (.get local-state LS-LOCAL-ASSIGNMENTS) {})
+        assigned-executors (defaulted (ls-local-assignments local-state) {})
         now (current-time-secs)
         allocated (read-allocated-workers supervisor assigned-executors now)
         disallowed (keys (filter-val
@@ -442,7 +439,7 @@
                                                                   assignment-versions)
           storm-code-map (read-storm-code-locations assignments-snapshot)
           downloaded-storm-ids (set (read-downloaded-storm-ids conf))
-          existing-assignment (.get local-state LS-LOCAL-ASSIGNMENTS)
+          existing-assignment (ls-local-assignments local-state)
           all-assignment (read-assignments assignments-snapshot
                                            (:assignment-id supervisor)
                                            existing-assignment
@@ -472,8 +469,7 @@
                                 (set (keys new-assignment)))]
         (.killedWorker isupervisor (int p)))
       (.assigned isupervisor (keys new-assignment))
-      (.put local-state
-            LS-LOCAL-ASSIGNMENTS
+      (ls-local-assignments! local-state
             new-assignment)
       (reset! (:assignment-versions supervisor) versions)
       (reset! (:curr-assignment supervisor) new-assignment)
@@ -780,10 +776,10 @@
       (prepare [this conf local-dir]
         (reset! conf-atom conf)
         (let [state (LocalState. local-dir)
-              curr-id (if-let [id (.get state LS-ID)]
+              curr-id (if-let [id (ls-supervisor-id state)]
                         id
                         (generate-supervisor-id))]
-          (.put state LS-ID curr-id)
+          (ls-supervisor-id! state curr-id)
           (reset! id-atom curr-id))
         )
       (confirmAssigned [this port]

storm-core/src/clj/backtype/storm/daemon/worker.clj

@@ -15,7 +15,7 @@
 ;; limitations under the License.
 (ns backtype.storm.daemon.worker
   (:use [backtype.storm.daemon common])
-  (:use [backtype.storm config log util timer])
+  (:use [backtype.storm config log util timer local-state])
   (:require [backtype.storm.daemon [executor :as executor]])
   (:require [backtype.storm [disruptor :as disruptor] [cluster :as cluster]])
   (:require [clojure.set :as set])
@@ -68,19 +68,9 @@
 
 (defn do-heartbeat [worker]
   (let [conf (:conf worker)
-        hb (mk-local-worker-heartbeat
-             (current-time-secs)
-             (:storm-id worker)
-             (:executors worker)
-             (:port worker))
         state (worker-state conf (:worker-id worker))]
-    (log-debug "Doing heartbeat " (pr-str hb))
     ;; do the local-file-system heartbeat.
-    (.put state
-        LS-WORKER-HEARTBEAT
-        hb
-        false
-        )
+    (ls-worker-heartbeat! state (current-time-secs) (:storm-id worker) (:executors worker) (:port worker))
     (.cleanup state 60) ; this is just in case supervisor is down so that disk doesn't fill up.
                          ; it shouldn't take supervisor 120 seconds between listing dir and reading it
 

storm-core/src/clj/backtype/storm/local_state.clj

@@ -0,0 +1,99 @@
+;; Licensed to the Apache Software Foundation (ASF) under one
+;; or more contributor license agreements.  See the NOTICE file
+;; distributed with this work for additional information
+;; regarding copyright ownership.  The ASF licenses this file
+;; to you under the Apache License, Version 2.0 (the
+;; "License"); you may not use this file except in compliance
+;; with the License.  You may obtain a copy of the License at
+;;
+;; http://www.apache.org/licenses/LICENSE-2.0
+;;
+;; Unless required by applicable law or agreed to in writing, software
+;; distributed under the License is distributed on an "AS IS" BASIS,
+;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+;; See the License for the specific language governing permissions and
+;; limitations under the License.
+(ns backtype.storm.local-state
+  (:use [backtype.storm log util])
+  (:import [backtype.storm.generated StormTopology
+            InvalidTopologyException GlobalStreamId
+            LSSupervisorId LSApprovedWorkers
+            LSSupervisorAssignments LocalAssignment
+            ExecutorInfo LSWorkerHeartbeat])
+  (:import [backtype.storm.utils LocalState]))
+
+(def LS-WORKER-HEARTBEAT "worker-heartbeat")
+(def LS-ID "supervisor-id")
+(def LS-LOCAL-ASSIGNMENTS "local-assignments")
+(def LS-APPROVED-WORKERS "approved-workers")
+
+(defn ls-supervisor-id!
+  [^LocalState local-state ^String id]
+    (.put local-state LS-ID (LSSupervisorId. id)))
+
+(defn ls-supervisor-id
+  [^LocalState local-state]
+  (if-let [super-id (.get local-state LS-ID)]
+    (.get_supervisor_id super-id)))
+
+(defn ls-approved-workers!
+  [^LocalState local-state workers]
+    (.put local-state LS-APPROVED-WORKERS (LSApprovedWorkers. workers)))
+
+(defn ls-approved-workers
+  [^LocalState local-state]
+  (if-let [tmp (.get local-state LS-APPROVED-WORKERS)]
+    (into {} (.get_approved_workers tmp))))
+
+(defn ->ExecutorInfo
+  [[low high]] (ExecutorInfo. low high))
+
+(defn ->ExecutorInfo-list
+  [executors]
+  (map ->ExecutorInfo executors))
+
+(defn ->executor-list
+  [executors]
+  (into [] 
+    (for [exec-info executors] 
+      [(.get_task_start exec-info) (.get_task_end exec-info)])))
+
+(defn ->LocalAssignment
+  [{storm-id :storm-id executors :executors}]
+  (LocalAssignment. storm-id (->ExecutorInfo-list executors)))
+
+(defn mk-local-assignment
+  [storm-id executors]
+  {:storm-id storm-id :executors executors})
+
+(defn ->local-assignment
+  [^LocalAssignment thrift-local-assignment]
+    (mk-local-assignment
+      (.get_topology_id thrift-local-assignment)
+      (->executor-list (.get_executors thrift-local-assignment))))
+
+(defn ls-local-assignments!
+  [^LocalState local-state assignments]
+    (let [local-assignment-map (map-val ->LocalAssignment assignments)]
+    (.put local-state LS-LOCAL-ASSIGNMENTS 
+          (LSSupervisorAssignments. local-assignment-map))))
+
+(defn ls-local-assignments
+  [^LocalState local-state]
+    (if-let [thrift-local-assignments (.get local-state LS-LOCAL-ASSIGNMENTS)]
+      (map-val
+        ->local-assignment
+        (.get_assignments thrift-local-assignments))))
+
+(defn ls-worker-heartbeat!
+  [^LocalState local-state time-secs storm-id executors port]
+  (.put local-state LS-WORKER-HEARTBEAT (LSWorkerHeartbeat. time-secs storm-id (->ExecutorInfo-list executors) port) false))
+
+(defn ls-worker-heartbeat 
+  [^LocalState local-state]
+  (if-let [worker-hb (.get local-state LS-WORKER-HEARTBEAT)]
+    {:time-secs (.get_time_secs worker-hb)
+     :storm-id (.get_topology_id worker-hb)
+     :executors (->executor-list (.get_executors worker-hb))
+     :port (.get_port worker-hb)}))
+

storm-core/src/clj/backtype/storm/testing.clj

@@ -42,7 +42,7 @@
   (:require [backtype.storm [zookeeper :as zk]])
   (:require [backtype.storm.messaging.loader :as msg-loader])
   (:require [backtype.storm.daemon.acker :as acker])
-  (:use [backtype.storm cluster util thrift config log]))
+  (:use [backtype.storm cluster util thrift config log local-state]))
 
 (defn feeder-spout
   [fields]
@@ -302,13 +302,13 @@
 (defn find-worker-id
   [supervisor-conf port]
   (let [supervisor-state (supervisor-state supervisor-conf)
-        worker->port (.get supervisor-state common/LS-APPROVED-WORKERS)]
+        worker->port (ls-approved-workers supervisor-state)]
     (first ((reverse-map worker->port) port))))
 
 (defn find-worker-port
   [supervisor-conf worker-id]
   (let [supervisor-state (supervisor-state supervisor-conf)
-        worker->port (.get supervisor-state common/LS-APPROVED-WORKERS)]
+        worker->port (ls-approved-workers supervisor-state)]
     (worker->port worker-id)))
 
 (defn mk-capture-shutdown-fn