poc

disperser/dataapi/operator_handlers.go

@@ -79,7 +79,6 @@ func checkIsOnlineAndProcessOperator(operatorStatus OperatorOnlineStatus, operat
 	var isOnline bool
 	var socket string
 	if operatorStatus.IndexedOperatorInfo != nil {
-		logger.Error("IndexedOperatorInfo is nil for operator %v", operatorStatus.OperatorInfo)
 		socket = core.OperatorSocket(operatorStatus.IndexedOperatorInfo.Socket).GetRetrievalSocket()
 		isOnline = checkIsOperatorOnline(socket)
 	}
@@ -104,6 +103,33 @@ func checkIsOnlineAndProcessOperator(operatorStatus OperatorOnlineStatus, operat
 	operatorOnlineStatusresultsChan <- metadata
 }
 
+func (s *server) probeOperatorPorts(ctx context.Context, operatorId string) (*OperatorPortCheckResponse, error) {
+	operatorInfo, err := s.subgraphClient.QueryOperatorInfoByOperatorId(context.Background(), operatorId)
+	if err != nil {
+		s.logger.Error("Failed to fetch operator", "error", err)
+		return &OperatorPortCheckResponse{}, err
+	}
+	retrieverSocket := core.OperatorSocket(operatorInfo.Socket).GetRetrievalSocket()
+	disperserSocket := core.OperatorSocket(operatorInfo.Socket).GetDispersalSocket()
+	retrieverStatus := checkIsOperatorOnline(retrieverSocket)
+	disperserStatus := checkIsOperatorOnline(disperserSocket)
+
+	// Log the online status
+	s.logger.Info("Operator port status", "retrieval", retrieverStatus, "retrieverSocket", retrieverSocket, "disperser", disperserStatus, "disperserSocket", disperserSocket)
+
+	// Create the metadata regardless of online status
+	portCheckResponse := &OperatorPortCheckResponse{
+		OperatorId:      operatorId,
+		DisperserSocket: disperserSocket,
+		RetrieverSocket: retrieverSocket,
+		DisperserStatus: disperserStatus,
+		RetrieverStatus: retrieverStatus,
+	}
+
+	// Send the metadata to the results channel
+	return portCheckResponse, nil
+}
+
 // method to check if operator is online
 // Note: This method is least intrusive way to check if operator is online
 // AlternateSolution: Should we add an endpt to check if operator is online?

disperser/dataapi/server.go

@@ -37,6 +37,7 @@ const (
 	// Cache control for responses.
 	// The time unit is second for max age.
 	maxOperatorsNonsigningPercentageAge = 10
+	maxOperatorPortCheckAge             = 600
 	maxNonSignerAge                     = 10
 	maxDeregisteredOperatorAage         = 10
 	maxThroughputAge                    = 10
@@ -138,6 +139,17 @@ type (
 		Data []*ServiceAvailability `json:"data"`
 	}
 
+	OperatorPortCheckRequest struct {
+		OperatorId string `json:"operator_id"`
+	}
+
+	OperatorPortCheckResponse struct {
+		OperatorId      string `json:"operator_id"`
+		DisperserSocket string `json:"disperser_socket"`
+		RetrieverSocket string `json:"retriever_socket"`
+		DisperserStatus bool   `json:"disperser_status"`
+		RetrieverStatus bool   `json:"retriever_status"`
+	}
 	ErrorResponse struct {
 		Error string `json:"error"`
 	}
@@ -230,6 +242,7 @@ func (s *server) Start() error {
 		operatorsInfo := v1.Group("/operators-info")
 		{
 			operatorsInfo.GET("/deregistered-operators", s.FetchDeregisteredOperators)
+			operatorsInfo.GET("/port-check", s.OperatorPortCheck)
 		}
 		metrics := v1.Group("/metrics")
 		{
@@ -586,6 +599,35 @@ func (s *server) FetchDeregisteredOperators(c *gin.Context) {
 	})
 }
 
+// OperatorPortCheck godoc
+//
+//	@Summary	Operator node reachability port check
+//	@Tags		OperatorsInfo
+//	@Produce	json
+//	@Param		operator_id		query		string		true	"Operator ID"
+//	@Success	200	{object}	ServiceAvailabilityResponse
+//	@Failure	400	{object}	ErrorResponse	"error: Bad request"
+//	@Failure	404	{object}	ErrorResponse	"error: Not found"
+//	@Failure	500	{object}	ErrorResponse	"error: Server error"
+//	@Router		/operators-info/port-check [get]
+func (s *server) OperatorPortCheck(c *gin.Context) {
+	timer := prometheus.NewTimer(prometheus.ObserverFunc(func(f float64) {
+		s.metrics.ObserveLatency("OperatorPortCheck", f*1000) // make milliseconds
+	}))
+	defer timer.ObserveDuration()
+
+	operatorId := c.DefaultQuery("operatorId", "")
+	portCheckResponse, err := s.probeOperatorPorts(c.Request.Context(), operatorId)
+	if err != nil {
+		s.metrics.IncrementFailedRequestNum("OperatorPortCheck")
+		errorResponse(c, err)
+		return
+	}
+
+	c.Writer.Header().Set(cacheControlParam, fmt.Sprintf("max-age=%d", maxOperatorPortCheckAge))
+	c.JSON(http.StatusOK, portCheckResponse)
+}
+
 // FetchDisperserServiceAvailability godoc
 //
 //	@Summary	Get status of EigenDA Disperser service.

disperser/dataapi/subgraph_client.go

@@ -27,6 +27,7 @@ type (
 		QueryBatchNonSigningInfoInInterval(ctx context.Context, startTime, endTime int64) ([]*BatchNonSigningInfo, error)
 		QueryOperatorQuorumEvent(ctx context.Context, startBlock, endBlock uint32) (*OperatorQuorumEvents, error)
 		QueryIndexedDeregisteredOperatorsForTimeWindow(ctx context.Context, days int32) (*IndexedDeregisteredOperatorState, error)
+		QueryOperatorInfoByOperatorId(ctx context.Context, operatorId string) (*core.IndexedOperatorInfo, error)
 	}
 	Batch struct {
 		Id              []byte
@@ -126,6 +127,22 @@ func (sc *subgraphClient) QueryOperatorsWithLimit(ctx context.Context, limit int
 	return operators, nil
 }
 
+func (sc *subgraphClient) QueryOperatorInfoByOperatorId(ctx context.Context, operatorId string) (*core.IndexedOperatorInfo, error) {
+	operatorInfo, err := sc.api.QueryOperatorInfoByOperatorIdAtBlockNumber(ctx, operatorId, 0)
+	if err != nil {
+		sc.logger.Error(fmt.Sprintf("failed to query operator info for operator %s", operatorId))
+		return nil, err
+	}
+
+	indexedOperatorInfo, err := ConvertOperatorInfoGqlToIndexedOperatorInfo(operatorInfo)
+	if err != nil {
+		errorMessage := fmt.Sprintf("failed to convert operator info gql to indexed operator info for operator %s", operatorId)
+		sc.logger.Error(errorMessage)
+		return nil, err
+	}
+	return indexedOperatorInfo, nil
+}
+
 func (sc *subgraphClient) QueryBatchNonSigningInfoInInterval(ctx context.Context, startTime, endTime int64) ([]*BatchNonSigningInfo, error) {
 	batchNonSigningInfoGql, err := sc.api.QueryBatchNonSigningInfo(ctx, startTime, endTime)
 	if err != nil {

go.mod

@@ -7,7 +7,6 @@ toolchain go1.21.1
 require (
 	github.com/Layr-Labs/eigenda/api v0.0.0
 	github.com/Layr-Labs/eigensdk-go v0.1.6-0.20240414172936-84d5bc10f72f
-	github.com/Ullaakut/nmap v2.0.2+incompatible
 	github.com/aws/aws-sdk-go-v2 v1.26.0
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.9
 	github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue v1.13.12

go.sum

@@ -27,8 +27,6 @@ github.com/PuerkitoBio/urlesc v0.0.0-20170810143723-de5bf2ad4578/go.mod h1:uGdko
 github.com/Shopify/goreferrer v0.0.0-20181106222321-ec9c9a553398/go.mod h1:a1uqRtAwp2Xwc6WNPJEufxJ7fx3npB4UV/JOLmbu5I0=
 github.com/StackExchange/wmi v1.2.1 h1:VIkavFPXSjcnS+O8yTq7NI32k0R5Aj+v39y29VYDOSA=
 github.com/StackExchange/wmi v1.2.1/go.mod h1:rcmrprowKIVzvc+NUiLncP2uuArMWLCbu9SBzvHz7e8=
-github.com/Ullaakut/nmap v2.0.2+incompatible h1:edw45QpSQBQ2B/Hqfg86Bt5rrK79tp/fAcqIHyNSdQs=
-github.com/Ullaakut/nmap v2.0.2+incompatible/go.mod h1:fkC066hwfcoKwlI7DS2ARTggSVtBTZYCjVH1TzuTMaQ=
 github.com/VictoriaMetrics/fastcache v1.12.1 h1:i0mICQuojGDL3KblA7wUNlY5lOK6a4bwt3uRKnkZU40=
 github.com/VictoriaMetrics/fastcache v1.12.1/go.mod h1:tX04vaqcNoQeGLD+ra5pU5sWkuxnzWhEzLwhP9w653o=
 github.com/ajg/form v1.5.1/go.mod h1:uL1WgH+h2mgNtvBq0339dVnzXdBETtL2LeUXaIv25UY=