[DataApi][Stage1] Deregistered operators endpt for LameDuck Operators

disperser/dataapi/operatorsInfo_handlers.go

@@ -0,0 +1,110 @@
+package dataapi
+
+import (
+	"context"
+	"net"
+	"sort"
+	"time"
+
+	"github.com/Layr-Labs/eigenda/common"
+	"github.com/Layr-Labs/eigenda/core"
+)
+
+type OperatorOnlineStatus struct {
+	OperatorInfo        *Operator
+	IndexedOperatorInfo *core.IndexedOperatorInfo
+}
+
+var (
+	// TODO: this should be configurable
+	numWorkers                      = 10
+	operatorOnlineStatusChan        chan OperatorOnlineStatus
+	operatorOnlineStatusresultsChan chan *DeregisteredOperatorMetadata
+)
+
+func (s *server) getDeregisterdOperatorForDays(ctx context.Context, days int32) ([]*DeregisteredOperatorMetadata, error) {
+	// Track Time taken to get deregistered operators
+	startTime := time.Now()
+
+	indexedDeregisteredOperatorState, err := s.subgraphClient.QueryIndexedDeregisteredOperatorsForTimeWindow(ctx, days)
+	if err != nil {
+		return nil, err
+	}
+
+	// Convert the map to a slice.
+	operators := indexedDeregisteredOperatorState.Operators
+
+	operatorOnlineStatusChan = make(chan OperatorOnlineStatus, len(operators))
+	operatorOnlineStatusresultsChan = make(chan *DeregisteredOperatorMetadata, len(operators))
+	processOperatorsInParallel(indexedDeregisteredOperatorState, operatorOnlineStatusChan, operatorOnlineStatusresultsChan, s.logger)
+
+	// Collect results of work done
+	DeregisteredOperatorMetadata := make([]*DeregisteredOperatorMetadata, 0, len(operators))
+	for range operators {
+		metadata := <-operatorOnlineStatusresultsChan
+		DeregisteredOperatorMetadata = append(DeregisteredOperatorMetadata, metadata)
+	}
+
+	// Log the time taken
+	s.logger.Info("Time taken to get deregistered operators for days: %v", time.Since(startTime))
+	sort.Slice(DeregisteredOperatorMetadata, func(i, j int) bool {
+		return DeregisteredOperatorMetadata[i].BlockNumber < DeregisteredOperatorMetadata[j].BlockNumber
+	})
+
+	return DeregisteredOperatorMetadata, nil
+}
+
+// method to check if operator is online
+func checkIsOperatorOnline(ipAddress string) bool {
+	timeout := time.Second * 10
+	conn, err := net.DialTimeout("tcp", ipAddress, timeout)
+	if err != nil {
+		return false
+	}
+	defer conn.Close() // Close the connection after checking
+	return true
+}
+
+// Helper Function to Process Operators in Parallel
+func processOperatorsInParallel(deRegisteredOperatorState *IndexedDeregisteredOperatorState, operatorOnlineStatusChan chan OperatorOnlineStatus, operatorOnlineStatusresultsChan chan<- *DeregisteredOperatorMetadata, logger common.Logger) {
+
+	operators := deRegisteredOperatorState.Operators
+	// Start worker goroutines
+	for i := 0; i < numWorkers; i++ {
+		go func() {
+			for item := range operatorOnlineStatusChan {
+				ipAddress := core.OperatorSocket(item.IndexedOperatorInfo.Socket).GetRetrievalSocket()
+				isOnline := checkIsOperatorOnline(ipAddress)
+
+				// Log the online status
+				if isOnline {
+					logger.Debug("Operator %v is online at %s", item.IndexedOperatorInfo, ipAddress)
+				} else {
+					logger.Debug("Operator %v is offline at %s", item.IndexedOperatorInfo, ipAddress)
+				}
+
+				// Create the metadata regardless of online status
+				metadata := &DeregisteredOperatorMetadata{
+					OperatorId:  string(item.OperatorInfo.OperatorId[:]),
+					BlockNumber: uint(item.OperatorInfo.BlockNumber),
+					IpAddress:   ipAddress,
+					IsOnline:    isOnline,
+				}
+
+				// Send the metadata to the results channel
+				operatorOnlineStatusresultsChan <- metadata
+			}
+		}()
+	}
+
+	// Send work to the workers
+	for _, operatorInfo := range operators {
+		operatorOnlineStatus := OperatorOnlineStatus{
+			OperatorInfo:        operatorInfo.Metadata,
+			IndexedOperatorInfo: operatorInfo.IndexedOperatorInfo,
+		}
+		operatorOnlineStatusChan <- operatorOnlineStatus
+	}
+	close(operatorOnlineStatusChan) // Close the channel after sending all tasks
+
+}

