Adds open tracing support #89

roobalimsab · 2019-09-24T06:39:42Z

This PR is to add open tracing support to Ziggurat. Open tracing enables to identify the time spent at various phases of the program flow.

Hence if the env variables are set for tracer, the entire behaviour of the handler function will be traced.

Tracing has been added to 3 flows as part of this PR:

Normal basic consume
Retry via rabbitmq
Produce to another kafka topic

mjayprateek · 2019-09-24T07:23:20Z

@roobalimsab Can you add some details to the PR description, to give us some context?

mjayprateek · 2019-09-30T08:07:19Z

src/ziggurat/mapper.clj


-(defn mapper-func [mapper-fn channels]
+(defn mapper-func [mapper-fn channels headers]


@roobalimsab Why can't we have headers in message-payload?

README.md

theanirudhvyas · 2019-09-30T09:22:38Z

README.md

+
+To enable tracing, the following config needs to be added to the `config.edn` under `:ziggurat` key.
+
+```clojure


this is not clojure format, edn format :p

the configuration below also has the same format. thats why used it. should i change it to :edn?

project.clj

src/ziggurat/messaging/producer.clj

mjayprateek · 2019-09-30T09:24:37Z

src/ziggurat/streams.clj

+                 (.start))]
+    (try
+      (.activate (.scopeManager tracer) span)
+      ((mapper-func handler-fn channels (:headers message-payload)) (->MessagePayload (:value message-payload) topic-entity))


@roobalimsab You could have put (:headers message-payload) in MessagePayload itself. That way you would not have needed to add an extra headers parameter everywhere in the code.

@theanirudhvyas WDYT?

we did not want to mess up with the existing message payload. we were not sure if the message payload will already have any headers. that is why kept it separate. if you don't see any issues, then we could merge it

the message-payload does not have any headers, you can merge it. Also, instead of calling it headers why not call it what it exactly is - tracing-ids

those are the kafka headers that we are passing across. hence the name

test/ziggurat/messaging/producer_test.clj

src/ziggurat/mapper.clj

src/ziggurat/tracer.clj

theanirudhvyas · 2019-09-30T09:52:41Z

src/ziggurat/messaging/consumer.clj

@@ -95,7 +95,7 @@
      (start-subscriber* (lch/open connection)
                         (get-in-config [:jobs :instant :prefetch-count])
                         (prefixed-queue-name topic-entity (get-in-config [:rabbit-mq :instant :queue-name]))
-                         (mpr/mapper-func mapper-fn channels)
+                         (mpr/mapper-func mapper-fn channels {})


If we are adding trace information to the rabbitmq headers, why are we not passing those to the mapper-func?

it only needs to be added when the message is put in rabbit the very first time. from there on it gets carried forward. so it is not needed in the subscriber. the producer will take care of it

src/ziggurat/messaging/producer.clj

theanirudhvyas · 2019-10-01T11:00:43Z

src/ziggurat/mapper.clj

@@ -80,7 +80,7 @@
            (sentry/report-error sentry-reporter e (str "Channel execution failed for " topic-entity-name " and for channel " channel-name))
            (metrics/increment-count default-namespace failure-metric additional-tags)))))))

-(defrecord MessagePayload [message topic-entity])
+(defrecord MessagePayload [message topic-entity headers])

 (s/defschema message-payload-schema


since you are adding headers to messagePayload, the schema will also change

the schema is used to validate the message at the consumer end right? we are not passing headers anywhere in the consumers. thats why did not add it here

theanirudhvyas · 2019-10-01T11:09:26Z

src/ziggurat/streams.clj

+      (.activate (.scopeManager tracer) span)
+      ((mapper-func handler-fn channels) (->MessagePayload (:value message-payload) topic-entity (:headers message-payload)))
+      (catch Exception e
+        (log/error "Exception while executing handler function " e))


why is there a generic exception catch here? This will lead to skipping the message and data loss right?

oh ya. this was added for testing the error scenario. will remove the catch

theanirudhvyas · 2019-10-01T11:10:55Z

src/ziggurat/streams.clj

+                 (.buildSpan "Message-Handler")
+                 (.asChildOf parent-ctx)
+                 (.withTag (.getKey Tags/SPAN_KIND) Tags/SPAN_KIND_CONSUMER)
+                 (.withTag (.getKey Tags/COMPONENT) "lambda")


what does this tag signify?? shouldn't it be ziggurat instead of lambda?

this is just to classify the consumers. yes it should be ziggurat.

theanirudhvyas · 2019-10-01T11:13:22Z

src/ziggurat/streams.clj

+                 (.start))]
+    (try
+      (.activate (.scopeManager tracer) span)
+      ((mapper-func handler-fn channels) (->MessagePayload (:value message-payload) topic-entity (:headers message-payload)))


instead of (:value message-payload) it should be (:value message)

the same with headers

theanirudhvyas · 2019-10-01T11:15:12Z

src/ziggurat/streams.clj

@@ -92,6 +97,22 @@
        additional-tags  {:topic_name topic-entity-name}]
    (.transform stream-builder (transformer-supplier metric-namespace oldest-processed-message-in-s additional-tags) (into-array [(.name (store-supplier-builder))]))))

+(defn- traced-handler-fn [handler-fn channels message-payload topic-entity]
+  (let [parent-ctx (TracingKafkaUtils/extractSpanContext (:headers message-payload) tracer)


right now the headers contain only the tracing headers. Will this still work when other information is added to the kafka headers?

yes this will work with all kafka headers

theanirudhvyas · 2019-10-01T11:17:41Z

src/ziggurat/mapper.clj

@@ -80,7 +80,7 @@
            (sentry/report-error sentry-reporter e (str "Channel execution failed for " topic-entity-name " and for channel " channel-name))
            (metrics/increment-count default-namespace failure-metric additional-tags)))))))

-(defrecord MessagePayload [message topic-entity])
+(defrecord MessagePayload [message topic-entity headers])

 (s/defschema message-payload-schema
  {:message                      s/Any


you'll also need to change the defschema as you changed the MessagePayload

is it mandatory? because we dont set it in consumers

theanirudhvyas · 2019-10-01T11:21:17Z

src/ziggurat/messaging/consumer.clj

@@ -21,7 +21,7 @@
    (catch Exception e
      (log/info "old message format read, converting to message-payload: " message)
      (let [retry-count (:retry-count message)
-            message-payload (mpr/->MessagePayload (dissoc message :retry-count) topic-entity)]
+            message-payload (mpr/->MessagePayload (dissoc message :retry-count) topic-entity {})]


why are we passing empty {} here? why not have headers as an optional parameter?

ok have added that as optional param to the schema

theanirudhvyas · 2019-10-01T11:22:55Z

src/ziggurat/messaging/producer.clj

@@ -99,7 +106,7 @@
  (let [{:keys [exchange-name]} (:instant (rabbitmq-config))
        topic-entity (:topic-entity message-payload)
        exchange-name (prefixed-channel-name topic-entity channel exchange-name)]
-    (publish exchange-name message-payload)))
+    (publish exchange-name message-payload nil)))


instead of passing nil, make it a multi-arity function

this was already there. removed the nil

theanirudhvyas · 2019-10-01T11:25:23Z

src/ziggurat/timestamp_transformer.clj

@@ -25,7 +25,7 @@
           (let [message-time (.timestamp processor-context)]
             (when (message-to-process? message-time oldest-processed-message-in-s)
               (calculate-and-report-kafka-delay metric-namespace message-time additional-tags)
-               (KeyValue/pair record-key record-value))))
+               (KeyValue/pair record-key {:value record-value :headers (.headers processor-context)}))))


IMO this message transformation should happen in streams.clj and not in timestamp_transformer. It does not make sense to me.

have extracted out a separate header_transformer.

theanirudhvyas · 2019-10-01T11:28:47Z

test/ziggurat/fixtures.clj

@@ -31,6 +33,17 @@
  (f)
  (mount/stop))

+(defn mount-tracer []
+  (with-redefs [tracer/create-tracer (fn [] (MockTracer.))]


instead of a mockTracer why can we not have a noOp tracer here?

mockTracer provides utility methods to check the number of spans completed in the test and so on. hence it is used to test the spans and corresponding tags

roobalimsab force-pushed the open-tracing branch from 11242ec to a0f425f Compare September 26, 2019 04:09

mjayprateek reviewed Sep 30, 2019

View reviewed changes