Add distributed tracing support

config/config.exs

@@ -211,6 +211,10 @@ config :ex_cldr,
 
 config :ex_json_schema, :remote_schema_resolver, {Archethic.Utils, :local_schema_resolver!}
 
+config :opentelemetry,
+  resource: [service: %{name: "archethic"}],
+  traces_exporter: :otlp
+
 # Import environment specific config. This must remain at the bottom
 # of this file so it overrides the configuration defined above.
 import_config("#{Mix.env()}.exs")

config/dev.exs

@@ -183,3 +183,7 @@ config :archethic, :throttle,
     period: 1000,
     limit: System.get_env("ARCHETHIC_THROTTLE_IP_AND_PATH", "5000") |> String.to_integer()
   ]
+
+config :opentelemetry_exporter,
+  otlp_protocol: :http_protobuf,
+  otlp_endpoint: "http://localhost:4318"

config/prod.exs

@@ -293,3 +293,9 @@ config :archethic, :throttle,
     period: 1000,
     limit: System.get_env("ARCHETHIC_THROTTLE_IP_AND_PATH", "20") |> String.to_integer()
   ]
+
+if System.get_env("ARCHETHIC_OTLP_ENDPOINT") do
+  config :opentelemetry_exporter,
+    otlp_protocol: :http_protobuf,
+    otlp_endpoint: System.fetch("ARCHETHIC_OTLP_ENDPOINT")
+end

lib/archethic.ex

@@ -33,6 +33,7 @@ defmodule Archethic do
   alias Archethic.TransactionChain.TransactionInput
 
   require Logger
+  require OpenTelemetry.Tracer
 
   @doc """
     Returns true if a node is up and false if it is down
@@ -123,10 +124,23 @@ defmodule Archethic do
       contract_context: contract_context
     }
 
+    mining_span = OpenTelemetry.Tracer.start_span("mining")
+    OpenTelemetry.Tracer.set_current_span(mining_span)
+    OpenTelemetry.Span.set_attribute(mining_span, "address", Base.encode16(tx.address))
+
+    OpenTelemetry.Span.set_attribute(
+      mining_span,
+      "node",
+      welcome_node_key |> P2P.get_node_info!() |> Node.endpoint()
+    )
+
+    trace = Archethic.Utils.inject_propagated_context()
+    :persistent_term.put({:initial_mining_span, tx.address}, mining_span)
+
     Task.Supervisor.async_stream_nolink(
       Archethic.TaskSupervisor,
       validation_nodes,
-      &P2P.send_message(&1, message),
+      &P2P.send_message(&1, message, trace: trace),
       ordered: false,
       on_timeout: :kill_task,
       timeout: Message.get_timeout(message) + 2000
@@ -186,7 +200,7 @@ defmodule Archethic do
             transaction_type: type
           )
 
-        _ ->
+        :ok ->
           :ok
       end
     end)

lib/archethic/beacon_chain/network_coordinates.ex

@@ -270,7 +270,7 @@ defmodule Archethic.BeaconChain.NetworkCoordinates do
       TaskSupervisor,
       beacon_nodes,
       fn node ->
-        P2P.send_message(node, %GetNetworkStats{summary_time: summary_time}, timeout)
+        P2P.send_message(node, %GetNetworkStats{summary_time: summary_time}, timeout: timeout)
       end,
       timeout: timeout + 1_000,
       ordered: false,

lib/archethic/beacon_chain/subset/p2p_sampling.ex

@@ -45,7 +45,7 @@ defmodule Archethic.BeaconChain.Subset.P2PSampling do
   defp do_sample_p2p_view(node = %Node{}, timeout) do
     start_time = System.monotonic_time(:millisecond)
 
-    case P2P.send_message(node, %Ping{}, timeout) do
+    case P2P.send_message(node, %Ping{}, timeout: timeout) do
       {:ok, %Ok{}} ->
         end_time = System.monotonic_time(:millisecond)
         end_time - start_time

lib/archethic/mining.ex

@@ -47,9 +47,24 @@ defmodule Archethic.Mining do
           transaction :: Transaction.t(),
           welcome_node_public_key :: Crypto.key(),
           validation_node_public_keys :: list(Crypto.key()),
-          contract_context :: nil | Contract.Context.t()
+          contract_context :: nil | Contract.Context.t(),
+          request_metadata :: %{}
         ) :: {:ok, pid()}
-  def start(tx = %Transaction{}, welcome_node_public_key, [_ | []], contract_context) do
+  def start(
+        tx,
+        welcome_node_public_key,
+        validation_node_keys,
+        contract_context,
+        request_metadata \\ %{}
+      )
+
+  def start(
+        tx = %Transaction{},
+        welcome_node_public_key,
+        [_ | []],
+        contract_context,
+        _
+      ) do
     StandaloneWorkflow.start_link(
       transaction: tx,
       welcome_node: P2P.get_node_info!(welcome_node_public_key),
@@ -61,7 +76,8 @@ defmodule Archethic.Mining do
         tx = %Transaction{},
         welcome_node_public_key,
         validation_node_public_keys,
-        contract_context
+        contract_context,
+        request_metadata
       )
       when is_binary(welcome_node_public_key) and is_list(validation_node_public_keys) do
     DynamicSupervisor.start_child(WorkerSupervisor, {
@@ -70,7 +86,8 @@ defmodule Archethic.Mining do
       welcome_node: P2P.get_node_info!(welcome_node_public_key),
       validation_nodes: Enum.map(validation_node_public_keys, &P2P.get_node_info!/1),
       node_public_key: Crypto.last_node_public_key(),
-      contract_context: contract_context
+      contract_context: contract_context,
+      request_metadata: request_metadata
     })
   end