disperser/dataapi/server.go

@@ -80,6 +80,18 @@ type (
 		Operators       map[string]OperatorNonsigningPercentageMetrics `json:"operators"`
 	}
 
+	DeregisteredOperatorMetadata struct {
+		OperatorId  string `json:"operator_id"`
+		BlockNumber uint   `json:"block_number"`
+		IpAddress   string `json:"ip_address"`
+		IsOnline    bool   `json:"is_online"`
+	}
+
+	DeregisteredOperatorsResponse struct {
+		Meta Meta                            `json:"meta"`
+		Data []*DeregisteredOperatorMetadata `json:"data"`
+	}
+
 	ErrorResponse struct {
 		Error string `json:"error"`
 	}
@@ -141,6 +153,10 @@ func (s *server) Start() error {
 			feed.GET("/blobs", s.FetchBlobsHandler)
 			feed.GET("/blobs/:blob_key", s.FetchBlobHandler)
 		}
+		operatorsInfo := v1.Group("/operatorsInfo")
+		{
+			operatorsInfo.GET("/deRegisteredOperators", s.FetchDeregisteredOperators)
+		}
 		metrics := v1.Group("/metrics")
 		{
 			metrics.GET("/", s.FetchMetricsHandler)
@@ -402,6 +418,49 @@ func (s *server) FetchOperatorsNonsigningPercentageHandler(c *gin.Context) {
 	c.JSON(http.StatusOK, metric)
 }
 
+// FetchDeregisteredOperators godoc
+//
+//	@Summary	Fetch list of DeregisteredOperators for days
+//	@Tags		OperatorsInfo
+//	@Produce	json
+//	@Success	200		{object}	BlobsResponse
+//	@Failure	400		{object}	ErrorResponse	"error: Bad request"
+//	@Failure	404		{object}	ErrorResponse	"error: Not found"
+//	@Failure	500		{object}	ErrorResponse	"error: Server error"
+//	@Router		/operatorsInfo/deRegisteredState [get]
+func (s *server) FetchDeregisteredOperators(c *gin.Context) {
+	timer := prometheus.NewTimer(prometheus.ObserverFunc(func(f float64) {
+		s.metrics.ObserveLatency("FetchDeregisteredOperators", f*1000) // make milliseconds
+	}))
+	defer timer.ObserveDuration()
+
+	// Get query parameters
+	// Default Value 14 days
+	days := c.DefaultQuery("days", "14") // If not specified, defaults to 14
+
+	// Convert days to integer
+	daysInt, err := strconv.Atoi(days)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid 'days' parameter"})
+		return
+	}
+
+	operatorMetadatas, err := s.getDeregisterdOperatorForDays(c.Request.Context(), int32(daysInt))
+	if err != nil {
+		s.metrics.IncrementFailedRequestNum("FetchDeregisteredOperators")
+		errorResponse(c, err)
+		return
+	}
+
+	s.metrics.IncrementSuccessfulRequestNum("FetchDeregisteredOperators")
+	c.JSON(http.StatusOK, DeregisteredOperatorsResponse{
+		Meta: Meta{
+			Size: len(operatorMetadatas),
+		},
+		Data: operatorMetadatas,
+	})
+}
+
 func (s *server) getBlobMetadataByBatchesWithLimit(ctx context.Context, limit int) ([]*Batch, []*disperser.BlobMetadata, error) {
 	var (
 		blobMetadatas   = make([]*disperser.BlobMetadata, 0)