feat(telemetry): implement telemetry for aggregator (#1077)

Co-authored-by: Avila Gastón <[email protected]>
Co-authored-by: Julian Ventura <[email protected]>
Co-authored-by: Julian Ventura <[email protected]>
Co-authored-by: avilagaston9 <[email protected]>
Co-authored-by: Mario Rugiero <[email protected]>

Author: 5 people

aggregator/internal/pkg/aggregator.go

@@ -12,7 +12,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/yetanotherco/aligned_layer/metrics"
 
-	"github.com/Layr-Labs/eigensdk-go/chainio/clients"
 	sdkclients "github.com/Layr-Labs/eigensdk-go/chainio/clients"
 	"github.com/Layr-Labs/eigensdk-go/logging"
 	"github.com/Layr-Labs/eigensdk-go/services/avsregistry"
@@ -80,8 +79,12 @@ type Aggregator struct {
 
 	logger logging.Logger
 
+	// Metrics
 	metricsReg *prometheus.Registry
 	metrics    *metrics.Metrics
+
+	// Telemetry
+	telemetry *Telemetry
 }
 
 func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error) {
@@ -119,7 +122,7 @@ func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error
 	aggregatorPrivateKey := aggregatorConfig.EcdsaConfig.PrivateKey
 
 	logger := aggregatorConfig.BaseConfig.Logger
-	clients, err := clients.BuildAll(chainioConfig, aggregatorPrivateKey, logger)
+	clients, err := sdkclients.BuildAll(chainioConfig, aggregatorPrivateKey, logger)
 	if err != nil {
 		logger.Errorf("Cannot create sdk clients", "err", err)
 		return nil, err
@@ -148,6 +151,9 @@ func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error
 	reg := prometheus.NewRegistry()
 	aggregatorMetrics := metrics.NewMetrics(aggregatorConfig.Aggregator.MetricsIpPortAddress, reg, logger)
 
+	// Telemetry
+	aggregatorTelemetry := NewTelemetry(aggregatorConfig.Aggregator.TelemetryIpPortAddress, logger)
+
 	nextBatchIndex := uint32(0)
 
 	aggregator := Aggregator{
@@ -169,6 +175,7 @@ func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error
 		logger:                logger,
 		metricsReg:            reg,
 		metrics:               aggregatorMetrics,
+		telemetry:             aggregatorTelemetry,
 	}
 
 	return &aggregator, nil
@@ -209,11 +216,20 @@ func (agg *Aggregator) Start(ctx context.Context) error {
 const MaxSentTxRetries = 5
 
 func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsAggregationServiceResponse) {
+	agg.taskMutex.Lock()
+	agg.AggregatorConfig.BaseConfig.Logger.Info("- Locked Resources: Fetching task data")
+	batchIdentifierHash := agg.batchesIdentifierHashByIdx[blsAggServiceResp.TaskIndex]
+	batchData := agg.batchDataByIdentifierHash[batchIdentifierHash]
+	taskCreatedBlock := agg.batchCreatedBlockByIdx[blsAggServiceResp.TaskIndex]
+	agg.taskMutex.Unlock()
+	agg.AggregatorConfig.BaseConfig.Logger.Info("- Unlocked Resources: Fetching task data")
+
+	// Finish task trace once the task is processed (either successfully or not)
+	defer agg.telemetry.FinishTrace(batchData.BatchMerkleRoot)
+
 	if blsAggServiceResp.Err != nil {
-		agg.taskMutex.Lock()
-		batchIdentifierHash := agg.batchesIdentifierHashByIdx[blsAggServiceResp.TaskIndex]
+		agg.telemetry.LogTaskError(batchData.BatchMerkleRoot, blsAggServiceResp.Err)
 		agg.logger.Error("BlsAggregationServiceResponse contains an error", "err", blsAggServiceResp.Err, "batchIdentifierHash", hex.EncodeToString(batchIdentifierHash[:]))
-		agg.taskMutex.Unlock()
 		return
 	}
 	nonSignerPubkeys := []servicemanager.BN254G1Point{}
@@ -236,13 +252,7 @@ func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsA
 		NonSignerStakeIndices:        blsAggServiceResp.NonSignerStakeIndices,
 	}
 
-	agg.taskMutex.Lock()
-	agg.AggregatorConfig.BaseConfig.Logger.Info("- Locked Resources: Fetching merkle root")
-	batchIdentifierHash := agg.batchesIdentifierHashByIdx[blsAggServiceResp.TaskIndex]
-	batchData := agg.batchDataByIdentifierHash[batchIdentifierHash]
-	taskCreatedBlock := agg.batchCreatedBlockByIdx[blsAggServiceResp.TaskIndex]
-	agg.AggregatorConfig.BaseConfig.Logger.Info("- Unlocked Resources: Fetching merkle root")
-	agg.taskMutex.Unlock()
+	agg.telemetry.LogQuorumReached(batchData.BatchMerkleRoot)
 
 	agg.logger.Info("Threshold reached", "taskIndex", blsAggServiceResp.TaskIndex,
 		"batchIdentifierHash", "0x"+hex.EncodeToString(batchIdentifierHash[:]))
@@ -278,6 +288,7 @@ func (agg *Aggregator) handleBlsAggServiceResponse(blsAggServiceResp blsagg.BlsA
 		"merkleRoot", "0x"+hex.EncodeToString(batchData.BatchMerkleRoot[:]),
 		"senderAddress", "0x"+hex.EncodeToString(batchData.SenderAddress[:]),
 		"batchIdentifierHash", "0x"+hex.EncodeToString(batchIdentifierHash[:]))
+	agg.telemetry.LogTaskError(batchData.BatchMerkleRoot, err)
 }
 
 // / Sends response to contract and waits for transaction receipt
@@ -294,6 +305,7 @@ func (agg *Aggregator) sendAggregatedResponse(batchIdentifierHash [32]byte, batc
 	if err != nil {
 		agg.walletMutex.Unlock()
 		agg.logger.Infof("- Unlocked Wallet Resources: Error sending aggregated response for batch %s. Error: %s", hex.EncodeToString(batchIdentifierHash[:]), err)
+		agg.telemetry.LogTaskError(batchMerkleRoot, err)
 		return nil, err
 	}
 
@@ -303,6 +315,7 @@ func (agg *Aggregator) sendAggregatedResponse(batchIdentifierHash [32]byte, batc
 	receipt, err := utils.WaitForTransactionReceipt(
 		agg.AggregatorConfig.BaseConfig.EthRpcClient, context.Background(), *txHash)
 	if err != nil {
+		agg.telemetry.LogTaskError(batchMerkleRoot, err)
 		return nil, err
 	}
 
@@ -312,6 +325,7 @@ func (agg *Aggregator) sendAggregatedResponse(batchIdentifierHash [32]byte, batc
 }
 
 func (agg *Aggregator) AddNewTask(batchMerkleRoot [32]byte, senderAddress [20]byte, taskCreatedBlock uint32) {
+	agg.telemetry.InitNewTrace(batchMerkleRoot)
 	batchIdentifier := append(batchMerkleRoot[:], senderAddress[:]...)
 	var batchIdentifierHash = *(*[32]byte)(crypto.Keccak256(batchIdentifier))
 

aggregator/internal/pkg/server.go

@@ -49,7 +49,6 @@ func (agg *Aggregator) ProcessOperatorSignedTaskResponseV2(signedTaskResponse *t
 		"SenderAddress", "0x"+hex.EncodeToString(signedTaskResponse.SenderAddress[:]),
 		"BatchIdentifierHash", "0x"+hex.EncodeToString(signedTaskResponse.BatchIdentifierHash[:]),
 		"operatorId", hex.EncodeToString(signedTaskResponse.OperatorId[:]))
-
 	taskIndex := uint32(0)
 	ok := false
 
@@ -71,6 +70,7 @@ func (agg *Aggregator) ProcessOperatorSignedTaskResponseV2(signedTaskResponse *t
 		*reply = 1
 		return nil
 	}
+	agg.telemetry.LogOperatorResponse(signedTaskResponse.BatchMerkleRoot, signedTaskResponse.OperatorId)
 
 	// Don't wait infinitely if it can't answer
 	// Create a context with a timeout of 5 seconds

aggregator/internal/pkg/telemetry.go

@@ -0,0 +1,134 @@
+package pkg
+
+import (
+	"bytes"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/Layr-Labs/eigensdk-go/logging"
+)
+
+type TraceMessage struct {
+	MerkleRoot string `json:"merkle_root"`
+}
+
+type OperatorResponseMessage struct {
+	MerkleRoot string `json:"merkle_root"`
+	OperatorId string `json:"operator_id"`
+}
+type QuorumReachedMessage struct {
+	MerkleRoot string `json:"merkle_root"`
+}
+
+type TaskErrorMessage struct {
+	MerkleRoot string `json:"merkle_root"`
+	TaskError  string `json:"error"`
+}
+
+type Telemetry struct {
+	client  http.Client
+	baseURL url.URL
+	logger  logging.Logger
+}
+
+func NewTelemetry(serverAddress string, logger logging.Logger) *Telemetry {
+	client := http.Client{}
+
+	baseURL := url.URL{
+		Scheme: "http",
+		Host:   serverAddress,
+	}
+	logger.Info("[Telemetry] Starting Telemetry client.", "server_address",
+		serverAddress)
+
+	return &Telemetry{
+		client:  client,
+		baseURL: baseURL,
+		logger:  logger,
+	}
+}
+
+func (t *Telemetry) InitNewTrace(batchMerkleRoot [32]byte) {
+	body := TraceMessage{
+		MerkleRoot: fmt.Sprintf("0x%s", hex.EncodeToString(batchMerkleRoot[:])),
+	}
+	if err := t.sendTelemetryMessage("/api/initTaskTrace", body); err != nil {
+		t.logger.Error("[Telemetry] Error in InitNewTrace", "error", err)
+	}
+}
+
+func (t *Telemetry) LogOperatorResponse(batchMerkleRoot [32]byte, operatorId [32]byte) {
+	body := OperatorResponseMessage{
+		MerkleRoot: fmt.Sprintf("0x%s", hex.EncodeToString(batchMerkleRoot[:])),
+		OperatorId: fmt.Sprintf("0x%s", hex.EncodeToString(operatorId[:])),
+	}
+	if err := t.sendTelemetryMessage("/api/operatorResponse", body); err != nil {
+		t.logger.Error("[Telemetry] Error in LogOperatorResponse", "error", err)
+	}
+}
+
+func (t *Telemetry) LogQuorumReached(batchMerkleRoot [32]byte) {
+	body := QuorumReachedMessage{
+		MerkleRoot: fmt.Sprintf("0x%s", hex.EncodeToString(batchMerkleRoot[:])),
+	}
+	if err := t.sendTelemetryMessage("/api/quorumReached", body); err != nil {
+		t.logger.Error("[Telemetry] Error in LogQuorumReached", "error", err)
+	}
+}
+
+func (t *Telemetry) LogTaskError(batchMerkleRoot [32]byte, taskError error) {
+	body := TaskErrorMessage{
+		MerkleRoot: fmt.Sprintf("0x%s", hex.EncodeToString(batchMerkleRoot[:])),
+		TaskError:  taskError.Error(),
+	}
+	if err := t.sendTelemetryMessage("/api/taskError", body); err != nil {
+		t.logger.Error("[Telemetry] Error in LogTaskError", "error", err)
+	}
+}
+
+func (t *Telemetry) FinishTrace(batchMerkleRoot [32]byte) {
+	// In order to wait for all operator responses, even if the quorum is reached, this function has a delayed execution
+	go func() {
+		time.Sleep(10 * time.Second)
+		body := TraceMessage{
+			MerkleRoot: fmt.Sprintf("0x%s", hex.EncodeToString(batchMerkleRoot[:])),
+		}
+		if err := t.sendTelemetryMessage("/api/finishTaskTrace", body); err != nil {
+			t.logger.Error("[Telemetry] Error in FinishTrace", "error", err)
+		}
+	}()
+}
+
+func (t *Telemetry) sendTelemetryMessage(endpoint string, message interface{}) error {
+	encodedBody, err := json.Marshal(message)
+	if err != nil {
+		t.logger.Error("[Telemetry] Error marshalling JSON", "error", err)
+		return fmt.Errorf("error marshalling JSON: %w", err)
+	}
+
+	t.logger.Info("[Telemetry] Sending message.", "endpoint", endpoint, "message", message)
+
+	fullURL := t.baseURL.ResolveReference(&url.URL{Path: endpoint})
+
+	resp, err := t.client.Post(fullURL.String(), "application/json", bytes.NewBuffer(encodedBody))
+	if err != nil {
+		t.logger.Error("[Telemetry] Error sending POST request", "error", err)
+		return fmt.Errorf("error making POST request: %w", err)
+	}
+	defer resp.Body.Close()
+
+	respBody, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.logger.Error("[Telemetry] Error reading response body", "error", err)
+		return fmt.Errorf("error reading response body: %w", err)
+	}
+
+	t.logger.Info("[Telemetry] Response received", "status", resp.Status, "response_body", string(respBody))
+
+	return nil
+}

config-files/config-aggregator.yaml

@@ -34,6 +34,7 @@ aggregator:
   avs_service_manager_address: 0xc3e53F4d16Ae77Db1c982e75a937B9f60FE63690
   enable_metrics: true
   metrics_ip_port_address: localhost:9091
+  telemetry_ip_port_address: localhost:4001
 ## Operator Configurations
 # operator:
 #   aggregator_rpc_server_ip_port_address: localhost:8090

core/config/aggregator.go

@@ -2,10 +2,11 @@ package config
 
 import (
 	"errors"
-	sdkutils "github.com/Layr-Labs/eigensdk-go/utils"
-	"github.com/ethereum/go-ethereum/common"
 	"log"
 	"os"
+
+	sdkutils "github.com/Layr-Labs/eigensdk-go/utils"
+	"github.com/ethereum/go-ethereum/common"
 )
 
 type AggregatorConfig struct {
@@ -18,6 +19,7 @@ type AggregatorConfig struct {
 		AvsServiceManagerAddress      common.Address
 		EnableMetrics                 bool
 		MetricsIpPortAddress          string
+		TelemetryIpPortAddress        string
 	}
 }
 
@@ -28,6 +30,7 @@ type AggregatorConfigFromYaml struct {
 		AvsServiceManagerAddress      common.Address `yaml:"avs_service_manager_address"`
 		EnableMetrics                 bool           `yaml:"enable_metrics"`
 		MetricsIpPortAddress          string         `yaml:"metrics_ip_port_address"`
+		TelemetryIpPortAddress        string         `yaml:"telemetry_ip_port_address"`
 	} `yaml:"aggregator"`
 }
 
@@ -68,6 +71,7 @@ func NewAggregatorConfig(configFilePath string) *AggregatorConfig {
 			AvsServiceManagerAddress      common.Address
 			EnableMetrics                 bool
 			MetricsIpPortAddress          string
+			TelemetryIpPortAddress        string
 		}(aggregatorConfigFromYaml.Aggregator),
 	}
 }

