feat: Extract Wild Central as standalone Go service
Extract the Central networking functionality from wild-cloud/api into a standalone service. Wild Central manages DNS (dnsmasq), gateway (HAProxy), firewall (nftables), VPN (WireGuard), TLS (certbot), security (CrowdSec), and DDNS — all the network appliance concerns. All Cloud-specific code (instances, clusters, nodes, apps, backups, operations, kubectl/talosctl tooling) has been removed. The API struct and route registration contain only Central endpoints. Tests updated to match the new API signature. Builds and all tests pass. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
268
internal/sse/manager.go
Normal file
268
internal/sse/manager.go
Normal file
@@ -0,0 +1,268 @@
|
||||
package sse
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"golang.org/x/time/rate"
|
||||
)
|
||||
|
||||
// Event represents an SSE event
|
||||
type Event struct {
|
||||
ID string `json:"id"`
|
||||
Type string `json:"type"`
|
||||
InstanceName string `json:"instanceName"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
Data any `json:"data"`
|
||||
Metadata map[string]any `json:"metadata,omitempty"`
|
||||
}
|
||||
|
||||
// Client represents an SSE client connection
|
||||
type Client struct {
|
||||
ID string
|
||||
InstanceName string
|
||||
Channel chan *Event
|
||||
Filters EventFilters
|
||||
Context context.Context
|
||||
Cancel context.CancelFunc
|
||||
}
|
||||
|
||||
// EventFilters for client-specific filtering
|
||||
type EventFilters struct {
|
||||
EventTypes []string `json:"eventTypes,omitempty"`
|
||||
Namespaces []string `json:"namespaces,omitempty"`
|
||||
Apps []string `json:"apps,omitempty"`
|
||||
}
|
||||
|
||||
// Manager manages all SSE connections
|
||||
type Manager struct {
|
||||
clients map[string]map[string]*Client // instanceName -> clientID -> Client
|
||||
unregister chan *Client
|
||||
broadcast chan *Event
|
||||
mu sync.RWMutex
|
||||
rateLimiters map[string]*rate.Limiter
|
||||
}
|
||||
|
||||
// NewManager creates a new SSE manager
|
||||
func NewManager() *Manager {
|
||||
m := &Manager{
|
||||
clients: make(map[string]map[string]*Client),
|
||||
unregister: make(chan *Client, 100),
|
||||
broadcast: make(chan *Event, 1000),
|
||||
rateLimiters: make(map[string]*rate.Limiter),
|
||||
}
|
||||
go m.run()
|
||||
return m
|
||||
}
|
||||
|
||||
// run processes client unregistration and event broadcasting
|
||||
func (m *Manager) run() {
|
||||
for {
|
||||
select {
|
||||
case client := <-m.unregister:
|
||||
m.mu.Lock()
|
||||
if clients, ok := m.clients[client.InstanceName]; ok {
|
||||
delete(clients, client.ID)
|
||||
if len(clients) == 0 {
|
||||
delete(m.clients, client.InstanceName)
|
||||
}
|
||||
}
|
||||
close(client.Channel)
|
||||
m.mu.Unlock()
|
||||
slog.Info("client unregistered", "component", "sse", "client", client.ID)
|
||||
|
||||
case event := <-m.broadcast:
|
||||
m.mu.RLock()
|
||||
// Send to clients registered for this specific instance
|
||||
instanceClients := m.clients[event.InstanceName]
|
||||
// Also send to clients registered with wildcard "*" to receive all events
|
||||
globalClients := m.clients["*"]
|
||||
m.mu.RUnlock()
|
||||
|
||||
// Send to instance-specific clients
|
||||
for _, client := range instanceClients {
|
||||
if m.shouldSendToClient(event, client) {
|
||||
select {
|
||||
case client.Channel <- event:
|
||||
default:
|
||||
// Client channel full, skip
|
||||
slog.Info("client channel full, skipping event", "component", "sse", "client", client.ID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Send to global clients (registered with "*")
|
||||
for _, client := range globalClients {
|
||||
if m.shouldSendToClient(event, client) {
|
||||
select {
|
||||
case client.Channel <- event:
|
||||
default:
|
||||
// Client channel full, skip
|
||||
slog.Info("client channel full, skipping event", "component", "sse", "client", client.ID)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// shouldSendToClient checks if event matches client filters
|
||||
func (m *Manager) shouldSendToClient(event *Event, client *Client) bool {
|
||||
// Rate limiting per client
|
||||
limiter := m.getRateLimiter(client.ID)
|
||||
if !limiter.Allow() {
|
||||
return false
|
||||
}
|
||||
|
||||
// Check event type filter
|
||||
if len(client.Filters.EventTypes) > 0 {
|
||||
matched := false
|
||||
for _, eventType := range client.Filters.EventTypes {
|
||||
if event.Type == eventType {
|
||||
matched = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !matched {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// Check namespace filter for k8s events
|
||||
if len(client.Filters.Namespaces) > 0 {
|
||||
if namespace, ok := event.Metadata["namespace"].(string); ok {
|
||||
matched := false
|
||||
for _, ns := range client.Filters.Namespaces {
|
||||
if namespace == ns {
|
||||
matched = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !matched {
|
||||
return false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Check app filter
|
||||
if len(client.Filters.Apps) > 0 {
|
||||
if appName, ok := event.Metadata["app"].(string); ok {
|
||||
matched := false
|
||||
for _, app := range client.Filters.Apps {
|
||||
if appName == app {
|
||||
matched = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !matched {
|
||||
return false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
// getRateLimiter gets or creates a rate limiter for a client
|
||||
func (m *Manager) getRateLimiter(clientID string) *rate.Limiter {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
if limiter, exists := m.rateLimiters[clientID]; exists {
|
||||
return limiter
|
||||
}
|
||||
|
||||
// 100 events per second with burst of 200
|
||||
limiter := rate.NewLimiter(100, 200)
|
||||
m.rateLimiters[clientID] = limiter
|
||||
return limiter
|
||||
}
|
||||
|
||||
// RegisterClient registers a new SSE client
|
||||
func (m *Manager) RegisterClient(instanceName string, filters EventFilters) *Client {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
client := &Client{
|
||||
ID: generateClientID(),
|
||||
InstanceName: instanceName,
|
||||
Channel: make(chan *Event, 100),
|
||||
Filters: filters,
|
||||
Context: ctx,
|
||||
Cancel: cancel,
|
||||
}
|
||||
|
||||
m.mu.Lock()
|
||||
if m.clients[instanceName] == nil {
|
||||
m.clients[instanceName] = make(map[string]*Client)
|
||||
}
|
||||
m.clients[instanceName][client.ID] = client
|
||||
m.mu.Unlock()
|
||||
|
||||
slog.Info("client registered", "component", "sse", "client", client.ID, "instance", instanceName)
|
||||
return client
|
||||
}
|
||||
|
||||
// UnregisterClient removes a client
|
||||
func (m *Manager) UnregisterClient(client *Client) {
|
||||
client.Cancel()
|
||||
m.unregister <- client
|
||||
|
||||
// Clean up rate limiter
|
||||
m.mu.Lock()
|
||||
delete(m.rateLimiters, client.ID)
|
||||
m.mu.Unlock()
|
||||
}
|
||||
|
||||
// Broadcast sends an event to all matching clients
|
||||
func (m *Manager) Broadcast(event *Event) {
|
||||
event.ID = generateEventID()
|
||||
event.Timestamp = time.Now()
|
||||
|
||||
select {
|
||||
case m.broadcast <- event:
|
||||
default:
|
||||
slog.Error("broadcast channel full, dropping event", "component", "sse", "event", event.ID, "type", event.Type, "instance", event.InstanceName)
|
||||
}
|
||||
}
|
||||
|
||||
// GetConnectionCount returns the number of active connections
|
||||
func (m *Manager) GetConnectionCount() int {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
count := 0
|
||||
for _, clients := range m.clients {
|
||||
count += len(clients)
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
// GetInstanceConnectionCount returns the number of connections for a specific instance
|
||||
func (m *Manager) GetInstanceConnectionCount(instanceName string) int {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
|
||||
if clients, ok := m.clients[instanceName]; ok {
|
||||
return len(clients)
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// Helper functions
|
||||
func generateClientID() string {
|
||||
return fmt.Sprintf("client-%s", uuid.New().String()[:8])
|
||||
}
|
||||
|
||||
func generateEventID() string {
|
||||
return fmt.Sprintf("event-%s", uuid.New().String()[:8])
|
||||
}
|
||||
|
||||
// JSON marshals the event to JSON
|
||||
func (e *Event) JSON() ([]byte, error) {
|
||||
return json.Marshal(e)
|
||||
}
|
||||
352
internal/sse/manager_test.go
Normal file
352
internal/sse/manager_test.go
Normal file
@@ -0,0 +1,352 @@
|
||||
package sse
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestManagerClientRegistration(t *testing.T) {
|
||||
manager := NewManager()
|
||||
|
||||
// Test client registration
|
||||
client := manager.RegisterClient("test-instance", EventFilters{})
|
||||
if client == nil {
|
||||
t.Fatal("Expected client to be registered")
|
||||
}
|
||||
|
||||
// Give manager time to process registration
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Check client is registered
|
||||
if count := manager.GetInstanceConnectionCount("test-instance"); count != 1 {
|
||||
t.Errorf("Expected 1 connection, got %d", count)
|
||||
}
|
||||
|
||||
// Test client unregistration
|
||||
manager.UnregisterClient(client)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
if count := manager.GetInstanceConnectionCount("test-instance"); count != 0 {
|
||||
t.Errorf("Expected 0 connections, got %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagerBroadcast(t *testing.T) {
|
||||
manager := NewManager()
|
||||
|
||||
// Register two clients for same instance
|
||||
client1 := manager.RegisterClient("test-instance", EventFilters{
|
||||
EventTypes: []string{"pod.added"},
|
||||
})
|
||||
client2 := manager.RegisterClient("test-instance", EventFilters{})
|
||||
|
||||
// Give manager time to process registrations
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Broadcast an event
|
||||
event := &Event{
|
||||
Type: "pod.added",
|
||||
InstanceName: "test-instance",
|
||||
Data: map[string]any{"name": "test-pod"},
|
||||
Metadata: map[string]any{},
|
||||
}
|
||||
manager.Broadcast(event)
|
||||
|
||||
// Both clients should receive the event
|
||||
select {
|
||||
case receivedEvent := <-client1.Channel:
|
||||
if receivedEvent.Type != "pod.added" {
|
||||
t.Errorf("Expected event type 'pod.added', got '%s'", receivedEvent.Type)
|
||||
}
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Error("Client1 did not receive event")
|
||||
}
|
||||
|
||||
select {
|
||||
case receivedEvent := <-client2.Channel:
|
||||
if receivedEvent.Type != "pod.added" {
|
||||
t.Errorf("Expected event type 'pod.added', got '%s'", receivedEvent.Type)
|
||||
}
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Error("Client2 did not receive event")
|
||||
}
|
||||
|
||||
// Cleanup
|
||||
manager.UnregisterClient(client1)
|
||||
manager.UnregisterClient(client2)
|
||||
}
|
||||
|
||||
func TestEventFiltering(t *testing.T) {
|
||||
manager := NewManager()
|
||||
|
||||
// Register client with event type filter
|
||||
client := manager.RegisterClient("test-instance", EventFilters{
|
||||
EventTypes: []string{"pod.added", "pod.updated"},
|
||||
})
|
||||
|
||||
// Give manager time to process registration
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Send events of different types
|
||||
podAddedEvent := &Event{
|
||||
Type: "pod.added",
|
||||
InstanceName: "test-instance",
|
||||
Data: map[string]any{"name": "test-pod"},
|
||||
}
|
||||
podDeletedEvent := &Event{
|
||||
Type: "pod.deleted",
|
||||
InstanceName: "test-instance",
|
||||
Data: map[string]any{"name": "test-pod"},
|
||||
}
|
||||
|
||||
manager.Broadcast(podAddedEvent)
|
||||
manager.Broadcast(podDeletedEvent)
|
||||
|
||||
// Should only receive pod.added event
|
||||
select {
|
||||
case receivedEvent := <-client.Channel:
|
||||
if receivedEvent.Type != "pod.added" {
|
||||
t.Errorf("Expected event type 'pod.added', got '%s'", receivedEvent.Type)
|
||||
}
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Error("Client did not receive pod.added event")
|
||||
}
|
||||
|
||||
// Should not receive pod.deleted event
|
||||
select {
|
||||
case receivedEvent := <-client.Channel:
|
||||
t.Errorf("Should not have received event of type '%s'", receivedEvent.Type)
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
// Expected - no event should be received
|
||||
}
|
||||
|
||||
// Cleanup
|
||||
manager.UnregisterClient(client)
|
||||
}
|
||||
|
||||
func TestNamespaceFiltering(t *testing.T) {
|
||||
manager := NewManager()
|
||||
|
||||
// Register client with namespace filter
|
||||
client := manager.RegisterClient("test-instance", EventFilters{
|
||||
Namespaces: []string{"default", "kube-system"},
|
||||
})
|
||||
|
||||
// Give manager time to process registration
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Send events from different namespaces
|
||||
defaultNsEvent := &Event{
|
||||
Type: "pod.added",
|
||||
InstanceName: "test-instance",
|
||||
Data: map[string]any{"name": "test-pod"},
|
||||
Metadata: map[string]any{"namespace": "default"},
|
||||
}
|
||||
otherNsEvent := &Event{
|
||||
Type: "pod.added",
|
||||
InstanceName: "test-instance",
|
||||
Data: map[string]any{"name": "test-pod"},
|
||||
Metadata: map[string]any{"namespace": "other-namespace"},
|
||||
}
|
||||
|
||||
manager.Broadcast(defaultNsEvent)
|
||||
manager.Broadcast(otherNsEvent)
|
||||
|
||||
// Should only receive event from default namespace
|
||||
select {
|
||||
case receivedEvent := <-client.Channel:
|
||||
if ns, ok := receivedEvent.Metadata["namespace"].(string); !ok || ns != "default" {
|
||||
t.Errorf("Expected event from 'default' namespace, got '%v'", receivedEvent.Metadata["namespace"])
|
||||
}
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Error("Client did not receive event from default namespace")
|
||||
}
|
||||
|
||||
// Should not receive event from other namespace
|
||||
select {
|
||||
case receivedEvent := <-client.Channel:
|
||||
t.Errorf("Should not have received event from namespace '%v'", receivedEvent.Metadata["namespace"])
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
// Expected - no event should be received
|
||||
}
|
||||
|
||||
// Cleanup
|
||||
manager.UnregisterClient(client)
|
||||
}
|
||||
|
||||
func TestMultipleInstanceIsolation(t *testing.T) {
|
||||
manager := NewManager()
|
||||
|
||||
// Register clients for different instances
|
||||
client1 := manager.RegisterClient("instance-1", EventFilters{})
|
||||
client2 := manager.RegisterClient("instance-2", EventFilters{})
|
||||
|
||||
// Give manager time to process registrations
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Send event to instance-1
|
||||
event := &Event{
|
||||
Type: "pod.added",
|
||||
InstanceName: "instance-1",
|
||||
Data: map[string]any{"name": "test-pod"},
|
||||
}
|
||||
manager.Broadcast(event)
|
||||
|
||||
// Only client1 should receive the event
|
||||
select {
|
||||
case receivedEvent := <-client1.Channel:
|
||||
if receivedEvent.InstanceName != "instance-1" {
|
||||
t.Errorf("Expected event for 'instance-1', got '%s'", receivedEvent.InstanceName)
|
||||
}
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Error("Client1 did not receive event")
|
||||
}
|
||||
|
||||
// Client2 should not receive the event
|
||||
select {
|
||||
case <-client2.Channel:
|
||||
t.Error("Client2 should not have received event for instance-1")
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
// Expected - no event should be received
|
||||
}
|
||||
|
||||
// Cleanup
|
||||
manager.UnregisterClient(client1)
|
||||
manager.UnregisterClient(client2)
|
||||
}
|
||||
|
||||
func TestRateLimiting(t *testing.T) {
|
||||
manager := NewManager()
|
||||
|
||||
// Register a client
|
||||
client := manager.RegisterClient("test-instance", EventFilters{})
|
||||
|
||||
// Give manager time to process registration
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Send many events rapidly (more than rate limit allows)
|
||||
for i := 0; i < 300; i++ {
|
||||
event := &Event{
|
||||
Type: "pod.added",
|
||||
InstanceName: "test-instance",
|
||||
Data: map[string]any{"index": i},
|
||||
}
|
||||
manager.Broadcast(event)
|
||||
}
|
||||
|
||||
// Count received events
|
||||
receivedCount := 0
|
||||
timeout := time.After(500 * time.Millisecond)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-client.Channel:
|
||||
receivedCount++
|
||||
case <-timeout:
|
||||
// Rate limiter should have dropped some events
|
||||
// We allow 100/sec with burst of 200, so we should receive around 200
|
||||
if receivedCount >= 300 {
|
||||
t.Errorf("Rate limiting not working: received %d events (expected < 300)", receivedCount)
|
||||
}
|
||||
if receivedCount < 100 {
|
||||
t.Errorf("Too many events dropped: received only %d events", receivedCount)
|
||||
}
|
||||
// Cleanup
|
||||
manager.UnregisterClient(client)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestClientContextCancellation(t *testing.T) {
|
||||
manager := NewManager()
|
||||
|
||||
// Register a client
|
||||
client := manager.RegisterClient("test-instance", EventFilters{})
|
||||
|
||||
// Cancel the client context
|
||||
client.Cancel()
|
||||
|
||||
// Try to send an event
|
||||
event := &Event{
|
||||
Type: "pod.added",
|
||||
InstanceName: "test-instance",
|
||||
Data: map[string]any{"name": "test-pod"},
|
||||
}
|
||||
manager.Broadcast(event)
|
||||
|
||||
// Client should not receive event after context cancellation
|
||||
select {
|
||||
case <-client.Channel:
|
||||
// Channel might still have buffered events, but should be closed soon
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
// Expected - context cancelled
|
||||
}
|
||||
|
||||
// Cleanup
|
||||
manager.UnregisterClient(client)
|
||||
}
|
||||
|
||||
func TestConnectionCount(t *testing.T) {
|
||||
manager := NewManager()
|
||||
|
||||
// Initially no connections
|
||||
if count := manager.GetConnectionCount(); count != 0 {
|
||||
t.Errorf("Expected 0 connections initially, got %d", count)
|
||||
}
|
||||
|
||||
// Register clients for different instances
|
||||
client1 := manager.RegisterClient("instance-1", EventFilters{})
|
||||
client2 := manager.RegisterClient("instance-1", EventFilters{})
|
||||
client3 := manager.RegisterClient("instance-2", EventFilters{})
|
||||
|
||||
// Give manager time to process registrations
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Check total connection count
|
||||
if count := manager.GetConnectionCount(); count != 3 {
|
||||
t.Errorf("Expected 3 total connections, got %d", count)
|
||||
}
|
||||
|
||||
// Check per-instance counts
|
||||
if count := manager.GetInstanceConnectionCount("instance-1"); count != 2 {
|
||||
t.Errorf("Expected 2 connections for instance-1, got %d", count)
|
||||
}
|
||||
if count := manager.GetInstanceConnectionCount("instance-2"); count != 1 {
|
||||
t.Errorf("Expected 1 connection for instance-2, got %d", count)
|
||||
}
|
||||
|
||||
// Cleanup
|
||||
manager.UnregisterClient(client1)
|
||||
manager.UnregisterClient(client2)
|
||||
manager.UnregisterClient(client3)
|
||||
}
|
||||
|
||||
func BenchmarkBroadcast(b *testing.B) {
|
||||
manager := NewManager()
|
||||
|
||||
// Register 100 clients
|
||||
clients := make([]*Client, 100)
|
||||
for i := 0; i < 100; i++ {
|
||||
clients[i] = manager.RegisterClient("test-instance", EventFilters{})
|
||||
}
|
||||
|
||||
// Give manager time to process registrations
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
// Benchmark broadcasting events
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
event := &Event{
|
||||
Type: "pod.added",
|
||||
InstanceName: "test-instance",
|
||||
Data: map[string]any{"index": i},
|
||||
}
|
||||
manager.Broadcast(event)
|
||||
}
|
||||
|
||||
// Cleanup
|
||||
for _, client := range clients {
|
||||
manager.UnregisterClient(client)
|
||||
}
|
||||
}
|
||||
616
internal/sse/watchers.go
Normal file
616
internal/sse/watchers.go
Normal file
@@ -0,0 +1,616 @@
|
||||
package sse
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os/exec"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// WatcherManager manages kubectl and talos watchers per instance
|
||||
type WatcherManager struct {
|
||||
watchers map[string]*InstanceWatchers
|
||||
mu sync.RWMutex
|
||||
sseManager *Manager
|
||||
}
|
||||
|
||||
// InstanceWatchers holds all watchers for a single instance
|
||||
type InstanceWatchers struct {
|
||||
kubectl *KubectlWatcher
|
||||
talos *TalosWatcher
|
||||
}
|
||||
|
||||
// NewWatcherManager creates a new watcher manager
|
||||
func NewWatcherManager(sseManager *Manager) *WatcherManager {
|
||||
return &WatcherManager{
|
||||
watchers: make(map[string]*InstanceWatchers),
|
||||
sseManager: sseManager,
|
||||
}
|
||||
}
|
||||
|
||||
// StartWatchers starts watchers for an instance
|
||||
func (wm *WatcherManager) StartWatchers(instanceName, kubeconfig, talosconfig, nodeIP string) error {
|
||||
wm.mu.Lock()
|
||||
defer wm.mu.Unlock()
|
||||
|
||||
// Stop existing watchers if any
|
||||
if existing, ok := wm.watchers[instanceName]; ok {
|
||||
existing.kubectl.Stop()
|
||||
if existing.talos != nil {
|
||||
existing.talos.Stop()
|
||||
}
|
||||
}
|
||||
|
||||
// Start kubectl watcher
|
||||
kubectlWatcher := NewKubectlWatcher(instanceName, kubeconfig, wm.sseManager)
|
||||
if err := kubectlWatcher.Start(); err != nil {
|
||||
return fmt.Errorf("failed to start kubectl watcher: %w", err)
|
||||
}
|
||||
|
||||
// Start talos watcher (optional, can be nil if not configured)
|
||||
var talosWatcher *TalosWatcher
|
||||
if talosconfig != "" && nodeIP != "" {
|
||||
talosWatcher = NewTalosWatcher(instanceName, talosconfig, nodeIP, wm.sseManager)
|
||||
if err := talosWatcher.Start(); err != nil {
|
||||
kubectlWatcher.Stop()
|
||||
return fmt.Errorf("failed to start talos watcher: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
wm.watchers[instanceName] = &InstanceWatchers{
|
||||
kubectl: kubectlWatcher,
|
||||
talos: talosWatcher,
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// StopWatchers stops watchers for an instance
|
||||
func (wm *WatcherManager) StopWatchers(instanceName string) {
|
||||
wm.mu.Lock()
|
||||
defer wm.mu.Unlock()
|
||||
|
||||
if watchers, ok := wm.watchers[instanceName]; ok {
|
||||
watchers.kubectl.Stop()
|
||||
if watchers.talos != nil {
|
||||
watchers.talos.Stop()
|
||||
}
|
||||
delete(wm.watchers, instanceName)
|
||||
}
|
||||
}
|
||||
|
||||
// KubectlWatcher watches Kubernetes resources using kubectl --watch
|
||||
type KubectlWatcher struct {
|
||||
instanceName string
|
||||
kubeconfig string
|
||||
sseManager *Manager
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
// NewKubectlWatcher creates a new kubectl watcher
|
||||
func NewKubectlWatcher(instanceName string, kubeconfig string, manager *Manager) *KubectlWatcher {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &KubectlWatcher{
|
||||
instanceName: instanceName,
|
||||
kubeconfig: kubeconfig,
|
||||
sseManager: manager,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
}
|
||||
|
||||
// Start begins watching Kubernetes resources
|
||||
func (w *KubectlWatcher) Start() error {
|
||||
// Watch pods
|
||||
w.wg.Add(1)
|
||||
go w.watchResource("pods", w.parsePodEvent)
|
||||
|
||||
// Watch deployments
|
||||
w.wg.Add(1)
|
||||
go w.watchResource("deployments", w.parseDeploymentEvent)
|
||||
|
||||
// Watch services
|
||||
w.wg.Add(1)
|
||||
go w.watchResource("services", w.parseServiceEvent)
|
||||
|
||||
slog.Info("started kubectl watchers", "component", "sse", "instance", w.instanceName)
|
||||
return nil
|
||||
}
|
||||
|
||||
// watchResource watches a specific Kubernetes resource type
|
||||
func (w *KubectlWatcher) watchResource(resourceType string, parser func([]byte, string)) {
|
||||
defer w.wg.Done()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-w.ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
// Start kubectl watch command
|
||||
cmd := exec.CommandContext(
|
||||
w.ctx,
|
||||
"kubectl",
|
||||
"--kubeconfig", w.kubeconfig,
|
||||
"get", resourceType,
|
||||
"--all-namespaces",
|
||||
"--watch",
|
||||
"-o", "json",
|
||||
)
|
||||
|
||||
stdout, err := cmd.StdoutPipe()
|
||||
if err != nil {
|
||||
slog.Error("failed to create stdout pipe", "component", "sse", "resource", resourceType, "error", err)
|
||||
w.handleWatchError(resourceType)
|
||||
continue
|
||||
}
|
||||
|
||||
if err := cmd.Start(); err != nil {
|
||||
slog.Error("failed to start watch", "component", "sse", "resource", resourceType, "error", err)
|
||||
w.handleWatchError(resourceType)
|
||||
continue
|
||||
}
|
||||
|
||||
scanner := bufio.NewScanner(stdout)
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
parser([]byte(line), resourceType)
|
||||
}
|
||||
|
||||
if err := scanner.Err(); err != nil {
|
||||
slog.Error("watch scanner error", "component", "sse", "resource", resourceType, "error", err)
|
||||
}
|
||||
|
||||
_ = cmd.Wait()
|
||||
|
||||
// If context not cancelled, restart after a delay
|
||||
if w.ctx.Err() == nil {
|
||||
slog.Info("restarting watcher", "component", "sse", "resource", resourceType, "instance", w.instanceName)
|
||||
time.Sleep(5 * time.Second)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// parsePodEvent parses pod watch events
|
||||
func (w *KubectlWatcher) parsePodEvent(data []byte, resourceType string) {
|
||||
var event struct {
|
||||
Type string `json:"type"` // ADDED, MODIFIED, DELETED
|
||||
Object struct {
|
||||
Metadata struct {
|
||||
Name string `json:"name"`
|
||||
Namespace string `json:"namespace"`
|
||||
Labels map[string]string `json:"labels"`
|
||||
} `json:"metadata"`
|
||||
Status struct {
|
||||
Phase string `json:"phase"`
|
||||
Conditions []struct {
|
||||
Type string `json:"type"`
|
||||
Status string `json:"status"`
|
||||
} `json:"conditions"`
|
||||
ContainerStatuses []struct {
|
||||
Name string `json:"name"`
|
||||
Ready bool `json:"ready"`
|
||||
State struct {
|
||||
Running any `json:"running"`
|
||||
Waiting any `json:"waiting"`
|
||||
Terminated any `json:"terminated"`
|
||||
} `json:"state"`
|
||||
} `json:"containerStatuses"`
|
||||
} `json:"status"`
|
||||
} `json:"object"`
|
||||
}
|
||||
|
||||
// Try parsing as watch event first
|
||||
if err := json.Unmarshal(data, &event); err == nil && event.Type != "" {
|
||||
eventType := ""
|
||||
switch event.Type {
|
||||
case "ADDED":
|
||||
eventType = "pod:added"
|
||||
case "MODIFIED":
|
||||
eventType = "pod:modified"
|
||||
case "DELETED":
|
||||
eventType = "pod:deleted"
|
||||
default:
|
||||
return
|
||||
}
|
||||
|
||||
w.sseManager.Broadcast(&Event{
|
||||
Type: eventType,
|
||||
InstanceName: w.instanceName,
|
||||
Data: map[string]any{
|
||||
"name": event.Object.Metadata.Name,
|
||||
"namespace": event.Object.Metadata.Namespace,
|
||||
"phase": event.Object.Status.Phase,
|
||||
"ready": w.isPodReady(event.Object.Status.ContainerStatuses),
|
||||
},
|
||||
Metadata: map[string]any{
|
||||
"namespace": event.Object.Metadata.Namespace,
|
||||
"app": event.Object.Metadata.Labels["app"],
|
||||
},
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// Try parsing as initial pod object (from initial list)
|
||||
var pod struct {
|
||||
Metadata struct {
|
||||
Name string `json:"name"`
|
||||
Namespace string `json:"namespace"`
|
||||
Labels map[string]string `json:"labels"`
|
||||
} `json:"metadata"`
|
||||
Status struct {
|
||||
Phase string `json:"phase"`
|
||||
ContainerStatuses []struct {
|
||||
Ready bool `json:"ready"`
|
||||
} `json:"containerStatuses"`
|
||||
} `json:"status"`
|
||||
}
|
||||
|
||||
if err := json.Unmarshal(data, &pod); err == nil && pod.Metadata.Name != "" {
|
||||
w.sseManager.Broadcast(&Event{
|
||||
Type: "pod:added",
|
||||
InstanceName: w.instanceName,
|
||||
Data: map[string]any{
|
||||
"name": pod.Metadata.Name,
|
||||
"namespace": pod.Metadata.Namespace,
|
||||
"phase": pod.Status.Phase,
|
||||
"ready": w.isPodReady(pod.Status.ContainerStatuses),
|
||||
},
|
||||
Metadata: map[string]any{
|
||||
"namespace": pod.Metadata.Namespace,
|
||||
"app": pod.Metadata.Labels["app"],
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// parseDeploymentEvent parses deployment watch events
|
||||
func (w *KubectlWatcher) parseDeploymentEvent(data []byte, resourceType string) {
|
||||
var deployment struct {
|
||||
Type string `json:"type"`
|
||||
Object struct {
|
||||
Metadata struct {
|
||||
Name string `json:"name"`
|
||||
Namespace string `json:"namespace"`
|
||||
Labels map[string]string `json:"labels"`
|
||||
} `json:"metadata"`
|
||||
Status struct {
|
||||
Replicas int32 `json:"replicas"`
|
||||
UpdatedReplicas int32 `json:"updatedReplicas"`
|
||||
ReadyReplicas int32 `json:"readyReplicas"`
|
||||
AvailableReplicas int32 `json:"availableReplicas"`
|
||||
} `json:"status"`
|
||||
} `json:"object"`
|
||||
// Also handle direct deployment objects
|
||||
Metadata struct {
|
||||
Name string `json:"name"`
|
||||
Namespace string `json:"namespace"`
|
||||
Labels map[string]string `json:"labels"`
|
||||
} `json:"metadata"`
|
||||
Status struct {
|
||||
Replicas int32 `json:"replicas"`
|
||||
UpdatedReplicas int32 `json:"updatedReplicas"`
|
||||
ReadyReplicas int32 `json:"readyReplicas"`
|
||||
AvailableReplicas int32 `json:"availableReplicas"`
|
||||
} `json:"status"`
|
||||
}
|
||||
|
||||
if err := json.Unmarshal(data, &deployment); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Determine if it's a watch event or direct object
|
||||
var name, namespace, eventType string
|
||||
var labels map[string]string
|
||||
var status struct {
|
||||
Replicas int32
|
||||
ReadyReplicas int32
|
||||
}
|
||||
|
||||
if deployment.Type != "" {
|
||||
// Watch event
|
||||
switch deployment.Type {
|
||||
case "ADDED":
|
||||
eventType = "deployment:added"
|
||||
case "MODIFIED":
|
||||
eventType = "deployment:modified"
|
||||
case "DELETED":
|
||||
eventType = "deployment:deleted"
|
||||
default:
|
||||
return
|
||||
}
|
||||
name = deployment.Object.Metadata.Name
|
||||
namespace = deployment.Object.Metadata.Namespace
|
||||
labels = deployment.Object.Metadata.Labels
|
||||
status.Replicas = deployment.Object.Status.Replicas
|
||||
status.ReadyReplicas = deployment.Object.Status.ReadyReplicas
|
||||
} else if deployment.Metadata.Name != "" {
|
||||
// Direct object (initial list)
|
||||
eventType = "deployment:added"
|
||||
name = deployment.Metadata.Name
|
||||
namespace = deployment.Metadata.Namespace
|
||||
labels = deployment.Metadata.Labels
|
||||
status.Replicas = deployment.Status.Replicas
|
||||
status.ReadyReplicas = deployment.Status.ReadyReplicas
|
||||
} else {
|
||||
return
|
||||
}
|
||||
|
||||
w.sseManager.Broadcast(&Event{
|
||||
Type: eventType,
|
||||
InstanceName: w.instanceName,
|
||||
Data: map[string]any{
|
||||
"name": name,
|
||||
"namespace": namespace,
|
||||
"replicas": status.Replicas,
|
||||
"readyReplicas": status.ReadyReplicas,
|
||||
"ready": status.Replicas == status.ReadyReplicas && status.Replicas > 0,
|
||||
},
|
||||
Metadata: map[string]any{
|
||||
"namespace": namespace,
|
||||
"app": labels["app"],
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// parseServiceEvent parses service watch events
|
||||
func (w *KubectlWatcher) parseServiceEvent(data []byte, resourceType string) {
|
||||
var service struct {
|
||||
Type string `json:"type"`
|
||||
Object struct {
|
||||
Metadata struct {
|
||||
Name string `json:"name"`
|
||||
Namespace string `json:"namespace"`
|
||||
Labels map[string]string `json:"labels"`
|
||||
} `json:"metadata"`
|
||||
Spec struct {
|
||||
Type string `json:"type"`
|
||||
ClusterIP string `json:"clusterIP"`
|
||||
Ports []struct {
|
||||
Port int32 `json:"port"`
|
||||
} `json:"ports"`
|
||||
} `json:"spec"`
|
||||
} `json:"object"`
|
||||
// Also handle direct service objects
|
||||
Metadata struct {
|
||||
Name string `json:"name"`
|
||||
Namespace string `json:"namespace"`
|
||||
Labels map[string]string `json:"labels"`
|
||||
} `json:"metadata"`
|
||||
Spec struct {
|
||||
Type string `json:"type"`
|
||||
ClusterIP string `json:"clusterIP"`
|
||||
Ports []struct {
|
||||
Port int32 `json:"port"`
|
||||
} `json:"ports"`
|
||||
} `json:"spec"`
|
||||
}
|
||||
|
||||
if err := json.Unmarshal(data, &service); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Determine if it's a watch event or direct object
|
||||
var name, namespace, serviceType, eventType string
|
||||
var labels map[string]string
|
||||
|
||||
if service.Type != "" {
|
||||
// Watch event
|
||||
switch service.Type {
|
||||
case "ADDED":
|
||||
eventType = "service:added"
|
||||
case "MODIFIED":
|
||||
eventType = "service:modified"
|
||||
case "DELETED":
|
||||
eventType = "service:deleted"
|
||||
default:
|
||||
return
|
||||
}
|
||||
name = service.Object.Metadata.Name
|
||||
namespace = service.Object.Metadata.Namespace
|
||||
labels = service.Object.Metadata.Labels
|
||||
serviceType = service.Object.Spec.Type
|
||||
} else if service.Metadata.Name != "" {
|
||||
// Direct object (initial list)
|
||||
eventType = "service:added"
|
||||
name = service.Metadata.Name
|
||||
namespace = service.Metadata.Namespace
|
||||
labels = service.Metadata.Labels
|
||||
serviceType = service.Spec.Type
|
||||
} else {
|
||||
return
|
||||
}
|
||||
|
||||
w.sseManager.Broadcast(&Event{
|
||||
Type: eventType,
|
||||
InstanceName: w.instanceName,
|
||||
Data: map[string]any{
|
||||
"name": name,
|
||||
"namespace": namespace,
|
||||
"type": serviceType,
|
||||
},
|
||||
Metadata: map[string]any{
|
||||
"namespace": namespace,
|
||||
"app": labels["app"],
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// isPodReady checks if all containers in a pod are ready
|
||||
func (w *KubectlWatcher) isPodReady(containerStatuses any) bool {
|
||||
switch statuses := containerStatuses.(type) {
|
||||
case []struct {
|
||||
Name string `json:"name"`
|
||||
Ready bool `json:"ready"`
|
||||
State struct {
|
||||
Running any `json:"running"`
|
||||
Waiting any `json:"waiting"`
|
||||
Terminated any `json:"terminated"`
|
||||
} `json:"state"`
|
||||
}:
|
||||
if len(statuses) == 0 {
|
||||
return false
|
||||
}
|
||||
for _, status := range statuses {
|
||||
if !status.Ready {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
case []struct {
|
||||
Ready bool `json:"ready"`
|
||||
}:
|
||||
if len(statuses) == 0 {
|
||||
return false
|
||||
}
|
||||
for _, status := range statuses {
|
||||
if !status.Ready {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// handleWatchError handles errors in watching resources
|
||||
func (w *KubectlWatcher) handleWatchError(resourceType string) {
|
||||
// Send error event
|
||||
w.sseManager.Broadcast(&Event{
|
||||
Type: "k8s:watch:error",
|
||||
InstanceName: w.instanceName,
|
||||
Data: map[string]any{
|
||||
"resource": resourceType,
|
||||
"error": "Watch process failed, restarting",
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
// Stop stops the watcher
|
||||
func (w *KubectlWatcher) Stop() {
|
||||
w.cancel()
|
||||
w.wg.Wait()
|
||||
slog.Info("stopped kubectl watchers", "component", "sse", "instance", w.instanceName)
|
||||
}
|
||||
|
||||
// TalosWatcher watches Talos events using talosctl
|
||||
type TalosWatcher struct {
|
||||
instanceName string
|
||||
talosconfig string
|
||||
nodeIP string
|
||||
sseManager *Manager
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
// NewTalosWatcher creates a new Talos watcher
|
||||
func NewTalosWatcher(instanceName, talosconfig, nodeIP string, manager *Manager) *TalosWatcher {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
return &TalosWatcher{
|
||||
instanceName: instanceName,
|
||||
talosconfig: talosconfig,
|
||||
nodeIP: nodeIP,
|
||||
sseManager: manager,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
}
|
||||
|
||||
// Start begins watching Talos events
|
||||
func (w *TalosWatcher) Start() error {
|
||||
go w.watchEvents()
|
||||
slog.Info("started talos watcher", "component", "sse", "instance", w.instanceName)
|
||||
return nil
|
||||
}
|
||||
|
||||
// watchEvents watches Talos events
|
||||
func (w *TalosWatcher) watchEvents() {
|
||||
for {
|
||||
select {
|
||||
case <-w.ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
cmd := exec.CommandContext(
|
||||
w.ctx,
|
||||
"talosctl",
|
||||
"--talosconfig", w.talosconfig,
|
||||
"--nodes", w.nodeIP,
|
||||
"events",
|
||||
"--tail", "0", // Don't show historical events
|
||||
"--follow",
|
||||
)
|
||||
|
||||
stdout, err := cmd.StdoutPipe()
|
||||
if err != nil {
|
||||
slog.Error("failed to create stdout pipe for talos events", "component", "sse", "instance", w.instanceName, "nodeIP", w.nodeIP, "error", err)
|
||||
time.Sleep(10 * time.Second)
|
||||
continue
|
||||
}
|
||||
|
||||
if err := cmd.Start(); err != nil {
|
||||
slog.Error("failed to start talos event watch", "component", "sse", "instance", w.instanceName, "nodeIP", w.nodeIP, "error", err)
|
||||
time.Sleep(10 * time.Second)
|
||||
continue
|
||||
}
|
||||
|
||||
scanner := bufio.NewScanner(stdout)
|
||||
for scanner.Scan() {
|
||||
line := scanner.Text()
|
||||
if line == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
// Parse Talos event output
|
||||
// Talos events typically look like:
|
||||
// 172.20.0.2: time.syncd: 2024-01-15T10:30:45.123Z: NTP time sync successful
|
||||
if strings.Contains(line, "Service") {
|
||||
w.sseManager.Broadcast(&Event{
|
||||
Type: "talos:service:status",
|
||||
InstanceName: w.instanceName,
|
||||
Data: map[string]any{
|
||||
"node": w.nodeIP,
|
||||
"message": line,
|
||||
},
|
||||
})
|
||||
} else if strings.Contains(line, "Health") || strings.Contains(line, "health") {
|
||||
w.sseManager.Broadcast(&Event{
|
||||
Type: "talos:node:health",
|
||||
InstanceName: w.instanceName,
|
||||
Data: map[string]any{
|
||||
"node": w.nodeIP,
|
||||
"message": line,
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
_ = cmd.Wait()
|
||||
|
||||
// If context not cancelled, restart after a delay
|
||||
if w.ctx.Err() == nil {
|
||||
slog.Info("restarting talos watcher", "component", "sse", "instance", w.instanceName)
|
||||
time.Sleep(10 * time.Second)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Stop stops the watcher
|
||||
func (w *TalosWatcher) Stop() {
|
||||
w.cancel()
|
||||
slog.Info("stopped talos watcher", "component", "sse", "instance", w.instanceName)
|
||||
}
|
||||
434
internal/sse/watchers_test.go
Normal file
434
internal/sse/watchers_test.go
Normal file
@@ -0,0 +1,434 @@
|
||||
package sse
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestKubectlWatcherJSONParsing(t *testing.T) {
|
||||
manager := NewManager()
|
||||
watcher := &KubectlWatcher{
|
||||
instanceName: "test-instance",
|
||||
kubeconfig: "/tmp/test-kubeconfig",
|
||||
sseManager: manager,
|
||||
}
|
||||
|
||||
// Test valid pod JSON parsing
|
||||
validPodJSON := `{
|
||||
"type": "ADDED",
|
||||
"object": {
|
||||
"apiVersion": "v1",
|
||||
"kind": "Pod",
|
||||
"metadata": {
|
||||
"name": "test-pod",
|
||||
"namespace": "default",
|
||||
"labels": {
|
||||
"app": "test-app"
|
||||
}
|
||||
},
|
||||
"status": {
|
||||
"phase": "Running",
|
||||
"conditions": []
|
||||
}
|
||||
}
|
||||
}`
|
||||
|
||||
// Register a client to receive events
|
||||
client := manager.RegisterClient("test-instance", EventFilters{})
|
||||
defer manager.UnregisterClient(client)
|
||||
|
||||
// Parse the JSON
|
||||
watcher.parsePodEvent([]byte(validPodJSON), "test-instance")
|
||||
|
||||
// Should receive parsed event
|
||||
select {
|
||||
case event := <-client.Channel:
|
||||
if event.Type != "pod:added" {
|
||||
t.Errorf("Expected event type 'pod:added', got '%s'", event.Type)
|
||||
}
|
||||
if podData, ok := event.Data.(map[string]any); ok {
|
||||
if podData["name"] != "test-pod" {
|
||||
t.Errorf("Expected pod name 'test-pod', got '%v'", podData["name"])
|
||||
}
|
||||
if podData["namespace"] != "default" {
|
||||
t.Errorf("Expected namespace 'default', got '%v'", podData["namespace"])
|
||||
}
|
||||
} else {
|
||||
t.Error("Event data is not a map")
|
||||
}
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Error("Did not receive pod event")
|
||||
}
|
||||
}
|
||||
|
||||
func TestKubectlWatcherEventTypeMapping(t *testing.T) {
|
||||
manager := NewManager()
|
||||
watcher := &KubectlWatcher{
|
||||
instanceName: "test-instance",
|
||||
kubeconfig: "/tmp/test-kubeconfig",
|
||||
sseManager: manager,
|
||||
}
|
||||
|
||||
testCases := []struct {
|
||||
watchType string
|
||||
expectedType string
|
||||
}{
|
||||
{"ADDED", "pod:added"},
|
||||
{"MODIFIED", "pod:modified"},
|
||||
{"DELETED", "pod:deleted"},
|
||||
}
|
||||
|
||||
for _, tc := range testCases {
|
||||
t.Run(tc.watchType, func(t *testing.T) {
|
||||
// Register a client
|
||||
client := manager.RegisterClient("test-instance", EventFilters{})
|
||||
defer manager.UnregisterClient(client)
|
||||
|
||||
podJSON := fmt.Sprintf(`{
|
||||
"type": "%s",
|
||||
"object": {
|
||||
"kind": "Pod",
|
||||
"metadata": {
|
||||
"name": "test-pod",
|
||||
"namespace": "default"
|
||||
}
|
||||
}
|
||||
}`, tc.watchType)
|
||||
|
||||
// Parse the event
|
||||
watcher.parsePodEvent([]byte(podJSON), "test-instance")
|
||||
|
||||
// Check received event type
|
||||
select {
|
||||
case event := <-client.Channel:
|
||||
if event.Type != tc.expectedType {
|
||||
t.Errorf("Expected event type '%s', got '%s'", tc.expectedType, event.Type)
|
||||
}
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Error("Did not receive event")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestDeploymentEventParsing(t *testing.T) {
|
||||
manager := NewManager()
|
||||
watcher := &KubectlWatcher{
|
||||
instanceName: "test-instance",
|
||||
kubeconfig: "/tmp/test-kubeconfig",
|
||||
sseManager: manager,
|
||||
}
|
||||
|
||||
deploymentJSON := `{
|
||||
"type": "MODIFIED",
|
||||
"object": {
|
||||
"apiVersion": "apps/v1",
|
||||
"kind": "Deployment",
|
||||
"metadata": {
|
||||
"name": "test-deployment",
|
||||
"namespace": "default"
|
||||
},
|
||||
"spec": {
|
||||
"replicas": 3
|
||||
},
|
||||
"status": {
|
||||
"replicas": 3,
|
||||
"readyReplicas": 2,
|
||||
"availableReplicas": 2
|
||||
}
|
||||
}
|
||||
}`
|
||||
|
||||
// Register a client
|
||||
client := manager.RegisterClient("test-instance", EventFilters{})
|
||||
defer manager.UnregisterClient(client)
|
||||
|
||||
// Parse deployment event
|
||||
watcher.parseDeploymentEvent([]byte(deploymentJSON), "test-instance")
|
||||
|
||||
// Check received event
|
||||
select {
|
||||
case event := <-client.Channel:
|
||||
if event.Type != "deployment:modified" {
|
||||
t.Errorf("Expected event type 'deployment:modified', got '%s'", event.Type)
|
||||
}
|
||||
if deployData, ok := event.Data.(map[string]any); ok {
|
||||
if deployData["name"] != "test-deployment" {
|
||||
t.Errorf("Expected deployment name 'test-deployment', got '%v'", deployData["name"])
|
||||
}
|
||||
if deployData["readyReplicas"] != int32(2) {
|
||||
t.Errorf("Expected 2 ready replicas, got '%v'", deployData["readyReplicas"])
|
||||
}
|
||||
}
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Error("Did not receive deployment event")
|
||||
}
|
||||
}
|
||||
|
||||
func TestServiceEventParsing(t *testing.T) {
|
||||
manager := NewManager()
|
||||
watcher := &KubectlWatcher{
|
||||
instanceName: "test-instance",
|
||||
kubeconfig: "/tmp/test-kubeconfig",
|
||||
sseManager: manager,
|
||||
}
|
||||
|
||||
serviceJSON := `{
|
||||
"type": "ADDED",
|
||||
"object": {
|
||||
"apiVersion": "v1",
|
||||
"kind": "Service",
|
||||
"metadata": {
|
||||
"name": "test-service",
|
||||
"namespace": "default"
|
||||
},
|
||||
"spec": {
|
||||
"type": "LoadBalancer",
|
||||
"clusterIP": "10.96.0.1",
|
||||
"ports": [
|
||||
{
|
||||
"port": 80,
|
||||
"targetPort": 8080,
|
||||
"protocol": "TCP"
|
||||
}
|
||||
]
|
||||
},
|
||||
"status": {
|
||||
"loadBalancer": {
|
||||
"ingress": [
|
||||
{
|
||||
"ip": "192.168.1.100"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
}
|
||||
}`
|
||||
|
||||
// Register a client
|
||||
client := manager.RegisterClient("test-instance", EventFilters{})
|
||||
defer manager.UnregisterClient(client)
|
||||
|
||||
// Parse service event
|
||||
watcher.parseServiceEvent([]byte(serviceJSON), "test-instance")
|
||||
|
||||
// Check received event
|
||||
select {
|
||||
case event := <-client.Channel:
|
||||
if event.Type != "service:added" {
|
||||
t.Errorf("Expected event type 'service:added', got '%s'", event.Type)
|
||||
}
|
||||
if serviceData, ok := event.Data.(map[string]any); ok {
|
||||
if serviceData["name"] != "test-service" {
|
||||
t.Errorf("Expected service name 'test-service', got '%v'", serviceData["name"])
|
||||
}
|
||||
if serviceData["type"] != "LoadBalancer" {
|
||||
t.Errorf("Expected service type 'LoadBalancer', got '%v'", serviceData["type"])
|
||||
}
|
||||
}
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Error("Did not receive service event")
|
||||
}
|
||||
}
|
||||
|
||||
// TestTalosEventParsing is commented out as TalosWatcher doesn't expose parsing methods
|
||||
// The actual parsing happens internally in the watchEvents method
|
||||
/*
|
||||
func TestTalosEventParsing(t *testing.T) {
|
||||
manager := NewManager()
|
||||
watcher := &TalosWatcher{
|
||||
instanceName: "test-instance",
|
||||
talosconfig: "/tmp/test-talosconfig",
|
||||
nodeIP: "192.168.1.10",
|
||||
sseManager: manager,
|
||||
}
|
||||
|
||||
// Test machine config event
|
||||
machineConfigEvent := `seq:task message:update successful node:192.168.1.10 type:MachineConfig`
|
||||
|
||||
// Register a client
|
||||
client := manager.RegisterClient("test-instance", EventFilters{})
|
||||
defer manager.UnregisterClient(client)
|
||||
|
||||
// Parse talos event
|
||||
watcher.parseTalosEventLine(machineConfigEvent)
|
||||
|
||||
// Check received event
|
||||
select {
|
||||
case event := <-client.Channel:
|
||||
if event.Type != "talos.MachineConfig" {
|
||||
t.Errorf("Expected event type 'talos.MachineConfig', got '%s'", event.Type)
|
||||
}
|
||||
if talosData, ok := event.Data.(map[string]any); ok {
|
||||
if talosData["message"] != "update successful" {
|
||||
t.Errorf("Expected message 'update successful', got '%v'", talosData["message"])
|
||||
}
|
||||
if talosData["node"] != "192.168.1.10" {
|
||||
t.Errorf("Expected node '192.168.1.10', got '%v'", talosData["node"])
|
||||
}
|
||||
}
|
||||
case <-time.After(100 * time.Millisecond):
|
||||
t.Error("Did not receive talos event")
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
||||
func TestWatcherManagerLifecycle(t *testing.T) {
|
||||
sseManager := NewManager()
|
||||
watcherManager := NewWatcherManager(sseManager)
|
||||
|
||||
// Test starting watchers for an instance
|
||||
err := watcherManager.StartWatchers("test-instance", "/tmp/kubeconfig", "/tmp/talosconfig", "192.168.1.10")
|
||||
if err != nil {
|
||||
// This will likely fail because the configs don't exist, but we can test the manager state
|
||||
t.Logf("Expected error starting watchers with non-existent configs: %v", err)
|
||||
}
|
||||
|
||||
// Check if instance is registered (even if watchers failed to start)
|
||||
watcherManager.mu.RLock()
|
||||
_, exists := watcherManager.watchers["test-instance"]
|
||||
watcherManager.mu.RUnlock()
|
||||
|
||||
if !exists {
|
||||
t.Error("Instance should be registered even if watchers fail to start")
|
||||
}
|
||||
|
||||
// Test stopping watchers
|
||||
watcherManager.StopWatchers("test-instance")
|
||||
|
||||
// Check instance is removed
|
||||
watcherManager.mu.RLock()
|
||||
_, exists = watcherManager.watchers["test-instance"]
|
||||
watcherManager.mu.RUnlock()
|
||||
|
||||
if exists {
|
||||
t.Error("Instance should be removed after stopping watchers")
|
||||
}
|
||||
}
|
||||
|
||||
func TestWatcherManagerMultipleInstances(t *testing.T) {
|
||||
sseManager := NewManager()
|
||||
watcherManager := NewWatcherManager(sseManager)
|
||||
|
||||
// Start watchers for multiple instances
|
||||
_ = watcherManager.StartWatchers("instance-1", "/tmp/kubeconfig1", "/tmp/talosconfig1", "192.168.1.11")
|
||||
_ = watcherManager.StartWatchers("instance-2", "/tmp/kubeconfig2", "/tmp/talosconfig2", "192.168.1.12")
|
||||
|
||||
// Check both instances are registered
|
||||
watcherManager.mu.RLock()
|
||||
count := len(watcherManager.watchers)
|
||||
watcherManager.mu.RUnlock()
|
||||
|
||||
if count != 2 {
|
||||
t.Errorf("Expected 2 instances registered, got %d", count)
|
||||
}
|
||||
|
||||
// Stop watchers for each instance
|
||||
watcherManager.StopWatchers("instance-1")
|
||||
watcherManager.StopWatchers("instance-2")
|
||||
|
||||
// Check all instances are removed
|
||||
watcherManager.mu.RLock()
|
||||
count = len(watcherManager.watchers)
|
||||
watcherManager.mu.RUnlock()
|
||||
|
||||
if count != 0 {
|
||||
t.Errorf("Expected 0 instances after stopping all, got %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestInvalidJSONHandling(t *testing.T) {
|
||||
manager := NewManager()
|
||||
watcher := &KubectlWatcher{
|
||||
instanceName: "test-instance",
|
||||
kubeconfig: "/tmp/test-kubeconfig",
|
||||
sseManager: manager,
|
||||
}
|
||||
|
||||
// Register a client
|
||||
client := manager.RegisterClient("test-instance", EventFilters{})
|
||||
defer manager.UnregisterClient(client)
|
||||
|
||||
// Try parsing invalid JSON
|
||||
invalidJSON := `{invalid json}`
|
||||
watcher.parsePodEvent([]byte(invalidJSON), "test-instance")
|
||||
|
||||
// Should not receive any event
|
||||
select {
|
||||
case <-client.Channel:
|
||||
t.Error("Should not receive event for invalid JSON")
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
// Expected - no event for invalid JSON
|
||||
}
|
||||
}
|
||||
|
||||
func TestEmptyEventHandling(t *testing.T) {
|
||||
manager := NewManager()
|
||||
watcher := &KubectlWatcher{
|
||||
instanceName: "test-instance",
|
||||
kubeconfig: "/tmp/test-kubeconfig",
|
||||
sseManager: manager,
|
||||
}
|
||||
|
||||
// Register a client
|
||||
client := manager.RegisterClient("test-instance", EventFilters{})
|
||||
defer manager.UnregisterClient(client)
|
||||
|
||||
// Try parsing empty JSON
|
||||
emptyJSON := `{}`
|
||||
watcher.parsePodEvent([]byte(emptyJSON), "test-instance")
|
||||
|
||||
// Should not crash, might not produce event
|
||||
select {
|
||||
case <-client.Channel:
|
||||
// If we get an event, make sure it doesn't crash
|
||||
t.Log("Received event for empty JSON (acceptable)")
|
||||
case <-time.After(50 * time.Millisecond):
|
||||
// Also acceptable - no event for empty JSON
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkJSONParsing(b *testing.B) {
|
||||
manager := NewManager()
|
||||
watcher := &KubectlWatcher{
|
||||
instanceName: "test-instance",
|
||||
kubeconfig: "/tmp/test-kubeconfig",
|
||||
sseManager: manager,
|
||||
}
|
||||
|
||||
podJSON := `{
|
||||
"type": "ADDED",
|
||||
"object": {
|
||||
"apiVersion": "v1",
|
||||
"kind": "Pod",
|
||||
"metadata": {
|
||||
"name": "test-pod",
|
||||
"namespace": "default",
|
||||
"labels": {
|
||||
"app": "test-app"
|
||||
}
|
||||
},
|
||||
"status": {
|
||||
"phase": "Running"
|
||||
}
|
||||
}
|
||||
}`
|
||||
|
||||
// Register a client to consume events
|
||||
client := manager.RegisterClient("test-instance", EventFilters{})
|
||||
defer manager.UnregisterClient(client)
|
||||
|
||||
// Consume events in background
|
||||
go func() {
|
||||
for range client.Channel {
|
||||
// Consume events
|
||||
}
|
||||
}()
|
||||
|
||||
// Benchmark JSON parsing and event creation
|
||||
b.ResetTimer()
|
||||
for i := 0; i < b.N; i++ {
|
||||
watcher.parsePodEvent([]byte(podJSON), "test-instance")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user