Merge branch 'master' into feat/cannon-smoke-test

.github/workflows/goreleaser.yaml

@@ -9,7 +9,9 @@ jobs:
   goreleaser:
     permissions:
       contents: write
-    runs-on: ubuntu-latest
+    runs-on:
+      - self-hosted-ghr
+      - size-l-x64
     steps:
       - name: Checkout
         uses: actions/checkout@v3

.gitignore

@@ -12,4 +12,5 @@ GeoLite2-City.mmdb
 __debug_bin*
 .vscode/launch.json
 sage.yaml
-.goreleaser.yaml.new
+.goreleaser.yaml.new
+.aider*

README.md

@@ -2,6 +2,10 @@
 
 Ethereum network monitoring with collection clients and a centralized server for data pipelining.
 
+
+> [!IMPORTANT]  
+> EthPandaOps publishes all the data we collect with Xatu. Check it out [here](https://github.com/ethpandaops/xatu-data)
+
 ## Overview
 
 Xatu can run in multiple modes. Each mode can be run independently. The following diagram shows the different modes and how they interact with each other and other services.

deploy/local/docker-compose/grafana/datasources/datasources.yaml

@@ -30,4 +30,11 @@ datasources:
       sslmode: disable
       tlsSkipVerify: true
       postgresVersion: 1500
-      database: xatu
+      database: xatu
+  - name: Tempo
+    type: tempo
+    access: proxy
+    uid: EbPG8fYoz
+    url: http://tempo:3200
+    jsonData:
+      httpMethod: GET

deploy/local/docker-compose/tempo.yaml

@@ -0,0 +1,59 @@
+stream_over_http_enabled: true
+server:
+  http_listen_port: 3200
+  log_level: info
+
+query_frontend:
+  search:
+    duration_slo: 5s
+    throughput_bytes_slo: 1.073741824e+09
+  trace_by_id:
+    duration_slo: 5s
+
+distributor:
+  receivers:                           # this configuration will listen on all ports and protocols that tempo is capable of.
+    jaeger:                            # the receives all come from the OpenTelemetry collector.  more configuration information can
+      protocols:                       # be found there: https://github.com/open-telemetry/opentelemetry-collector/tree/main/receiver
+        thrift_http:                   #
+        grpc:                          # for a production deployment you should only enable the receivers you need!
+        thrift_binary:
+        thrift_compact:
+    zipkin:
+    otlp:
+      protocols:
+        http:
+        grpc:
+    opencensus:
+
+ingester:
+  max_block_duration: 5m               # cut the headblock when this much time passes. this is being set for demo purposes and should probably be left alone normally
+
+compactor:
+  compaction:
+    block_retention: 15m                # overall Tempo trace retention. set for demo purposes
+
+metrics_generator:
+  registry:
+    external_labels:
+      source: tempo
+      cluster: docker-compose
+  storage:
+    path: /var/tempo/generator/wal
+    remote_write:
+      - url: http://prometheus:9090/api/v1/write
+        send_exemplars: true
+  traces_storage:
+    path: /var/tempo/generator/traces
+
+storage:
+  trace:
+    backend: local                     # backend configuration to use
+    wal:
+      path: /var/tempo/wal             # where to store the wal locally
+    local:
+      path: /var/tempo/blocks
+
+overrides:
+  defaults:
+    metrics_generator:
+      processors: [service-graphs, span-metrics, local-blocks] # enables metrics generator

deploy/local/docker-compose/vector-http-kafka.yaml

@@ -17,7 +17,6 @@ transforms:
     inputs:
       - xatu_server_events_http
     route:
-      beacon_p2p_attestation: .event.name == "BEACON_P2P_ATTESTATION"
       blockprint_block_classification: .event.name == "BLOCKPRINT_BLOCK_CLASSIFICATION"
       eth_v1_beacon_blob_sidecar: .event.name == "BEACON_API_ETH_V1_BEACON_BLOB_SIDECAR"
       eth_v1_beacon_committee: .event.name == "BEACON_API_ETH_V1_BEACON_COMMITTEE"
@@ -66,6 +65,7 @@ transforms:
       libp2p_trace_gossipsub_beacon_block: .event.name == "LIBP2P_TRACE_GOSSIPSUB_BEACON_BLOCK"
       libp2p_trace_gossipsub_beacon_attestation: .event.name == "LIBP2P_TRACE_GOSSIPSUB_BEACON_ATTESTATION"
       libp2p_trace_gossipsub_blob_sidecar: .event.name == "LIBP2P_TRACE_GOSSIPSUB_BLOB_SIDECAR"
+      beacon_api_eth_v1_beacon_validators: .event.name == "BEACON_API_ETH_V1_BEACON_VALIDATORS"
 sinks:
   metrics:
     type: prometheus_exporter
@@ -633,22 +633,6 @@ sinks:
       codec: json
     librdkafka_options:
       message.max.bytes: "10485760" # 10MB
-  beacon_p2p_attestation_kafka:
-    type: kafka
-    buffer:
-      max_events: 500000
-    batch:
-      timeout_secs: 0.5
-    inputs:
-      - xatu_server_events_router.beacon_p2p_attestation
-    bootstrap_servers: "${KAFKA_BROKERS}"
-    key_field: "event.id"
-    topic: beacon-p2p-attestation
-    compression: snappy
-    healthcheck:
-      enabled: true
-    encoding:
-      codec: json
   beacon_api_eth_v1_proposer_duty_kafka:
     type: kafka
     buffer:
@@ -873,3 +857,19 @@ sinks:
       enabled: true
     encoding:
       codec: json
+  beacon_api_eth_v1_beacon_validators_kafka:
+    type: kafka
+    buffer:
+      max_events: 500000
+    batch:
+      timeout_secs: 0.5
+    inputs:
+      - xatu_server_events_router.beacon_api_eth_v1_beacon_validators
+    bootstrap_servers: "${KAFKA_BROKERS}"
+    key_field: "event.id"
+    topic: beacon-api-eth-v1-beacon-validators
+    compression: snappy
+    healthcheck:
+      enabled: true
+    encoding:
+      codec: json

deploy/local/docker-compose/vector-kafka-clickhouse-libp2p.yaml

@@ -365,7 +365,6 @@ transforms:
         log(., level: "error", rate_limit_secs: 60)
       }
       .transient = .data.transient
