remove testing physical dispatching in alerting tests

pkg/test/alerting/alert_router_gen.go

@@ -3,13 +3,10 @@ package alerting
 import (
 	"context"
 	"fmt"
-	"io"
 	"math/rand"
-	"net/http"
+	"net/url"
 	"os"
 	"path"
-	"strings"
-	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -44,32 +41,36 @@ type MockIntegrationWebhookServer struct {
 	Webhook    string
 	Port       int
 	Addr       string
-	*sync.RWMutex
-	Buffer []*config.WebhookMessage
+	// *sync.RWMutex
+	// Buffer []*config.WebhookMessage
 }
 
-func (m *MockIntegrationWebhookServer) WriteBuffer(msg *config.WebhookMessage) {
-	m.Lock()
-	defer m.Unlock()
-	m.Buffer = append(m.Buffer, msg)
-}
-
-func (m *MockIntegrationWebhookServer) ClearBuffer() {
-	m.Lock()
-	defer m.Unlock()
-	m.Buffer = m.Buffer[:0]
-}
-
-func (m *MockIntegrationWebhookServer) GetBuffer() []*config.WebhookMessage {
-	m.RLock()
-	defer m.RUnlock()
-	return lo.Map(m.Buffer, func(msg *config.WebhookMessage, _ int) *config.WebhookMessage {
-		return msg
-	})
-}
+// func (m *MockIntegrationWebhookServer) WriteBuffer(msg *config.WebhookMessage) {
+// 	m.Lock()
+// 	defer m.Unlock()
+// 	m.Buffer = append(m.Buffer, msg)
+// }
+
+// func (m *MockIntegrationWebhookServer) ClearBuffer() {
+// 	m.Lock()
+// 	defer m.Unlock()
+// 	m.Buffer = m.Buffer[:0]
+// }
+
+// func (m *MockIntegrationWebhookServer) GetBuffer() []*config.WebhookMessage {
+// 	m.RLock()
+// 	defer m.RUnlock()
+// 	return lo.Map(m.Buffer, func(msg *config.WebhookMessage, _ int) *config.WebhookMessage {
+// 		return msg
+// 	})
+// }
 
 func (m *MockIntegrationWebhookServer) GetWebhook() string {
-	return "http://" + path.Join(m.Addr, m.Webhook)
+	webhook := url.URL{
+		Scheme: "http",
+		Host:   fmt.Sprintf("%s:%d", m.Addr, m.Port),
+	}
+	return webhook.String()
 }
 
 func (m *MockIntegrationWebhookServer) Endpoint() *alertingv1.AlertEndpoint {
@@ -85,60 +86,21 @@ func (m *MockIntegrationWebhookServer) Endpoint() *alertingv1.AlertEndpoint {
 	}
 }
 
-func CreateWebhookServer(e *test.Environment, num int) []*MockIntegrationWebhookServer {
+func CreateWebhookServer(num int) []*MockIntegrationWebhookServer {
 	var servers []*MockIntegrationWebhookServer
 	for i := 0; i < num; i++ {
-		servers = append(servers, NewWebhookMemoryServer(e, "webhook"))
+		servers = append(servers, NewWebhookMemoryServer("webhook"))
 	}
 	return servers
 }
 
-func NewWebhookMemoryServer(e *test.Environment, webHookRoute string) *MockIntegrationWebhookServer {
-	port := freeport.GetFreePort()
-	buf := []*config.WebhookMessage{}
-	mu := &sync.RWMutex{}
-	mux := http.NewServeMux()
-	res := &MockIntegrationWebhookServer{
+func NewWebhookMemoryServer(webHookRoute string) *MockIntegrationWebhookServer {
+	return &MockIntegrationWebhookServer{
+		EndpointId: shared.NewAlertingRefId("webhook"),
 		Webhook:    webHookRoute,
-		Port:       port,
-		Buffer:     buf,
-		RWMutex:    mu,
-		EndpointId: uuid.New().String(),
-	}
-	if !strings.HasPrefix(webHookRoute, "/") {
-		webHookRoute = "/" + webHookRoute
-	}
-	mux.HandleFunc(webHookRoute, func(w http.ResponseWriter, r *http.Request) {
-		data, err := io.ReadAll(r.Body)
-		if err != nil {
-			panic(err)
-		}
-		var msg config.WebhookMessage
-		err = yaml.Unmarshal(data, &msg)
-		if err != nil {
-			panic(err)
-		}
-		res.WriteBuffer(&msg)
-	})
-	webhookServer := &http.Server{
-		Addr:           fmt.Sprintf("localhost:%d", port),
-		Handler:        mux,
-		ReadTimeout:    30 * time.Second,
-		WriteTimeout:   30 * time.Second,
-		MaxHeaderBytes: 1 << 20,
+		Port:       3000,
+		Addr:       "127.0.0.1",
 	}
-	res.Addr = webhookServer.Addr
-
-	go func() {
-		err := webhookServer.ListenAndServe()
-		if err != http.ErrServerClosed {
-			panic(err)
-		}
-	}()
-	context.AfterFunc(e.Context(), func() {
-		webhookServer.Shutdown(context.Background())
-	})
-	return res
 }
 
 type NamespaceSubTreeTestcase struct {

plugins/alerting/pkg/alerting/alarms/v1/runner.go

@@ -21,6 +21,7 @@ type Runner struct {
 func NewRunner() *Runner {
 	return &Runner{
 		systemConditionUpdateListeners: make(map[string]*EvaluatorContext),
+		systemConditionMu:              &sync.Mutex{},
 	}
 }
 

plugins/alerting/pkg/alerting/alarms/v1/streams.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	alertingv1 "github.com/rancher/opni/pkg/apis/alerting/v1"
@@ -155,8 +156,9 @@ func (p *AlarmServerComponent) onSystemConditionCreate(conditionId, conditionNam
 		evaluator.EvaluateLoop()
 	}()
 	p.runner.AddSystemConfigListener(conditionId, &EvaluatorContext{
-		Ctx:    evaluator.evaluationCtx,
-		Cancel: evaluator.cancelEvaluation,
+		Ctx:     evaluator.evaluationCtx,
+		Cancel:  evaluator.cancelEvaluation,
+		running: &atomic.Bool{},
 	})
 	return nil
 }
@@ -247,8 +249,9 @@ func (p *AlarmServerComponent) onDownstreamCapabilityConditionCreate(conditionId
 		evaluator.EvaluateLoop()
 	}()
 	p.runner.AddSystemConfigListener(conditionId, &EvaluatorContext{
-		Ctx:    evaluator.evaluationCtx,
-		Cancel: evaluator.cancelEvaluation,
+		Ctx:     evaluator.evaluationCtx,
+		Cancel:  evaluator.cancelEvaluation,
+		running: &atomic.Bool{},
 	})
 	return nil
 }
@@ -485,8 +488,9 @@ func (p *AlarmServerComponent) onCortexClusterStatusCreate(conditionId, conditio
 		evaluator.EvaluateLoop()
 	}()
 	p.runner.AddSystemConfigListener(conditionId, &EvaluatorContext{
-		Ctx:    evaluator.evaluationCtx,
-		Cancel: evaluator.cancelEvaluation,
+		Ctx:     evaluator.evaluationCtx,
+		Cancel:  evaluator.cancelEvaluation,
+		running: &atomic.Bool{},
 	})
 	return nil
 }

plugins/alerting/pkg/alerting/system.go

@@ -201,7 +201,7 @@ func (p *Plugin) handleDriverNotifications() {
 			p.logger.Info("shutting down cluster driver update handler")
 			return
 		case client := <-p.clusterNotifier:
-			p.logger.Infof("updating alerting client based on cluster status : %v", client)
+			p.logger.Info("updating alerting client based on cluster status")
 			serverCfg := server.Config{
 				Client: client.Clone(),
 			}