Pooling

cmd/agent/main.go

@@ -45,7 +45,31 @@ func main() {
 	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer cancel()
 	g := glpi.New(cfg.GLPIURL, cfg.GLPIAPIVersion, cfg.GLPIClientID, cfg.GLPIClientSecret, cfg.GLPIUsername, cfg.GLPIPassword, cfg.GLPITimeout)
-	o := ollama.New(cfg.OllamaURL, cfg.OllamaModel, cfg.OllamaEmbeddingModel, cfg.CommunicationLanguage, cfg.CommunicationStyle, cfg.OllamaTimeout, cfg.OllamaNumPredict, cfg.OllamaKeepAlive, cfg.OllamaThink, cfg.OllamaMaxConcurrent, cfg.OllamaJSONRetries)
+	nodes := make([]ollama.NodeConfig, 0, len(cfg.OllamaURLs))
+	for i, nodeURL := range cfg.OllamaURLs {
+		name := ""
+		if i < len(cfg.OllamaNodeNames) {
+			name = cfg.OllamaNodeNames[i]
+		}
+		weight := 1
+		if i < len(cfg.OllamaNodeWeights) {
+			weight = cfg.OllamaNodeWeights[i]
+		}
+		nodes = append(nodes, ollama.NodeConfig{Name: name, URL: nodeURL, Weight: weight})
+	}
+	o, err := ollama.NewPool(ollama.PoolConfig{
+		Nodes: nodes, RoutingMode: cfg.OllamaRoutingMode, NodeMaxInflight: cfg.OllamaNodeMaxInflight,
+		HealthInterval: cfg.OllamaNodeHealthInterval, FailureCooldown: cfg.OllamaNodeFailureCooldown,
+		NodeRequestTimeout: cfg.OllamaNodeRequestTimeout, FailoverEnabled: cfg.OllamaFailoverEnabled,
+		FailoverAttempts: cfg.OllamaFailoverAttempts, RequireSameModelDigest: cfg.OllamaRequireSameDigest,
+		RequireEmbeddingModel: cfg.OllamaRequireEmbeddingModel, Model: cfg.OllamaModel, EmbeddingModel: cfg.OllamaEmbeddingModel,
+	}, cfg.OllamaModel, cfg.OllamaEmbeddingModel, cfg.CommunicationLanguage, cfg.CommunicationStyle, cfg.OllamaNumPredict, cfg.OllamaKeepAlive, cfg.OllamaThink, cfg.OllamaJSONRetries)
+	if err != nil {
+		slog.Error("Ollama pool configuration failed", "error", err)
+		os.Exit(1)
+	}
+	o.Start(ctx)
+	slog.Info("Ollama pool configured", "nodes", len(nodes), "routing", cfg.OllamaRoutingMode, "max_inflight_per_node", cfg.OllamaNodeMaxInflight, "failover", cfg.OllamaFailoverEnabled, "failover_attempts", cfg.OllamaFailoverAttempts, "require_same_digest", cfg.OllamaRequireSameDigest)
 	if err := g.ValidateContract(ctx); err != nil {
 		slog.Error("GLPI API contract validation failed", "error", err)
 		os.Exit(1)
@@ -92,7 +116,7 @@ func main() {
 	}
 	contextCollector := contextdata.New(cfg, g, kuma)
 	svc := agent.New(cfg, g, o, k, l, st, q, m, contextCollector)
-	web, err := webui.New(cfg, m, st, q, k, svc)
+	web, err := webui.New(cfg, m, st, q, k, svc, o)
 	if err != nil {
 		slog.Error("web UI initialization failed", "error", err)
 		os.Exit(1)
@@ -110,6 +134,9 @@ func main() {
 	// Ticket polling/workers remain paused until the local index is ready.
 	go func() {
 		slog.Info("knowledge initialization started in background", "knowledge_dir", cfg.KnowledgeDir, "rag_enabled", cfg.RAGEnabled)
+		if err := waitForOllamaPool(ctx, o, cfg.OllamaNodeHealthInterval); err != nil {
+			return
+		}
 		if err := k.Initialize(ctx); err != nil {
 			slog.Error("knowledge store initialization failed; web UI remains available", "error", err, "knowledge_dir", cfg.KnowledgeDir, "data_dir", cfg.DataDir, "rag_enabled", cfg.RAGEnabled)
 			return
@@ -143,6 +170,39 @@ func main() {
 	slog.Info("shutdown complete")
 }
 
+func waitForOllamaPool(ctx context.Context, client *ollama.Client, retryInterval time.Duration) error {
+	if retryInterval < 2*time.Second {
+		retryInterval = 5 * time.Second
+	}
+	for {
+		err := client.Ping(ctx)
+		if err == nil {
+			statuses := client.NodeStatuses()
+			healthy := 0
+			for _, status := range statuses {
+				if status.Healthy && status.Compatible {
+					healthy++
+				}
+			}
+			slog.Info("Ollama pool ready", "healthy_nodes", healthy, "nodes", len(statuses), "routing", client.RoutingMode())
+			return nil
+		}
+		slog.Warn("waiting for compatible Ollama pool", "retry_in", retryInterval.String(), "error", err)
+		timer := time.NewTimer(retryInterval)
+		select {
+		case <-ctx.Done():
+			if !timer.Stop() {
+				select {
+				case <-timer.C:
+				default:
+				}
+			}
+			return ctx.Err()
+		case <-timer.C:
+		}
+	}
+}
+
 func maxDuration(a, b time.Duration) time.Duration {
 	if a > b {
 		return a