feat: implement decision-based routing with plugin architecture

Signed-off-by: bitliu <[email protected]>

Author: Xunzhuo

src/semantic-router/pkg/config/loader.go

@@ -14,6 +14,10 @@ var (
 	configOnce sync.Once
 	configErr  error
 	configMu   sync.RWMutex
+
+	// Config change notification channel
+	configUpdateCh chan *RouterConfig
+	configUpdateMu sync.Mutex
 )
 
 // Load loads the configuration from the specified YAML file once and caches it globally.
@@ -64,10 +68,20 @@ func Parse(configPath string) (*RouterConfig, error) {
 // Replace replaces the globally cached config. It is safe for concurrent readers.
 func Replace(newCfg *RouterConfig) {
 	configMu.Lock()
-	defer configMu.Unlock()
 	config = newCfg
-	// Do not reset configOnce to avoid racing re-parses via LoadConfig; callers should use ParseConfigFile for fresher reads.
 	configErr = nil
+	configMu.Unlock()
+
+	// Notify listeners of config change
+	configUpdateMu.Lock()
+	if configUpdateCh != nil {
+		select {
+		case configUpdateCh <- newCfg:
+		default:
+			// Channel full or no listener, skip
+		}
+	}
+	configUpdateMu.Unlock()
 }
 
 // Get returns the current configuration
@@ -76,3 +90,15 @@ func Get() *RouterConfig {
 	defer configMu.RUnlock()
 	return config
 }
+
+// WatchConfigUpdates returns a channel that receives config updates
+// Only one watcher is supported at a time
+func WatchConfigUpdates() <-chan *RouterConfig {
+	configUpdateMu.Lock()
+	defer configUpdateMu.Unlock()
+
+	if configUpdateCh == nil {
+		configUpdateCh = make(chan *RouterConfig, 1)
+	}
+	return configUpdateCh
+}

src/semantic-router/pkg/extproc/server.go

@@ -160,11 +160,11 @@ func (rs *RouterService) Process(stream ext_proc.ExternalProcessor_ProcessServer
 // watchConfigAndReload watches the config file and reloads router on changes.
 func (s *Server) watchConfigAndReload(ctx context.Context) {
 	// Check if we're using Kubernetes config source
-	// If so, skip file watching as config is managed by the Kubernetes controller
 	cfg := config.Get()
 	if cfg != nil && cfg.ConfigSource == config.ConfigSourceKubernetes {
-		logging.Infof("ConfigSource is kubernetes, skipping file watcher")
-		<-ctx.Done() // Just wait for context cancellation
+		logging.Infof("ConfigSource is kubernetes, watching for config updates from controller")
+		// Watch for config updates from the Kubernetes controller
+		s.watchKubernetesConfigUpdates(ctx)
 		return
 	}
 
@@ -244,3 +244,35 @@ func (s *Server) watchConfigAndReload(ctx context.Context) {
 		}
 	}
 }
+
+// watchKubernetesConfigUpdates watches for config updates from the Kubernetes controller
+func (s *Server) watchKubernetesConfigUpdates(ctx context.Context) {
+	updateCh := config.WatchConfigUpdates()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case newCfg := <-updateCh:
+			if newCfg == nil {
+				continue
+			}
+
+			// Build a new router with the updated config
+			// Note: We pass the configPath but NewOpenAIRouter will use the global config
+			newRouter, err := NewOpenAIRouter(s.configPath)
+			if err != nil {
+				logging.LogEvent("config_reload_failed", map[string]interface{}{
+					"source": "kubernetes",
+					"error":  err.Error(),
+				})
+				continue
+			}
+
+			s.service.Swap(newRouter)
+			logging.LogEvent("config_reloaded", map[string]interface{}{
+				"source": "kubernetes",
+			})
+		}
+	}
+}