otel-collector.yaml

@@ -0,0 +1,27 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:4317
+processors:
+extensions:
+  health_check: {}
+exporters:
+  otlp:
+    endpoint: jaeger:4317
+    tls:
+      insecure: true
+  debug:
+
+service:
+  extensions: [health_check]
+  pipelines:
+    traces:
+      receivers: [otlp]
+      processors: []
+      exporters: [otlp]
+
+    metrics:
+      receivers: [ otlp ]
+      processors: [ ]
+      exporters: [ debug ]

telemetry-docker-compose.yaml

@@ -0,0 +1,15 @@
+version: "3"
+
+services:
+  otel-collector:
+    image: otel/opentelemetry-collector-contrib:0.107.0
+    command: ["--config=/etc/otel-collector.yaml"]
+    volumes:
+      - ./otel-collector.yaml:/etc/otel-collector.yaml
+    ports:
+      - "4317:4317"
+
+  jaeger:
+    image: jaegertracing/all-in-one:1.60
+    ports:
+      - "16686:16686"

telemetry_api/.env.dev

@@ -1,2 +1,3 @@
 ALIGNED_CONFIG_FILE="../contracts/script/output/devnet/alignedlayer_deployment_output.json"
+OPERATOR_FETCHER_WAIT_TIME_MS=5000
 ENVIRONMENT=devnet

telemetry_api/config/config.exs

@@ -30,6 +30,16 @@ config :logger, :console,
 # Use Jason for JSON parsing in Phoenix
 config :phoenix, :json_library, Jason
 
+# https://opentelemetry.io/docs/languages/erlang/exporters/#setting-up-the-collector
+config :opentelemetry_exporter,
+  otlp_protocol: :grpc,
+  otlp_endpoint: "http://localhost:4317"
+
+config :opentelemetry,
+  resource: %{service: %{name: "telemetry_api"}},
+  span_processor: :batch,
+  traces_exporter: :otlp
+
 # Import environment specific config. This must remain at the bottom
 # of this file so it overrides the configuration defined above.
 import_config "#{config_env()}.exs"

telemetry_api/config/dev.exs

@@ -65,3 +65,6 @@ config :phoenix, :plug_init_mode, :runtime
 
 # Configure ethereumex url
 config :ethereumex, url: "http://localhost:8545"
+
+# For development, we use the stdout exporter to ensure everything is working properly
+# config :opentelemetry, traces_exporter: {:otel_exporter_stdout, []}