"Implement retry logic for key operations to enhance resilience and error handling in plugin initialization."

Signed-off-by: Osmany Montero <osmontero@icloud.com>

Author: osmontero

plugins/inputs/auth.go

@@ -3,9 +3,9 @@ package main
 import (
 	"context"
 	"crypto/tls"
+	"fmt"
 	"github.com/threatwinds/go-sdk/catcher"
 	"github.com/threatwinds/go-sdk/plugins"
-	"os"
 	"strings"
 	"sync"
 	"time"
@@ -16,7 +16,7 @@ import (
 	"google.golang.org/grpc/metadata"
 )
 
-const maxMessageSize = 1024 * 1024 * 1024
+const maxMessageSize = 20 * 1024 * 1024 // 20MB
 
 type LogAuthService struct {
 	Mutex              *sync.Mutex
@@ -57,8 +57,9 @@ func (auth *LogAuthService) syncKeys(typ agent.ConnectorType) {
 	internalKey := pConfig.Get("internalKey").String()
 
 	if agentManager == "" {
-		_ = catcher.Error("agentManager config is empty", nil, nil)
-		os.Exit(1)
+		_ = catcher.Error("Could not sync keys. This is a common occurrence during the startup process and typically resolves on its own after a short while.", fmt.Errorf("configuration is empty"), nil)
+		// Don't exit, just return and retry later
+		return
 	}
 
 	tlsConfig := &tls.Config{
@@ -68,7 +69,7 @@ func (auth *LogAuthService) syncKeys(typ agent.ConnectorType) {
 	tlsCredentials := credentials.NewTLS(tlsConfig)
 	conn, err := grpc.NewClient(agentManager, grpc.WithTransportCredentials(tlsCredentials), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMessageSize)))
 	if err != nil {
-		_ = catcher.Error("cannot to connect to agent manager", err, nil)
+		_ = catcher.Error("Could not sync keys. This is a common occurrence during the startup process and typically resolves on its own after a short while.", err, nil)
 		return
 	}
 	defer func() {

plugins/inputs/handlers.go

@@ -7,7 +7,6 @@ import (
 	"github.com/threatwinds/go-sdk/utils"
 	"net"
 	"net/http"
-	"os"
 	"time"
 
 	"github.com/gin-gonic/gin"
@@ -19,6 +18,10 @@ import (
 )
 
 func startHTTPServer(middlewares *Middlewares, cert string, key string) {
+	// Retry logic for starting HTTP server
+	maxRetries := 3
+	retryDelay := 2 * time.Second
+
 	gin.SetMode(gin.ReleaseMode)
 
 	router := gin.Default()
@@ -27,10 +30,26 @@ func startHTTPServer(middlewares *Middlewares, cert string, key string) {
 	router.GET("/v1/ping", Ping)
 	router.GET("/v1/health", func(c *gin.Context) { c.Status(http.StatusOK) })
 
-	err := router.RunTLS(":8080", cert, key)
-	if err != nil {
-		_ = catcher.Error("failed to start http server", err, nil)
-		os.Exit(1)
+	for retry := 0; retry < maxRetries; retry++ {
+		err := router.RunTLS(":8080", cert, key)
+		if err == nil {
+			break
+		}
+
+		_ = catcher.Error("failed to start http server, retrying", err, map[string]any{
+			"retry":      retry + 1,
+			"maxRetries": maxRetries,
+		})
+
+		if retry < maxRetries-1 {
+			time.Sleep(retryDelay)
+			// Increase delay for next retry
+			retryDelay *= 2
+		} else {
+			// If all retries failed, log the error and return
+			_ = catcher.Error("all retries failed when starting http server", err, nil)
+			return
+		}
 	}
 }
 
@@ -115,11 +134,32 @@ type integration struct {
 	plugins.UnimplementedIntegrationServer
 }
 
-func startGRPCServer(middlewares *Middlewares, cert string, key string) {
-	transportCredentials, err := credentials.NewServerTLSFromFile(cert, key)
-	if err != nil {
-		_ = catcher.Error("failed to create credentials", err, nil)
-		os.Exit(1)
+func startGRPCServer(middlewares *Middlewares, cert string, key string) error {
+	// Retry logic for creating credentials
+	maxRetries := 3
+	retryDelay := 2 * time.Second
+	var transportCredentials credentials.TransportCredentials
+	var err error
+
+	for retry := 0; retry < maxRetries; retry++ {
+		transportCredentials, err = credentials.NewServerTLSFromFile(cert, key)
+		if err == nil {
+			break
+		}
+
+		_ = catcher.Error("failed to create credentials, retrying", err, map[string]any{
+			"retry":      retry + 1,
+			"maxRetries": maxRetries,
+		})
+
+		if retry < maxRetries-1 {
+			time.Sleep(retryDelay)
+			// Increase delay for next retry
+			retryDelay *= 2
+		} else {
+			// If all retries failed, log the error and return
+			return catcher.Error("all retries failed when creating credentials", err, nil)
+		}
 	}
 
 	server := grpc.NewServer(
@@ -135,16 +175,37 @@ func startGRPCServer(middlewares *Middlewares, cert string, key string) {
 	grpcHealth.RegisterHealthServer(server, healthServer)
 	healthServer.SetServingStatus("", grpcHealth.HealthCheckResponse_SERVING)
 
-	listener, err := net.Listen("tcp", "0.0.0.0:50051")
-	if err != nil {
-		_ = catcher.Error("failed to listen to grpc", err, nil)
-		os.Exit(1)
+	// Retry logic for listening to gRPC
+	retryDelay = 2 * time.Second
+	var listener net.Listener
+
+	for retry := 0; retry < maxRetries; retry++ {
+		listener, err = net.Listen("tcp", "0.0.0.0:50051")
+		if err == nil {
+			break
+		}
+
+		_ = catcher.Error("failed to listen to grpc, retrying", err, map[string]any{
+			"retry":      retry + 1,
+			"maxRetries": maxRetries,
+		})
+
+		if retry < maxRetries-1 {
+			time.Sleep(retryDelay)
+			// Increase delay for next retry
+			retryDelay *= 2
+		} else {
+			// If all retries failed, log the error and return
+			return catcher.Error("all retries failed when listening to grpc", err, nil)
+		}
 	}
 
+	// Serve with error handling
 	if err := server.Serve(listener); err != nil {
-		_ = catcher.Error("failed to serve grpc", err, nil)
-		os.Exit(1)
+		return catcher.Error("failed to serve grpc", err, nil)
 	}
+
+	return nil
 }
 
 func (i *integration) ProcessLog(srv plugins.Integration_ProcessLogServer) error {

plugins/inputs/health.go

@@ -3,6 +3,8 @@ package main
 import (
 	"context"
 	"crypto/tls"
+	"fmt"
+	"github.com/threatwinds/go-sdk/catcher"
 	"github.com/threatwinds/go-sdk/plugins"
 	"time"
 
@@ -12,6 +14,8 @@ import (
 	"google.golang.org/grpc/metadata"
 )
 
+const healthMaxMessageSize = 5 * 1024 * 1024 // 5MB
+
 func CheckAgentManagerHealth() {
 	tlsConfig := &tls.Config{
 		InsecureSkipVerify: true,
@@ -24,11 +28,14 @@ func CheckAgentManagerHealth() {
 		internalKey := pConfig.Get("internalKey").String()
 
 		if agentManager == "" {
-			panic("agentManager config is empty")
+			_ = catcher.Error("Could not connect to the Agent Manager. This is a common occurrence during the startup process and typically resolves on its own after a short while.", fmt.Errorf("configuration is empty"), nil)
+			time.Sleep(5 * time.Second)
+			continue
 		}
 
-		conn, err := grpc.NewClient(agentManager, grpc.WithTransportCredentials(tlsCredentials), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(maxMessageSize)))
+		conn, err := grpc.NewClient(agentManager, grpc.WithTransportCredentials(tlsCredentials), grpc.WithDefaultCallOptions(grpc.MaxCallRecvMsgSize(healthMaxMessageSize)))
 		if err != nil {
+			_ = catcher.Error("Could not connect to the Agent Manager. This is a common occurrence during the startup process and typically resolves on its own after a short while.", err, nil)
 			time.Sleep(5 * time.Second)
 			continue
 		}
@@ -43,6 +50,7 @@ func CheckAgentManagerHealth() {
 		if err != nil {
 			cancel()
 			_ = conn.Close()
+			_ = catcher.Error("Could not connect to the Agent Manager. This is a common occurrence during the startup process and typically resolves on its own after a short while.", err, nil)
 			time.Sleep(5 * time.Second)
 			continue
 		}
@@ -52,9 +60,5 @@ func CheckAgentManagerHealth() {
 			_ = conn.Close()
 			break
 		}
-
-		cancel()
-		_ = conn.Close()
-		time.Sleep(5 * time.Second)
 	}
 }

plugins/inputs/main.go

@@ -7,6 +7,7 @@ import (
 	"github.com/threatwinds/go-sdk/utils"
 	"os"
 	"runtime"
+	"time"
 )
 
 const defaultTenant string = "ce66672c-e36d-4761-a8c8-90058fee1a24"
@@ -16,20 +17,44 @@ var localLogsChannel chan *plugins.Log
 func main() {
 	mode := plugins.GetCfg().Env.Mode
 	if mode != "worker" {
-		os.Exit(0)
+		return
 	}
 
 	CheckAgentManagerHealth()
 
 	autService := NewLogAuthService()
-	go autService.SyncAuth()
+	go func() {
+		autService.SyncAuth()
+	}()
 
 	middlewares := NewMiddlewares(autService)
 
-	cert, key, err := loadCerts()
-	if err != nil {
-		_ = catcher.Error("cannot load certificates", err, nil)
-		os.Exit(1)
+	// Retry logic for loading certificates
+	maxRetries := 3
+	retryDelay := 2 * time.Second
+	var cert, key string
+	var err error
+
+	for retry := 0; retry < maxRetries; retry++ {
+		cert, key, err = loadCerts()
+		if err == nil {
+			break
+		}
+
+		_ = catcher.Error("cannot load certificates, retrying", err, map[string]any{
+			"retry":      retry + 1,
+			"maxRetries": maxRetries,
+		})
+
+		if retry < maxRetries-1 {
+			time.Sleep(retryDelay)
+			// Increase delay for next retry
+			retryDelay *= 2
+		} else {
+			// If all retries failed, log the error and return
+			_ = catcher.Error("all retries failed when loading certificates", err, nil)
+			return
+		}
 	}
 
 	cpu := runtime.NumCPU()
@@ -41,7 +66,7 @@ func main() {
 	}
 
 	go startHTTPServer(middlewares, cert, key)
-	startGRPCServer(middlewares, cert, key)
+	_ = startGRPCServer(middlewares, cert, key)
 }
 
 func loadCerts() (string, string, error) {