mk-supervisor
(defserverfn mk-supervisor [conf shared-context ^ISupervisor isupervisor] (log-message "Starting Supervisor with conf " conf) (.prepare isupervisor conf (supervisor-isupervisor-dir conf)) ;;初始化supervisor-id,并存在localstate中(参考ISupervisor的实现)
(FileUtils/cleanDirectory (File. (supervisor-tmp-dir conf))) ;;清空本机的supervisor目录 (let [supervisor (supervisor-data conf shared-context isupervisor)
;;创建两个event-manager,用于在后台执行function [event-manager processes-event-manager :as managers] [(event/event-manager false) (event/event-manager false)] sync-processes (partial sync-processes supervisor) ;;partial sync-process ;;mk-synchronize-supervisor, mk-supervisor的主要工作,参考下面
synchronize-supervisor (mk-synchronize-supervisor supervisor sync-processes event-manager processes-event-manager) ;;定义生成supervisor hb的funciton
heartbeat-fn (fn [] (.supervisor-heartbeat! (:storm-cluster-state supervisor) (:supervisor-id supervisor) (SupervisorInfo. (current-time-secs) (:my-hostname supervisor) (:assignment-id supervisor) (keys @(:curr-assignment supervisor)) ;; used ports (.getMetadata isupervisor) (conf SUPERVISOR-SCHEDULER-META) ((:uptime supervisor)))))] ;;先调用heartbeat-fn发送一次supervisor的hb
;;接着使用schedule-recurring去定期调用heartbeat-fn更新hb
(heartbeat-fn)
;; should synchronize supervisor so it doesn't launch anything after being down (optimization)
(schedule-recurring (:timer supervisor)
0
(conf SUPERVISOR-HEARTBEAT-FREQUENCY-SECS)
heartbeat-fn))
mk-synchronize-supervisor
supervisor很简单, 主要管两件事,
当assignment发生变化时, 从nimbus同步topology的代码到本地
当assignment发生变化时, check workers状态, 保证被分配的work的状态都是valid
两个需求,
1. 当assignment发生变化时触发
怎样通过zookeeper的watcher实现这个反复触发机制, 参考Storm-源码分析- Storm中Zookeeper的使用
2. 因为比较耗时, 后台执行
创建两个event-manager, 分别用于后台执行mk-synchronize-supervisor和sync-processes
mk-synchronize-supervisor, 比较特别的是内部用了一个有名字的匿名函数this来封装这个函数体
刚开始看到非常诧异, 其实目的是为了可以在sync-callback中将这个函数add到event-manager里面去
即每次被调用, 都需要再一次把sync-callback注册到zk, 以保证下次可以被继续触发
(defn mk-synchronize-supervisor [supervisor sync-processes event-manager processes-event-manager] (fn this [] (let [conf (:conf supervisor) storm-cluster-state (:storm-cluster-state supervisor) ^ISupervisor isupervisor (:isupervisor supervisor) ^LocalState local-state (:local-state supervisor) ;;本地缓存数据库 sync-callback (fn [& ignored] (.add event-manager this)) ;;生成callback函数(后台执行mk-synchronize-supervisor) assignments-snapshot (assignments-snapshot storm-cluster-state sync-callback) ;;读取assignments,并注册callback,在zk->assignment发生变化时被触发 storm-code-map (read-storm-code-locations assignments-snapshot) ;;从哪儿下载topology code downloaded-storm-ids (set (read-downloaded-storm-ids conf)) ;;已经下载了哪些topology all-assignment (read-assignments ;;supervisor的port上被分配了哪些executors assignments-snapshot (:assignment-id supervisor)) ;;supervisor-id new-assignment (->> all-assignment ;;new=all,因为confirmAssigned没有具体实现,always返回true (filter-key #(.confirmAssigned isupervisor %))) assigned-storm-ids (assigned-storm-ids-from-port-assignments new-assignment) ;;supervisor上被分配的topology id集合 existing-assignment (.get local-state LS-LOCAL-ASSIGNMENTS)] ;;从local-state数据库里面读出当前保存的local assignments ;;下载新分配的topology代码 (doseq [[storm-id master-code-dir] storm-code-map] (when (and (not (downloaded-storm-ids storm-id)) (assigned-storm-ids storm-id)) (download-storm-code conf storm-id master-code-dir)))
(.put local-state ;;把new-assignment存到local-state数据库中 LS-LOCAL-ASSIGNMENTS new-assignment) (reset! (:curr-assignment supervisor) new-assignment) ;;把new-assignment cache到supervisor对象中
;;删除无用的topology code
;;remove any downloaded code that's no longer assigned or active (doseq [storm-id downloaded-storm-ids] (when-not (assigned-storm-ids storm-id) (log-message "Removing code for storm id " storm-id) (rmr (supervisor-stormdist-root conf storm-id)) ))
;;后台执行sync-processes
(.add processes-event-manager sync-processes)
)))
sync-processes
sync-processes用于管理workers, 比如处理不正常的worker或dead worker, 并创建新的workers
首先从本地读出workers的hb, 来判断work状况, shutdown所有状态非valid的workers
并为被assignment, 而worker状态非valid的slot, 创建新的worker
(defn sync-processes [supervisor]
(let [conf (:conf supervisor)
^LocalState local-state (:local-state supervisor)
assigned-executors (defaulted (.get local-state LS-LOCAL-ASSIGNMENTS) {})
now (current-time-secs)
allocated (read-allocated-workers supervisor assigned-executors now) ;;1.读取当前worker的状况
keepers (filter-val ;;找出状态为valid的worker
(fn [[state _]] (= state :valid))
allocated)
keep-ports (set (for [[id [_ hb]] keepers] (:port hb))) ;;keepers的ports集合
;;select-keys-pred(pred map), 对map中的key使用pred进行过滤
;;找出assigned-executors中executor的port, 哪些不属于keep-ports,
;;即找出新被assign的workers或那些虽被assign但状态不是valid的workers(dead或没有start)
;;这些executors需要从新分配到新的worker上去 reassign-executors (select-keys-pred (complement keep-ports) assigned-executors) new-worker-ids (into {} (for [port (keys reassign-executors)] ;;为reassign-executors的port产生新的worker-id [port (uuid)])) ] ;; 1. to kill are those in allocated that are dead or disallowed ;; 2. kill the ones that should be dead ;; - read pids, kill -9 and individually remove file ;; - rmr heartbeat dir, rmdir pid dir, rmdir id dir (catch exception and log) ;; 3. of the rest, figure out what assignments aren't yet satisfied ;; 4. generate new worker ids, write new "approved workers" to LS ;; 5. create local dir for worker id ;; 5. launch new workers (give worker-id, port, and supervisor-id) ;; 6. wait for workers launch (doseq [[id [state heartbeat]] allocated] (when (not= :valid state) ;;shutdown所有状态不是valid的worker (shutdown-worker supervisor id))) (doseq [id (vals new-worker-ids)] (local-mkdirs (worker-pids-root conf id))) ;;为新的worker创建目录, 并加到local-state的LS-APPROVED-WORKERS中 (.put local-state LS-APPROVED-WORKERS ;;更新的approved worker, 状态为valid的 + new workers (merge (select-keys (.get local-state LS-APPROVED-WORKERS) ;;现有approved worker中状态为valid (keys keepers)) (zipmap (vals new-worker-ids) (keys new-worker-ids)) ;;new workers )) (wait-for-workers-launch ;;2.wait-for-workers-launch conf (dofor [[port assignment] reassign-executors] (let [id (new-worker-ids port)] (launch-worker supervisor (:storm-id assignment) port id) id))) ))
1. read-allocated-workers
(defn read-allocated-workers "Returns map from worker id to worker heartbeat. if the heartbeat is nil, then the worker is dead (timed out or never wrote heartbeat)" [supervisor assigned-executors now] (let [conf (:conf supervisor) ^LocalState local-state (:local-state supervisor)
;从local-state中读出每个worker的hb, 当然每个worker进程会不断的更新本地hb id->heartbeat (read-worker-heartbeats conf)
approved-ids (set (keys (.get local-state LS-APPROVED-WORKERS)))] ;;从local-state读出approved的worker
(into
{}
(dofor [[id hb] id->heartbeat] ;;根据hb来判断worker的当前状态
(let [state (cond
(or (not (contains? approved-ids id))
(not (matches-an-assignment? hb assigned-executors)))
:disallowed ;;不被允许
(not hb)
:not-started ;;无hb,没有start
(> (- now (:time-secs hb))
(conf SUPERVISOR-WORKER-TIMEOUT-SECS))
:timed-out ;;超时,dead
true
:valid)]
(log-debug "Worker " id " is " state ": " (pr-str hb) " at supervisor time-secs " now)
[id [state hb]] ;;返回每个worker的当前state和hb
))
)))
2. wait-for-workers-launch
对reassign-executors中的每个new_work_id调用launch-worker
最终调用wait-for-workers-launch, 等待worder被成功launch
逻辑也比较简单, check hb, 如果没有就不停的sleep, 至到超时, 打印failed to start
(defn- wait-for-worker-launch [conf id start-time] (let [state (worker-state conf id)] (loop [] (let [hb (.get state LS-WORKER-HEARTBEAT)] (when (and (not hb) (< (- (current-time-secs) start-time) (conf SUPERVISOR-WORKER-START-TIMEOUT-SECS) )) (log-message id " still hasn't started") (Time/sleep 500) (recur) ))) (when-not (.get state LS-WORKER-HEARTBEAT) (log-message "Worker " id " failed to start") ))) (defn- wait-for-workers-launch [conf ids] (let [start-time (current-time-secs)] (doseq [id ids] (wait-for-worker-launch conf id start-time)) ))
本文章摘自博客园，原文发布日期：2013-06-28