-      .unique_key = seahash(.event.id)
       .updated_date_time = to_unix_timestamp(now())
       del(.event)
       del(.meta)
@@ -466,7 +465,6 @@ transforms:
         log(., level: "error", rate_limit_secs: 60)
       }
       .transient = .data.transient
-      .unique_key = seahash(.event.id)
       .updated_date_time = to_unix_timestamp(now())
       del(.event)
       del(.meta)
@@ -492,7 +490,6 @@ transforms:
       }
       .peer_id_unique_key = seahash(peer_id_key)
       .protocol = .data.protocol
-      .unique_key = seahash(.event.id)
       .updated_date_time = to_unix_timestamp(now())
       del(.event)
       del(.meta)
@@ -518,7 +515,6 @@ transforms:
         log(., level: "error", rate_limit_secs: 60)
       }
       .peer_id_unique_key = seahash(peer_id_key)
-      .unique_key = seahash(.event.id)
       .updated_date_time = to_unix_timestamp(now())
       del(.event)
       del(.meta)
@@ -560,7 +556,6 @@ transforms:
         log(., level: "error", rate_limit_secs: 60)
       }
       .peer_id_unique_key = seahash(peer_id_key)
-      .unique_key = seahash(.event.id)
       .updated_date_time = to_unix_timestamp(now())
       del(.event)
       del(.meta)
@@ -588,7 +583,6 @@ transforms:
         log(., level: "error", rate_limit_secs: 60)
       }
       .peer_id_unique_key = seahash(peer_id_key)
-      .unique_key = seahash(.event.id)
 
       .proposer_index = .data.proposer_index
 
@@ -681,7 +675,6 @@ transforms:
         log(., level: "error", rate_limit_secs: 60)
       }
       .peer_id_unique_key = seahash(peer_id_key)
-      .unique_key = seahash(.event.id)
 
       .attesting_validator_index = .meta.client.additional_data.attesting_validator.index
       .attesting_validator_committee_index = .meta.client.additional_data.attesting_validator.committee_index
@@ -782,7 +775,6 @@ transforms:
     inputs:
       - xatu_server_events_router.libp2p_trace_gossipsub_blob_sidecar
     source: |-
-      .unique_key = seahash(.event.id)
       .updated_date_time = to_unix_timestamp(now())
       event_date_time, err = parse_timestamp(.event.date_time, format: "%+");
       if err == null {
@@ -1366,7 +1358,6 @@ transforms:
         log(., level: "error", rate_limit_secs: 60)
       }
       .peer_id_unique_key = seahash(peer_id_key)
-      .unique_key = seahash(.event.id)
 
       if .data.error != null {
         .error = .data.error