;; 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.ui.core
  (:use compojure.core)
  (:use [clojure.java.shell :only [sh]])
  (:use ring.middleware.reload
        ring.middleware.multipart-params)
  (:use [ring.middleware.json :only [wrap-json-params]])
  (:use [hiccup core page-helpers])
  (:use [backtype.storm config util log stats tuple zookeeper converter])
  (:use [backtype.storm.ui helpers])
  (:use [backtype.storm.daemon [common :only [ACKER-COMPONENT-ID ACKER-INIT-STREAM-ID ACKER-ACK-STREAM-ID
                                              ACKER-FAIL-STREAM-ID mk-authorization-handler
                                              start-metrics-reporters]]])
  (:import [backtype.storm.utils Utils]
           [backtype.storm.generated NimbusSummary])
  (:use [clojure.string :only [blank? lower-case trim split]])
  (:import [backtype.storm.generated ExecutorSpecificStats
            ExecutorStats ExecutorSummary ExecutorInfo TopologyInfo SpoutStats BoltStats
            ErrorInfo ClusterSummary SupervisorSummary TopologySummary
            Nimbus$Client StormTopology GlobalStreamId RebalanceOptions
            KillOptions GetInfoOptions NumErrorsChoice DebugOptions TopologyPageInfo
            TopologyStats CommonAggregateStats ComponentAggregateStats
            ComponentType BoltAggregateStats SpoutAggregateStats
            ExecutorAggregateStats SpecificAggregateStats ComponentPageInfo
            LogConfig LogLevel LogLevelAction])
  (:import [backtype.storm.security.auth AuthUtils ReqContext])
  (:import [backtype.storm.generated AuthorizationException ProfileRequest ProfileAction NodeInfo])
  (:import [backtype.storm.security.auth AuthUtils])
  (:import [backtype.storm.utils VersionInfo])
  (:import [backtype.storm Config])
  (:import [java.io File])
  (:require [compojure.route :as route]
            [compojure.handler :as handler]
            [ring.util.response :as resp]
            [backtype.storm [thrift :as thrift]])
  (:require [metrics.meters :refer [defmeter mark!]])
  (:import [org.apache.commons.lang StringEscapeUtils])
  (:import [org.apache.logging.log4j Level])
  (:gen-class))

(def ^:dynamic *STORM-CONF* (read-storm-config))
(def ^:dynamic *UI-ACL-HANDLER* (mk-authorization-handler (*STORM-CONF* NIMBUS-AUTHORIZER) *STORM-CONF*))
(def ^:dynamic *UI-IMPERSONATION-HANDLER* (mk-authorization-handler (*STORM-CONF* NIMBUS-IMPERSONATION-AUTHORIZER) *STORM-CONF*))
(def http-creds-handler (AuthUtils/GetUiHttpCredentialsPlugin *STORM-CONF*))

(defmeter ui:num-cluster-configuration-http-requests)
(defmeter ui:num-cluster-summary-http-requests)
(defmeter ui:num-nimbus-summary-http-requests)
(defmeter ui:num-supervisor-summary-http-requests)
(defmeter ui:num-all-topologies-summary-http-requests)
(defmeter ui:num-topology-page-http-requests)
(defmeter ui:num-build-visualization-http-requests)
(defmeter ui:num-mk-visualization-data-http-requests)
(defmeter ui:num-component-page-http-requests)
(defmeter ui:num-log-config-http-requests)
(defmeter ui:num-activate-topology-http-requests)
(defmeter ui:num-deactivate-topology-http-requests)
(defmeter ui:num-debug-topology-http-requests)
(defmeter ui:num-component-op-response-http-requests)
(defmeter ui:num-topology-op-response-http-requests)
(defmeter ui:num-topology-op-response-http-requests)
(defmeter ui:num-topology-op-response-http-requests)
(defmeter ui:num-main-page-http-requests)

