feat(sentry): Fetch proposer duties

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

@@ -314,8 +314,9 @@ transforms:
       canonical_beacon_block_voluntary_exit: .event.name == "BEACON_API_ETH_V2_BEACON_BLOCK_VOLUNTARY_EXIT"
       canonical_beacon_block_withdrawal: .event.name == "BEACON_API_ETH_V2_BEACON_BLOCK_WITHDRAWAL"
       canonical_beacon_block: .event.name == "BEACON_API_ETH_V2_BEACON_BLOCK_V2" && .meta.client.additional_data.finalized_when_requested == true
-      canonical_beacon_proposer_duty: .event.name == "BEACON_API_ETH_V1_PROPOSER_DUTY"
+      canonical_beacon_proposer_duty: .event.name == "BEACON_API_ETH_V1_PROPOSER_DUTY" && .meta.client.additional_data.state_id == "finalized"
       eth_v1_beacon_committee: .event.name == "BEACON_API_ETH_V1_BEACON_COMMITTEE"
+      eth_v1_proposer_duty: .event.name == "BEACON_API_ETH_V1_PROPOSER_DUTY" && .meta.client.additional_data.state_id == "head"
       eth_v1_events_attestation_v2: .event.name == "BEACON_API_ETH_V1_EVENTS_ATTESTATION_V2"
       eth_v1_events_attestation: .event.name == "BEACON_API_ETH_V1_EVENTS_ATTESTATION"
       eth_v1_events_blob_sidecar: .event.name == "BEACON_API_ETH_V1_EVENTS_BLOB_SIDECAR"
@@ -353,6 +354,7 @@ transforms:
       - xatu_server_events_router.canonical_beacon_block_withdrawal
       - xatu_server_events_router.canonical_beacon_proposer_duty
       - xatu_server_events_router.eth_v1_beacon_committee
+      - xatu_server_events_router.eth_v1_proposer_duty
       - xatu_server_events_router.eth_v1_events_attestation
       - xatu_server_events_router.eth_v1_events_attestation_v2
       - xatu_server_events_router.eth_v1_events_blob_sidecar
@@ -1687,7 +1689,49 @@ transforms:
       del(.event)
       del(.meta)
       del(.data)
-
+  eth_v1_proposer_duty_formatted:
+    type: remap
+    inputs:
+      - xatu_server_events_router.eth_v1_proposer_duty
+    source: |-
+      event_date_time, err = parse_timestamp(.event.date_time, format: "%+");
+      if err == null {
+        .event_date_time = to_unix_timestamp(event_date_time, unit: "milliseconds")
+      } else {
+        .error = err
+        .error_description = "failed to parse event date time"
+        log(., level: "error", rate_limit_secs: 60)
+      }
+      .slot = .meta.client.additional_data.slot.number
+      slot_start_date_time, err = parse_timestamp(.meta.client.additional_data.slot.start_date_time, format: "%+");
+      if err == null {
+        .slot_start_date_time = to_unix_timestamp(slot_start_date_time)
+      } else {
+        .error = err
+        .error_description = "failed to parse slot start date time"
+        log(., level: "error", rate_limit_secs: 60)
+      }
+      .epoch = .meta.client.additional_data.epoch.number
+      epoch_start_date_time, err = parse_timestamp(.meta.client.additional_data.epoch.start_date_time, format: "%+");
+      if err == null {
+        .epoch_start_date_time = to_unix_timestamp(epoch_start_date_time)
+      } else {
+        .error = err
+        .error_description = "failed to parse epoch start date time"
+        log(., level: "error", rate_limit_secs: 60)
+      }
+      .proposer_validator_index = .data.validator_index
+      .proposer_pubkey = .data.pubkey
+      key, err = .slot + .proposer_validator_index + .proposer_pubkey + .meta.client.name
+      if err != null {
+        .error = err
+        .error_description = "failed to generate unique key"
+      }
+      .unique_key = seahash(key)
+      .updated_date_time = to_unix_timestamp(now())
+      del(.event)
+      del(.meta)
+      del(.data)
 
 sinks:
   metrics:
@@ -2177,6 +2221,26 @@ sinks:
     healthcheck:
       enabled: true
     skip_unknown_fields: false
