     (handler {:component component :stream-id stream-id :events values})))
 
 
+(defn pipeline-stats-wrapper [component stream-id pipeline]
+  (let [job-name (-> component :config :job :job-name)
+        prefix   (str "pipeline." job-name "." (name stream-id))
+        track-time-name2 (str prefix ".pipeline-processing.time")]
+    (fn [state events]
+      (track-time track-time-name2
+                  (pipeline state events)))))
+
+
+(defn stream-stats-wrapper [component stream-id handler]
+  (let [job-name (-> component :config :job :job-name)
+        prefix   (str "pipeline." job-name "." (name stream-id))
+        track-time-name (str prefix ".overall-processing.time")
+        total-size-in (count-tracker (str prefix ".in.total-size"))
+        dist-size-in (distribution-tracker (str prefix ".in.size"))
+        size-in-tracker (invariant #(let [sz (count %)]
+                                      (total-size-in sz)
+                                      (dist-size-in sz)))]
+    (fn [events]
+      (track-time track-time-name
+       (handler (map size-in-tracker events))))))
+
+
 (defn- standard-stream-wrappers
   [component stream-id handler]
   (let [{:keys [format input-partitions]} (stream-config component stream-id)]
     (->> handler
          (message-only-wrapper)
          (format-wrapper format)
+         (stream-stats-wrapper component stream-id)
          (partition-filter-wrapper input-partitions))))
 
 

       (component-stream-processor-wrapper component stream-id handler)

       :samsara-factory
       (let [handler' (handler (:config component))]
         (topic-process-wrapper component stream-id handler'))

       :raw
       handler))))

(ns samsara-core.kernel2
  (:refer-clojure :exclude [var-get var-set])
  (:require [samsara.trackit :refer [track-time track-rate new-registry
                                     count-tracker distribution-tracker
                                     track-pass-count] :as trk])
  (:require [clojure.string :as str]
            [com.stuartsierra.component :as component]
            [moebius
             [core :refer [inject-if where]]
             [kv :as kv]]
            [samsara.utils :refer [from-json invariant to-json]]
            [taoensso.timbre :as log]))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;;                                                                            ;;