(defn assert-authorized-user
  ([op]
    (assert-authorized-user op nil))
  ([op topology-conf]
    (let [context (ReqContext/context)]
      (if (.isImpersonating context)
        (if *UI-IMPERSONATION-HANDLER*
            (if-not (.permit *UI-IMPERSONATION-HANDLER* context op topology-conf)
              (let [principal (.principal context)
                    real-principal (.realPrincipal context)
                    user (if principal (.getName principal) "unknown")
                    real-user (if real-principal (.getName real-principal) "unknown")
                    remote-address (.remoteAddress context)]
                (throw (AuthorizationException.
                         (str "user '" real-user "' is not authorized to impersonate user '" user "' from host '" remote-address "'. Please
                         see SECURITY.MD to learn how to configure impersonation ACL.")))))
          (log-warn " principal " (.realPrincipal context) " is trying to impersonate " (.principal context) " but "
            NIMBUS-IMPERSONATION-AUTHORIZER " has no authorizer configured. This is a potential security hole.
            Please see SECURITY.MD to learn how to configure an impersonation authorizer.")))

      (if *UI-ACL-HANDLER*
       (if-not (.permit *UI-ACL-HANDLER* context op topology-conf)
         (let [principal (.principal context)
               user (if principal (.getName principal) "unknown")]
           (throw (AuthorizationException.
                   (str "UI request '" op "' for '" user "' user is not authorized")))))))))


(defn assert-authorized-profiler-action
  [op]
  (if-not (*STORM-CONF* WORKER-PROFILER-ENABLED)
    (throw (AuthorizationException.
             (str "UI request for profiler action '" op "' is disabled.")))))


(defn executor-summary-type
  [topology ^ExecutorSummary s]
  (component-type topology (.get_component_id s)))

(defn is-ack-stream
  [stream]
  (let [acker-streams
        [ACKER-INIT-STREAM-ID
         ACKER-ACK-STREAM-ID
         ACKER-FAIL-STREAM-ID]]
    (every? #(not= %1 stream) acker-streams)))

(defn spout-summary?
  [topology s]
  (= :spout (executor-summary-type topology s)))

(defn bolt-summary?
  [topology s]
  (= :bolt (executor-summary-type topology s)))

(defn group-by-comp
  [summs]
  (let [ret (group-by #(.get_component_id ^ExecutorSummary %) summs)]
    (into (sorted-map) ret )))

(defn logviewer-link [host fname secure?]
  (if (and secure? (*STORM-CONF* LOGVIEWER-HTTPS-PORT))
    (url-format "https://%s:%s/log?file=%s"
      host
      (*STORM-CONF* LOGVIEWER-HTTPS-PORT)
      fname)
    (url-format "http://%s:%s/log?file=%s"
      host
      (*STORM-CONF* LOGVIEWER-PORT)
      fname)))

(defn event-log-link
  [topology-id component-id host port secure?]
  (logviewer-link host (event-logs-filename topology-id port) secure?))

(defn worker-log-link [host port topology-id secure?]
  (let [fname (logs-filename topology-id port)]
    (logviewer-link host fname secure?)))

(defn nimbus-log-link [host port]
  (url-format "http://%s:%s/daemonlog?file=nimbus.log" host (*STORM-CONF* LOGVIEWER-PORT) port))

(defn get-error-time
  [error]
  (if error
    (time-delta (.get_error_time_secs ^ErrorInfo error))))

(defn get-error-data
  [error]
  (if error
    (error-subset (.get_error ^ErrorInfo error))
    ""))

(defn get-error-port
  [error]
  (if error
    (.get_port ^ErrorInfo error)
    ""))

(defn get-error-host
  [error]
  (if error
    (.get_host ^ErrorInfo error)
    ""))

(defn get-error-time
  [error]
  (if error
    (.get_error_time_secs ^ErrorInfo error)
    ""))

(defn worker-dump-link [host port topology-id]
  (url-format "http://%s:%s/dumps/%s/%s"
              (url-encode host)
              (*STORM-CONF* LOGVIEWER-PORT)
              (url-encode topology-id)
              (str (url-encode host) ":" (url-encode port))))

(defn stats-times
  [stats-map]
  (sort-by #(Integer/parseInt %)
           (-> stats-map
               clojurify-structure
               (dissoc ":all-time")
               keys)))

(defn window-hint
  [window]
  (if (= window ":all-time")
    "All time"
    (pretty-uptime-sec window)))

(defn sanitize-stream-name
  [name]
  (let [sym-regex #"(?![A-Za-z_\-:\.])."]
    (str
     (if (re-find #"^[A-Za-z]" name)
       (clojure.string/replace name sym-regex "_")
       (clojure.string/replace (str \s name) sym-regex "_"))
     (hash name))))

(defn sanitize-transferred
  [transferred]
  (into {}
        (for [[time, stream-map] transferred]
          [time, (into {}
                       (for [[stream, trans] stream-map]
                         [(sanitize-stream-name stream), trans]))])))

(defn visualization-data
  [spout-bolt spout-comp-summs bolt-comp-summs window storm-id]
  (let [components (for [[id spec] spout-bolt]
            [id
             (let [inputs (.get_inputs (.get_common spec))
                   bolt-summs (get bolt-comp-summs id)
                   spout-summs (get spout-comp-summs id)
                   bolt-cap (if bolt-summs
                              (compute-bolt-capacity bolt-summs)
                              0)]
               {:type (if bolt-summs "bolt" "spout")
                :capacity bolt-cap
                :latency (if bolt-summs
                           (get-in
                             (bolt-streams-stats bolt-summs true)
                             [:process-latencies window])
                           (get-in
                             (spout-streams-stats spout-summs true)
                             [:complete-latencies window]))
                :transferred (or
                               (get-in
                                 (spout-streams-stats spout-summs true)
                                 [:transferred window])
                               (get-in
                                 (bolt-streams-stats bolt-summs true)
                                 [:transferred window]))
                :stats (let [mapfn (fn [dat]
                                     (map (fn [^ExecutorSummary summ]
                                            {:host (.get_host summ)
                                             :port (.get_port summ)
                                             :uptime_secs (.get_uptime_secs summ)
                                             :transferred (if-let [stats (.get_stats summ)]
                                                            (sanitize-transferred (.get_transferred stats)))})
                                          dat))]
                         (if bolt-summs
                           (mapfn bolt-summs)
                           (mapfn spout-summs)))
                :link (url-format "/component.html?id=%s&topology_id=%s" id storm-id)
                :inputs (for [[global-stream-id group] inputs]
                          {:component (.get_componentId global-stream-id)
                           :stream (.get_streamId global-stream-id)
                           :sani-stream (sanitize-stream-name (.get_streamId global-stream-id))
                           :grouping (clojure.core/name (thrift/grouping-type group))})})])]
    (into {} (doall components))))

(defn stream-boxes [datmap]
  (let [filter-fn (mk-include-sys-fn true)
        streams
        (vec (doall (distinct
                     (apply concat
                            (for [[k v] datmap]
                              (for [m (get v :inputs)]
                                {:stream (get m :stream)
                                 :sani-stream (get m :sani-stream)
                                 :checked (is-ack-stream (get m :stream))}))))))]
    (map (fn [row]
           {:row row}) (partition 4 4 nil streams))))

(defn- get-topology-info
  ([^Nimbus$Client nimbus id]
    (.getTopologyInfo nimbus id))
  ([^Nimbus$Client nimbus id options]
    (.getTopologyInfoWithOpts nimbus id options)))

(defn mk-visualization-data
  [id window include-sys?]
  (thrift/with-configured-nimbus-connection
    nimbus
    (let [window (if window window ":all-time")
          topology (.getTopology ^Nimbus$Client nimbus id)
          spouts (.get_spouts topology)
          bolts (.get_bolts topology)
          summ (->> (doto
                      (GetInfoOptions.)
                      (.set_num_err_choice NumErrorsChoice/NONE))
                    (.getTopologyInfoWithOpts ^Nimbus$Client nimbus id))
          execs (.get_executors summ)
          spout-summs (filter (partial spout-summary? topology) execs)
          bolt-summs (filter (partial bolt-summary? topology) execs)
          spout-comp-summs (group-by-comp spout-summs)
          bolt-comp-summs (group-by-comp bolt-summs)
          bolt-comp-summs (filter-key (mk-include-sys-fn include-sys?)
                                      bolt-comp-summs)]
      (visualization-data
       (merge (hashmap-to-persistent spouts)
              (hashmap-to-persistent bolts))
       spout-comp-summs bolt-comp-summs window id))))

(defn validate-tplg-submit-params [params]
  (let [tplg-jar-file (params :topologyJar)
        tplg-config (if (not-nil? (params :topologyConfig)) (from-json (params :topologyConfig)))]
    (cond
     (nil? tplg-jar-file) {:valid false :error "missing topology jar file"}
     (nil? tplg-config) {:valid false :error "missing topology config"}
     (nil? (tplg-config "topologyMainClass")) {:valid false :error "topologyMainClass missing in topologyConfig"}
     :else {:valid true})))

(defn run-tplg-submit-cmd [tplg-jar-file tplg-config user]
  (let [tplg-main-class (if (not-nil? tplg-config) (trim (tplg-config "topologyMainClass")))
        tplg-main-class-args (if (not-nil? tplg-config) (tplg-config "topologyMainClassArgs"))
        storm-home (System/getProperty "storm.home")
        storm-conf-dir (str storm-home file-path-separator "conf")
        storm-log-dir (if (not-nil? (*STORM-CONF* "storm.log.dir")) (*STORM-CONF* "storm.log.dir")
                          (str storm-home file-path-separator "logs"))
        storm-libs (str storm-home file-path-separator "lib" file-path-separator "*")
        java-cmd (str (System/getProperty "java.home") file-path-separator "bin" file-path-separator "java")
        storm-cmd (str storm-home file-path-separator "bin" file-path-separator "storm")
        tplg-cmd-response (apply sh
                            (flatten
                              [storm-cmd "jar" tplg-jar-file tplg-main-class
                                (if (not-nil? tplg-main-class-args) tplg-main-class-args [])
                                (if (not= user "unknown") (str "-c storm.doAsUser=" user) [])]))]
    (log-message "tplg-cmd-response " tplg-cmd-response)
    (cond
     (= (tplg-cmd-response :exit) 0) {"status" "success"}
     (and (not= (tplg-cmd-response :exit) 0)
          (not-nil? (re-find #"already exists on cluster" (tplg-cmd-response :err)))) {"status" "failed" "error" "Topology with the same name exists in cluster"}
          (not= (tplg-cmd-response :exit) 0) {"status" "failed" "error" (clojure.string/trim-newline (tplg-cmd-response :err))}
          :else {"status" "success" "response" "topology deployed"}
          )))

(defn cluster-configuration []
  (thrift/with-configured-nimbus-connection nimbus
    (.getNimbusConf ^Nimbus$Client nimbus)))

(defn topology-history-info
  ([user]
    (thrift/with-configured-nimbus-connection nimbus
      (topology-history-info (.getTopologyHistory ^Nimbus$Client nimbus user) user)))
  ([history user]
    {"topo-history"
     (into [] (.get_topo_ids history))}))

(defn cluster-summary
  ([user]
     (thrift/with-configured-nimbus-connection nimbus
        (cluster-summary (.getClusterInfo ^Nimbus$Client nimbus) user)))
  ([^ClusterSummary summ user]
     (let [sups (.get_supervisors summ)
           used-slots (reduce + (map #(.get_num_used_workers ^SupervisorSummary %) sups))
           total-slots (reduce + (map #(.get_num_workers ^SupervisorSummary %) sups))
           free-slots (- total-slots used-slots)
           topologies (.get_topologies_size summ)
           total-tasks (->> (.get_topologies summ)
                            (map #(.get_num_tasks ^TopologySummary %))
                            (reduce +))
           total-executors (->> (.get_topologies summ)
                                (map #(.get_num_executors ^TopologySummary %))
                                (reduce +))]
       {"user" user
        "stormVersion" (str (VersionInfo/getVersion))
        "supervisors" (count sups)
        "topologies" topologies
        "slotsTotal" total-slots
        "slotsUsed"  used-slots
        "slotsFree" free-slots
        "executorsTotal" total-executors
        "tasksTotal" total-tasks })))

(defn convert-to-nimbus-summary[nimbus-seed]
  (let [[host port] (.split nimbus-seed ":")]
    {
      "host" host
      "port" port
      "nimbusLogLink" (nimbus-log-link host port)
      "status" "Offline"
      "version" "Not applicable"
      "nimbusUpTime" "Not applicable"
      "nimbusUptimeSeconds" "Not applicable"}
    ))

(defn nimbus-summary
  ([]
    (thrift/with-configured-nimbus-connection nimbus
      (nimbus-summary
        (.get_nimbuses (.getClusterInfo ^Nimbus$Client nimbus)))))
  ([nimbuses]
    (let [nimbus-seeds (set (map #(str %1 ":" (*STORM-CONF* NIMBUS-THRIFT-PORT)) (set (*STORM-CONF* NIMBUS-SEEDS))))
          alive-nimbuses (set (map #(str (.get_host %1) ":" (.get_port %1)) nimbuses))
          offline-nimbuses (clojure.set/difference nimbus-seeds alive-nimbuses)
          offline-nimbuses-summary (map #(convert-to-nimbus-summary %1) offline-nimbuses)]
      {"nimbuses"
       (concat offline-nimbuses-summary
       (for [^NimbusSummary n nimbuses
             :let [uptime (.get_uptime_secs n)]]
         {
          "host" (.get_host n)
          "port" (.get_port n)
          "nimbusLogLink" (nimbus-log-link (.get_host n) (.get_port n))
          "status" (if (.is_isLeader n) "Leader" "Not a Leader")
          "version" (.get_version n)
          "nimbusUpTime" (pretty-uptime-sec uptime)
          "nimbusUpTimeSeconds" uptime}))})))

(defn supervisor-summary
  ([]
   (thrift/with-configured-nimbus-connection nimbus
                (supervisor-summary
                  (.get_supervisors (.getClusterInfo ^Nimbus$Client nimbus)))))
  ([summs]
   {"supervisors"
    (for [^SupervisorSummary s summs]
      {"id" (.get_supervisor_id s)
       "host" (.get_host s)
       "uptime" (pretty-uptime-sec (.get_uptime_secs s))
       "uptimeSeconds" (.get_uptime_secs s)
       "slotsTotal" (.get_num_workers s)
       "slotsUsed" (.get_num_used_workers s)
       "totalMem" (get (.get_total_resources s) Config/SUPERVISOR_MEMORY_CAPACITY_MB)
       "totalCpu" (get (.get_total_resources s) Config/SUPERVISOR_CPU_CAPACITY)
       "usedMem" (.get_used_mem s)
       "usedCpu" (.get_used_cpu s)
       "version" (.get_version s)})
    "schedulerDisplayResource" (*STORM-CONF* Config/SCHEDULER_DISPLAY_RESOURCE)}))

(defn all-topologies-summary
  ([]
   (thrift/with-configured-nimbus-connection
     nimbus
     (all-topologies-summary
       (.get_topologies (.getClusterInfo ^Nimbus$Client nimbus)))))
  ([summs]
   {"topologies"
    (for [^TopologySummary t summs]
      {
       "id" (.get_id t)
       "encodedId" (url-encode (.get_id t))
       "owner" (.get_owner t)
       "name" (.get_name t)
       "status" (.get_status t)
       "uptime" (pretty-uptime-sec (.get_uptime_secs t))
       "uptimeSeconds" (.get_uptime_secs t)
       "tasksTotal" (.get_num_tasks t)
       "workersTotal" (.get_num_workers t)
       "executorsTotal" (.get_num_executors t)
       "replicationCount" (.get_replication_count t)
       "schedulerInfo" (.get_sched_status t)
       "requestedMemOnHeap" (.get_requested_memonheap t)
       "requestedMemOffHeap" (.get_requested_memoffheap t)
       "requestedTotalMem" (+ (.get_requested_memonheap t) (.get_requested_memoffheap t))
       "requestedCpu" (.get_requested_cpu t)
       "assignedMemOnHeap" (.get_assigned_memonheap t)
       "assignedMemOffHeap" (.get_assigned_memoffheap t)
       "assignedTotalMem" (+ (.get_assigned_memonheap t) (.get_assigned_memoffheap t))
       "assignedCpu" (.get_assigned_cpu t)})
    "schedulerDisplayResource" (*STORM-CONF* Config/SCHEDULER_DISPLAY_RESOURCE)}))

(defn topology-stats [window stats]
  (let [times (stats-times (:emitted stats))
        display-map (into {} (for [t times] [t pretty-uptime-sec]))
        display-map (assoc display-map ":all-time" (fn [_] "All time"))]
    (for [w (concat times [":all-time"])
          :let [disp ((display-map w) w)]]
      {"windowPretty" disp
       "window" w
       "emitted" (get-in stats [:emitted w])
       "transferred" (get-in stats [:transferred w])
       "completeLatency" (float-str (get-in stats [:complete-latencies w]))
       "acked" (get-in stats [:acked w])
       "failed" (get-in stats [:failed w])})))

(defn build-visualization [id window include-sys?]
  (thrift/with-configured-nimbus-connection nimbus
    (let [window (if window window ":all-time")
          topology-info (->> (doto
                               (GetInfoOptions.)
                               (.set_num_err_choice NumErrorsChoice/ONE))
                             (.getTopologyInfoWithOpts ^Nimbus$Client nimbus
                                                       id))
          storm-topology (.getTopology ^Nimbus$Client nimbus id)
          spout-executor-summaries (filter (partial spout-summary? storm-topology) (.get_executors topology-info))
          bolt-executor-summaries (filter (partial bolt-summary? storm-topology) (.get_executors topology-info))
          spout-comp-id->executor-summaries (group-by-comp spout-executor-summaries)
          bolt-comp-id->executor-summaries (group-by-comp bolt-executor-summaries)
          bolt-comp-id->executor-summaries (filter-key (mk-include-sys-fn include-sys?) bolt-comp-id->executor-summaries)
          id->spout-spec (.get_spouts storm-topology)
          id->bolt (.get_bolts storm-topology)
          visualizer-data (visualization-data (merge (hashmap-to-persistent id->spout-spec)
                                                     (hashmap-to-persistent id->bolt))
                                              spout-comp-id->executor-summaries
                                              bolt-comp-id->executor-summaries
                                              window
                                              id)]
       {"visualizationTable" (stream-boxes visualizer-data)})))

(defn- get-error-json
  [topo-id error-info secure?]
  (let [host (get-error-host error-info)
        port (get-error-port error-info)]
    {"lastError" (get-error-data error-info)
     "errorTime" (get-error-time error-info)
     "errorHost" host
     "errorPort" port
     "errorLapsedSecs" (get-error-time error-info)
     "errorWorkerLogLink" (worker-log-link host port topo-id secure?)}))

(defn- common-agg-stats-json
  "Returns a JSON representation of a common aggregated statistics."
  [^CommonAggregateStats common-stats]
  {"executors" (.get_num_executors common-stats)
   "tasks" (.get_num_tasks common-stats)
   "emitted" (.get_emitted common-stats)
   "transferred" (.get_transferred common-stats)
   "acked" (.get_acked common-stats)
   "failed" (.get_failed common-stats)})

(defmulti comp-agg-stats-json
  "Returns a JSON representation of aggregated statistics."
  (fn [_ _ [id ^ComponentAggregateStats s]] (.get_type s)))

(defmethod comp-agg-stats-json ComponentType/SPOUT
  [topo-id secure? [id ^ComponentAggregateStats s]]
  (let [^SpoutAggregateStats ss (.. s get_specific_stats get_spout)
        cs (.get_common_stats s)]
    (merge
      (common-agg-stats-json cs)
      (get-error-json topo-id (.get_last_error s) secure?)
      {"spoutId" id
       "encodedSpoutId" (url-encode id)
       "completeLatency" (float-str (.get_complete_latency_ms ss))})))

(defmethod comp-agg-stats-json ComponentType/BOLT
  [topo-id secure? [id ^ComponentAggregateStats s]]
  (let [^BoltAggregateStats ss (.. s get_specific_stats get_bolt)
        cs (.get_common_stats s)]
    (merge
      (common-agg-stats-json cs)
      (get-error-json topo-id (.get_last_error s) secure?)
      {"boltId" id
       "encodedBoltId" (url-encode id)
       "capacity" (float-str (.get_capacity ss))
       "executeLatency" (float-str (.get_execute_latency_ms ss))
       "executed" (.get_executed ss)
       "processLatency" (float-str (.get_process_latency_ms ss))})))

(defn- unpack-topology-page-info
  "Unpacks the serialized object to data structures"
  [^TopologyPageInfo topo-info window secure?]
  (let [id (.get_id topo-info)
        ^TopologyStats topo-stats (.get_topology_stats topo-info)
        stat->window->number
          {:emitted (.get_window_to_emitted topo-stats)
           :transferred (.get_window_to_transferred topo-stats)
           :complete-latencies (.get_window_to_complete_latencies_ms topo-stats)
           :acked (.get_window_to_acked topo-stats)
           :failed (.get_window_to_failed topo-stats)}
        topo-stats (topology-stats window stat->window->number)
        [debugEnabled
         samplingPct] (if-let [debug-opts (.get_debug_options topo-info)]
                        [(.is_enable debug-opts)
                         (.get_samplingpct debug-opts)])
        uptime (.get_uptime_secs topo-info)]
    {"id" id
     "encodedId" (url-encode id)
     "owner" (.get_owner topo-info)
     "name" (.get_name topo-info)
     "status" (.get_status topo-info)
     "uptime" (pretty-uptime-sec uptime)
     "uptimeSeconds" uptime
     "tasksTotal" (.get_num_tasks topo-info)
     "workersTotal" (.get_num_workers topo-info)
     "executorsTotal" (.get_num_executors topo-info)
     "schedulerInfo" (.get_sched_status topo-info)
     "requestedMemOnHeap" (.get_requested_memonheap topo-info)
     "requestedMemOffHeap" (.get_requested_memoffheap topo-info)
     "requestedCpu" (.get_requested_cpu topo-info)
     "assignedMemOnHeap" (.get_assigned_memonheap topo-info)
     "assignedMemOffHeap" (.get_assigned_memoffheap topo-info)
     "assignedTotalMem" (+ (.get_assigned_memonheap topo-info) (.get_assigned_memoffheap topo-info))
     "assignedCpu" (.get_assigned_cpu topo-info)
     "topologyStats" topo-stats
     "spouts" (map (partial comp-agg-stats-json id secure?)
                   (.get_id_to_spout_agg_stats topo-info))
     "bolts" (map (partial comp-agg-stats-json id secure?)
                  (.get_id_to_bolt_agg_stats topo-info))
     "configuration" (.get_topology_conf topo-info)
     "debug" (or debugEnabled false)
     "samplingPct" (or samplingPct 10)
     "replicationCount" (.get_replication_count topo-info)}))

(defn exec-host-port
  [executors]
  (for [^ExecutorSummary e executors]
    {"host" (.get_host e)
     "port" (.get_port e)}))

(defn worker-host-port
  "Get the set of all worker host/ports"
  [id]
  (thrift/with-configured-nimbus-connection nimbus
    (distinct (exec-host-port (.get_executors (get-topology-info nimbus id))))))

(defn topology-page [id window include-sys? user secure?]
  (thrift/with-configured-nimbus-connection nimbus
    (let [window (if window window ":all-time")
          window-hint (window-hint window)
          topo-page-info (.getTopologyPageInfo ^Nimbus$Client nimbus
                                               id
                                               window
                                               include-sys?)
          topology-conf (from-json (.get_topology_conf topo-page-info))
          msg-timeout (topology-conf TOPOLOGY-MESSAGE-TIMEOUT-SECS)]
      (merge
       (unpack-topology-page-info topo-page-info window secure?)
       {"user" user
        "window" window
        "windowHint" window-hint
        "msgTimeout" msg-timeout
        "configuration" topology-conf
        "visualizationTable" []
        "schedulerDisplayResource" (*STORM-CONF* Config/SCHEDULER_DISPLAY_RESOURCE)}))))

(defn component-errors
  [errors-list topology-id secure?]
  (let [errors (->> errors-list
                    (sort-by #(.get_error_time_secs ^ErrorInfo %))
                    reverse)]
    {"componentErrors"
     (for [^ErrorInfo e errors]
       {"time" (* 1000 (long (.get_error_time_secs e)))
        "errorHost" (.get_host e)
        "errorPort"  (.get_port e)
        "errorWorkerLogLink"  (worker-log-link (.get_host e)
                                               (.get_port e)
                                               topology-id
                                               secure?)
        "errorLapsedSecs" (get-error-time e)
        "error" (.get_error e)})}))

(defmulti unpack-comp-agg-stat
  (fn [[_ ^ComponentAggregateStats s]] (.get_type s)))

(defmethod unpack-comp-agg-stat ComponentType/BOLT
  [[window ^ComponentAggregateStats s]]
  (let [^CommonAggregateStats comm-s (.get_common_stats s)
        ^SpecificAggregateStats spec-s (.get_specific_stats s)
        ^BoltAggregateStats bolt-s (.get_bolt spec-s)]
    {"window" window
     "windowPretty" (window-hint window)
     "emitted" (.get_emitted comm-s)
     "transferred" (.get_transferred comm-s)
     "acked" (.get_acked comm-s)
     "failed" (.get_failed comm-s)
     "executeLatency" (float-str (.get_execute_latency_ms bolt-s))
     "processLatency"  (float-str (.get_process_latency_ms bolt-s))
     "executed" (.get_executed bolt-s)
     "capacity" (float-str (.get_capacity bolt-s))}))

(defmethod unpack-comp-agg-stat ComponentType/SPOUT
  [[window ^ComponentAggregateStats s]]
  (let [^CommonAggregateStats comm-s (.get_common_stats s)
        ^SpecificAggregateStats spec-s (.get_specific_stats s)
        ^SpoutAggregateStats spout-s (.get_spout spec-s)]
    {"window" window
     "windowPretty" (window-hint window)
     "emitted" (.get_emitted comm-s)
     "transferred" (.get_transferred comm-s)
     "acked" (.get_acked comm-s)
     "failed" (.get_failed comm-s)
     "completeLatency" (float-str (.get_complete_latency_ms spout-s))}))

(defn- unpack-bolt-input-stat
  [[^GlobalStreamId s ^ComponentAggregateStats stats]]
  (let [^SpecificAggregateStats sas (.get_specific_stats stats)
        ^BoltAggregateStats bas (.get_bolt sas)
        ^CommonAggregateStats cas (.get_common_stats stats)
        comp-id (.get_componentId s)]
    {"component" comp-id
     "encodedComponentId" (url-encode comp-id)
     "stream" (.get_streamId s)
     "executeLatency" (float-str (.get_execute_latency_ms bas))
     "processLatency" (float-str (.get_process_latency_ms bas))
     "executed" (nil-to-zero (.get_executed bas))
     "acked" (nil-to-zero (.get_acked cas))
     "failed" (nil-to-zero (.get_failed cas))}))

(defmulti unpack-comp-output-stat
  (fn [[_ ^ComponentAggregateStats s]] (.get_type s)))

(defmethod unpack-comp-output-stat ComponentType/BOLT
  [[stream-id ^ComponentAggregateStats stats]]
  (let [^CommonAggregateStats cas (.get_common_stats stats)]
    {"stream" stream-id
     "emitted" (nil-to-zero (.get_emitted cas))
     "transferred" (nil-to-zero (.get_transferred cas))}))

(defmethod unpack-comp-output-stat ComponentType/SPOUT
  [[stream-id ^ComponentAggregateStats stats]]
  (let [^CommonAggregateStats cas (.get_common_stats stats)
        ^SpecificAggregateStats spec-s (.get_specific_stats stats)
        ^SpoutAggregateStats spout-s (.get_spout spec-s)]
    {"stream" stream-id
     "emitted" (nil-to-zero (.get_emitted cas))
     "transferred" (nil-to-zero (.get_transferred cas))
     "completeLatency" (float-str (.get_complete_latency_ms spout-s))
     "acked" (nil-to-zero (.get_acked cas))
     "failed" (nil-to-zero (.get_failed cas))}))

(defmulti unpack-comp-exec-stat
  (fn [_ _ ^ComponentAggregateStats cas] (.get_type (.get_stats ^ExecutorAggregateStats cas))))

(defmethod unpack-comp-exec-stat ComponentType/BOLT
  [topology-id secure? ^ExecutorAggregateStats eas]
  (let [^ExecutorSummary summ (.get_exec_summary eas)
        ^ExecutorInfo info (.get_executor_info summ)
        ^ComponentAggregateStats stats (.get_stats eas)
        ^SpecificAggregateStats ss (.get_specific_stats stats)
        ^BoltAggregateStats bas (.get_bolt ss)
        ^CommonAggregateStats cas (.get_common_stats stats)
        host (.get_host summ)
        port (.get_port summ)
        exec-id (pretty-executor-info info)
        uptime (.get_uptime_secs summ)]
    {"id" exec-id
     "encodedId" (url-encode exec-id)
     "uptime" (pretty-uptime-sec uptime)
     "uptimeSeconds" uptime
     "host" host
     "port" port
     "emitted" (nil-to-zero (.get_emitted cas))
     "transferred" (nil-to-zero (.get_transferred cas))
     "capacity" (float-str (nil-to-zero (.get_capacity bas)))
     "executeLatency" (float-str (.get_execute_latency_ms bas))
     "executed" (nil-to-zero (.get_executed bas))
     "processLatency" (float-str (.get_process_latency_ms bas))
     "acked" (nil-to-zero (.get_acked cas))
     "failed" (nil-to-zero (.get_failed cas))
     "workerLogLink" (worker-log-link host port topology-id secure?)}))

(defmethod unpack-comp-exec-stat ComponentType/SPOUT
  [topology-id secure? ^ExecutorAggregateStats eas]
  (let [^ExecutorSummary summ (.get_exec_summary eas)
        ^ExecutorInfo info (.get_executor_info summ)
        ^ComponentAggregateStats stats (.get_stats eas)
        ^SpecificAggregateStats ss (.get_specific_stats stats)
        ^SpoutAggregateStats sas (.get_spout ss)
        ^CommonAggregateStats cas (.get_common_stats stats)
        host (.get_host summ)
        port (.get_port summ)
        exec-id (pretty-executor-info info)
        uptime (.get_uptime_secs summ)]
    {"id" exec-id
     "encodedId" (url-encode exec-id)
     "uptime" (pretty-uptime-sec uptime)
     "uptimeSeconds" uptime
     "host" host
     "port" port
     "emitted" (nil-to-zero (.get_emitted cas))
     "transferred" (nil-to-zero (.get_transferred cas))
     "completeLatency" (float-str (.get_complete_latency_ms sas))
     "acked" (nil-to-zero (.get_acked cas))
     "failed" (nil-to-zero (.get_failed cas))
     "workerLogLink" (worker-log-link host port topology-id secure?)}))

(defmulti unpack-component-page-info
  "Unpacks component-specific info to clojure data structures"
  (fn [^ComponentPageInfo info & _]
    (.get_component_type info)))

(defmethod unpack-component-page-info ComponentType/BOLT
  [^ComponentPageInfo info topology-id window include-sys? secure?]
  (merge
    {"boltStats" (map unpack-comp-agg-stat (.get_window_to_stats info))
     "inputStats" (map unpack-bolt-input-stat (.get_gsid_to_input_stats info))
     "outputStats" (map unpack-comp-output-stat (.get_sid_to_output_stats info))
     "executorStats" (map (partial unpack-comp-exec-stat topology-id secure?)
                          (.get_exec_stats info))}
    (-> info .get_errors (component-errors topology-id secure?))))

(defmethod unpack-component-page-info ComponentType/SPOUT
  [^ComponentPageInfo info topology-id window include-sys? secure?]
  (merge
    {"spoutSummary" (map unpack-comp-agg-stat (.get_window_to_stats info))
     "outputStats" (map unpack-comp-output-stat (.get_sid_to_output_stats info))
     "executorStats" (map (partial unpack-comp-exec-stat topology-id secure?)
                          (.get_exec_stats info))}
    (-> info .get_errors (component-errors topology-id secure?))))

(defn get-active-profile-actions
  [nimbus topology-id component]
  (let [profile-actions  (.getComponentPendingProfileActions nimbus
                                               topology-id
                                               component
                                 ProfileAction/JPROFILE_STOP)
        latest-profile-actions (map clojurify-profile-request profile-actions)
        active-actions (map (fn [profile-action]
                              {"host" (:host profile-action)
                               "port" (str (:port profile-action))
                               "dumplink" (worker-dump-link (:host profile-action) (str (:port profile-action)) topology-id)
                               "timestamp" (str (- (:timestamp profile-action) (System/currentTimeMillis)))})
                            latest-profile-actions)]
    (log-message "Latest-active actions are: " (pr active-actions))
    active-actions))

(defn component-page
  [topology-id component window include-sys? user secure?]
  (thrift/with-configured-nimbus-connection nimbus
    (let [window (or window ":all-time")
          window-hint (window-hint window)
          comp-page-info (.getComponentPageInfo ^Nimbus$Client nimbus
                                                topology-id
                                                component
                                                window
                                                include-sys?)
          topology-conf (from-json (.getTopologyConf ^Nimbus$Client nimbus
                                                     topology-id))
          msg-timeout (topology-conf TOPOLOGY-MESSAGE-TIMEOUT-SECS)
          [debugEnabled
           samplingPct] (if-let [debug-opts (.get_debug_options comp-page-info)]
                          [(.is_enable debug-opts)
                           (.get_samplingpct debug-opts)])]
      (assoc
       (unpack-component-page-info comp-page-info
                                   topology-id
                                   window
                                   include-sys?
                                   secure?)
       "user" user
       "id" component
       "encodedId" (url-encode component)
       "name" (.get_topology_name comp-page-info)
       "executors" (.get_num_executors comp-page-info)
       "tasks" (.get_num_tasks comp-page-info)
       "topologyId" topology-id
       "topologyStatus" (.get_topology_status comp-page-info)
       "encodedTopologyId" (url-encode topology-id)
       "window" window
       "componentType" (-> comp-page-info .get_component_type str lower-case)
       "windowHint" window-hint
       "debug" (or debugEnabled false)
       "samplingPct" (or samplingPct 10)
       "eventLogLink" (event-log-link topology-id
                                      component
                                      (.get_eventlog_host comp-page-info)
                                      (.get_eventlog_port comp-page-info)
                                      secure?)
       "profileActionEnabled" (*STORM-CONF* WORKER-PROFILER-ENABLED)
       "profilerActive" (if (*STORM-CONF* WORKER-PROFILER-ENABLED)
                          (get-active-profile-actions nimbus topology-id component)
                          [])))))
    
(defn- level-to-dict [level]
  (if level
    (let [timeout (.get_reset_log_level_timeout_secs level)
          timeout-epoch (.get_reset_log_level_timeout_epoch level)
          target-level (.get_target_log_level level)
          reset-level (.get_reset_log_level level)]
          {"target_level" (.toString (Level/toLevel target-level))
           "reset_level" (.toString (Level/toLevel reset-level))
           "timeout" timeout
           "timeout_epoch" timeout-epoch})))

(defn log-config [topology-id]
  (thrift/with-configured-nimbus-connection
    nimbus
    (let [log-config (.getLogConfig ^Nimbus$Client nimbus topology-id)
          named-logger-levels (into {}
                                (for [[key val] (.get_named_logger_level log-config)]
                                  [(str key) (level-to-dict val)]))]
      {"namedLoggerLevels" named-logger-levels})))

(defn topology-config [topology-id]
  (thrift/with-configured-nimbus-connection nimbus
    (from-json (.getTopologyConf ^Nimbus$Client nimbus topology-id))))

(defn topology-op-response [topology-id op]
  {"topologyOperation" op,
   "topologyId" topology-id,
   "status" "success"
   })

(defn component-op-response [topology-id component-id op]
  {"topologyOperation" op,
   "topologyId" topology-id,
   "componentId" component-id,
   "status" "success"
   })

(defn check-include-sys?
  [sys?]
  (if (or (nil? sys?) (= "false" sys?)) false true))

(def http-creds-handler (AuthUtils/GetUiHttpCredentialsPlugin *STORM-CONF*))

(defn populate-context!
  "Populate the Storm RequestContext from an servlet-request. This should be called in each handler"
  [servlet-request]
    (when http-creds-handler
      (.populateContext http-creds-handler (ReqContext/context) servlet-request)))

(defn get-user-name
  [servlet-request]
  (.getUserName http-creds-handler servlet-request))

(defroutes main-routes
  (GET "/api/v1/cluster/configuration" [& m]
    (mark! ui:num-cluster-configuration-http-requests)
    (json-response (cluster-configuration)
                   (:callback m) :serialize-fn identity))
  (GET "/api/v1/cluster/summary" [:as {:keys [cookies servlet-request]} & m]
    (mark! ui:num-cluster-summary-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "getClusterInfo")
    (let [user (get-user-name servlet-request)]
      (json-response (assoc (cluster-summary user)
                          "bugtracker-url" (*STORM-CONF* UI-PROJECT-BUGTRACKER-URL)
                          "central-log-url" (*STORM-CONF* UI-CENTRAL-LOGGING-URL)) (:callback m))))
  (GET "/api/v1/nimbus/summary" [:as {:keys [cookies servlet-request]} & m]
    (mark! ui:num-nimbus-summary-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "getClusterInfo")
    (json-response (nimbus-summary) (:callback m)))
  (GET "/api/v1/history/summary" [:as {:keys [cookies servlet-request]} & m]
    (let [user (.getUserName http-creds-handler servlet-request)]
      (json-response (topology-history-info user) (:callback m))))
  (GET "/api/v1/supervisor/summary" [:as {:keys [cookies servlet-request]} & m]
    (mark! ui:num-supervisor-summary-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "getClusterInfo")
    (json-response (assoc (supervisor-summary)
                     "logviewerPort" (*STORM-CONF* LOGVIEWER-PORT)) (:callback m)))
  (GET "/api/v1/topology/summary" [:as {:keys [cookies servlet-request]} & m]
    (mark! ui:num-all-topologies-summary-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "getClusterInfo")
    (json-response (all-topologies-summary) (:callback m)))
  (GET  "/api/v1/topology-workers/:id" [:as {:keys [cookies servlet-request]} id & m]
    (let [id (url-decode id)]
      (json-response {"hostPortList" (worker-host-port id)
                      "logviewerPort" (*STORM-CONF* LOGVIEWER-PORT)} (:callback m))))
  (GET "/api/v1/topology/:id" [:as {:keys [cookies servlet-request scheme]} id & m]
    (mark! ui:num-topology-page-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "getTopology" (topology-config id))
    (let [user (get-user-name servlet-request)]
      (json-response (topology-page id (:window m) (check-include-sys? (:sys m)) user (= scheme :https)) (:callback m))))
  (GET "/api/v1/topology/:id/visualization-init" [:as {:keys [cookies servlet-request]} id & m]
    (mark! ui:num-build-visualization-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "getTopology" (topology-config id))
    (json-response (build-visualization id (:window m) (check-include-sys? (:sys m))) (:callback m)))
  (GET "/api/v1/topology/:id/visualization" [:as {:keys [cookies servlet-request]} id & m]
    (mark! ui:num-mk-visualization-data-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "getTopology" (topology-config id))
    (json-response (mk-visualization-data id (:window m) (check-include-sys? (:sys m))) (:callback m)))
  (GET "/api/v1/topology/:id/component/:component" [:as {:keys [cookies servlet-request scheme]} id component & m]
    (mark! ui:num-component-page-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "getTopology" (topology-config id))
    (let [user (get-user-name servlet-request)]
      (json-response
          (component-page id component (:window m) (check-include-sys? (:sys m)) user (= scheme :https))
          (:callback m))))
  (GET "/api/v1/topology/:id/logconfig" [:as {:keys [cookies servlet-request]} id & m]
    (mark! ui:num-log-config-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "getTopology" (topology-config id))
       (json-response (log-config id) (:callback m)))
  (POST "/api/v1/topology/:id/activate" [:as {:keys [cookies servlet-request]} id & m]
    (mark! ui:num-activate-topology-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "activate" (topology-config id))
    (thrift/with-configured-nimbus-connection nimbus
       (let [tplg (->> (doto
                        (GetInfoOptions.)
                        (.set_num_err_choice NumErrorsChoice/NONE))
                      (.getTopologyInfoWithOpts ^Nimbus$Client nimbus id))
            name (.get_name tplg)]
        (.activate nimbus name)
        (log-message "Activating topology '" name "'")))
    (json-response (topology-op-response id "activate") (m "callback")))
  (POST "/api/v1/topology/:id/deactivate" [:as {:keys [cookies servlet-request]} id & m]
    (mark! ui:num-deactivate-topology-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "deactivate" (topology-config id))
    (thrift/with-configured-nimbus-connection nimbus
        (let [tplg (->> (doto
                        (GetInfoOptions.)
                        (.set_num_err_choice NumErrorsChoice/NONE))
                      (.getTopologyInfoWithOpts ^Nimbus$Client nimbus id))
            name (.get_name tplg)]
        (.deactivate nimbus name)
        (log-message "Deactivating topology '" name "'")))
    (json-response (topology-op-response id "deactivate") (m "callback")))
  (POST "/api/v1/topology/:id/debug/:action/:spct" [:as {:keys [cookies servlet-request]} id action spct & m]
    (mark! ui:num-debug-topology-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "debug" (topology-config id))
    (thrift/with-configured-nimbus-connection nimbus
        (let [tplg (->> (doto
                        (GetInfoOptions.)
                        (.set_num_err_choice NumErrorsChoice/NONE))
                   (.getTopologyInfoWithOpts ^Nimbus$Client nimbus id))
            name (.get_name tplg)
            enable? (= "enable" action)]
        (.debug nimbus name "" enable? (Integer/parseInt spct))
        (log-message "Debug topology [" name "] action [" action "] sampling pct [" spct "]")))
     (json-response (topology-op-response id (str "debug/" action)) (m "callback")))
  (POST "/api/v1/topology/:id/component/:component/debug/:action/:spct" [:as {:keys [cookies servlet-request]} id component action spct & m]
    (mark! ui:num-component-op-response-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "debug" (topology-config id))
    (thrift/with-configured-nimbus-connection nimbus
      (let [tplg (->> (doto
                        (GetInfoOptions.)
                        (.set_num_err_choice NumErrorsChoice/NONE))
                   (.getTopologyInfoWithOpts ^Nimbus$Client nimbus id))
            name (.get_name tplg)
            enable? (= "enable" action)]
        (.debug nimbus name component enable? (Integer/parseInt spct))
        (log-message "Debug topology [" name "] component [" component "] action [" action "] sampling pct [" spct "]")))
    (json-response (component-op-response id component (str "/debug/" action)) (m "callback")))
  (POST "/api/v1/topology/:id/rebalance/:wait-time" [:as {:keys [cookies servlet-request]} id wait-time & m]
    (mark! ui:num-topology-op-response-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "rebalance" (topology-config id))
    (thrift/with-configured-nimbus-connection nimbus
      (let [tplg (->> (doto
                        (GetInfoOptions.)
                        (.set_num_err_choice NumErrorsChoice/NONE))
                      (.getTopologyInfoWithOpts ^Nimbus$Client nimbus id))
            name (.get_name tplg)
            rebalance-options (m "rebalanceOptions")
            options (RebalanceOptions.)]
        (.set_wait_secs options (Integer/parseInt wait-time))
        (if (and (not-nil? rebalance-options) (contains? rebalance-options "numWorkers"))
          (.set_num_workers options (Integer/parseInt (.toString (rebalance-options "numWorkers")))))
        (if (and (not-nil? rebalance-options) (contains? rebalance-options "executors"))
          (doseq [keyval (rebalance-options "executors")]
            (.put_to_num_executors options (key keyval) (Integer/parseInt (.toString (val keyval))))))
        (.rebalance nimbus name options)
        (log-message "Rebalancing topology '" name "' with wait time: " wait-time " secs")))
    (json-response (topology-op-response id "rebalance") (m "callback")))
  (POST "/api/v1/topology/:id/kill/:wait-time" [:as {:keys [cookies servlet-request]} id wait-time & m]
    (mark! ui:num-topology-op-response-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "killTopology" (topology-config id))
    (thrift/with-configured-nimbus-connection nimbus
      (let [tplg (->> (doto
                        (GetInfoOptions.)
                        (.set_num_err_choice NumErrorsChoice/NONE))
                      (.getTopologyInfoWithOpts ^Nimbus$Client nimbus id))
            name (.get_name tplg)
            options (KillOptions.)]
        (.set_wait_secs options (Integer/parseInt wait-time))
        (.killTopologyWithOpts nimbus name options)
        (log-message "Killing topology '" name "' with wait time: " wait-time " secs")))
    (json-response (topology-op-response id "kill") (m "callback")))
  (POST "/api/v1/topology/:id/logconfig" [:as {:keys [cookies servlet-request]} id namedLoggerLevels & m]
    (mark! ui:num-topology-op-response-http-requests)
    (populate-context! servlet-request)
    (assert-authorized-user "setLogConfig" (topology-config id))
    (thrift/with-configured-nimbus-connection
      nimbus
      (let [new-log-config (LogConfig.)]
        (doseq [[key level] namedLoggerLevels]
            (let [logger-name (str key)
                  target-level (.get level "target_level")
                  timeout (or (.get level "timeout") 0)
                  named-logger-level (LogLevel.)]
              ;; if target-level is nil, do not set it, user wants to clear
              (log-message "The target level for " logger-name " is " target-level)
              (if (nil? target-level)
                (do
                  (.set_action named-logger-level LogLevelAction/REMOVE)
                  (.unset_target_log_level named-logger-level))
                (do
                  (.set_action named-logger-level LogLevelAction/UPDATE)
                  ;; the toLevel here ensures the string we get is valid
                  (.set_target_log_level named-logger-level (.name (Level/toLevel target-level)))
                  (.set_reset_log_level_timeout_secs named-logger-level timeout)))
              (log-message "Adding this " logger-name " " named-logger-level " to " new-log-config)
              (.put_to_named_logger_level new-log-config logger-name named-logger-level)))
        (log-message "Setting topology " id " log config " new-log-config)
        (.setLogConfig nimbus id new-log-config)
        (json-response (log-config id) (m "callback")))))

  (GET "/api/v1/topology/:id/profiling/start/:host-port/:timeout"
       [:as {:keys [servlet-request]} id host-port timeout & m]
       (thrift/with-configured-nimbus-connection nimbus
         (let [user (.getUserName http-creds-handler servlet-request)
               topology-conf (from-json
                              (.getTopologyConf ^Nimbus$Client nimbus id))]
           (assert-authorized-user "setWorkerProfiler" (topology-config id))
           (assert-authorized-profiler-action "start"))

         (let [[host, port] (split host-port #":")
               nodeinfo (NodeInfo. host (set [(Long. port)]))
               timestamp (+ (System/currentTimeMillis) (* 60000 (Long. timeout)))
               request (ProfileRequest. nodeinfo
                                        ProfileAction/JPROFILE_STOP)]
           (.set_time_stamp request timestamp)
           (.setWorkerProfiler nimbus id request)
           (json-response {"status" "ok"
                           "id" host-port
                           "timeout" timeout
                           "dumplink" (worker-dump-link
                                       host
                                       port
                                       id)}
                          (m "callback")))))

  (GET "/api/v1/topology/:id/profiling/stop/:host-port"
       [:as {:keys [servlet-request]} id host-port & m]
       (thrift/with-configured-nimbus-connection nimbus
         (let [user (.getUserName http-creds-handler servlet-request)
               topology-conf (from-json
                              (.getTopologyConf ^Nimbus$Client nimbus id))]
           (assert-authorized-user "setWorkerProfiler" (topology-config id))
           (assert-authorized-profiler-action "stop"))
         (let [[host, port] (split host-port #":")
               nodeinfo (NodeInfo. host (set [(Long. port)]))
               timestamp 0
               request (ProfileRequest. nodeinfo
                                        ProfileAction/JPROFILE_STOP)]
           (.set_time_stamp request timestamp)
           (.setWorkerProfiler nimbus id request)
           (json-response {"status" "ok"
                           "id" host-port}
                          (m "callback")))))
  
  (GET "/api/v1/topology/:id/profiling/dumpprofile/:host-port"
       [:as {:keys [servlet-request]} id host-port & m]
       (thrift/with-configured-nimbus-connection nimbus
         (let [user (.getUserName http-creds-handler servlet-request)
               topology-conf (from-json
                              (.getTopologyConf ^Nimbus$Client nimbus id))]
           (assert-authorized-user "setWorkerProfiler" (topology-config id))
           (assert-authorized-profiler-action "dumpprofile"))
         (let [[host, port] (split host-port #":")
               nodeinfo (NodeInfo. host (set [(Long. port)]))
               timestamp (System/currentTimeMillis)
               request (ProfileRequest. nodeinfo
                                        ProfileAction/JPROFILE_DUMP)]
           (.set_time_stamp request timestamp)
           (.setWorkerProfiler nimbus id request)
           (json-response {"status" "ok"
                           "id" host-port}
                          (m "callback")))))

  (GET "/api/v1/topology/:id/profiling/dumpjstack/:host-port"
       [:as {:keys [servlet-request]} id host-port & m]
       (thrift/with-configured-nimbus-connection nimbus
         (let [user (.getUserName http-creds-handler servlet-request)
               topology-conf (from-json
                              (.getTopologyConf ^Nimbus$Client nimbus id))]
           (assert-authorized-user "setWorkerProfiler" (topology-config id))
           (assert-authorized-profiler-action "dumpjstack"))
         (let [[host, port] (split host-port #":")
               nodeinfo (NodeInfo. host (set [(Long. port)]))
               timestamp (System/currentTimeMillis)
               request (ProfileRequest. nodeinfo
                                        ProfileAction/JSTACK_DUMP)]
           (.set_time_stamp request timestamp)
           (.setWorkerProfiler nimbus id request)
           (json-response {"status" "ok"
                           "id" host-port}
                          (m "callback")))))

  (GET "/api/v1/topology/:id/profiling/restartworker/:host-port"
       [:as {:keys [servlet-request]} id host-port & m]
       (thrift/with-configured-nimbus-connection nimbus
         (let [user (.getUserName http-creds-handler servlet-request)
               topology-conf (from-json
                              (.getTopologyConf ^Nimbus$Client nimbus id))]
           (assert-authorized-user "setWorkerProfiler" (topology-config id))
           (assert-authorized-profiler-action "restartworker"))
         (let [[host, port] (split host-port #":")
               nodeinfo (NodeInfo. host (set [(Long. port)]))
               timestamp (System/currentTimeMillis)
               request (ProfileRequest. nodeinfo
                                        ProfileAction/JVM_RESTART)]
           (.set_time_stamp request timestamp)
           (.setWorkerProfiler nimbus id request)
           (json-response {"status" "ok"
                           "id" host-port}
                          (m "callback")))))
       
  (GET "/api/v1/topology/:id/profiling/dumpheap/:host-port"
       [:as {:keys [servlet-request]} id host-port & m]
       (thrift/with-configured-nimbus-connection nimbus
         (let [user (.getUserName http-creds-handler servlet-request)
               topology-conf (from-json
                              (.getTopologyConf ^Nimbus$Client nimbus id))]
           (assert-authorized-user "setWorkerProfiler" (topology-config id))
           (assert-authorized-profiler-action "dumpheap"))
         (let [[host, port] (split host-port #":")
               nodeinfo (NodeInfo. host (set [(Long. port)]))
               timestamp (System/currentTimeMillis)
               request (ProfileRequest. nodeinfo
                                        ProfileAction/JMAP_DUMP)]
           (.set_time_stamp request timestamp)
           (.setWorkerProfiler nimbus id request)
           (json-response {"status" "ok"
                           "id" host-port}
                          (m "callback")))))
  
  (GET "/" [:as {cookies :cookies}]
    (mark! ui:num-main-page-http-requests)
    (resp/redirect "/index.html"))
  (route/resources "/")
  (route/not-found "Page not found"))

(defn catch-errors
  [handler]
  (fn [request]
    (try
      (handler request)
      (catch Exception ex
        (json-response (exception->json ex) ((:query-params request) "callback") :status 500)))))

(def app
  (handler/site (-> main-routes
                    (wrap-json-params)
                    (wrap-multipart-params)
                    (wrap-reload '[backtype.storm.ui.core])
                    requests-middleware
                    catch-errors)))

(defn start-server!
  []
  (try
    (let [conf *STORM-CONF*
          header-buffer-size (int (.get conf UI-HEADER-BUFFER-BYTES))
          filters-confs [{:filter-class (conf UI-FILTER)
                          :filter-params (conf UI-FILTER-PARAMS)}]
          https-port (if (not-nil? (conf UI-HTTPS-PORT)) (conf UI-HTTPS-PORT) 0)
          https-ks-path (conf UI-HTTPS-KEYSTORE-PATH)
          https-ks-password (conf UI-HTTPS-KEYSTORE-PASSWORD)
          https-ks-type (conf UI-HTTPS-KEYSTORE-TYPE)
          https-key-password (conf UI-HTTPS-KEY-PASSWORD)
          https-ts-path (conf UI-HTTPS-TRUSTSTORE-PATH)
          https-ts-password (conf UI-HTTPS-TRUSTSTORE-PASSWORD)
          https-ts-type (conf UI-HTTPS-TRUSTSTORE-TYPE)
          https-want-client-auth (conf UI-HTTPS-WANT-CLIENT-AUTH)
          https-need-client-auth (conf UI-HTTPS-NEED-CLIENT-AUTH)]
      (start-metrics-reporters)
      (storm-run-jetty {:port (conf UI-PORT)
                        :host (conf UI-HOST)
                        :https-port https-port
                        :configurator (fn [server]
                                        (config-ssl server
                                                    https-port
                                                    https-ks-path
                                                    https-ks-password
                                                    https-ks-type
                                                    https-key-password
                                                    https-ts-path
                                                    https-ts-password
                                                    https-ts-type
                                                    https-need-client-auth
                                                    https-want-client-auth)
                                        (doseq [connector (.getConnectors server)]
                                          (.setRequestHeaderSize connector header-buffer-size))
                                        (config-filter server app filters-confs))}))
   (catch Exception ex
     (log-error ex))))

(defn -main [] (start-server!))