feat(telemetry): Add url data to operator updates on telemetry service (#1136)

Co-authored-by: JuArce <[email protected]>
Co-authored-by: Avila Gastón <[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

core/config/base.go

@@ -33,6 +33,8 @@ type BaseConfig struct {
 	EthRpcClientFallback         eth.InstrumentedClient
 	EthWsClient                  eth.InstrumentedClient
 	EthWsClientFallback          eth.InstrumentedClient
+	EthRpcUrlFallback            string
+	EthWsUrlFallback             string
 	EigenMetricsIpPortAddress    string
 	ChainId                      *big.Int
 }
@@ -149,6 +151,8 @@ func NewBaseConfig(configFilePath string) *BaseConfig {
 		EthRpcClientFallback:         *ethRpcClientFallback,
 		EthWsClient:                  *ethWsClient,
 		EthWsClientFallback:          *ethWsClientFallback,
+		EthRpcUrlFallback:            baseConfigFromYaml.EthRpcUrlFallback,
+		EthWsUrlFallback:             baseConfigFromYaml.EthWsUrlFallback,
 		EigenMetricsIpPortAddress:    baseConfigFromYaml.EigenMetricsIpPortAddress,
 		ChainId:                      chainId,
 	}

operator/cmd/actions/start.go

@@ -1,18 +1,13 @@
 package actions
 
 import (
-	"bytes"
 	"context"
-	"encoding/json"
 	"log"
-	"net/http"
 
 	sdkutils "github.com/Layr-Labs/eigensdk-go/utils"
-	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/urfave/cli/v2"
 	"github.com/yetanotherco/aligned_layer/core/config"
 	operator "github.com/yetanotherco/aligned_layer/operator/pkg"
-	"golang.org/x/crypto/sha3"
 )
 
 var StartFlags = []cli.Flag{
@@ -39,44 +34,11 @@ func operatorMain(ctx *cli.Context) error {
 		return err
 	}
 
-	// hash version
-	hash := sha3.NewLegacyKeccak256()
-	hash.Write([]byte(ctx.App.Version))
-
-	// get hash
-	version := hash.Sum(nil)
-
-	// sign version
-	signature, err := crypto.Sign(version[:], operatorConfig.EcdsaConfig.PrivateKey)
-	if err != nil {
-		return err
-	}
-
-	body := map[string]interface{}{
-		"version":   ctx.App.Version,
-		"signature": signature,
-	}
-	bodyBuffer := new(bytes.Buffer)
-
-	bodyReader := json.NewEncoder(bodyBuffer)
-	err = bodyReader.Encode(body)
+	err = operator.SendTelemetryData(ctx)
 	if err != nil {
 		return err
 	}
 
-	// send version to operator tracker server
-	endpoint := operatorConfig.Operator.OperatorTrackerIpPortAddress + "/versions"
-	operator.Logger.Info("Sending version to operator tracker server: ", "endpoint", endpoint)
-
-	res, err := http.Post(endpoint, "application/json",
-		bodyBuffer)
-	if err != nil {
-		// Dont prevent operator from starting if operator tracker server is down
-		operator.Logger.Warn("Error sending version to metrics server: ", "err", err)
-	} else if res.StatusCode != http.StatusCreated && res.StatusCode != http.StatusNoContent {
-		operator.Logger.Warn("Error sending version to operator tracker server: ", "status_code", res.StatusCode)
-	}
-
 	operator.Logger.Info("Operator starting...")
 	err = operator.Start(context.Background())
 	if err != nil {

operator/pkg/operator.go

@@ -8,13 +8,16 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"net/http"
 	"os"
 	"path/filepath"
 	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/urfave/cli/v2"
 	"github.com/yetanotherco/aligned_layer/operator/risc_zero"
+	"golang.org/x/crypto/sha3"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/yetanotherco/aligned_layer/metrics"
@@ -590,3 +593,65 @@ func (o *Operator) SignTaskResponse(batchIdentifierHash [32]byte) *bls.Signature
 	responseSignature := *o.Config.BlsConfig.KeyPair.SignMessage(batchIdentifierHash)
 	return &responseSignature
 }
+
+func (o *Operator) SendTelemetryData(ctx *cli.Context) error {
+	// hash version
+	hash := sha3.NewLegacyKeccak256()
+	hash.Write([]byte(ctx.App.Version))
+
+	// get hash
+	version := hash.Sum(nil)
+
+	// sign version
+	signature, err := crypto.Sign(version[:], o.Config.EcdsaConfig.PrivateKey)
+	if err != nil {
+		return err
+	}
+	ethRpcUrl, err := BaseUrlOnly(o.Config.BaseConfig.EthRpcUrl)
+	if err != nil {
+		return err
+	}
+	ethRpcUrlFallback, err := BaseUrlOnly(o.Config.BaseConfig.EthRpcUrlFallback)
+	if err != nil {
+		return err
+	}
+	ethWsUrl, err := BaseUrlOnly(o.Config.BaseConfig.EthWsUrl)
+	if err != nil {
+		return err
+	}
+	ethWsUrlFallback, err := BaseUrlOnly(o.Config.BaseConfig.EthWsUrlFallback)
+	if err != nil {
+		return err
+	}
+
+	body := map[string]interface{}{
+		"version":              ctx.App.Version,
+		"signature":            signature,
+		"eth_rpc_url":          ethRpcUrl,
+		"eth_rpc_url_fallback": ethRpcUrlFallback,
+		"eth_ws_url":           ethWsUrl,
+		"eth_ws_url_fallback":  ethWsUrlFallback,
+	}
+
+	bodyBuffer := new(bytes.Buffer)
+
+	bodyReader := json.NewEncoder(bodyBuffer)
+	err = bodyReader.Encode(body)
+	if err != nil {
+		return err
+	}
+
+	// send version to operator tracker server
+	endpoint := o.Config.Operator.OperatorTrackerIpPortAddress + "/versions"
+	o.Logger.Info("Sending version to operator tracker server: ", "endpoint", endpoint)
+
+	res, err := http.Post(endpoint, "application/json", bodyBuffer)
+	if err != nil {
+		// Dont prevent operator from starting if operator tracker server is down
+		o.Logger.Warn("Error sending version to metrics server: ", "err", err)
+	} else if res.StatusCode != http.StatusCreated && res.StatusCode != http.StatusNoContent {
+		o.Logger.Warn("Error sending version to operator tracker server: ", "status_code", res.StatusCode)
+	}
+
+	return nil
+}

operator/pkg/utils.go

@@ -0,0 +1,33 @@
+package operator
+
+import (
+	"fmt"
+	"regexp"
+	"strings"
+)
+
+func BaseUrlOnly(input string) (string, error) {
+	// Define a regex pattern to match the URL format
+	// The pattern captures the scheme, host, and path
+	pattern := `^(?P<scheme>[^:]+)://(?P<host>[^/]+)(?P<path>/.*)?$`
+	re := regexp.MustCompile(pattern)
+
+	matches := re.FindStringSubmatch(input)
+
+	if matches == nil {
+		return "", fmt.Errorf("invalid URL: %s", input)
+	}
+
+	host := matches[2]
+	path := matches[3]
+
+	// If the path is not empty, append the path without the last segment (api_key)
+	if path != "" {
+		pathSegments := strings.Split(path, "/")
+		if len(pathSegments) > 1 {
+			return host + strings.Join(pathSegments[:len(pathSegments)-1], "/"), nil
+		}
+	}
+
+	return host, nil
+}

operator/pkg/utils_test.go

@@ -0,0 +1,47 @@
+package operator
+
+import (
+	"testing"
+)
+
+func TestBaseUrlOnlyHappyPath(t *testing.T) {
+	// Format "<protocol>://<base_url>/<api_key>"
+
+	urls := [...][2]string{
+		{"http://localhost:8545/asdfoij2a7831has89%342jddav98j2748", "localhost:8545"},
+		{"ws://test.com/23r2f98hkjva0udhvi1j%342jddav98j2748", "test.com"},
+		{"http://localhost:8545", "localhost:8545"},
+		{"https://myservice.com/holesky/ApiKey", "myservice.com/holesky"},
+	}
+
+	for _, pair := range urls {
+		url := pair[0]
+		expectedBaseUrl := pair[1]
+
+		baseUrl, err := BaseUrlOnly(url)
+
+		if err != nil {
+			t.Errorf("Unexpected error for URL %s: %v", url, err)
+		}
+
+		if baseUrl != expectedBaseUrl {
+			t.Errorf("Expected base URL %s, got %s for URL %s", expectedBaseUrl, baseUrl, url)
+		}
+	}
+}
+
+func TestBaseUrlOnlyFailureCases(t *testing.T) {
+
+	urls := [...]string{
+		"localhost:8545/asdfoij2a7831has89%342jddav98j2748",
+		"this-is-all-wrong",
+	}
+
+	for _, url := range urls {
+		baseUrl, err := BaseUrlOnly(url)
+
+		if err == nil {
+			t.Errorf("An error was expected, but received %s", baseUrl)
+		}
+	}
+}

telemetry_api/lib/telemetry_api/operators.ex

@@ -112,38 +112,33 @@ defmodule TelemetryApi.Operators do
   end
 
   @doc """
-  Updates an operator's version.
+  Updates an operator.
 
   ## Examples
 
-      iex> update_operator_version(%{field: value})
+      iex> update_operator(some_version, some_signature, %{field: value})
       {:ok, %Ecto.Changeset{}}
 
-      iex> update_operator_version(%{field: bad_value})
+      iex> update_operator(some_version, invalid_signature, %{field:  value})
       {:error, "Some status", "Some message"}
 
   """
-  def update_operator_version(%{"version" => version, "signature" => signature}) do
+  def update_operator(version, signature, changes) do
     with {:ok, address} <- SignatureVerifier.recover_address(version, signature) do
       address = "0x" <> address
-      # We only want to allow changes on version
-      changes = %{
-        version: version
-      }
-
       case Repo.get(Operator, address) do
         nil ->
           {:error, :bad_request,
            "Provided address does not correspond to any registered operator"}
 
         operator ->
-          operator |> Operator.changeset(changes) |> Repo.insert_or_update()
+          update_operator(operator, changes)
       end
     end
   end
 
   @doc """
-  Updates a operator.
+  Updates an operator.
 
   ## Examples
 

telemetry_api/lib/telemetry_api/operators/operator.ex

@@ -9,14 +9,29 @@ defmodule TelemetryApi.Operators.Operator do
     field :name, :string
     field :version, :string
     field :status, :string
+    field :eth_rpc_url, :string
+    field :eth_rpc_url_fallback, :string
+    field :eth_ws_url, :string
+    field :eth_ws_url_fallback, :string
 
     timestamps(type: :utc_datetime)
   end
 
   @doc false
   def changeset(operator, attrs) do
     operator
-    |> cast(attrs, [:address, :id, :stake, :name, :version, :status])
+    |> cast(attrs, [
+      :address,
+      :id,
+      :stake,
+      :name,
+      :version,
+      :status,
+      :eth_rpc_url,
+      :eth_rpc_url_fallback,
+      :eth_ws_url,
+      :eth_ws_url_fallback
+    ])
     |> validate_required([:address, :id, :name, :stake])
   end
 end

telemetry_api/lib/telemetry_api_web/controllers/operator_controller.ex

@@ -11,8 +11,8 @@ defmodule TelemetryApiWeb.OperatorController do
     render(conn, :index, operators: operators)
   end
 
-  def create_or_update(conn, operator_params) do
-    with {:ok, %Operator{} = operator} <- Operators.update_operator_version(operator_params) do
+  def create_or_update(conn, %{"version" => version, "signature" => signature} = attrs) do
+    with {:ok, %Operator{} = operator} <- Operators.update_operator(version, signature, attrs) do
       conn
       |> put_status(:created)
       |> put_resp_header("location", ~p"/api/operators/#{operator}")

telemetry_api/lib/telemetry_api_web/controllers/operator_json.ex

@@ -22,7 +22,11 @@ defmodule TelemetryApiWeb.OperatorJSON do
       stake: operator.stake,
       name: operator.name,
       version: operator.version,
-      status: operator.status
+      status: operator.status,
+      eth_rpc_url: operator.eth_rpc_url,
+      eth_rpc_url_fallback: operator.eth_rpc_url_fallback,
+      eth_ws_url: operator.eth_ws_url,
+      eth_ws_url_fallback: operator.eth_ws_url_fallback
     }
   end
 end

telemetry_api/priv/repo/migrations/20241002135112_add_operator_url_data.exs

@@ -0,0 +1,12 @@
+defmodule TelemetryApi.Repo.Migrations.AddOperatorUrlData do
+  use Ecto.Migration
+
+  def change do
+    alter table(:operators) do
+      add :eth_rpc_url, :string
+      add :eth_rpc_url_fallback, :string
+      add :eth_ws_url, :string
+      add :eth_ws_url_fallback, :string
+    end
+  end
+end