+  eth_v1_proposer_duty_clickhouse:
+    type: clickhouse
+    inputs:
+      - eth_v1_proposer_duty_formatted
+    database: default
+    endpoint: "${CLICKHOUSE_ENDPOINT}"
+    table: beacon_api_eth_v1_proposer_duty
+    auth:
+      strategy: basic
+      user: "${CLICKHOUSE_USER}"
+      password: "${CLICKHOUSE_PASSWORD}"
+    batch:
+      max_bytes: 52428800
+      max_events: 200000
+      timeout_secs: 1
+    buffer:
+      max_events: 200000
+    healthcheck:
+      enabled: true
+    skip_unknown_fields: false
   canonical_beacon_block_elaborated_attestation_clickhouse:
     type: clickhouse
     inputs:

deploy/migrations/clickhouse/030_v1_proposer_duties.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS beacon_api_eth_v1_proposer_duty ON CLUSTER '{cluster}';
+DROP TABLE IF EXISTS default.beacon_api_eth_v1_proposer_duty_local ON CLUSTER '{cluster}';

deploy/migrations/clickhouse/030_v1_proposer_duties.up.sql

@@ -0,0 +1,73 @@
+CREATE TABLE default.beacon_api_eth_v1_proposer_duty_local on cluster '{cluster}'
+(
+    unique_key Int64,
+    updated_date_time DateTime CODEC(DoubleDelta, ZSTD(1)),
+    event_date_time DateTime64(3) CODEC(DoubleDelta, ZSTD(1)),
+    slot UInt32 CODEC(DoubleDelta, ZSTD(1)),
+    slot_start_date_time DateTime CODEC(DoubleDelta, ZSTD(1)),
+    epoch UInt32 CODEC(DoubleDelta, ZSTD(1)),
+    epoch_start_date_time DateTime CODEC(DoubleDelta, ZSTD(1)),
+    proposer_validator_index UInt32 CODEC(ZSTD(1)),
+    proposer_pubkey String CODEC(ZSTD(1)),
+    meta_client_name LowCardinality(String),
+    meta_client_id String CODEC(ZSTD(1)),
+    meta_client_version LowCardinality(String),
+    meta_client_implementation LowCardinality(String),
+    meta_client_os LowCardinality(String),
+    meta_client_ip Nullable(IPv6) CODEC(ZSTD(1)),
+    meta_client_geo_city LowCardinality(String) CODEC(ZSTD(1)),
+    meta_client_geo_country LowCardinality(String) CODEC(ZSTD(1)),
+    meta_client_geo_country_code LowCardinality(String) CODEC(ZSTD(1)),
+    meta_client_geo_continent_code LowCardinality(String)  CODEC(ZSTD(1)),
+    meta_client_geo_longitude Nullable(Float64)  CODEC(ZSTD(1)),
+    meta_client_geo_latitude Nullable(Float64) CODEC(ZSTD(1)),
+    meta_client_geo_autonomous_system_number Nullable(UInt32) CODEC(ZSTD(1)),
+    meta_client_geo_autonomous_system_organization Nullable(String) CODEC(ZSTD(1)),
+    meta_network_id Int32 CODEC(DoubleDelta, ZSTD(1)),
+    meta_network_name LowCardinality(String),
+    meta_consensus_version LowCardinality(String),
+    meta_consensus_version_major LowCardinality(String),
+    meta_consensus_version_minor LowCardinality(String),
+    meta_consensus_version_patch LowCardinality(String),
+    meta_consensus_implementation LowCardinality(String),
+    meta_labels Map(String, String) CODEC(ZSTD(1))
+) Engine = ReplicatedReplacingMergeTree('/clickhouse/{installation}/{cluster}/tables/{shard}/{database}/{table}', '{replica}', updated_date_time)
+PARTITION BY toStartOfMonth(slot_start_date_time)
+ORDER BY (slot_start_date_time, unique_key, meta_network_name);
+
+ALTER TABLE default.beacon_api_eth_v1_proposer_duty_local ON CLUSTER '{cluster}'
+MODIFY COMMENT 'Contains a proposer duty from a beacon block.',
+COMMENT COLUMN unique_key 'Unique key for the row generated from seahash',
+COMMENT COLUMN updated_date_time 'When this row was last updated',
+COMMENT COLUMN event_date_time 'When the client fetched the beacon block from a beacon node',
+COMMENT COLUMN slot 'The slot number from beacon block payload',
+COMMENT COLUMN slot_start_date_time 'The wall clock time when the slot started',
+COMMENT COLUMN epoch 'The epoch number from beacon block payload',
+COMMENT COLUMN epoch_start_date_time 'The wall clock time when the epoch started',
+COMMENT COLUMN proposer_validator_index 'The validator index from the proposer duty payload',
+COMMENT COLUMN proposer_pubkey 'The BLS public key of the validator from the proposer duty payload',
+COMMENT COLUMN meta_client_name 'Name of the client that generated the event',
+COMMENT COLUMN meta_client_id 'Unique Session ID of the client that generated the event. This changes every time the client is restarted.',
+COMMENT COLUMN meta_client_version 'Version of the client that generated the event',
+COMMENT COLUMN meta_client_implementation 'Implementation of the client that generated the event',
+COMMENT COLUMN meta_client_os 'Operating system of the client that generated the event',
+COMMENT COLUMN meta_client_ip 'IP address of the client that generated the event',
+COMMENT COLUMN meta_client_geo_city 'City of the client that generated the event',
+COMMENT COLUMN meta_client_geo_country 'Country of the client that generated the event',
+COMMENT COLUMN meta_client_geo_country_code 'Country code of the client that generated the event',
+COMMENT COLUMN meta_client_geo_continent_code 'Continent code of the client that generated the event',
+COMMENT COLUMN meta_client_geo_longitude 'Longitude of the client that generated the event',
+COMMENT COLUMN meta_client_geo_latitude 'Latitude of the client that generated the event',
+COMMENT COLUMN meta_client_geo_autonomous_system_number 'Autonomous system number of the client that generated the event',
+COMMENT COLUMN meta_client_geo_autonomous_system_organization 'Autonomous system organization of the client that generated the event',
+COMMENT COLUMN meta_network_id 'Ethereum network ID',
+COMMENT COLUMN meta_network_name 'Ethereum network name',
+COMMENT COLUMN meta_consensus_version 'Ethereum consensus client version that generated the event',
+COMMENT COLUMN meta_consensus_version_major 'Ethereum consensus client major version that generated the event',
+COMMENT COLUMN meta_consensus_version_minor 'Ethereum consensus client minor version that generated the event',
+COMMENT COLUMN meta_consensus_version_patch 'Ethereum consensus client patch version that generated the event',
+COMMENT COLUMN meta_consensus_implementation 'Ethereum consensus client implementation that generated the event',
+COMMENT COLUMN meta_labels 'Labels associated with the event';
+
+CREATE TABLE beacon_api_eth_v1_proposer_duty on cluster '{cluster}' AS beacon_api_eth_v1_proposer_duty_local
+ENGINE = Distributed('{cluster}', default, beacon_api_eth_v1_proposer_duty_local, rand());

