remove temporary function and use pattern matching instead

Author: Knappek

lib/rig/event_stream/kafka_to_filter.ex

@@ -5,6 +5,8 @@ defmodule Rig.EventStream.KafkaToFilter do
   """
   use Rig.KafkaConsumerSetup
 
+  import RIG.Tracing.CloudEvent
+
   alias Rig.EventFilter
   alias RIG.Tracing
 
@@ -19,10 +21,10 @@ 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_temp "kafka_to_filter", cloud_event do
+        Tracing.CloudEvent.with_child_span "kafka_to_filter", cloud_event do
           cloud_event =
             cloud_event
-            |> Tracing.append_context_with_mode(Tracing.context(), :private)
+            |> Tracing.append_context(Tracing.context(), :private)
 
           Logger.debug(fn -> inspect(cloud_event) end)
           EventFilter.forward_event(cloud_event)

lib/rig/event_stream/kafka_to_http.ex

@@ -5,6 +5,8 @@ defmodule Rig.EventStream.KafkaToHttp do
   """
   use Rig.KafkaConsumerSetup, [:targets]
 
+  import RIG.Tracing.CloudEvent
+
   alias HTTPoison
   alias RIG.Tracing
   alias RigCloudEvents.CloudEvent

lib/rig/event_stream/kinesis_to_filter.ex

@@ -4,6 +4,8 @@ defmodule Rig.EventStream.KinesisToFilter do
 
   """
 
+  import RIG.Tracing.CloudEvent
+
   alias Rig.EventFilter
   alias RIG.Tracing
   alias RigCloudEvents.CloudEvent
@@ -23,7 +25,7 @@ defmodule Rig.EventStream.KinesisToFilter do
         Tracing.CloudEvent.with_child_span "kinesis_to_filter", cloud_event do
           cloud_event =
             cloud_event
-            |> Tracing.append_context_with_mode(Tracing.context(), :private)
+            |> Tracing.append_context(Tracing.context(), :private)
 
           Logger.debug(fn -> inspect(cloud_event.parsed) end)
           EventFilter.forward_event(cloud_event)

lib/rig/tracing.ex

@@ -71,35 +71,23 @@ defmodule RIG.Tracing do
 
   # ---
 
-  @spec append_context_with_mode(CloudEvent.t(), t(), mode :: atom()) ::
-          CloudEvent.t()
-  def append_context_with_mode(%CloudEvent{} = cloudevent, context, mode) do
-    cloudevent =
-      cloudevent.json
-      |> Jason.decode!()
-      |> append_context(context)
-      |> CloudEvent.parse!()
-
-    case mode do
-      :private ->
-        Logger.debug(fn -> "private mode, remove tracestate." end)
-        remove_tracestate(cloudevent)
+  def append_context(a, b, mode \\ :public)
 
-      _ ->
-        cloudevent
-    end
+  @spec append_context(CloudEvent.t(), t(), mode :: atom()) :: CloudEvent.t()
+  def append_context(%CloudEvent{} = cloudevent, context, mode) do
+    cloudevent.json
+    |> Jason.decode!()
+    |> append_context(context, mode)
+    |> CloudEvent.parse!()
   end
 
-  # 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
-    {:ok, cloudevent} =
-      cloudevent
-      |> append_context(context)
-      |> Cloudevents.to_json()
-      |> CloudEvent.parse()
+  @spec append_context(map, t()) :: map
+  def append_context(%{} = map, context, mode) do
+    cloudevent =
+      Enum.reduce(context, map, fn trace_header, acc ->
+        {key, val} = trace_header
+        Map.put(acc, key, val)
+      end)
 
     case mode do
       :private ->
@@ -113,23 +101,17 @@ defmodule RIG.Tracing do
 
   # ---
 
-  def append_context(a, b, mode \\ :public)
-
-  @spec append_context(map, t()) :: map
-  def append_context(%{} = map, context, _mode) do
-    Enum.reduce(context, map, fn trace_header, acc ->
-      {key, val} = trace_header
-      Map.put(acc, key, val)
-    end)
-  end
-
-  # ---
-
   @spec remove_tracestate(CloudEvent.t()) :: CloudEvent.t()
   defp remove_tracestate(%CloudEvent{} = cloudevent) do
     cloudevent.json
     |> Jason.decode!()
     |> Map.delete("tracestate")
     |> CloudEvent.parse!()
   end
+
+  @spec remove_tracestate(map) :: map
+  defp remove_tracestate(%{} = cloudevent) do
+    cloudevent
+    |> Map.delete("tracestate")
+  end
 end

lib/rig/tracing/cloud_event.ex

@@ -2,6 +2,7 @@ defmodule RIG.Tracing.CloudEvent do
   @moduledoc """
   Distributed Tracing instrumenter for cloudevents
   """
+  alias RigCloudEvents.CloudEvent
   alias RigCloudEvents.Parser.PartialParser
 
   @doc "Like Opencensus.Trace.with_child_span (https://hexdocs.pm/opencensus_elixir/Opencensus.Trace.html#with_child_span/3),
@@ -24,19 +25,7 @@ defmodule RIG.Tracing.CloudEvent do
       })
 
     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
+      tracecontext = compute_context(unquote(event))
 
       parent_span_ctx = :oc_propagation_http_tracecontext.from_headers(tracecontext)
 
@@ -58,54 +47,37 @@ defmodule RIG.Tracing.CloudEvent do
 
   # ---
 
-  # 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,
-        module: module,
-        file: file,
-        function: function
-      })
-
-    quote do
-      tracecontext = []
-
-      tracecontext =
-        case Map.get(unquote(event), "traceparent") do
-          nil -> tracecontext
-          traceparent -> Enum.concat(tracecontext, %{"traceparent" => traceparent})
-        end
-
-      tracecontext =
-        case Map.get(unquote(event), "tracestate") do
-          nil -> tracecontext
-          tracestate -> Enum.concat(tracecontext, %{"tracestate" => tracestate})
-        end
+  def compute_context(%CloudEvent{} = event) do
+    tracecontext = []
 
-      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)
-        })
+    tracecontext =
+      case PartialParser.context_attribute(event.parsed, "traceparent") do
+        {:ok, traceparent} -> Enum.concat(tracecontext, %{"traceparent" => traceparent})
+        _ -> tracecontext
+      end
 
-      :ocp.with_span_ctx(new_span_ctx)
+    case PartialParser.context_attribute(event.parsed, "tracestate") do
+      {:ok, tracestate} -> Enum.concat(tracecontext, %{"tracestate" => tracestate})
+      _ -> tracecontext
+    end
+  end
 
-      try do
-        unquote(block)
-      after
-        :oc_trace.finish_span(new_span_ctx)
-        :ocp.with_span_ctx(parent_span_ctx)
+  def compute_context(%{} = event) do
+    tracecontext = []
+    tracecontext =
+      case Map.get(event, "traceparent") do
+        nil -> tracecontext
+        traceparent -> Enum.concat(tracecontext, %{"traceparent" => traceparent})
       end
+
+    case Map.get(event, "tracestate") do
+      nil -> tracecontext
+      tracestate -> Enum.concat(tracecontext, %{"tracestate" => tracestate})
     end
   end
 
+  # ---
+
   defp format_function(nil), do: nil
   defp format_function({name, arity}), do: "#{name}/#{arity}"