fix: /workers_status api fetchWorker logics (#625)

cmd/main.go

@@ -91,8 +91,19 @@ var command = cobra.Command{
 
 		var settlementCaller *vsl.SettlementCaller
 
-		// Apply database migrations for all modules except the broadcaster.
+		// Broadcaster and RSS Only Node does not need Redis, DB and Network Params Client
 		if module != BroadcasterArg && !config.IsRSSComponentOnly(configFile) {
+			// Init a Redis client.
+			if configFile.Redis == nil {
+				zap.L().Error("redis configFile is missing")
+				return fmt.Errorf("redis configFile is missing")
+			}
+
+			redisClient, err = redis.NewClient(*configFile.Redis)
+			if err != nil {
+				return fmt.Errorf("new redis client: %w", err)
+			}
+
 			databaseClient, err = dialer.Dial(cmd.Context(), configFile.Database)
 			if err != nil {
 				return fmt.Errorf("dial database: %w", err)
@@ -111,17 +122,6 @@ var command = cobra.Command{
 				return fmt.Errorf("migrate database: %w", err)
 			}
 
-			// Init a Redis client.
-			if configFile.Redis == nil {
-				zap.L().Error("redis configFile is missing")
-				return fmt.Errorf("redis configFile is missing")
-			}
-
-			redisClient, err = redis.NewClient(*configFile.Redis)
-			if err != nil {
-				return fmt.Errorf("new redis client: %w", err)
-			}
-
 			vslClient, err := parameter.InitVSLClient()
 			if err != nil {
 				return fmt.Errorf("init vsl client: %w", err)

internal/node/component/info/component.go

@@ -3,7 +3,6 @@ package info
 import (
 	"context"
 	"fmt"
-	"net/http"
 
 	"github.com/labstack/echo/v4"
 	"github.com/redis/rueidis"
@@ -12,6 +11,7 @@ import (
 	"github.com/rss3-network/node/internal/database"
 	"github.com/rss3-network/node/internal/node/component"
 	"github.com/rss3-network/node/provider/ethereum/contract/vsl"
+	"github.com/rss3-network/node/provider/httpx"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
@@ -24,7 +24,7 @@ type Component struct {
 	databaseClient      database.Client
 	redisClient         rueidis.Client
 	networkParamsCaller *vsl.NetworkParamsCaller
-	httpClient          *http.Client
+	httpClient          httpx.Client
 }
 
 const Name = "info"
@@ -36,12 +36,17 @@ func (c *Component) Name() string {
 var _ component.Component = (*Component)(nil)
 
 func NewComponent(_ context.Context, apiServer *echo.Echo, config *config.File, databaseClient database.Client, redisClient rueidis.Client, networkParamsCaller *vsl.NetworkParamsCaller) component.Component {
+	httpxClient, err := httpx.NewHTTPClient()
+	if err != nil {
+		return nil
+	}
+
 	c := &Component{
 		config:              config,
 		databaseClient:      databaseClient,
 		redisClient:         redisClient,
 		networkParamsCaller: networkParamsCaller,
-		httpClient:          http.DefaultClient,
+		httpClient:          httpxClient,
 	}
 
 	operators := apiServer.Group("/operators")

internal/node/component/info/handler_node.go

@@ -291,32 +291,16 @@ func (c *Component) sendRequest(ctx context.Context, path string, result any) er
 
 	internalURL.Path = path
 
-	req, err := http.NewRequestWithContext(ctx, http.MethodGet, internalURL.String(), nil)
+	body, err := c.httpClient.Fetch(ctx, internalURL.String())
 	if err != nil {
-		return fmt.Errorf("new request: %w", err)
+		return fmt.Errorf("fetch request: %w", err)
 	}
+	defer body.Close()
 
-	req.Header.Set("Content-Type", "application/json")
-
-	resp, err := c.httpClient.Do(req)
-	if err != nil {
-		return fmt.Errorf("do request: %w", err)
-	}
-
-	defer func() {
-		_ = resp.Body.Close()
-	}()
-
-	if err = json.NewDecoder(resp.Body).Decode(&result); err != nil {
+	if err = json.NewDecoder(body).Decode(&result); err != nil {
 		return fmt.Errorf("decode response: %w", err)
 	}
 
-	if resp.StatusCode != http.StatusOK {
-		marshal, _ := json.Marshal(result)
-
-		return fmt.Errorf("unexpected status: %s, response: %s", resp.Status, string(marshal))
-	}
-
 	return nil
 }
 

internal/node/component/info/handler_worker.go

@@ -5,10 +5,13 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"net/url"
+	"path"
 	"sync"
 
 	"github.com/labstack/echo/v4"
 	"github.com/rss3-network/node/config"
+	rssx "github.com/rss3-network/node/internal/node/component/rss"
 	"github.com/rss3-network/node/internal/node/monitor"
 	"github.com/rss3-network/node/schema/worker"
 	"github.com/rss3-network/node/schema/worker/decentralized"
@@ -24,9 +27,9 @@ type WorkerResponse struct {
 }
 
 type ComponentInfo struct {
-	Decentralized []*WorkerInfo `json:"decentralized"`
-	RSS           *WorkerInfo   `json:"rss"`
-	Federated     []*WorkerInfo `json:"federated"`
+	Decentralized []*WorkerInfo `json:"decentralized,omitempty"`
+	RSS           *WorkerInfo   `json:"rss,omitempty"`
+	Federated     []*WorkerInfo `json:"federated,omitempty"`
 }
 
 type WorkerInfo struct {
@@ -46,50 +49,10 @@ func (c *Component) GetWorkersStatus(ctx echo.Context) error {
 	workerCount := config.CalculateWorkerCount(c.config)
 	workerInfoChan := make(chan *WorkerInfo, workerCount)
 
-	var response *WorkerResponse
-
-	switch {
-	case c.redisClient != nil && len(c.config.Component.Decentralized) > 0:
-		// Fetch all worker info concurrently.
-		c.fetchAllWorkerInfo(ctx, workerInfoChan)
-
-		// Build the worker response.
-		response = c.buildWorkerResponse(workerInfoChan)
-	case c.config.Component.RSS != nil:
-		m := c.config.Component.RSS
-
-		response = &WorkerResponse{
-			Data: ComponentInfo{
-				RSS: &WorkerInfo{
-					WorkerID: m.ID,
-					Network:  m.Network,
-					Worker:   m.Worker,
-					Tags:     rss.ToTagsMap[m.Worker.(rss.Worker)],
-					Platform: rss.ToPlatformMap[m.Worker.(rss.Worker)].String(),
-					Status:   worker.StatusReady},
-			},
-		}
-	case len(c.config.Component.Federated) > 0:
-		f := c.config.Component.Federated[0]
-		switch f.Worker {
-		case federated.Core:
-			response = &WorkerResponse{
-				Data: ComponentInfo{
-					RSS: &WorkerInfo{
-						WorkerID: f.ID,
-						Network:  f.Network,
-						Worker:   f.Worker,
-						Tags:     federated.ToTagsMap[federated.Core],
-						Platform: rss.ToPlatformMap[f.Worker.(rss.Worker)].String(),
-						Status:   worker.StatusReady},
-				},
-			}
-		default:
-			return nil
-		}
-	default:
-		return nil
-	}
+	// Fetch the status of all workers concurrently
+	c.fetchAllWorkerInfo(ctx, workerInfoChan)
+
+	response := c.buildWorkerResponse(workerInfoChan)
 
 	return ctx.JSON(http.StatusOK, response)
 }
@@ -108,7 +71,19 @@ func (c *Component) fetchAllWorkerInfo(ctx echo.Context, workerInfoChan chan<- *
 		}(w)
 	}
 
-	modules := append(append(c.config.Component.Decentralized, c.config.Component.RSS), c.config.Component.Federated...)
+	modules := make([]*config.Module, 0, config.CalculateWorkerCount(c.config))
+
+	if len(c.config.Component.Decentralized) > 0 {
+		modules = append(modules, c.config.Component.Decentralized...)
+	}
+
+	if len(c.config.Component.Federated) > 0 {
+		modules = append(modules, c.config.Component.Federated...)
+	}
+
+	if c.config.Component.RSS != nil {
+		modules = append(modules, c.config.Component.RSS)
+	}
 
 	for _, m := range modules {
 		if m.Network.Protocol() == network.RSSProtocol {
@@ -129,17 +104,23 @@ func (c *Component) fetchAllWorkerInfo(ctx echo.Context, workerInfoChan chan<- *
 // buildWorkerResponse builds the worker response from the worker info channel
 func (c *Component) buildWorkerResponse(workerInfoChan <-chan *WorkerInfo) *WorkerResponse {
 	response := &WorkerResponse{
-		Data: ComponentInfo{
-			Decentralized: []*WorkerInfo{},
-			RSS:           &WorkerInfo{},
-			Federated:     []*WorkerInfo{},
-		},
+		Data: ComponentInfo{},
+	}
+
+	if len(c.config.Component.Decentralized) > 0 {
+		response.Data.Decentralized = []*WorkerInfo{}
+	}
+
+	if len(c.config.Component.Federated) > 0 {
+		response.Data.Federated = []*WorkerInfo{}
 	}
 
 	for workerInfo := range workerInfoChan {
 		switch workerInfo.Network.Protocol() {
 		case network.RSSProtocol:
-			response.Data.RSS = workerInfo
+			if c.config.Component.RSS != nil {
+				response.Data.RSS = workerInfo
+			}
 		case network.EthereumProtocol, network.FarcasterProtocol, network.ArweaveProtocol, network.NearProtocol:
 			response.Data.Decentralized = append(response.Data.Decentralized, workerInfo)
 		case network.ActivityPubProtocol:
@@ -162,8 +143,18 @@ func (c *Component) fetchWorkerInfo(ctx context.Context, module *config.Module)
 		}
 	}
 
-	// Fetch status and progress from a specific worker by id.
-	status, workerProgress := c.getWorkerStatusAndProgressByID(ctx, module.ID)
+	var (
+		status         worker.Status
+		workerProgress monitor.WorkerProgress
+	)
+
+	if module.Network.Protocol() == network.RSSProtocol {
+		// Check RSS worker health status
+		status, _ = c.checkRSSWorkerHealth(ctx, module)
+	} else {
+		// Fetch decentralized or federated worker status and progress from a specific worker by id.
+		status, workerProgress = c.getWorkerStatusAndProgressByID(ctx, module.ID)
+	}
 
 	workerInfo := &WorkerInfo{
 		WorkerID: module.ID,
@@ -211,6 +202,39 @@ func (c *Component) fetchWorkerInfo(ctx context.Context, module *config.Module)
 	return workerInfo
 }
 
+// checkRSSWorkerHealth checks the health of the RSS worker by `healthz` api.
+func (c *Component) checkRSSWorkerHealth(ctx context.Context, module *config.Module) (worker.Status, error) {
+	baseURL, err := url.Parse(module.EndpointID)
+	if err != nil {
+		zap.L().Error("invalid RSS endpoint", zap.String("endpoint", module.EndpointID))
+		return worker.StatusUnhealthy, err
+	}
+
+	baseURL.Path = path.Join(baseURL.Path, "healthz")
+
+	// Parse RSS options from module parameters
+	option, err := rssx.NewOption(module.Parameters)
+	if err != nil {
+		zap.L().Error("parse config parameters", zap.Error(err))
+		return worker.StatusUnhealthy, err
+	}
+
+	if option.Authentication.AccessKey != "" {
+		query := baseURL.Query()
+		query.Set("key", option.Authentication.AccessKey)
+		baseURL.RawQuery = query.Encode()
+	}
+
+	body, err := c.httpClient.Fetch(ctx, baseURL.String())
+	if err != nil {
+		zap.L().Error("fetch RSS healthz", zap.String("endpoint", baseURL.String()), zap.Error(err))
+		return worker.StatusUnhealthy, err
+	}
+	defer body.Close()
+
+	return worker.StatusReady, nil
+}
+
 // getWorkerStatusAndProgressByID gets both worker status and progress from Redis cache by worker ID.
 func (c *Component) getWorkerStatusAndProgressByID(ctx context.Context, workerID string) (worker.Status, monitor.WorkerProgress) {
 	if c.redisClient == nil {

internal/node/indexer/server.go

@@ -326,11 +326,6 @@ func NewServer(ctx context.Context, config *config.Module, databaseClient databa
 		if err != nil {
 			return nil, fmt.Errorf("new near monitorClient: %w", err)
 		}
-	case network.RSSProtocol:
-		instance.monitorClient, err = monitor.NewRssClient(config.EndpointID, config.Parameters)
-		if err != nil {
-			return nil, fmt.Errorf("new rss monitorClient: %w", err)
-		}
 	}
 
 	if err := instance.initializeMeter(); err != nil {