wip

Author: sitole

packages/client-proxy/internal/edge/api/api.gen.go

@@ -1,90 +1,57 @@
 package handlers
 
 import (
+	"context"
+	"errors"
+	"net/http"
+	"time"
+
 	"github.com/gin-gonic/gin"
+	"go.uber.org/zap"
+
+	"github.com/e2b-dev/infra/packages/proxy/internal/edge/orchestrators"
+	orchestratorinfo "github.com/e2b-dev/infra/packages/shared/pkg/grpc/orchestrator-info"
 )
 
 func (a *APIStore) V1ServiceDiscoveryNodeDrain(c *gin.Context, nodeId string) {
-	/*
-		findCtx, findCtxCancel := context.WithTimeout(c, 5*time.Second)
-		defer findCtxCancel()
-
-		logger := a.logger.With(zap.String("node_id", nodeId))
-
-		node, err := a.serviceDiscovery.GetNodeById(findCtx, nodeId)
-		if err != nil {
-			logger.Error("failed to get node by id", zap.Error(err))
-
-			if errors.Is(err, service_discovery.NodeNotFoundErr) {
-				a.sendAPIStoreError(c, http.StatusNotFound, "node not found")
-			} else {
-				a.sendAPIStoreError(c, http.StatusInternalServerError, "failed to get node by id")
-			}
-
-			return
-		}
+	err := a.sendNodeRequest(c, nodeId, orchestratorinfo.ServiceInfoStatus_OrchestratorDraining)
+	if err != nil {
+		a.sendAPIStoreError(c, http.StatusBadRequest, "Error when calling service discovery node")
+		return
+	}
 
-		currentNodeId := a.serviceDiscovery.GetSelfNodeId()
-		if nodeId != currentNodeId {
-			logger.Info("sending draining node request to neighbor", zap.String("node_ip", node.NodeIp))
+	c.Status(http.StatusOK)
+}
 
-			err := a.serviceDiscoveryCallNeighborNodes(c, nodeId, node.NodeIp, node.NodePort, "drain")
-			if err != nil {
-				logger.Error("failed to call node drain", zap.Error(err))
-				a.sendAPIStoreError(c, http.StatusInternalServerError, "failed to call neighbor node")
-				return
-			}
+func (a *APIStore) sendNodeRequest(ctx context.Context, nodeId string, status orchestratorinfo.ServiceInfoStatus) error {
+	findCtx, findCtxCancel := context.WithTimeout(ctx, 5*time.Second)
+	defer findCtxCancel()
 
-			logger.Info("cluster node drain request delivered")
-			c.Status(http.StatusOK)
-			return
-		}
+	logger := a.logger.With(zap.String("node_id", nodeId))
 
-		// handle self drain process
-		if a.selfDrainHandler == nil {
-			logger.Error("self drain handler is not set")
-			a.sendAPIStoreError(c, http.StatusInternalServerError, "self drain handler is not configured")
-			return
+	// try to find orchestrator node first
+	o, err := a.orchestratorPool.GetOrchestrator(nodeId)
+	if err != nil {
+		if !errors.Is(err, orchestrators.ErrOrchestratorNotFound) {
+			logger.Warn("Failed to get orchestrator", zap.Error(err))
+			return errors.New("Error when getting orchestrator node")
 		}
+	}
 
-		logger.Info("starting self-drain process")
+	if o != nil {
+		logger.Info("found orchestrator node, calling drain request")
+		_, err = o.Client.Info.ServiceStatusOverride(
+			findCtx, &orchestratorinfo.ServiceStatusChangeRequest{ServiceStatus: status},
+		)
 
-		err = (*a.selfDrainHandler)()
 		if err != nil {
-			logger.Error("failed to start self drain process", zap.Error(err))
-			a.sendAPIStoreError(c, http.StatusInternalServerError, "failed to start self drain process")
-			return
+			logger.Error("failed to drain orchestrator node", zap.Error(err))
+			return errors.New("Failed to drain orchestrator node")
 		}
 
-		a.healthStatus = api.Draining
-		c.Status(http.StatusOK)
-
-	*/
-}
-
-/*
-func (a *APIStore) serviceDiscoveryCallNeighborNodes(ctx context.Context, nodeId string, nodeIp string, nodePort int, callMethod string) error {
-	// todo: add authorization when implemented
-	// update / drain
-	reqUrl := fmt.Sprintf("http://%s:%d/v1/service-discovery/nodes/%s/%s", nodeIp, nodePort, nodeId, callMethod)
-	req, err := http.NewRequest(http.MethodPost, reqUrl, nil)
-	if err != nil {
-		return err
-	}
-
-	reqCtx, reqCtxCancel := context.WithTimeout(ctx, 5*time.Second)
-	defer reqCtxCancel()
-
-	req.WithContext(reqCtx)
-	res, err := http.DefaultClient.Do(req)
-	if err != nil {
-		return err
-	}
-
-	if res.StatusCode != http.StatusOK {
-		return fmt.Errorf("failed to call neighbor node: %s", res.Status)
+		return nil
 	}
 
-	return nil
+	// todo: call edge api node
+	return errors.New("Failed to find node, it must be edge api node")
 }
-*/

packages/client-proxy/internal/edge/api/spec.gen.go

@@ -0,0 +1,19 @@
+package handlers
+
+import (
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+
+	orchestratorinfo "github.com/e2b-dev/infra/packages/shared/pkg/grpc/orchestrator-info"
+)
+
+func (a *APIStore) V1ServiceDiscoveryNodeKill(c *gin.Context, nodeId string) {
+	err := a.sendNodeRequest(c, nodeId, orchestratorinfo.ServiceInfoStatus_OrchestratorUnhealthy)
+	if err != nil {
+		a.sendAPIStoreError(c, http.StatusBadRequest, "Error when calling service discovery node")
+		return
+	}
+
+	c.Status(http.StatusOK)
+}

packages/client-proxy/internal/edge/handlers/service-discovery-node-drain.go

@@ -106,6 +106,7 @@ func (o *Orchestrator) sync() {
 		select {
 		case <-o.ctx.Done():
 			zap.L().Info("context done", zap.String("orchestrator sync id", o.ServiceId))
+			return
 		case <-ticker.C:
 			o.syncRun()
 		}