You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by kn...@apache.org on 2016/10/11 02:04:36 UTC
[50/55] [abbrv] [partial] storm git commit: Ported successfully.
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/cluster.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/cluster.clj b/storm-core/src/clj/backtype/storm/cluster.clj
deleted file mode 100644
index 914fa5b..0000000
--- a/storm-core/src/clj/backtype/storm/cluster.clj
+++ /dev/null
@@ -1,708 +0,0 @@
-;; 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 org.apache.storm.cluster
- (:import [org.apache.zookeeper.data Stat ACL Id]
- [org.apache.storm.generated SupervisorInfo Assignment StormBase ClusterWorkerHeartbeat ErrorInfo Credentials NimbusSummary
- LogConfig ProfileAction ProfileRequest NodeInfo]
- [java.io Serializable])
- (:import [org.apache.zookeeper KeeperException KeeperException$NoNodeException ZooDefs ZooDefs$Ids ZooDefs$Perms])
- (:import [org.apache.curator.framework CuratorFramework])
- (:import [org.apache.storm.utils Utils])
- (:import [org.apache.storm.cluster ClusterState ClusterStateContext ClusterStateListener ConnectionState])
- (:import [java.security MessageDigest])
- (:import [org.apache.zookeeper.server.auth DigestAuthenticationProvider])
- (:import [org.apache.storm.nimbus NimbusInfo])
- (:use [org.apache.storm util log config converter])
- (:require [org.apache.storm [zookeeper :as zk]])
- (:require [org.apache.storm.daemon [common :as common]]))
-
-(defn mk-topo-only-acls
- [topo-conf]
- (let [payload (.get topo-conf STORM-ZOOKEEPER-TOPOLOGY-AUTH-PAYLOAD)]
- (when (Utils/isZkAuthenticationConfiguredTopology topo-conf)
- [(first ZooDefs$Ids/CREATOR_ALL_ACL)
- (ACL. ZooDefs$Perms/READ (Id. "digest" (DigestAuthenticationProvider/generateDigest payload)))])))
-
-(defnk mk-distributed-cluster-state
- [conf :auth-conf nil :acls nil :context (ClusterStateContext.)]
- (let [clazz (Class/forName (or (conf STORM-CLUSTER-STATE-STORE)
- "org.apache.storm.cluster_state.zookeeper_state_factory"))
- state-instance (.newInstance clazz)]
- (log-debug "Creating cluster state: " (.toString clazz))
- (or (.mkState state-instance conf auth-conf acls context)
- nil)))
-
-(defprotocol StormClusterState
- (assignments [this callback])
- (assignment-info [this storm-id callback])
- (assignment-info-with-version [this storm-id callback])
- (assignment-version [this storm-id callback])
- ;returns key information under /storm/blobstore/key
- (blobstore-info [this blob-key])
- ;returns list of nimbus summaries stored under /stormroot/nimbuses/<nimbus-ids> -> <data>
- (nimbuses [this])
- ;adds the NimbusSummary to /stormroot/nimbuses/nimbus-id
- (add-nimbus-host! [this nimbus-id nimbus-summary])
-
- (active-storms [this])
- (storm-base [this storm-id callback])
- (get-worker-heartbeat [this storm-id node port])
- (get-worker-profile-requests [this storm-id nodeinfo thrift?])
- (get-topology-profile-requests [this storm-id thrift?])
- (set-worker-profile-request [this storm-id profile-request])
- (delete-topology-profile-requests [this storm-id profile-request])
- (executor-beats [this storm-id executor->node+port])
- (supervisors [this callback])
- (supervisor-info [this supervisor-id]) ;; returns nil if doesn't exist
- (setup-heartbeats! [this storm-id])
- (teardown-heartbeats! [this storm-id])
- (teardown-topology-errors! [this storm-id])
- (heartbeat-storms [this])
- (error-topologies [this])
- (backpressure-topologies [this])
- (set-topology-log-config! [this storm-id log-config])
- (topology-log-config [this storm-id cb])
- (worker-heartbeat! [this storm-id node port info])
- (remove-worker-heartbeat! [this storm-id node port])
- (supervisor-heartbeat! [this supervisor-id info])
- (worker-backpressure! [this storm-id node port info])
- (topology-backpressure [this storm-id callback])
- (setup-backpressure! [this storm-id])
- (remove-backpressure! [this storm-id])
- (remove-worker-backpressure! [this storm-id node port])
- (activate-storm! [this storm-id storm-base])
- (update-storm! [this storm-id new-elems])
- (remove-storm-base! [this storm-id])
- (set-assignment! [this storm-id info])
- ;; sets up information related to key consisting of nimbus
- ;; host:port and version info of the blob
- (setup-blobstore! [this key nimbusInfo versionInfo])
- (active-keys [this])
- (blobstore [this callback])
- (remove-storm! [this storm-id])
- (remove-blobstore-key! [this blob-key])
- (remove-key-version! [this blob-key])
- (report-error [this storm-id component-id node port error])
- (errors [this storm-id component-id])
- (last-error [this storm-id component-id])
- (set-credentials! [this storm-id creds topo-conf])
- (credentials [this storm-id callback])
- (disconnect [this]))
-
-(def ASSIGNMENTS-ROOT "assignments")
-(def CODE-ROOT "code")
-(def STORMS-ROOT "storms")
-(def SUPERVISORS-ROOT "supervisors")
-(def WORKERBEATS-ROOT "workerbeats")
-(def BACKPRESSURE-ROOT "backpressure")
-(def ERRORS-ROOT "errors")
-(def BLOBSTORE-ROOT "blobstore")
-; Stores the latest update sequence for a blob
-(def BLOBSTORE-MAX-KEY-SEQUENCE-NUMBER-ROOT "blobstoremaxkeysequencenumber")
-(def NIMBUSES-ROOT "nimbuses")
-(def CREDENTIALS-ROOT "credentials")
-(def LOGCONFIG-ROOT "logconfigs")
-(def PROFILERCONFIG-ROOT "profilerconfigs")
-
-(def ASSIGNMENTS-SUBTREE (str "/" ASSIGNMENTS-ROOT))
-(def STORMS-SUBTREE (str "/" STORMS-ROOT))
-(def SUPERVISORS-SUBTREE (str "/" SUPERVISORS-ROOT))
-(def WORKERBEATS-SUBTREE (str "/" WORKERBEATS-ROOT))
-(def BACKPRESSURE-SUBTREE (str "/" BACKPRESSURE-ROOT))
-(def ERRORS-SUBTREE (str "/" ERRORS-ROOT))
-;; Blobstore subtree /storm/blobstore
-(def BLOBSTORE-SUBTREE (str "/" BLOBSTORE-ROOT))
-(def BLOBSTORE-MAX-KEY-SEQUENCE-NUMBER-SUBTREE (str "/" BLOBSTORE-MAX-KEY-SEQUENCE-NUMBER-ROOT))
-(def NIMBUSES-SUBTREE (str "/" NIMBUSES-ROOT))
-(def CREDENTIALS-SUBTREE (str "/" CREDENTIALS-ROOT))
-(def LOGCONFIG-SUBTREE (str "/" LOGCONFIG-ROOT))
-(def PROFILERCONFIG-SUBTREE (str "/" PROFILERCONFIG-ROOT))
-
-(defn supervisor-path
- [id]
- (str SUPERVISORS-SUBTREE "/" id))
-
-(defn assignment-path
- [id]
- (str ASSIGNMENTS-SUBTREE "/" id))
-
-(defn blobstore-path
- [key]
- (str BLOBSTORE-SUBTREE "/" key))
-
-(defn blobstore-max-key-sequence-number-path
- [key]
- (str BLOBSTORE-MAX-KEY-SEQUENCE-NUMBER-SUBTREE "/" key))
-
-(defn nimbus-path
- [id]
- (str NIMBUSES-SUBTREE "/" id))
-
-(defn storm-path
- [id]
- (str STORMS-SUBTREE "/" id))
-
-(defn workerbeat-storm-root
- [storm-id]
- (str WORKERBEATS-SUBTREE "/" storm-id))
-
-(defn workerbeat-path
- [storm-id node port]
- (str (workerbeat-storm-root storm-id) "/" node "-" port))
-
-(defn backpressure-storm-root
- [storm-id]
- (str BACKPRESSURE-SUBTREE "/" storm-id))
-
-(defn backpressure-path
- [storm-id node port]
- (str (backpressure-storm-root storm-id) "/" node "-" port))
-
-(defn error-storm-root
- [storm-id]
- (str ERRORS-SUBTREE "/" storm-id))
-
-(defn error-path
- [storm-id component-id]
- (str (error-storm-root storm-id) "/" (url-encode component-id)))
-
-(def last-error-path-seg "last-error")
-
-(defn last-error-path
- [storm-id component-id]
- (str (error-storm-root storm-id)
- "/"
- (url-encode component-id)
- "-"
- last-error-path-seg))
-
-(defn credentials-path
- [storm-id]
- (str CREDENTIALS-SUBTREE "/" storm-id))
-
-(defn log-config-path
- [storm-id]
- (str LOGCONFIG-SUBTREE "/" storm-id))
-
-(defn profiler-config-path
- ([storm-id]
- (str PROFILERCONFIG-SUBTREE "/" storm-id))
- ([storm-id host port request-type]
- (str (profiler-config-path storm-id) "/" host "_" port "_" request-type)))
-
-(defn- issue-callback!
- [cb-atom]
- (let [cb @cb-atom]
- (reset! cb-atom nil)
- (when cb
- (cb))))
-
-(defn- issue-map-callback!
- [cb-atom id]
- (let [cb (@cb-atom id)]
- (swap! cb-atom dissoc id)
- (when cb
- (cb id))))
-
-(defn- maybe-deserialize
- [ser clazz]
- (when ser
- (Utils/deserialize ser clazz)))
-
-(defrecord TaskError [error time-secs host port])
-
-(defn- parse-error-path
- [^String p]
- (Long/parseLong (.substring p 1)))
-
-(defn convert-executor-beats
- "Ensures that we only return heartbeats for executors assigned to
- this worker."
- [executors worker-hb]
- (let [executor-stats (:executor-stats worker-hb)]
- (->> executors
- (map (fn [t]
- (if (contains? executor-stats t)
- {t {:time-secs (:time-secs worker-hb)
- :uptime (:uptime worker-hb)
- :stats (get executor-stats t)}})))
- (into {}))))
-
-;; Watches should be used for optimization. When ZK is reconnecting, they're not guaranteed to be called.
-(defnk mk-storm-cluster-state
- [cluster-state-spec :acls nil :context (ClusterStateContext.)]
- (let [[solo? cluster-state] (if (instance? ClusterState cluster-state-spec)
- [false cluster-state-spec]
- [true (mk-distributed-cluster-state cluster-state-spec :auth-conf cluster-state-spec :acls acls :context context)])
- assignment-info-callback (atom {})
- assignment-info-with-version-callback (atom {})
- assignment-version-callback (atom {})
- supervisors-callback (atom nil)
- backpressure-callback (atom {}) ;; we want to reigister a topo directory getChildren callback for all workers of this dir
- assignments-callback (atom nil)
- storm-base-callback (atom {})
- blobstore-callback (atom nil)
- credentials-callback (atom {})
- log-config-callback (atom {})
- state-id (.register
- cluster-state
- (fn [type path]
- (let [[subtree & args] (tokenize-path path)]
- (condp = subtree
- ASSIGNMENTS-ROOT (if (empty? args)
- (issue-callback! assignments-callback)
- (do
- (issue-map-callback! assignment-info-callback (first args))
- (issue-map-callback! assignment-version-callback (first args))
- (issue-map-callback! assignment-info-with-version-callback (first args))))
- SUPERVISORS-ROOT (issue-callback! supervisors-callback)
- BLOBSTORE-ROOT (issue-callback! blobstore-callback) ;; callback register for blobstore
- STORMS-ROOT (issue-map-callback! storm-base-callback (first args))
- CREDENTIALS-ROOT (issue-map-callback! credentials-callback (first args))
- LOGCONFIG-ROOT (issue-map-callback! log-config-callback (first args))
- BACKPRESSURE-ROOT (issue-map-callback! backpressure-callback (first args))
- ;; this should never happen
- (exit-process! 30 "Unknown callback for subtree " subtree args)))))]
- (doseq [p [ASSIGNMENTS-SUBTREE STORMS-SUBTREE SUPERVISORS-SUBTREE WORKERBEATS-SUBTREE ERRORS-SUBTREE BLOBSTORE-SUBTREE NIMBUSES-SUBTREE
- LOGCONFIG-SUBTREE BACKPRESSURE-SUBTREE]]
- (.mkdirs cluster-state p acls))
- (reify
- StormClusterState
-
- (assignments
- [this callback]
- (when callback
- (reset! assignments-callback callback))
- (.get_children cluster-state ASSIGNMENTS-SUBTREE (not-nil? callback)))
-
- (assignment-info
- [this storm-id callback]
- (when callback
- (swap! assignment-info-callback assoc storm-id callback))
- (clojurify-assignment (maybe-deserialize (.get_data cluster-state (assignment-path storm-id) (not-nil? callback)) Assignment)))
-
- (assignment-info-with-version
- [this storm-id callback]
- (when callback
- (swap! assignment-info-with-version-callback assoc storm-id callback))
- (let [{data :data version :version}
- (.get_data_with_version cluster-state (assignment-path storm-id) (not-nil? callback))]
- {:data (clojurify-assignment (maybe-deserialize data Assignment))
- :version version}))
-
- (assignment-version
- [this storm-id callback]
- (when callback
- (swap! assignment-version-callback assoc storm-id callback))
- (.get_version cluster-state (assignment-path storm-id) (not-nil? callback)))
-
- ;; blobstore state
- (blobstore
- [this callback]
- (when callback
- (reset! blobstore-callback callback))
- (.sync_path cluster-state BLOBSTORE-SUBTREE)
- (.get_children cluster-state BLOBSTORE-SUBTREE (not-nil? callback)))
-
- (nimbuses
- [this]
- (map #(maybe-deserialize (.get_data cluster-state (nimbus-path %1) false) NimbusSummary)
- (.get_children cluster-state NIMBUSES-SUBTREE false)))
-
- (add-nimbus-host!
- [this nimbus-id nimbus-summary]
- ;explicit delete for ephmeral node to ensure this session creates the entry.
- (.delete_node cluster-state (nimbus-path nimbus-id))
-
- (.add_listener cluster-state (reify ClusterStateListener
- (^void stateChanged[this ^ConnectionState newState]
- (log-message "Connection state listener invoked, zookeeper connection state has changed to " newState)
- (if (.equals newState ConnectionState/RECONNECTED)
- (do
- (log-message "Connection state has changed to reconnected so setting nimbuses entry one more time")
- ;explicit delete for ephmeral node to ensure this session creates the entry.
- (.delete_node cluster-state (nimbus-path nimbus-id))
- (.set_ephemeral_node cluster-state (nimbus-path nimbus-id) (Utils/serialize nimbus-summary) acls))))))
-
- (.set_ephemeral_node cluster-state (nimbus-path nimbus-id) (Utils/serialize nimbus-summary) acls))
-
- (setup-blobstore!
- [this key nimbusInfo versionInfo]
- (let [path (str (blobstore-path key) "/" (.toHostPortString nimbusInfo) "-" versionInfo)]
- (log-message "setup-path" path)
- (.mkdirs cluster-state (blobstore-path key) acls)
- ;we delete the node first to ensure the node gets created as part of this session only.
- (.delete_node_blobstore cluster-state (str (blobstore-path key)) (.toHostPortString nimbusInfo))
- (.set_ephemeral_node cluster-state path nil acls)))
-
- (blobstore-info
- [this blob-key]
- (let [path (blobstore-path blob-key)]
- (.sync_path cluster-state path)
- (.get_children cluster-state path false)))
-
- (active-storms
- [this]
- (.get_children cluster-state STORMS-SUBTREE false))
-
- (active-keys
- [this]
- (.get_children cluster-state BLOBSTORE-SUBTREE false))
-
- (heartbeat-storms
- [this]
- (.get_worker_hb_children cluster-state WORKERBEATS-SUBTREE false))
-
- (error-topologies
- [this]
- (.get_children cluster-state ERRORS-SUBTREE false))
-
- (backpressure-topologies
- [this]
- (.get_children cluster-state BACKPRESSURE-SUBTREE false))
-
- (get-worker-heartbeat
- [this storm-id node port]
- (let [worker-hb (.get_worker_hb cluster-state (workerbeat-path storm-id node port) false)]
- (if worker-hb
- (-> worker-hb
- (maybe-deserialize ClusterWorkerHeartbeat)
- clojurify-zk-worker-hb))))
-
- (executor-beats
- [this storm-id executor->node+port]
- ;; need to take executor->node+port in explicitly so that we don't run into a situation where a
- ;; long dead worker with a skewed clock overrides all the timestamps. By only checking heartbeats
- ;; with an assigned node+port, and only reading executors from that heartbeat that are actually assigned,
- ;; we avoid situations like that
- (let [node+port->executors (reverse-map executor->node+port)
- all-heartbeats (for [[[node port] executors] node+port->executors]
- (->> (get-worker-heartbeat this storm-id node port)
- (convert-executor-beats executors)
- ))]
- (apply merge all-heartbeats)))
-
- (supervisors
- [this callback]
- (when callback
- (reset! supervisors-callback callback))
- (.get_children cluster-state SUPERVISORS-SUBTREE (not-nil? callback)))
-
- (supervisor-info
- [this supervisor-id]
- (clojurify-supervisor-info (maybe-deserialize (.get_data cluster-state (supervisor-path supervisor-id) false) SupervisorInfo)))
-
- (topology-log-config
- [this storm-id cb]
- (when cb
- (swap! log-config-callback assoc storm-id cb))
- (maybe-deserialize (.get_data cluster-state (log-config-path storm-id) (not-nil? cb)) LogConfig))
-
- (set-topology-log-config!
- [this storm-id log-config]
- (.set_data cluster-state (log-config-path storm-id) (Utils/serialize log-config) acls))
-
- (set-worker-profile-request
- [this storm-id profile-request]
- (let [request-type (.get_action profile-request)
- host (.get_node (.get_nodeInfo profile-request))
- port (first (.get_port (.get_nodeInfo profile-request)))]
- (.set_data cluster-state
- (profiler-config-path storm-id host port request-type)
- (Utils/serialize profile-request)
- acls)))
-
- (get-topology-profile-requests
- [this storm-id thrift?]
- (let [path (profiler-config-path storm-id)
- requests (if (.node_exists cluster-state path false)
- (dofor [c (.get_children cluster-state path false)]
- (let [raw (.get_data cluster-state (str path "/" c) false)
- request (maybe-deserialize raw ProfileRequest)]
- (if thrift?
- request
- (clojurify-profile-request request)))))]
- requests))
-
- (delete-topology-profile-requests
- [this storm-id profile-request]
- (let [profile-request-inst (thriftify-profile-request profile-request)
- action (:action profile-request)
- host (:host profile-request)
- port (:port profile-request)]
- (.delete_node cluster-state
- (profiler-config-path storm-id host port action))))
-
- (get-worker-profile-requests
- [this storm-id node-info thrift?]
- (let [host (:host node-info)
- port (:port node-info)
- profile-requests (get-topology-profile-requests this storm-id thrift?)]
- (if thrift?
- (filter #(and (= host (.get_node (.get_nodeInfo %))) (= port (first (.get_port (.get_nodeInfo %)))))
- profile-requests)
- (filter #(and (= host (:host %)) (= port (:port %)))
- profile-requests))))
-
- (worker-heartbeat!
- [this storm-id node port info]
- (let [thrift-worker-hb (thriftify-zk-worker-hb info)]
- (if thrift-worker-hb
- (.set_worker_hb cluster-state (workerbeat-path storm-id node port) (Utils/serialize thrift-worker-hb) acls))))
-
- (remove-worker-heartbeat!
- [this storm-id node port]
- (.delete_worker_hb cluster-state (workerbeat-path storm-id node port)))
-
- (setup-heartbeats!
- [this storm-id]
- (.mkdirs cluster-state (workerbeat-storm-root storm-id) acls))
-
- (teardown-heartbeats!
- [this storm-id]
- (try-cause
- (.delete_worker_hb cluster-state (workerbeat-storm-root storm-id))
- (catch KeeperException e
- (log-warn-error e "Could not teardown heartbeats for " storm-id))))
-
- (worker-backpressure!
- [this storm-id node port on?]
- "if znode exists and to be not on?, delete; if exists and on?, do nothing;
- if not exists and to be on?, create; if not exists and not on?, do nothing"
- (let [path (backpressure-path storm-id node port)
- existed (.node_exists cluster-state path false)]
- (if existed
- (if (not on?)
- (.delete_node cluster-state path)) ;; delete the znode since the worker is not congested
- (if on?
- (.set_ephemeral_node cluster-state path nil acls))))) ;; create the znode since worker is congested
-
- (topology-backpressure
- [this storm-id callback]
- "if the backpresure/storm-id dir is not empty, this topology has throttle-on, otherwise throttle-off.
- The backpressure/storm-id dir may not exist if nimbus has shutdown the topology"
- (when callback
- (swap! backpressure-callback assoc storm-id callback))
- (let [path (backpressure-storm-root storm-id)
- children (if (.node_exists cluster-state path false)
- (.get_children cluster-state path (not-nil? callback))) ]
- (> (count children) 0)))
-
- (setup-backpressure!
- [this storm-id]
- (.mkdirs cluster-state (backpressure-storm-root storm-id) acls))
-
- (remove-backpressure!
- [this storm-id]
- (.delete_node cluster-state (backpressure-storm-root storm-id)))
-
- (remove-worker-backpressure!
- [this storm-id node port]
- (let [path (backpressure-path storm-id node port)
- existed (.node_exists cluster-state path false)]
- (if existed
- (.delete_node cluster-state (backpressure-path storm-id node port)))))
-
- (teardown-topology-errors!
- [this storm-id]
- (try-cause
- (.delete_node cluster-state (error-storm-root storm-id))
- (catch KeeperException e
- (log-warn-error e "Could not teardown errors for " storm-id))))
-
- (supervisor-heartbeat!
- [this supervisor-id info]
- (let [thrift-supervisor-info (thriftify-supervisor-info info)]
- (.set_ephemeral_node cluster-state (supervisor-path supervisor-id) (Utils/serialize thrift-supervisor-info) acls)))
-
- (activate-storm!
- [this storm-id storm-base]
- (let [thrift-storm-base (thriftify-storm-base storm-base)]
- (.set_data cluster-state (storm-path storm-id) (Utils/serialize thrift-storm-base) acls)))
-
- (update-storm!
- [this storm-id new-elems]
- (let [base (storm-base this storm-id nil)
- executors (:component->executors base)
- component->debug (:component->debug base)
- new-elems (update new-elems :component->executors (partial merge executors))
- new-elems (update new-elems :component->debug (partial merge-with merge component->debug))]
- (.set_data cluster-state (storm-path storm-id)
- (-> base
- (merge new-elems)
- thriftify-storm-base
- Utils/serialize)
- acls)))
-
- (storm-base
- [this storm-id callback]
- (when callback
- (swap! storm-base-callback assoc storm-id callback))
- (clojurify-storm-base (maybe-deserialize (.get_data cluster-state (storm-path storm-id) (not-nil? callback)) StormBase)))
-
- (remove-storm-base!
- [this storm-id]
- (.delete_node cluster-state (storm-path storm-id)))
-
- (set-assignment!
- [this storm-id info]
- (let [thrift-assignment (thriftify-assignment info)]
- (.set_data cluster-state (assignment-path storm-id) (Utils/serialize thrift-assignment) acls)))
-
- (remove-blobstore-key!
- [this blob-key]
- (log-debug "removing key" blob-key)
- (.delete_node cluster-state (blobstore-path blob-key)))
-
- (remove-key-version!
- [this blob-key]
- (.delete_node cluster-state (blobstore-max-key-sequence-number-path blob-key)))
-
- (remove-storm!
- [this storm-id]
- (.delete_node cluster-state (assignment-path storm-id))
- (.delete_node cluster-state (credentials-path storm-id))
- (.delete_node cluster-state (log-config-path storm-id))
- (.delete_node cluster-state (profiler-config-path storm-id))
- (remove-storm-base! this storm-id))
-
- (set-credentials!
- [this storm-id creds topo-conf]
- (let [topo-acls (mk-topo-only-acls topo-conf)
- path (credentials-path storm-id)
- thriftified-creds (thriftify-credentials creds)]
- (.set_data cluster-state path (Utils/serialize thriftified-creds) topo-acls)))
-
- (credentials
- [this storm-id callback]
- (when callback
- (swap! credentials-callback assoc storm-id callback))
- (clojurify-crdentials (maybe-deserialize (.get_data cluster-state (credentials-path storm-id) (not-nil? callback)) Credentials)))
-
- (report-error
- [this storm-id component-id node port error]
- (let [path (error-path storm-id component-id)
- last-error-path (last-error-path storm-id component-id)
- data (thriftify-error {:time-secs (current-time-secs) :error (stringify-error error) :host node :port port})
- _ (.mkdirs cluster-state path acls)
- ser-data (Utils/serialize data)
- _ (.mkdirs cluster-state path acls)
- _ (.create_sequential cluster-state (str path "/e") ser-data acls)
- _ (.set_data cluster-state last-error-path ser-data acls)
- to-kill (->> (.get_children cluster-state path false)
- (sort-by parse-error-path)
- reverse
- (drop 10))]
- (doseq [k to-kill]
- (.delete_node cluster-state (str path "/" k)))))
-
- (errors
- [this storm-id component-id]
- (let [path (error-path storm-id component-id)
- errors (if (.node_exists cluster-state path false)
- (dofor [c (.get_children cluster-state path false)]
- (if-let [data (-> (.get_data cluster-state
- (str path "/" c)
- false)
- (maybe-deserialize ErrorInfo)
- clojurify-error)]
- (map->TaskError data)))
- ())]
- (->> (filter not-nil? errors)
- (sort-by (comp - :time-secs)))))
-
- (last-error
- [this storm-id component-id]
- (let [path (last-error-path storm-id component-id)]
- (if (.node_exists cluster-state path false)
- (if-let [data (-> (.get_data cluster-state path false)
- (maybe-deserialize ErrorInfo)
- clojurify-error)]
- (map->TaskError data)))))
-
- (disconnect
- [this]
- (.unregister cluster-state state-id)
- (when solo?
- (.close cluster-state))))))
-
-;; daemons have a single thread that will respond to events
-;; start with initialize event
-;; callbacks add events to the thread's queue
-
-;; keeps in memory cache of the state, only for what client subscribes to. Any subscription is automatically kept in sync, and when there are changes, client is notified.
-;; master gives orders through state, and client records status in state (ephemerally)
-
-;; master tells nodes what workers to launch
-
-;; master writes this. supervisors and workers subscribe to this to understand complete topology. each storm is a map from nodes to workers to tasks to ports whenever topology changes everyone will be notified
-;; master includes timestamp of each assignment so that appropriate time can be given to each worker to start up
-;; /assignments/{storm id}
-
-;; which tasks they talk to, etc. (immutable until shutdown)
-;; everyone reads this in full to understand structure
-;; /tasks/{storm id}/{task id} ; just contains bolt id
-
-;; supervisors send heartbeats here, master doesn't subscribe but checks asynchronously
-;; /supervisors/status/{ephemeral node ids} ;; node metadata such as port ranges are kept here
-
-;; tasks send heartbeats here, master doesn't subscribe, just checks asynchronously
-;; /taskbeats/{storm id}/{ephemeral task id}
-
-;; contains data about whether it's started or not, tasks and workers subscribe to specific storm here to know when to shutdown
-;; master manipulates
-;; /storms/{storm id}
-
-;; Zookeeper flows:
-
-;; Master:
-;; job submit:
-;; 1. read which nodes are available
-;; 2. set up the worker/{storm}/{task} stuff (static)
-;; 3. set assignments
-;; 4. start storm - necessary in case master goes down, when goes back up can remember to take down the storm (2 states: on or off)
-
-;; Monitoring (or by checking when nodes go down or heartbeats aren't received):
-;; 1. read assignment
-;; 2. see which tasks/nodes are up
-;; 3. make new assignment to fix any problems
-;; 4. if a storm exists but is not taken down fully, ensure that storm takedown is launched (step by step remove tasks and finally remove assignments)
-
-;; masters only possible watches is on ephemeral nodes and tasks, and maybe not even
-
-;; Supervisor:
-;; 1. monitor /storms/* and assignments
-;; 2. local state about which workers are local
-;; 3. when storm is on, check that workers are running locally & start/kill if different than assignments
-;; 4. when storm is off, monitor tasks for workers - when they all die or don't hearbeat, kill the process and cleanup
-
-;; Worker:
-;; 1. On startup, start the tasks if the storm is on
-
-;; Task:
-;; 1. monitor assignments, reroute when assignments change
-;; 2. monitor storm (when storm turns off, error if assignments change) - take down tasks as master turns them off
-
-;; locally on supervisor: workers write pids locally on startup, supervisor deletes it on shutdown (associates pid with worker name)
-;; supervisor periodically checks to make sure processes are alive
-;; {rootdir}/workers/{storm id}/{worker id} ;; contains pid inside
-
-;; all tasks in a worker share the same cluster state
-;; workers, supervisors, and tasks subscribes to storm to know when it's started or stopped
-;; on stopped, master removes records in order (tasks need to subscribe to themselves to see if they disappear)
-;; when a master removes a worker, the supervisor should kill it (and escalate to kill -9)
-;; on shutdown, tasks subscribe to tasks that send data to them to wait for them to die. when node disappears, they can die
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/cluster_state/zookeeper_state_factory.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/cluster_state/zookeeper_state_factory.clj b/storm-core/src/clj/backtype/storm/cluster_state/zookeeper_state_factory.clj
deleted file mode 100644
index 061c9e8..0000000
--- a/storm-core/src/clj/backtype/storm/cluster_state/zookeeper_state_factory.clj
+++ /dev/null
@@ -1,165 +0,0 @@
-;; 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 org.apache.storm.cluster-state.zookeeper-state-factory
- (:import [org.apache.curator.framework.state ConnectionStateListener])
- (:import [org.apache.zookeeper KeeperException$NoNodeException KeeperException$NodeExistsException]
- [org.apache.storm.cluster ClusterState DaemonType])
- (:import [org.apache.storm.utils StormConnectionStateConverter])
- (:use [org.apache.storm cluster config log util])
- (:require [org.apache.storm [zookeeper :as zk]])
- (:gen-class
- :implements [org.apache.storm.cluster.ClusterStateFactory]))
-
-(defn -mkState [this conf auth-conf acls context]
- (let [zk (zk/mk-client conf (conf STORM-ZOOKEEPER-SERVERS) (conf STORM-ZOOKEEPER-PORT) :auth-conf auth-conf)]
- (zk/mkdirs zk (conf STORM-ZOOKEEPER-ROOT) acls)
- (.close zk))
- (let [callbacks (atom {})
- active (atom true)
- zk-writer (zk/mk-client conf
- (conf STORM-ZOOKEEPER-SERVERS)
- (conf STORM-ZOOKEEPER-PORT)
- :auth-conf auth-conf
- :root (conf STORM-ZOOKEEPER-ROOT)
- :watcher (fn [state type path]
- (when @active
- (when-not (= :connected state)
- (log-warn "Received event " state ":" type ":" path " with disconnected Writer Zookeeper."))
- (when-not (= :none type)
- (doseq [callback (vals @callbacks)]
- (callback type path))))))
- is-nimbus? (= (.getDaemonType context) DaemonType/NIMBUS)
- zk-reader (if is-nimbus?
- (zk/mk-client conf
- (conf STORM-ZOOKEEPER-SERVERS)
- (conf STORM-ZOOKEEPER-PORT)
- :auth-conf auth-conf
- :root (conf STORM-ZOOKEEPER-ROOT)
- :watcher (fn [state type path]
- (when @active
- (when-not (= :connected state)
- (log-warn "Received event " state ":" type ":" path " with disconnected Reader Zookeeper."))
- (when-not (= :none type)
- (doseq [callback (vals @callbacks)]
- (callback type path))))))
- zk-writer)]
- (reify
- ClusterState
-
- (register
- [this callback]
- (let [id (uuid)]
- (swap! callbacks assoc id callback)
- id))
-
- (unregister
- [this id]
- (swap! callbacks dissoc id))
-
- (set-ephemeral-node
- [this path data acls]
- (zk/mkdirs zk-writer (parent-path path) acls)
- (if (zk/exists zk-writer path false)
- (try-cause
- (zk/set-data zk-writer path data) ; should verify that it's ephemeral
- (catch KeeperException$NoNodeException e
- (log-warn-error e "Ephemeral node disappeared between checking for existing and setting data")
- (zk/create-node zk-writer path data :ephemeral acls)))
- (zk/create-node zk-writer path data :ephemeral acls)))
-
- (create-sequential
- [this path data acls]
- (zk/create-node zk-writer path data :sequential acls))
-
- (set-data
- [this path data acls]
- ;; note: this does not turn off any existing watches
- (if (zk/exists zk-writer path false)
- (zk/set-data zk-writer path data)
- (do
- (zk/mkdirs zk-writer (parent-path path) acls)
- (try-cause
- (zk/create-node zk-writer path data :persistent acls)
- (catch KeeperException$NodeExistsException e
- (zk/set-data zk-writer path data))))))
-
- (set-worker-hb
- [this path data acls]
- (.set_data this path data acls))
-
- (delete-node
- [this path]
- (zk/delete-node zk-writer path))
-
- (delete-worker-hb
- [this path]
- (.delete_node this path))
-
- (get-data
- [this path watch?]
- (zk/get-data zk-reader path watch?))
-
- (get-data-with-version
- [this path watch?]
- (zk/get-data-with-version zk-reader path watch?))
-
- (get-version
- [this path watch?]
- (zk/get-version zk-reader path watch?))
-
- (get-worker-hb
- [this path watch?]
- (.get_data this path watch?))
-
- (get-children
- [this path watch?]
- (zk/get-children zk-reader path watch?))
-
- (get-worker-hb-children
- [this path watch?]
- (.get_children this path watch?))
-
- (mkdirs
- [this path acls]
- (zk/mkdirs zk-writer path acls))
-
- (node-exists
- [this path watch?]
- (zk/exists-node? zk-reader path watch?))
-
- (add-listener
- [this listener]
- (let [curator-listener (reify ConnectionStateListener
- (stateChanged
- [this client newState]
- (.stateChanged listener (StormConnectionStateConverter/convert newState))))]
- (zk/add-listener zk-reader curator-listener)))
-
- (sync-path
- [this path]
- (zk/sync-path zk-writer path))
-
- (delete-node-blobstore
- [this path nimbus-host-port-info]
- (zk/delete-node-blobstore zk-writer path nimbus-host-port-info))
-
- (close
- [this]
- (reset! active false)
- (.close zk-writer)
- (if is-nimbus?
- (.close zk-reader))))))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/activate.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/activate.clj b/storm-core/src/clj/backtype/storm/command/activate.clj
deleted file mode 100644
index dc452e8..0000000
--- a/storm-core/src/clj/backtype/storm/command/activate.clj
+++ /dev/null
@@ -1,24 +0,0 @@
-;; 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 org.apache.storm.command.activate
- (:use [org.apache.storm thrift log])
- (:gen-class))
-
-(defn -main [name]
- (with-configured-nimbus-connection nimbus
- (.activate nimbus name)
- (log-message "Activated topology: " name)
- ))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/blobstore.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/blobstore.clj b/storm-core/src/clj/backtype/storm/command/blobstore.clj
deleted file mode 100644
index b1496db..0000000
--- a/storm-core/src/clj/backtype/storm/command/blobstore.clj
+++ /dev/null
@@ -1,162 +0,0 @@
-;; 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 org.apache.storm.command.blobstore
- (:import [java.io InputStream OutputStream]
- [org.apache.storm.generated SettableBlobMeta AccessControl AuthorizationException
- KeyNotFoundException]
- [org.apache.storm.blobstore BlobStoreAclHandler])
- (:use [org.apache.storm config]
- [clojure.string :only [split]]
- [clojure.tools.cli :only [cli]]
- [clojure.java.io :only [copy input-stream output-stream]]
- [org.apache.storm blobstore log util])
- (:gen-class))
-
-(defn update-blob-from-stream
- "Update a blob in the blob store from an InputStream"
- [key ^InputStream in]
- (with-configured-blob-client blobstore
- (let [out (.updateBlob blobstore key)]
- (try
- (copy in out)
- (.close out)
- (catch Exception e
- (log-message e)
- (.cancel out)
- (throw e))))))
-
-(defn create-blob-from-stream
- "Create a blob in the blob store from an InputStream"
- [key ^InputStream in ^SettableBlobMeta meta]
- (with-configured-blob-client blobstore
- (let [out (.createBlob blobstore key meta)]
- (try
- (copy in out)
- (.close out)
- (catch Exception e
- (.cancel out)
- (throw e))))))
-
-(defn read-blob
- "Read a blob in the blob store and write to an OutputStream"
- [key ^OutputStream out]
- (with-configured-blob-client blobstore
- (with-open [in (.getBlob blobstore key)]
- (copy in out))))
-
-(defn as-access-control
- "Convert a parameter to an AccessControl object"
- [param]
- (BlobStoreAclHandler/parseAccessControl (str param)))
-
-(defn as-acl
- [param]
- (map as-access-control (split param #",")))
-
-(defn access-control-str
- [^AccessControl acl]
- (BlobStoreAclHandler/accessControlToString acl))
-
-(defn read-cli [args]
- (let [[{file :file} [key] _] (cli args ["-f" "--file" :default nil])]
- (if file
- (with-open [f (output-stream file)]
- (read-blob key f))
- (read-blob key System/out))))
-
-(defn update-cli [args]
- (let [[{file :file} [key] _] (cli args ["-f" "--file" :default nil])]
- (if file
- (with-open [f (input-stream file)]
- (update-blob-from-stream key f))
- (update-blob-from-stream key System/in))
- (log-message "Successfully updated " key)))
-
-(defn create-cli [args]
- (let [[{file :file acl :acl replication-factor :replication-factor} [key] _] (cli args ["-f" "--file" :default nil]
- ["-a" "--acl" :default [] :parse-fn as-acl]
- ["-r" "--replication-factor" :default -1 :parse-fn parse-int])
- meta (doto (SettableBlobMeta. acl)
- (.set_replication_factor replication-factor))]
- (validate-key-name! key)
- (log-message "Creating " key " with ACL " (pr-str (map access-control-str acl)))
- (if file
- (with-open [f (input-stream file)]
- (create-blob-from-stream key f meta))
- (create-blob-from-stream key System/in meta))
- (log-message "Successfully created " key)))
-
-(defn delete-cli [args]
- (with-configured-blob-client blobstore
- (doseq [key args]
- (.deleteBlob blobstore key)
- (log-message "deleted " key))))
-
-(defn list-cli [args]
- (with-configured-blob-client blobstore
- (let [keys (if (empty? args) (iterator-seq (.listKeys blobstore)) args)]
- (doseq [key keys]
- (try
- (let [meta (.getBlobMeta blobstore key)
- version (.get_version meta)
- acl (.get_acl (.get_settable meta))]
- (log-message key " " version " " (pr-str (map access-control-str acl))))
- (catch AuthorizationException ae
- (if-not (empty? args) (log-error "ACCESS DENIED to key: " key)))
- (catch KeyNotFoundException knf
- (if-not (empty? args) (log-error key " NOT FOUND"))))))))
-
-(defn set-acl-cli [args]
- (let [[{set-acl :set} [key] _]
- (cli args ["-s" "--set" :default [] :parse-fn as-acl])]
- (with-configured-blob-client blobstore
- (let [meta (.getBlobMeta blobstore key)
- acl (.get_acl (.get_settable meta))
- new-acl (if set-acl set-acl acl)
- new-meta (SettableBlobMeta. new-acl)]
- (log-message "Setting ACL for " key " to " (pr-str (map access-control-str new-acl)))
- (.setBlobMeta blobstore key new-meta)))))
-
-(defn rep-cli [args]
- (let [sub-command (first args)
- new-args (rest args)]
- (with-configured-blob-client blobstore
- (condp = sub-command
- "--read" (let [key (first new-args)
- blob-replication (.getBlobReplication blobstore key)]
- (log-message "Current replication factor " blob-replication)
- blob-replication)
- "--update" (let [[{replication-factor :replication-factor} [key] _]
- (cli new-args ["-r" "--replication-factor" :parse-fn parse-int])]
- (if (nil? replication-factor)
- (throw (RuntimeException. (str "Please set the replication factor")))
- (let [blob-replication (.updateBlobReplication blobstore key replication-factor)]
- (log-message "Replication factor is set to " blob-replication)
- blob-replication)))
- :else (throw (RuntimeException. (str sub-command " is not a supported blobstore command")))))))
-
-(defn -main [& args]
- (let [command (first args)
- new-args (rest args)]
- (condp = command
- "cat" (read-cli new-args)
- "create" (create-cli new-args)
- "update" (update-cli new-args)
- "delete" (delete-cli new-args)
- "list" (list-cli new-args)
- "set-acl" (set-acl-cli new-args)
- "replication" (rep-cli new-args)
- :else (throw (RuntimeException. (str command " is not a supported blobstore command"))))))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/config_value.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/config_value.clj b/storm-core/src/clj/backtype/storm/command/config_value.clj
deleted file mode 100644
index 9bc3e92..0000000
--- a/storm-core/src/clj/backtype/storm/command/config_value.clj
+++ /dev/null
@@ -1,24 +0,0 @@
-;; 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 org.apache.storm.command.config-value
- (:use [org.apache.storm config log])
- (:gen-class))
-
-
-(defn -main [^String name]
- (let [conf (read-storm-config)]
- (println "VALUE:" (conf name))
- ))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/deactivate.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/deactivate.clj b/storm-core/src/clj/backtype/storm/command/deactivate.clj
deleted file mode 100644
index 4fd2c85..0000000
--- a/storm-core/src/clj/backtype/storm/command/deactivate.clj
+++ /dev/null
@@ -1,24 +0,0 @@
-;; 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 org.apache.storm.command.deactivate
- (:use [org.apache.storm thrift log])
- (:gen-class))
-
-(defn -main [name]
- (with-configured-nimbus-connection nimbus
- (.deactivate nimbus name)
- (log-message "Deactivated topology: " name)
- ))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/dev_zookeeper.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/dev_zookeeper.clj b/storm-core/src/clj/backtype/storm/command/dev_zookeeper.clj
deleted file mode 100644
index 96de02d..0000000
--- a/storm-core/src/clj/backtype/storm/command/dev_zookeeper.clj
+++ /dev/null
@@ -1,26 +0,0 @@
-;; 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 org.apache.storm.command.dev-zookeeper
- (:use [org.apache.storm zookeeper util config])
- (:gen-class))
-
-(defn -main [& args]
- (let [conf (read-storm-config)
- port (conf STORM-ZOOKEEPER-PORT)
- localpath (conf DEV-ZOOKEEPER-PATH)]
- (rmr localpath)
- (mk-inprocess-zookeeper localpath :port port)
- ))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/get_errors.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/get_errors.clj b/storm-core/src/clj/backtype/storm/command/get_errors.clj
deleted file mode 100644
index c267390..0000000
--- a/storm-core/src/clj/backtype/storm/command/get_errors.clj
+++ /dev/null
@@ -1,52 +0,0 @@
-;; 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 org.apache.storm.command.get-errors
- (:use [clojure.tools.cli :only [cli]])
- (:use [org.apache.storm thrift log])
- (:use [org.apache.storm util])
- (:require [org.apache.storm.daemon
- [nimbus :as nimbus]
- [common :as common]])
- (:import [org.apache.storm.generated GetInfoOptions NumErrorsChoice
- TopologySummary ErrorInfo])
- (:gen-class))
-
-(defn get-topology-id [name topologies]
- (let [topology (first (filter #(= (.get_name %1) name) topologies))]
- (when (not-nil? topology) (.get_id topology))))
-
-(defn get-component-errors
- [topology-errors]
- (apply hash-map (remove nil?
- (flatten (for [[comp-name comp-errors] topology-errors]
- (let [latest-error (when (not (empty? comp-errors)) (first comp-errors))]
- (if latest-error [comp-name (.get_error ^ErrorInfo latest-error)])))))))
-
-(defn -main [name]
- (with-configured-nimbus-connection nimbus
- (let [opts (doto (GetInfoOptions.)
- (.set_num_err_choice NumErrorsChoice/ONE))
- cluster-info (.getClusterInfo nimbus)
- topologies (.get_topologies cluster-info)
- topo-id (get-topology-id name topologies)
- topo-info (when (not-nil? topo-id) (.getTopologyInfoWithOpts nimbus topo-id opts))]
- (if (or (nil? topo-id) (nil? topo-info))
- (println (to-json {"Failure" (str "No topologies running with name " name)}))
- (let [topology-name (.get_name topo-info)
- topology-errors (.get_errors topo-info)]
- (println (to-json (hash-map
- "Topology Name" topology-name
- "Comp-Errors" (get-component-errors topology-errors)))))))))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/healthcheck.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/healthcheck.clj b/storm-core/src/clj/backtype/storm/command/healthcheck.clj
deleted file mode 100644
index d96d7b3..0000000
--- a/storm-core/src/clj/backtype/storm/command/healthcheck.clj
+++ /dev/null
@@ -1,88 +0,0 @@
-;; 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 org.apache.storm.command.healthcheck
- (:require [org.apache.storm
- [config :refer :all]
- [log :refer :all]]
- [clojure.java [io :as io]]
- [clojure [string :refer [split]]])
- (:gen-class))
-
-(defn interrupter
- "Interrupt a given thread after ms milliseconds."
- [thread ms]
- (let [interrupter (Thread.
- (fn []
- (try
- (Thread/sleep ms)
- (.interrupt thread)
- (catch InterruptedException e))))]
- (.start interrupter)
- interrupter))
-
-(defn check-output [lines]
- (if (some #(.startsWith % "ERROR") lines)
- :failed
- :success))
-
-(defn process-script [conf script]
- (let [script-proc (. (Runtime/getRuntime) (exec script))
- curthread (Thread/currentThread)
- interrupter-thread (interrupter curthread
- (conf STORM-HEALTH-CHECK-TIMEOUT-MS))]
- (try
- (.waitFor script-proc)
- (.interrupt interrupter-thread)
- (if (not (= (.exitValue script-proc) 0))
- :failed_with_exit_code
- (check-output (split
- (slurp (.getInputStream script-proc))
- #"\n+")))
- (catch InterruptedException e
- (println "Script" script "timed out.")
- :timeout)
- (catch Exception e
- (println "Script failed with exception: " e)
- :failed_with_exception)
- (finally (.interrupt interrupter-thread)))))
-
-(defn health-check [conf]
- (let [health-dir (absolute-healthcheck-dir conf)
- health-files (file-seq (io/file health-dir))
- health-scripts (filter #(and (.canExecute %)
- (not (.isDirectory %)))
- health-files)
- results (->> health-scripts
- (map #(.getAbsolutePath %))
- (map (partial process-script conf)))]
- (log-message
- (pr-str (map #'vector
- (map #(.getAbsolutePath %) health-scripts)
- results)))
- ; failed_with_exit_code is OK. We're mimicing Hadoop's health checks.
- ; We treat non-zero exit codes as indicators that the scripts failed
- ; to execute properly, not that the system is unhealthy, in which case
- ; we don't want to start killing things.
- (if (every? #(or (= % :failed_with_exit_code)
- (= % :success))
- results)
- 0
- 1)))
-
-(defn -main [& args]
- (let [conf (read-storm-config)]
- (System/exit
- (health-check conf))))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/heartbeats.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/heartbeats.clj b/storm-core/src/clj/backtype/storm/command/heartbeats.clj
deleted file mode 100644
index ff28cba..0000000
--- a/storm-core/src/clj/backtype/storm/command/heartbeats.clj
+++ /dev/null
@@ -1,52 +0,0 @@
-;; 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 org.apache.storm.command.heartbeats
- (:require [org.apache.storm
- [config :refer :all]
- [log :refer :all]
- [cluster :refer :all]
- [converter :refer :all]]
- [clojure.string :refer :all])
- (:import [org.apache.storm.generated ClusterWorkerHeartbeat]
- [org.apache.storm.utils Utils])
- (:gen-class))
-
-(defn -main [command path & args]
- (let [conf (read-storm-config)
- cluster (mk-distributed-cluster-state conf :auth-conf conf)]
- (println "Command: [" command "]")
- (condp = command
- "list"
- (let [message (join " \n" (.get_worker_hb_children cluster path false))]
- (log-message "list " path ":\n"
- message "\n"))
- "get"
- (log-message
- (if-let [hb (.get_worker_hb cluster path false)]
- (clojurify-zk-worker-hb
- (Utils/deserialize
- hb
- ClusterWorkerHeartbeat))
- "Nothing"))
-
- (log-message "Usage: heartbeats [list|get] path"))
-
- (try
- (.close cluster)
- (catch Exception e
- (log-message "Caught exception: " e " on close."))))
- (System/exit 0))
-
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/kill_topology.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/kill_topology.clj b/storm-core/src/clj/backtype/storm/command/kill_topology.clj
deleted file mode 100644
index 84e0a64..0000000
--- a/storm-core/src/clj/backtype/storm/command/kill_topology.clj
+++ /dev/null
@@ -1,29 +0,0 @@
-;; 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 org.apache.storm.command.kill-topology
- (:use [clojure.tools.cli :only [cli]])
- (:use [org.apache.storm thrift config log])
- (:import [org.apache.storm.generated KillOptions])
- (:gen-class))
-
-(defn -main [& args]
- (let [[{wait :wait} [name] _] (cli args ["-w" "--wait" :default nil :parse-fn #(Integer/parseInt %)])
- opts (KillOptions.)]
- (if wait (.set_wait_secs opts wait))
- (with-configured-nimbus-connection nimbus
- (.killTopologyWithOpts nimbus name opts)
- (log-message "Killed topology: " name)
- )))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/kill_workers.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/kill_workers.clj b/storm-core/src/clj/backtype/storm/command/kill_workers.clj
deleted file mode 100644
index 2670735..0000000
--- a/storm-core/src/clj/backtype/storm/command/kill_workers.clj
+++ /dev/null
@@ -1,33 +0,0 @@
-;; 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 org.apache.storm.command.kill-workers
- (:import [java.io File])
- (:use [org.apache.storm.daemon common])
- (:use [org.apache.storm util config])
- (:require [org.apache.storm.daemon
- [supervisor :as supervisor]])
- (:gen-class))
-
-(defn -main
- "Construct the supervisor-data from scratch and kill the workers on this supervisor"
- [& args]
- (let [conf (read-storm-config)
- conf (assoc conf STORM-LOCAL-DIR (. (File. (conf STORM-LOCAL-DIR)) getCanonicalPath))
- isupervisor (supervisor/standalone-supervisor)
- supervisor-data (supervisor/supervisor-data conf nil isupervisor)
- ids (supervisor/my-worker-ids conf)]
- (doseq [id ids]
- (supervisor/shutdown-worker supervisor-data id))))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/list.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/list.clj b/storm-core/src/clj/backtype/storm/command/list.clj
deleted file mode 100644
index 87975cd..0000000
--- a/storm-core/src/clj/backtype/storm/command/list.clj
+++ /dev/null
@@ -1,38 +0,0 @@
-;; 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 org.apache.storm.command.list
- (:use [org.apache.storm thrift log])
- (:import [org.apache.storm.generated TopologySummary])
- (:gen-class))
-
-(defn -main []
- (with-configured-nimbus-connection nimbus
- (let [cluster-info (.getClusterInfo nimbus)
- topologies (.get_topologies cluster-info)
- msg-format "%-20s %-10s %-10s %-12s %-10s"]
- (if (or (nil? topologies) (empty? topologies))
- (println "No topologies running.")
- (do
- (println (format msg-format "Topology_name" "Status" "Num_tasks" "Num_workers" "Uptime_secs"))
- (println "-------------------------------------------------------------------")
- (doseq [^TopologySummary topology topologies]
- (let [topology-name (.get_name topology)
- topology-status (.get_status topology)
- topology-num-tasks (.get_num_tasks topology)
- topology-num-workers (.get_num_workers topology)
- topology-uptime-secs (.get_uptime_secs topology)]
- (println (format msg-format topology-name topology-status topology-num-tasks
- topology-num-workers topology-uptime-secs)))))))))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/monitor.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/monitor.clj b/storm-core/src/clj/backtype/storm/command/monitor.clj
deleted file mode 100644
index 7fa9b2a..0000000
--- a/storm-core/src/clj/backtype/storm/command/monitor.clj
+++ /dev/null
@@ -1,37 +0,0 @@
-;; 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 org.apache.storm.command.monitor
- (:use [clojure.tools.cli :only [cli]])
- (:use [org.apache.storm.thrift :only [with-configured-nimbus-connection]])
- (:import [org.apache.storm.utils Monitor])
- (:gen-class)
- )
-
-(defn -main [& args]
- (let [[{interval :interval component :component stream :stream watch :watch} [name] _]
- (cli args ["-i" "--interval" :default 4 :parse-fn #(Integer/parseInt %)]
- ["-m" "--component" :default nil]
- ["-s" "--stream" :default "default"]
- ["-w" "--watch" :default "emitted"])
- mon (Monitor.)]
- (if interval (.set_interval mon interval))
- (if name (.set_topology mon name))
- (if component (.set_component mon component))
- (if stream (.set_stream mon stream))
- (if watch (.set_watch mon watch))
- (with-configured-nimbus-connection nimbus
- (.metrics mon nimbus)
- )))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/rebalance.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/rebalance.clj b/storm-core/src/clj/backtype/storm/command/rebalance.clj
deleted file mode 100644
index 3868091..0000000
--- a/storm-core/src/clj/backtype/storm/command/rebalance.clj
+++ /dev/null
@@ -1,46 +0,0 @@
-;; 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 org.apache.storm.command.rebalance
- (:use [clojure.tools.cli :only [cli]])
- (:use [org.apache.storm thrift config log])
- (:import [org.apache.storm.generated RebalanceOptions])
- (:gen-class))
-
-(defn- parse-executor [^String s]
- (let [eq-pos (.lastIndexOf s "=")
- name (.substring s 0 eq-pos)
- amt (.substring s (inc eq-pos))]
- {name (Integer/parseInt amt)}
- ))
-
-(defn -main [& args]
- (let [[{wait :wait executor :executor num-workers :num-workers} [name] _]
- (cli args ["-w" "--wait" :default nil :parse-fn #(Integer/parseInt %)]
- ["-n" "--num-workers" :default nil :parse-fn #(Integer/parseInt %)]
- ["-e" "--executor" :parse-fn parse-executor
- :assoc-fn (fn [previous key val]
- (assoc previous key
- (if-let [oldval (get previous key)]
- (merge oldval val)
- val)))])
- opts (RebalanceOptions.)]
- (if wait (.set_wait_secs opts wait))
- (if executor (.set_num_executors opts executor))
- (if num-workers (.set_num_workers opts num-workers))
- (with-configured-nimbus-connection nimbus
- (.rebalance nimbus name opts)
- (log-message "Topology " name " is rebalancing")
- )))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/set_log_level.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/set_log_level.clj b/storm-core/src/clj/backtype/storm/command/set_log_level.clj
deleted file mode 100644
index 7e1c3c5..0000000
--- a/storm-core/src/clj/backtype/storm/command/set_log_level.clj
+++ /dev/null
@@ -1,75 +0,0 @@
-;; 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 org.apache.storm.command.set-log-level
- (:use [clojure.tools.cli :only [cli]])
- (:use [org.apache.storm thrift log])
- (:import [org.apache.logging.log4j Level])
- (:import [org.apache.storm.generated LogConfig LogLevel LogLevelAction])
- (:gen-class))
-
-(defn- get-storm-id
- "Get topology id for a running topology from the topology name."
- [nimbus name]
- (let [info (.getClusterInfo nimbus)
- topologies (.get_topologies info)
- topology (first (filter (fn [topo] (= name (.get_name topo))) topologies))]
- (if topology
- (.get_id topology)
- (throw (.IllegalArgumentException (str name " is not a running topology"))))))
-
-(defn- parse-named-log-levels [action]
- "Parses [logger name]=[level string]:[optional timeout],[logger name2]...
-
- e.g. ROOT=DEBUG:30
- root logger, debug for 30 seconds
-
- org.apache.foo=WARN
- org.apache.foo set to WARN indefinitely"
- (fn [^String s]
- (let [log-args (re-find #"(.*)=([A-Z]+):?(\d*)" s)
- name (if (= action LogLevelAction/REMOVE) s (nth log-args 1))
- level (Level/toLevel (nth log-args 2))
- timeout-str (nth log-args 3)
- log-level (LogLevel.)]
- (if (= action LogLevelAction/REMOVE)
- (.set_action log-level action)
- (do
- (.set_action log-level action)
- (.set_target_log_level log-level (.toString level))
- (.set_reset_log_level_timeout_secs log-level
- (Integer. (if (= timeout-str "") "0" timeout-str)))))
- {name log-level})))
-
-(defn- merge-together [previous key val]
- (assoc previous key
- (if-let [oldval (get previous key)]
- (merge oldval val)
- val)))
-
-(defn -main [& args]
- (let [[{log-setting :log-setting remove-log-setting :remove-log-setting} [name] _]
- (cli args ["-l" "--log-setting"
- :parse-fn (parse-named-log-levels LogLevelAction/UPDATE)
- :assoc-fn merge-together]
- ["-r" "--remove-log-setting"
- :parse-fn (parse-named-log-levels LogLevelAction/REMOVE)
- :assoc-fn merge-together])
- log-config (LogConfig.)]
- (doseq [[log-name log-val] (merge log-setting remove-log-setting)]
- (.put_to_named_logger_level log-config log-name log-val))
- (log-message "Sent log config " log-config " for topology " name)
- (with-configured-nimbus-connection nimbus
- (.setLogConfig nimbus (get-storm-id nimbus name) log-config))))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/shell_submission.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/shell_submission.clj b/storm-core/src/clj/backtype/storm/command/shell_submission.clj
deleted file mode 100644
index 887ab3b..0000000
--- a/storm-core/src/clj/backtype/storm/command/shell_submission.clj
+++ /dev/null
@@ -1,34 +0,0 @@
-;; 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 org.apache.storm.command.shell-submission
- (:import [org.apache.storm StormSubmitter])
- (:use [org.apache.storm thrift util config log zookeeper])
- (:require [clojure.string :as str])
- (:gen-class))
-
-
-(defn -main [^String tmpjarpath & args]
- (let [conf (read-storm-config)
- ; since this is not a purpose to add to leader lock queue, passing nil as blob-store is ok
- zk-leader-elector (zk-leader-elector conf nil)
- leader-nimbus (.getLeader zk-leader-elector)
- host (.getHost leader-nimbus)
- port (.getPort leader-nimbus)
- no-op (.close zk-leader-elector)
- jarpath (StormSubmitter/submitJar conf tmpjarpath)
- args (concat args [host port jarpath])]
- (exec-command! (str/join " " args))
- ))
http://git-wip-us.apache.org/repos/asf/storm/blob/a6171bd6/storm-core/src/clj/backtype/storm/command/upload_credentials.clj
----------------------------------------------------------------------
diff --git a/storm-core/src/clj/backtype/storm/command/upload_credentials.clj b/storm-core/src/clj/backtype/storm/command/upload_credentials.clj
deleted file mode 100644
index f63bde4..0000000
--- a/storm-core/src/clj/backtype/storm/command/upload_credentials.clj
+++ /dev/null
@@ -1,35 +0,0 @@
-;; 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 org.apache.storm.command.upload-credentials
- (:use [clojure.tools.cli :only [cli]])
- (:use [org.apache.storm log util])
- (:import [org.apache.storm StormSubmitter])
- (:import [java.util Properties])
- (:import [java.io FileReader])
- (:gen-class))
-
-(defn read-map [file-name]
- (let [props (Properties. )
- _ (.load props (FileReader. file-name))]
- (clojurify-structure props)))
-
-(defn -main [& args]
- (let [[{cred-file :file} [name & rawCreds]] (cli args ["-f" "--file" :default nil])
- _ (when (and rawCreds (not (even? (.size rawCreds)))) (throw (RuntimeException. "Need an even number of arguments to make a map")))
- mapping (if rawCreds (apply assoc {} rawCreds) {})
- file-mapping (if (nil? cred-file) {} (read-map cred-file))]
- (StormSubmitter/pushCredentials name {} (merge file-mapping mapping))
- (log-message "Uploaded new creds to topology: " name)))