Feat: Embedded HTTP API Server #7

Author: Denis Tu

cmd/main.go

@@ -17,6 +17,7 @@ import (
 	kagentv1alpha2 "github.com/kagent-dev/khook/api/v1alpha2"
 	kclient "github.com/kagent-dev/khook/internal/client"
 	"github.com/kagent-dev/khook/internal/config"
+	"github.com/kagent-dev/khook/internal/sre"
 	"github.com/kagent-dev/khook/internal/workflow"
 )
 
@@ -76,8 +77,17 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Start SRE-IDE server
+	sreServer := sre.NewServer(8082, mgr.GetClient())
+	go func() {
+		ctx := context.Background()
+		if err := sreServer.Start(ctx); err != nil {
+			setupLog.Error(err, "problem running SRE-IDE server")
+		}
+	}()
+
 	// Add workflow coordinator to manage hooks and event processing
-	if err := mgr.Add(newWorkflowCoordinator(mgr)); err != nil {
+	if err := mgr.Add(newWorkflowCoordinator(mgr, sreServer)); err != nil {
 		setupLog.Error(err, "unable to add workflow coordinator")
 		os.Exit(1)
 	}
@@ -91,11 +101,15 @@ func main() {
 
 // workflowCoordinator manages the complete workflow lifecycle using proper services
 type workflowCoordinator struct {
-	mgr ctrl.Manager
+	mgr       ctrl.Manager
+	sreServer *sre.Server
 }
 
-func newWorkflowCoordinator(mgr ctrl.Manager) *workflowCoordinator {
-	return &workflowCoordinator{mgr: mgr}
+func newWorkflowCoordinator(mgr ctrl.Manager, sreServer *sre.Server) *workflowCoordinator {
+	return &workflowCoordinator{
+		mgr:       mgr,
+		sreServer: sreServer,
+	}
 }
 
 func (w *workflowCoordinator) NeedLeaderElection() bool { return true }
@@ -121,7 +135,7 @@ func (w *workflowCoordinator) Start(ctx context.Context) error {
 
 	// Create workflow coordinator
 	eventRecorder := w.mgr.GetEventRecorderFor("khook")
-	coordinator := workflow.NewCoordinator(k8s, w.mgr.GetClient(), kagentCli, eventRecorder)
+	coordinator := workflow.NewCoordinator(k8s, w.mgr.GetClient(), kagentCli, eventRecorder, w.sreServer)
 
 	// Start the coordinator
 	return coordinator.Start(ctx)

internal/pipeline/processor.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/kagent-dev/khook/api/v1alpha2"
 	"github.com/kagent-dev/khook/internal/interfaces"
+	"github.com/kagent-dev/khook/internal/sre"
 )
 
 // Processor handles the complete event processing pipeline
@@ -22,6 +23,7 @@ type Processor struct {
 	deduplicationManager interfaces.DeduplicationManager
 	kagentClient         interfaces.KagentClient
 	statusManager        interfaces.StatusManager
+	sreServer            interface{}
 	logger               logr.Logger
 }
 
@@ -31,12 +33,14 @@ func NewProcessor(
 	deduplicationManager interfaces.DeduplicationManager,
 	kagentClient interfaces.KagentClient,
 	statusManager interfaces.StatusManager,
+	sreServer interface{},
 ) *Processor {
 	return &Processor{
 		eventWatcher:         eventWatcher,
 		deduplicationManager: deduplicationManager,
 		kagentClient:         kagentClient,
 		statusManager:        statusManager,
+		sreServer:            sreServer,
 		logger:               log.Log.WithName("event-processor"),
 	}
 }
@@ -167,6 +171,19 @@ func (p *Processor) processEventMatch(ctx context.Context, match EventMatch) err
 		// Continue even if status recording fails
 	}
 
+	// Add alert to SRE server if available
+	p.logger.Info("Checking SRE server integration", "sreServer", p.sreServer != nil)
+	if p.sreServer != nil {
+		if sreServer, ok := p.sreServer.(*sre.Server); ok {
+			// Convert event to alert and add to SRE server
+			alert := sre.ConvertEventToAlert(match.Event, match.Hook, agentRef, response)
+			sreServer.AddAlert(alert)
+			p.logger.Info("Added alert to SRE server", "alertId", alert.ID)
+		} else {
+			p.logger.Error(nil, "Type assertion failed for SRE server", "sreServerType", fmt.Sprintf("%T", p.sreServer))
+		}
+	}
+
 	// Mark event as notified to suppress re-sending within suppression window
 	p.deduplicationManager.MarkNotified(hookRef, match.Event)