Merge branch 'main' into validator-manager-genserver-removal

Author: rodrigo-o

Makefile

@@ -66,7 +66,9 @@ kurtosis.setup.ethereum-package:
 
 # 💻 kurtosis.setup.grafana: @ Copies the grafana dashboards to the ethereum-package folder under grafana-config
 kurtosis.setup.grafana:
-	[ -d  $(KURTOSIS_GRAFANA_DASHBOARDS_DIR)/lambdaconsensus ] || mkdir $(KURTOSIS_GRAFANA_DASHBOARDS_DIR)/lambdaconsensus
+	[ -d  $(KURTOSIS_GRAFANA_DASHBOARDS_DIR)/lambdaconsensus ] && \
+		rm $(KURTOSIS_GRAFANA_DASHBOARDS_DIR)/lambdaconsensus/* || \
+		mkdir $(KURTOSIS_GRAFANA_DASHBOARDS_DIR)/lambdaconsensus
 	cp -r $(GRAFANA_DASHBOARDS_DIR)/* $(KURTOSIS_GRAFANA_DASHBOARDS_DIR)/lambdaconsensus
 
 #💻 kurtosis.setup.lambdaconsensus: @ Builds the node Docker for the kurtosis environment

README.md

@@ -332,7 +332,7 @@ make kurtosis.purge
 
 ## Live Metrics
 
-When running the node, metrics are available at [`http://localhost:9568/metrics`](http://localhost:9568/metrics) in Prometheus format.
+When running the node, use the `--metrics` flag to enable metrics at [`http://localhost:9568/metrics`](http://localhost:9568/metrics) in Prometheus format.
 
 ### Grafana
 

config/config.exs

@@ -20,8 +20,6 @@ config :lambda_ethereum_consensus, :fork, fork
 # Configure logging
 config :logger, level: :info, truncate: :infinity
 
-config :lambda_ethereum_consensus, LambdaEthereumConsensus.Telemetry, enable: true
-
 # Use Jason for JSON parsing in Phoenix
 config :phoenix, :json_library, Jason
 

config/runtime.exs

@@ -43,8 +43,8 @@ checkpoint_sync_url = Keyword.get(args, :checkpoint_sync_url)
 execution_endpoint = Keyword.get(args, :execution_endpoint, "http://localhost:8551")
 jwt_path = Keyword.get(args, :execution_jwt)
 testnet_dir = Keyword.get(args, :testnet_dir)
-enable_metrics = Keyword.get(args, :metrics, false)
-metrics_port = Keyword.get(args, :metrics_port, if(enable_metrics, do: 9568, else: nil))
+metrics_port = Keyword.get(args, :metrics_port, nil)
+enable_metrics = Keyword.get(args, :metrics, not is_nil(metrics_port))
 beacon_api_port = Keyword.get(args, :beacon_api_port, nil)
 enable_beacon_api = Keyword.get(args, :beacon_api, not is_nil(beacon_api_port))
 listen_addresses = Keyword.get_values(args, :listen_address)
@@ -188,9 +188,16 @@ block_time_ms =
 
 # Metrics
 
-config :lambda_ethereum_consensus, LambdaEthereumConsensus.Telemetry,
+config :lambda_ethereum_consensus, LambdaEthereumConsensus.PromEx,
+  disabled: not enable_metrics,
+  metrics_server: [
+    port: metrics_port || 9568,
+    auth_strategy: :none
+  ]
+
+config :lambda_ethereum_consensus, LambdaEthereumConsensus.PromExPlugin,
   block_processing_buckets: [0.5, 1.0, 1.5, 2, 4, 6, 8] |> Enum.map(&(&1 * block_time_ms)),
-  port: metrics_port
+  poll_rate: 15_000
 
 # Logging
 

lib/beacon_api/endpoint.ex

@@ -8,7 +8,6 @@ defmodule BeaconApi.Endpoint do
     json_decoder: Phoenix.json_library()
   )
 
-  plug(BeaconApi.MetricsExporter)
   plug(BeaconApi.Router)
   plug(Sentry.PlugContext)
 end

lib/beacon_api/metrics_exporter.ex

@@ -9,7 +9,6 @@ defmodule LambdaEthereumConsensus.Application do
 
   @impl true
   def start(_type, _args) do
-    BeaconApi.MetricsExporter.setup()
     # Configure sentry logger handler
     Logger.add_handlers(:lambda_ethereum_consensus)
     mode = get_operation_mode()
@@ -37,7 +36,6 @@ defmodule LambdaEthereumConsensus.Application do
     CheckpointStates.new()
 
     [
-      LambdaEthereumConsensus.Telemetry,
       LambdaEthereumConsensus.Store.Db,
       LambdaEthereumConsensus.Store.Blocks,
       LambdaEthereumConsensus.Store.BlockStates
@@ -48,6 +46,7 @@ defmodule LambdaEthereumConsensus.Application do
     get_children(:db) ++
       [
         BeaconApi.Endpoint,
+        LambdaEthereumConsensus.PromEx,
         LambdaEthereumConsensus.Beacon.BeaconNode
       ]
   end

lib/lambda_ethereum_consensus/application.ex

@@ -0,0 +1,14 @@
+defmodule LambdaEthereumConsensus.PromEx do
+  @moduledoc """
+  This module integrates the PromEx library. It sets up PromEx plugins and pre-built dashboards for the node.
+  """
+  use PromEx, otp_app: :lambda_ethereum_consensus
+
+  @impl true
+  def plugins() do
+    [
+      PromEx.Plugins.Beam,
+      LambdaEthereumConsensus.PromExPlugin
+    ]
+  end
+end

lib/lambda_ethereum_consensus/prom_ex.ex

@@ -0,0 +1,134 @@
+defmodule LambdaEthereumConsensus.PromExPlugin do
+  @moduledoc """
+  This module defines our custom PromEx plugin.
+  It contains all our custom metrics that are displayed on the `node` dashboard.
+  """
+
+  alias LambdaEthereumConsensus.Store.Db
+  use PromEx.Plugin
+
+  @impl true
+  def event_metrics(_opts) do
+    buckets =
+      Application.get_env(:lambda_ethereum_consensus, __MODULE__)
+      |> Keyword.fetch!(:block_processing_buckets)
+
+    [
+      Event.build(:peer_metrics, [
+        counter([:peers, :connection, :count], tags: [:result]),
+        counter([:peers, :challenge, :count], tags: [:result]),
+        counter([:network, :request, :count], tags: [:result, :type, :reason]),
+        counter([:network, :pubsub_peers, :count], tags: [:result]),
+        sum([:network, :pubsub_topic_active, :active], tags: [:topic]),
+        counter([:network, :pubsub_topics_graft, :count], tags: [:topic]),
+        counter([:network, :pubsub_topics_prune, :count], tags: [:topic]),
+        counter([:network, :pubsub_topics_deliver_message, :count], tags: [:topic]),
+        counter([:network, :pubsub_topics_duplicate_message, :count], tags: [:topic]),
+        counter([:network, :pubsub_topics_reject_message, :count], tags: [:topic]),
+        counter([:network, :pubsub_topics_un_deliverable_message, :count], tags: [:topic]),
+        counter([:network, :pubsub_topics_validate_message, :count], tags: [:topic]),
+        counter([:port, :message, :count], tags: [:function, :direction]),
+        sum([:network, :request, :blocks], tags: [:result, :type, :reason])
+      ]),
+      Event.build(:sync_metrics, [
+        last_value([:sync, :store, :slot], []),
+        last_value([:sync, :on_block, :slot], []),
+        distribution([:sync, :on_block, :stop, :duration],
+          reporter_options: [buckets: buckets],
+          unit: {:native, :millisecond}
+        ),
+        distribution([:sync, :on_block, :exception, :duration],
+          reporter_options: [buckets: buckets],
+          unit: {:native, :millisecond}
+        )
+      ]),
+      Event.build(:db_metrics, [
+        last_value([:db, :latency, :stop, :duration],
+          unit: {:native, :millisecond},
+          tags: [:module, :action]
+        ),
+        last_value([:db, :latency, :exception, :duration],
+          unit: {:native, :millisecond},
+          tags: [:module, :action]
+        ),
+        counter([:db, :latency, :stop, :count],
+          unit: {:native, :millisecond},
+          tags: [:module, :action]
+        )
+      ]),
+      Event.build(:libp2pport_metrics, [
+        last_value([:libp2pport, :handler, :stop, :duration],
+          unit: {:native, :millisecond},
+          tags: [:module, :action]
+        ),
+        last_value([:libp2pport, :handler, :exception, :duration],
+          unit: {:native, :millisecond},
+          tags: [:module, :action]
+        ),
+        counter([:libp2pport, :handler, :stop, :count],
+          unit: {:native, :millisecond},
+          tags: [:module, :action]
+        )
+      ]),
+      Event.build(:forkchoice_metrics, [
+        last_value([:fork_choice, :latency, :stop, :duration],
+          unit: {:native, :millisecond},
+          tags: [:handler, :transition, :operation]
+        ),
+        last_value([:fork_choice, :recompute_head, :stop, :duration],
+          unit: {:native, :millisecond}
+        ),
+        last_value([:fork_choice, :recompute_head, :exception, :duration],
+          unit: {:native, :millisecond}
+        )
+      ]),
+      Event.build(:blocks_status, [
+        last_value([:blocks, :status, :total],
+          tags: [:id, :mainstat, :color, :title, :detail__root]
+        ),
+        last_value([:blocks, :relationship, :total],
+          tags: [:id, :source, :target]
+        )
+      ])
+    ]
+  end
+
+  @impl true
+  def polling_metrics(_opts) do
+    poll_rate =
+      Application.get_env(:lambda_ethereum_consensus, __MODULE__)
+      |> Keyword.fetch!(:poll_rate)
+
+    [
+      Polling.build(:periodic_measurements, poll_rate, {__MODULE__, :periodic_measurements, []}, [
+        last_value([:db, :size, :total], unit: :byte),
+        last_value([:vm, :message_queue, :length], tags: [:process])
+      ])
+    ]
+  end
+
+  def periodic_measurements() do
+    message_queue_lengths()
+    db_size()
+  end
+
+  def db_size() do
+    db_size = Db.size()
+    :telemetry.execute([:db, :size], %{total: db_size})
+  end
+
+  defp register_queue_length(name, len) do
+    :telemetry.execute([:vm, :message_queue], %{length: len}, %{process: inspect(name)})
+  end
+
+  def message_queue_lengths() do
+    Process.list()
+    |> Enum.each(fn pid ->
+      case Process.info(pid, [:message_queue_len, :registered_name]) do
+        [message_queue_len: len, registered_name: name] -> register_queue_length(name, len)
+        [message_queue_len: len] -> register_queue_length(pid, len)
+        _ -> nil
+      end
+    end)
+  end
+end