docker-compose.yml

@@ -134,7 +134,6 @@ services:
   xatu-server:
     profiles:
       - ""
-    image: ethpandaops/xatu:latest
     command: server --config /etc/xatu-server/config.yaml
     build:
       context: .

pkg/cannon/deriver/beacon/eth/v1/proposer_duty.go

@@ -295,7 +295,9 @@ func (b *ProposerDutyDeriver) createEventFromProposerDuty(ctx context.Context, d
 }
 
 func (b *ProposerDutyDeriver) getAdditionalData(_ context.Context, duty *apiv1.ProposerDuty) (*xatu.ClientMeta_AdditionalEthV1ProposerDutyData, error) {
-	extra := &xatu.ClientMeta_AdditionalEthV1ProposerDutyData{}
+	extra := &xatu.ClientMeta_AdditionalEthV1ProposerDutyData{
+		StateId: xatuethv1.StateIDFinalized,
+	}
 
 	slot := b.beacon.Metadata().Wallclock().Slots().FromNumber(uint64(duty.Slot))
 	epoch := b.beacon.Metadata().Wallclock().Epochs().FromSlot(uint64(duty.Slot))

pkg/proto/eth/v1/state.go

@@ -0,0 +1,7 @@
+package v1
+
+var (
+	StateIDFinalized = "finalized"
+	StateIDHead      = "head"
+	StateIDJustified = "justified"
+)