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:

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

@@ -12,6 +12,7 @@ import (
 	"github.com/ethpandaops/xatu/pkg/cannon/ethereum"
 	"github.com/ethpandaops/xatu/pkg/cannon/iterator"
 	"github.com/ethpandaops/xatu/pkg/observability"
+	v1 "github.com/ethpandaops/xatu/pkg/proto/eth/v1"
 	xatuethv1 "github.com/ethpandaops/xatu/pkg/proto/eth/v1"
 	"github.com/ethpandaops/xatu/pkg/proto/xatu"
 	"github.com/google/uuid"
@@ -295,7 +296,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: v1.StateIDFinalized,
+	}
 
 	slot := b.beacon.Metadata().Wallclock().Slots().FromNumber(uint64(duty.Slot))
 	epoch := b.beacon.Metadata().Wallclock().Epochs().FromSlot(uint64(duty.Slot))