feat: Add Xatu Sentry (ethpandaops#466)

Adds support for [Xatu](https://github.com/ethpandaops/xatu) in Sentry
mode. Similarly to Ethereum Metrics Exporter, Xatu Sentry can be enabled
globally, or per client pair.

Note: This PR only enables "Sentry". Xatu Server/Clickhouse etc will
need to run outside of Kurtosis for this to be useful.

---------

Co-authored-by: Barnabas Busa <[email protected]>

README.md

@@ -231,6 +231,10 @@ participants:
   # Defaults to false
   ethereum_metrics_exporter_enabled: false
 
+  # Enables Xatu Sentry for this participant. Can be set globally.
+  # Defaults to false
+  xatu_sentry_enabled: false
+
   # Count of nodes to spin up for this participant
   # Default to 1
   count: 1
@@ -466,6 +470,31 @@ mev_params:
   # Optional parameters to send to the custom_flood script that sends reliable payloads
   custom_flood_params:
     interval_between_transactions: 1
+
+# Enables Xatu Sentry for all participants
+# Defaults to false
+xatu_sentry_enabled: false
+
+# Xatu Sentry params
+xatu_sentry_params:
+  # The image to use for Xatu Sentry
+  xatu_sentry_image: ethpandaops/xatu:latest
+  # GRPC Endpoint of Xatu Server to send events to
+  xatu_server_addr: localhost:8080
+  # Enables TLS to Xatu Server
+  xatu_server_tls: false
+  # Headers to add on to Xatu Server requests
+  xatu_server_headers: {}
+  # Beacon event stream topics to subscribe to
+  beacon_subscriptions:
+  - attestation
+  - block
+  - chain_reorg
+  - finalized_checkpoint
+  - head
+  - voluntary_exit
+  - contribution_and_proof
+  - blob_sidecar
 ```
 
 #### Example configurations

main.star

@@ -61,6 +61,7 @@ def run(plan, args={}):
     mev_params = args_with_right_defaults.mev_params
     parallel_keystore_generation = args_with_right_defaults.parallel_keystore_generation
     persistent = args_with_right_defaults.persistent
+    xatu_sentry_params = args_with_right_defaults.xatu_sentry_params
 
     grafana_datasource_config_template = read_file(
         static_files.GRAFANA_DATASOURCE_CONFIG_TEMPLATE_FILEPATH
@@ -94,6 +95,7 @@ def run(plan, args={}):
         args_with_right_defaults.global_client_log_level,
         jwt_file,
         persistent,
+        xatu_sentry_params,
         parallel_keystore_generation,
     )
 
@@ -107,12 +109,14 @@ def run(plan, args={}):
     all_el_client_contexts = []
     all_cl_client_contexts = []
     all_ethereum_metrics_exporter_contexts = []
+    all_xatu_sentry_contexts = []
     for participant in all_participants:
         all_el_client_contexts.append(participant.el_client_context)
         all_cl_client_contexts.append(participant.cl_client_context)
         all_ethereum_metrics_exporter_contexts.append(
             participant.ethereum_metrics_exporter_context
         )
+        all_xatu_sentry_contexts.append(participant.xatu_sentry_context)
 
     # Generate validator ranges
     validator_ranges_config_template = read_file(
@@ -405,6 +409,7 @@ def run(plan, args={}):
             all_cl_client_contexts,
             prometheus_additional_metrics_jobs,
             all_ethereum_metrics_exporter_contexts,
+            all_xatu_sentry_contexts,
         )
 
         plan.print("Launching grafana...")

network_params.yaml

@@ -15,6 +15,7 @@ participants:
     validator_count: null
     snooper_enabled: false
     ethereum_metrics_exporter_enabled: false
+    xatu_sentry_enabled: false
     el_min_cpu: 0
     el_max_cpu: 0
     el_min_mem: 0
@@ -76,3 +77,4 @@ mev_params:
   mev_flood_seconds_per_bundle: 15
 grafana_additional_dashboards: []
 persistent: false
+xatu_sentry_enabled: false

src/package_io/input_parser.star

@@ -56,6 +56,7 @@ ATTR_TO_BE_SKIPPED_AT_ROOT = (
     "goomy_blob_params",
     "tx_spammer_params",
     "custom_flood_params",
+    "xatu_sentry_params",
 )
 
 
@@ -75,6 +76,7 @@ def input_parser(plan, input_args):
     result["disable_peer_scoring"] = False
     result["goomy_blob_params"] = get_default_goomy_blob_params()
     result["assertoor_params"] = get_default_assertoor_params()
+    result["xatu_sentry_params"] = get_default_xatu_sentry_params()
     result["persistent"] = False
 
     for attr in input_args:
@@ -103,6 +105,10 @@ def input_parser(plan, input_args):
             for sub_attr in input_args["assertoor_params"]:
                 sub_value = input_args["assertoor_params"][sub_attr]
                 result["assertoor_params"][sub_attr] = sub_value
+        elif attr == "xatu_sentry_params":
+            for sub_attr in input_args["xatu_sentry_params"]:
+                sub_value = input_args["xatu_sentry_params"][sub_attr]
+                result["xatu_sentry_params"][sub_attr] = sub_value
 
     if result.get("disable_peer_scoring"):
         result = enrich_disable_peer_scoring(result)
@@ -165,6 +171,7 @@ def input_parser(plan, input_args):
                 ethereum_metrics_exporter_enabled=participant[
                     "ethereum_metrics_exporter_enabled"
                 ],
+                xatu_sentry_enabled=participant["xatu_sentry_enabled"],
                 prometheus_config=struct(
                     scrape_interval=participant["prometheus_config"]["scrape_interval"],
                     labels=participant["prometheus_config"]["labels"],
@@ -249,10 +256,18 @@ def input_parser(plan, input_args):
         mev_type=result["mev_type"],
         snooper_enabled=result["snooper_enabled"],
         ethereum_metrics_exporter_enabled=result["ethereum_metrics_exporter_enabled"],
+        xatu_sentry_enabled=result["xatu_sentry_enabled"],
         parallel_keystore_generation=result["parallel_keystore_generation"],
         grafana_additional_dashboards=result["grafana_additional_dashboards"],
         disable_peer_scoring=result["disable_peer_scoring"],
         persistent=result["persistent"],
+        xatu_sentry_params=struct(
+            xatu_sentry_image=result["xatu_sentry_params"]["xatu_sentry_image"],
+            xatu_server_addr=result["xatu_sentry_params"]["xatu_server_addr"],
+            xatu_server_headers=result["xatu_sentry_params"]["xatu_server_headers"],
+            beacon_subscriptions=result["xatu_sentry_params"]["beacon_subscriptions"],
+            xatu_server_tls=result["xatu_sentry_params"]["xatu_server_tls"],
+        ),
     )
 
 
@@ -333,6 +348,8 @@ def parse_network_params(input_args):
             "ethereum_metrics_exporter_enabled"
         ]
 
+        xatu_sentry_enabled = participant["xatu_sentry_enabled"]
+
         blobber_enabled = participant["blobber_enabled"]
         if blobber_enabled:
             # unless we are running lighthouse, we don't support blobber
@@ -352,6 +369,11 @@ def parse_network_params(input_args):
                     "ethereum_metrics_exporter_enabled"
                 ] = default_ethereum_metrics_exporter_enabled
 
+        if xatu_sentry_enabled == False:
+            default_xatu_sentry_enabled = result["xatu_sentry_enabled"]
+            if default_xatu_sentry_enabled:
+                participant["xatu_sentry_enabled"] = default_xatu_sentry_enabled
+
         validator_count = participant["validator_count"]
         if validator_count == None:
             default_validator_count = result["network_params"][
@@ -442,6 +464,7 @@ def default_input_args():
         "global_client_log_level": "info",
         "snooper_enabled": False,
         "ethereum_metrics_exporter_enabled": False,
+        "xatu_sentry_enabled": False,
         "parallel_keystore_generation": False,
         "disable_peer_scoring": False,
     }
@@ -501,6 +524,7 @@ def default_participant():
         "validator_count": None,
         "snooper_enabled": False,
         "ethereum_metrics_exporter_enabled": False,
+        "xatu_sentry_enabled": False,
         "count": 1,
         "prometheus_config": {
             "scrape_interval": "15s",
@@ -551,6 +575,25 @@ def get_default_assertoor_params():
     }
 
 
+def get_default_xatu_sentry_params():
+    return {
+        "xatu_sentry_image": "ethpandaops/xatu:latest",
+        "xatu_server_addr": "localhost:8080",
+        "xatu_server_headers": {},
+        "xatu_server_tls": False,
+        "beacon_subscriptions": [
+            "attestation",
+            "block",
+            "chain_reorg",
+            "finalized_checkpoint",
+            "head",
+            "voluntary_exit",
+            "contribution_and_proof",
+            "blob_sidecar",
+        ],
+    }
+
+
 def get_default_custom_flood_params():
     # this is a simple script that increases the balance of the coinbase address at a cadence
     return {"interval_between_transactions": 1}

src/participant.star

@@ -5,6 +5,7 @@ def new_participant(
     cl_client_context,
     snooper_engine_context,
     ethereum_metrics_exporter_context,
+    xatu_sentry_context,
 ):
     return struct(
         el_client_type=el_client_type,
@@ -13,4 +14,5 @@ def new_participant(
         cl_client_context=cl_client_context,
         snooper_engine_context=snooper_engine_context,
         ethereum_metrics_exporter_context=ethereum_metrics_exporter_context,
+        xatu_sentry_context=xatu_sentry_context,
     )

src/participant_network.star

@@ -31,6 +31,8 @@ ethereum_metrics_exporter = import_module(
     "./ethereum_metrics_exporter/ethereum_metrics_exporter_launcher.star"
 )
 
+xatu_sentry = import_module("./xatu_sentry/xatu_sentry_launcher.star")
+
 genesis_constants = import_module(
     "./prelaunch_data_generator/genesis_constants/genesis_constants.star"
 )
@@ -61,6 +63,7 @@ def launch_participant_network(
     global_log_level,
     jwt_file,
     persistent,
+    xatu_sentry_params,
     parallel_keystore_generation=False,
 ):
     num_participants = len(participants)
@@ -365,6 +368,7 @@ def launch_participant_network(
     all_snooper_engine_contexts = []
     all_cl_client_contexts = []
     all_ethereum_metrics_exporter_contexts = []
+    all_xatu_sentry_contexts = []
     preregistered_validator_keys_for_nodes = (
         validator_data.per_node_keystores
         if network_params.network == "kurtosis"
@@ -513,6 +517,29 @@ def launch_participant_network(
 
         all_ethereum_metrics_exporter_contexts.append(ethereum_metrics_exporter_context)
 
+        xatu_sentry_context = None
+
+        if participant.xatu_sentry_enabled:
+            pair_name = "{0}-{1}-{2}".format(index_str, cl_client_type, el_client_type)
+
+            xatu_sentry_service_name = "xatu-sentry-{0}".format(pair_name)
+
+            xatu_sentry_context = xatu_sentry.launch(
+                plan,
+                xatu_sentry_service_name,
+                cl_client_context,
+                xatu_sentry_params,
+                network_params,
+                pair_name,
+            )
+            plan.print(
+                "Successfully added {0} xatu sentry participants".format(
+                    xatu_sentry_context
+                )
+            )
+
+        all_xatu_sentry_contexts.append(xatu_sentry_context)
+
     plan.print("Successfully added {0} CL participants".format(num_participants))
 
     all_participants = []
@@ -533,6 +560,10 @@ def launch_participant_network(
             ethereum_metrics_exporter_context = all_ethereum_metrics_exporter_contexts[
                 index
             ]
+        xatu_sentry_context = None
+
+        if participant.xatu_sentry_enabled:
+            xatu_sentry_context = all_xatu_sentry_contexts[index]
 
         participant_entry = participant_module.new_participant(
             el_client_type,
@@ -541,6 +572,7 @@ def launch_participant_network(
             cl_client_context,
             snooper_engine_context,
             ethereum_metrics_exporter_context,
+            xatu_sentry_context,
         )
 
         all_participants.append(participant_entry)

src/prometheus/prometheus_launcher.star

@@ -25,12 +25,14 @@ def launch_prometheus(
     cl_client_contexts,
     additional_metrics_jobs,
     ethereum_metrics_exporter_contexts,
+    xatu_sentry_contexts,
 ):
     metrics_jobs = get_metrics_jobs(
         el_client_contexts,
         cl_client_contexts,
         additional_metrics_jobs,
         ethereum_metrics_exporter_contexts,
+        xatu_sentry_contexts,
     )
     prometheus_url = prometheus.run(
         plan, metrics_jobs, MIN_CPU, MAX_CPU, MIN_MEMORY, MAX_MEMORY
@@ -44,6 +46,7 @@ def get_metrics_jobs(
     cl_client_contexts,
     additional_metrics_jobs,
     ethereum_metrics_exporter_contexts,
+    xatu_sentry_contexts,
 ):
     metrics_jobs = []
     # Adding execution clients metrics jobs
@@ -159,6 +162,23 @@ def get_metrics_jobs(
                     },
                 )
             )
+    # Adding Xatu Sentry metrics jobs
+    for context in xatu_sentry_contexts:
+        if context != None:
+            metrics_jobs.append(
+                new_metrics_job(
+                    job_name="xatu-sentry-{0}".format(context.pair_name),
+                    endpoint="{}:{}".format(
+                        context.ip_addr,
+                        context.metrics_port_num,
+                    ),
+                    metrics_path="/metrics",
+                    labels={
+                        "pair": context.pair_name,
+                    },
+                )
+            )
+
     # Adding additional metrics jobs
     for job in additional_metrics_jobs:
         if job == None:

src/static_files/static_files.star

@@ -31,6 +31,12 @@ ASSERTOOR_TESTS_CONFIG_DIRPATH = (
     STATIC_FILES_DIRPATH + ASSERTOOR_CONFIG_DIRPATH + "/tests"
 )
 
+# xatu-sentry config
+XATU_SENTRY_CONFIG_DIRPATH = "/xatu-sentry-config"
+XATU_SENTRY_CONFIG_TEMPLATE_FILEPATH = (
+    STATIC_FILES_DIRPATH + XATU_SENTRY_CONFIG_DIRPATH + "/config.yaml.tmpl"
+)
+
 # Grafana config
 GRAFANA_CONFIG_DIRPATH = "/grafana-config"
 GRAFANA_DATASOURCE_CONFIG_TEMPLATE_FILEPATH = (

src/xatu_sentry/xatu_sentry_context.star

@@ -0,0 +1,10 @@
+def new_xatu_sentry_context(
+    ip_addr,
+    metrics_port_num,
+    pair_name,
+):
+    return struct(
+        ip_addr=ip_addr,
+        metrics_port_num=metrics_port_num,
+        pair_name=pair_name,
+    )