 (defn try-complete-job [pipeline event]
   (when (sentinel-found? event)
     (if (p-ext/drained? pipeline event)
-      (complete-job event)
+      (do (complete-job event)
+          (extensions/emit (:onyx.core/monitoring) {:event :peer/try-complete-job}))
       (p-ext/retry-message pipeline event (sentinel-id event))))
   event)
 
 (defn strip-sentinel
   [event]
   (if (= (:onyx/type (:onyx.core/task-map event)) :input)
-    (update-in event
-               [:onyx.core/batch]
-               (fn [batch]
-                 (remove (fn [v] (= :done (:message v)))
-                         batch)))
+    (do (extensions/emit (:onyx.core/monitoring) {:event :peer/strip-sentinel})
+        (update-in event
+                   [:onyx.core/batch]
+                   (fn [batch]
+                     (remove (fn [v] (= :done (:message v)))
+                             batch))))
     event))
 
 (defn collect-next-segments [f input]

              [rotating-seq.core :as rsc]
              [onyx.log.commands.common :as common]
              [onyx.log.entry :as entry]
              [onyx.static.planning :refer [find-task find-task-fast build-pred-fn]]
              [onyx.messaging.acking-daemon :as acker]
              [onyx.peer.pipeline-extensions :as p-ext]

            (swap! state dissoc k)
            (extensions/close-peer-connection (:onyx.core/messenger event) 
                                              event 
                                              (:link (get (:links snapshot) k)))
            (extensions/emit (:onyx.core/monitoring event) {:event :peer/gc-peer-link})))
        (catch InterruptedException e
          (throw e))
        (catch Throwable e
