307 - reverted some changes

Author: mmacai

lib/rig/event_stream/kafka_to_filter.ex

@@ -19,7 +19,7 @@ defmodule Rig.EventStream.KafkaToFilter do
   def kafka_handler(body, headers) do
     case Cloudevents.from_kafka_message(body, headers) do
       {:ok, cloud_event} ->
-        Tracing.CloudEvent.with_child_span "kafka_to_filter", cloud_event do
+        Tracing.CloudEvent.with_child_span_temp "kafka_to_filter", cloud_event do
           cloud_event =
             cloud_event
             |> Tracing.append_context_with_mode(Tracing.context(), :private)

lib/rig/tracing.ex

@@ -90,7 +90,8 @@ defmodule RIG.Tracing do
     end
   end
 
-  # Temporary function for Kafka as we are transitioning to new cloudevents library, once everything is migrated, we can remove function above
+  # Temporary function for Kafka as we are transitioning to new cloudevents library,
+  # once everything is migrated, we can remove function above
   @spec append_context_with_mode(Cloudevents.t(), t(), mode :: atom()) ::
           CloudEvent.t()
   def append_context_with_mode(cloudevent, context, mode) do

lib/rig/tracing/cloud_event.ex

@@ -2,6 +2,7 @@ defmodule RIG.Tracing.CloudEvent do
   @moduledoc """
   Distributed Tracing instrumenter for cloudevents
   """
+  alias RigCloudEvents.Parser.PartialParser
 
   @doc "Like Opencensus.Trace.with_child_span (https://hexdocs.pm/opencensus_elixir/Opencensus.Trace.html#with_child_span/3),
   but for cloudevents distributed tracing extension defined in https://github.com/cloudevents/spec/blob/master/extensions/distributed-tracing.md.
@@ -14,6 +15,56 @@ defmodule RIG.Tracing.CloudEvent do
     file = __CALLER__.file
     function = format_function(__CALLER__.function)
 
+    computed_attributes =
+      compute_attributes(attributes, %{
+        line: line,
+        module: module,
+        file: file,
+        function: function
+      })
+
+    quote do
+      tracecontext = []
+
+      tracecontext =
+        case PartialParser.context_attribute(unquote(event).parsed, "traceparent") do
+          {:ok, traceparent} -> Enum.concat(tracecontext, %{"traceparent" => traceparent})
+          _ -> tracecontext
+        end
+
+      tracecontext =
+        case PartialParser.context_attribute(unquote(event).parsed, "tracestate") do
+          {:ok, tracestate} -> Enum.concat(tracecontext, %{"tracestate" => tracestate})
+          _ -> tracecontext
+        end
+
+      parent_span_ctx = :oc_propagation_http_tracecontext.from_headers(tracecontext)
+
+      new_span_ctx =
+        :oc_trace.start_span(unquote(label), parent_span_ctx, %{
+          :attributes => unquote(computed_attributes)
+        })
+
+      :ocp.with_span_ctx(new_span_ctx)
+
+      try do
+        unquote(block)
+      after
+        :oc_trace.finish_span(new_span_ctx)
+        :ocp.with_span_ctx(parent_span_ctx)
+      end
+    end
+  end
+
+  # ---
+
+  # temporary function to handle new cloudevents library for Kafka
+  defmacro with_child_span_temp(label, event, attributes \\ quote(do: %{}), do: block) do
+    line = __CALLER__.line
+    module = __CALLER__.module
+    file = __CALLER__.file
+    function = format_function(__CALLER__.function)
+
     computed_attributes =
       compute_attributes(attributes, %{
         line: line,