feat: Enhance Pulsar Function Manager with cluster error handling (#57)

Author: freeznet

pkg/mcp/pftools/errors.go

@@ -1,8 +1,21 @@
 package pftools
 
-import "errors"
+import (
+	"errors"
+	"strings"
+
+	"github.com/apache/pulsar-client-go/pulsaradmin/pkg/rest"
+)
 
 var (
 	ErrFunctionNotFound = errors.New("function not found")
 	ErrNotOurMessage    = errors.New("not our message")
 )
+
+// IsClusterUnhealthy checks if an error indicates cluster health issues
+func IsClusterUnhealthy(err error) bool {
+	if restErr, ok := err.(rest.Error); ok {
+		return restErr.Code == 503 && strings.Contains(restErr.Reason, "no healthy upstream")
+	}
+	return false
+}

pkg/mcp/pftools/manager.go

@@ -89,23 +89,24 @@ func NewPulsarFunctionManager(snServer *Server, readOnly bool, options *ManagerO
 
 	// Create the manager
 	manager := &PulsarFunctionManager{
-		adminClient:       adminClient,
-		v2adminClient:     v2adminClient,
-		pulsarClient:      pulsarClient,
-		fnToToolMap:       make(map[string]*FunctionTool),
-		mutex:             sync.RWMutex{},
-		producerCache:     make(map[string]pulsarclient.Producer),
-		producerMutex:     sync.RWMutex{},
-		pollInterval:      options.PollInterval,
-		stopCh:            make(chan struct{}),
-		callInProgressMap: make(map[string]context.CancelFunc),
-		mcpServer:         snServer.MCPServer,
-		readOnly:          readOnly,
-		defaultTimeout:    options.DefaultTimeout,
-		circuitBreakers:   make(map[string]*CircuitBreaker),
-		tenantNamespaces:  options.TenantNamespaces,
-		strictExport:      options.StrictExport,
-		sessionID:         sessionID,
+		adminClient:         adminClient,
+		v2adminClient:       v2adminClient,
+		pulsarClient:        pulsarClient,
+		fnToToolMap:         make(map[string]*FunctionTool),
+		mutex:               sync.RWMutex{},
+		producerCache:       make(map[string]pulsarclient.Producer),
+		producerMutex:       sync.RWMutex{},
+		pollInterval:        options.PollInterval,
+		stopCh:              make(chan struct{}),
+		callInProgressMap:   make(map[string]context.CancelFunc),
+		mcpServer:           snServer.MCPServer,
+		readOnly:            readOnly,
+		defaultTimeout:      options.DefaultTimeout,
+		circuitBreakers:     make(map[string]*CircuitBreaker),
+		tenantNamespaces:    options.TenantNamespaces,
+		strictExport:        options.StrictExport,
+		sessionID:           sessionID,
+		clusterErrorHandler: options.ClusterErrorHandler,
 	}
 
 	return manager, nil
@@ -154,6 +155,11 @@ func (m *PulsarFunctionManager) updateFunctions() {
 	functions, err := m.getFunctionsList()
 	if err != nil {
 		log.Printf("Failed to get functions list: %v", err)
+
+		// Check if this is a cluster health error and invoke callback if configured
+		if IsClusterUnhealthy(err) && m.clusterErrorHandler != nil {
+			go m.clusterErrorHandler(m, err)
+		}
 		return
 	}
 

pkg/mcp/pftools/types.go

@@ -31,23 +31,24 @@ import (
 
 // PulsarFunctionManager manages the lifecycle of Pulsar Functions as MCP tools
 type PulsarFunctionManager struct {
-	adminClient       cmdutils.Client
-	v2adminClient     cmdutils.Client
-	pulsarClient      pulsar.Client
-	fnToToolMap       map[string]*FunctionTool
-	mutex             sync.RWMutex
-	producerCache     map[string]pulsar.Producer
-	producerMutex     sync.RWMutex
-	pollInterval      time.Duration
-	stopCh            chan struct{}
-	callInProgressMap map[string]context.CancelFunc
-	mcpServer         *server.MCPServer
-	readOnly          bool
-	defaultTimeout    time.Duration
-	circuitBreakers   map[string]*CircuitBreaker
-	tenantNamespaces  []string
-	strictExport      bool
-	sessionID         string
+	adminClient         cmdutils.Client
+	v2adminClient       cmdutils.Client
+	pulsarClient        pulsar.Client
+	fnToToolMap         map[string]*FunctionTool
+	mutex               sync.RWMutex
+	producerCache       map[string]pulsar.Producer
+	producerMutex       sync.RWMutex
+	pollInterval        time.Duration
+	stopCh              chan struct{}
+	callInProgressMap   map[string]context.CancelFunc
+	mcpServer           *server.MCPServer
+	readOnly            bool
+	defaultTimeout      time.Duration
+	circuitBreakers     map[string]*CircuitBreaker
+	tenantNamespaces    []string
+	strictExport        bool
+	sessionID           string
+	clusterErrorHandler ClusterErrorHandler
 }
 
 type FunctionTool struct {
@@ -84,13 +85,16 @@ const (
 	StateClosed
 )
 
+type ClusterErrorHandler func(*PulsarFunctionManager, error)
+
 type ManagerOptions struct {
-	PollInterval     time.Duration
-	DefaultTimeout   time.Duration
-	FailureThreshold int
-	ResetTimeout     time.Duration
-	TenantNamespaces []string
-	StrictExport     bool
+	PollInterval        time.Duration
+	DefaultTimeout      time.Duration
+	FailureThreshold    int
+	ResetTimeout        time.Duration
+	TenantNamespaces    []string
+	StrictExport        bool
+	ClusterErrorHandler ClusterErrorHandler
 }
 
 func DefaultManagerOptions() *ManagerOptions {

pkg/mcp/pulsar_functions_as_tools.go

@@ -66,6 +66,14 @@ func (s *Server) PulsarFunctionManagedMcpTools(readOnly bool, features []string,
 
 	options := pftools2.DefaultManagerOptions()
 
+	// Configure cluster error handler for graceful cleanup
+	options.ClusterErrorHandler = func(_ *pftools2.PulsarFunctionManager, err error) {
+		log.Printf("Cluster health error detected: %v", err)
+		log.Printf("Consider implementing cleanup logic here (e.g., stopping manager, notifying monitoring systems)")
+		// The calling service can implement specific cleanup logic here
+		// For example: stop the manager, send alerts, implement backoff strategies
+	}
+
 	if s.SNCloudSession.Ctx.Organization == "" || s.SNCloudSession.Ctx.PulsarInstance == "" || s.SNCloudSession.Ctx.PulsarCluster == "" {
 		log.Printf("Skipping Pulsar Functions as MCP Tools because both organization, pulsar instance and pulsar cluster are not set")
 		return