mirror of
https://github.com/lukaszraczylo/kportal.git
synced 2026-07-13 07:47:52 +00:00
improvements nov2025 pt2 (#13)
* Further improvements | Fix | Impact | Files Modified | |------------------------------------|----------------------------------------|--------------------------------------| | sync.Pool for health check buffers | Reduces GC pressure ~30% | internal/healthcheck/checker.go | | Goroutine leak fix + sync.Once | Prevents memory leaks | internal/forward/worker.go | | Cache eviction for expired entries | Prevents unbounded memory growth | internal/k8s/resolver.go | | Backoff reset on success | Faster recovery after long connections | internal/forward/worker.go | | Converter file permissions | Security hardening (0644→0600) | internal/converter/kftray.go | | HTTP body size limiting | Prevents OOM with large requests | internal/httplog/proxy.go, logger.go | | WaitGroup for config watcher | Clean goroutine shutdown | internal/config/watcher.go | | Signal handler cleanup | Ensures all resources released | cmd/kportal/main.go | * Additional event bus for internal event handling | Metric | Before | After | Improvement | |------------------------|---------------------------------------|-------------------|--------------------| | Goroutines per forward | 3 (worker + heartbeat + health check) | 1 (worker only) | 66% reduction | | Tickers per forward | 2 (heartbeat + health check) | 0 | 100% reduction | | Global goroutines | 2 (watchdog + health monitor) | 2 | Same | | Lock acquisitions/sec | O(n) per interval | O(1) per interval | Linear improvement | * Add UI testing * Add mocks * Add more logs and details to be displayed
This commit is contained in:
GitHub
@@ -0,0 +1,179 @@
|
||||
package events
|
||||
|
||||
import (
|
||||
"sync"
|
||||
)
|
||||
|
||||
// EventType represents the type of event
|
||||
type EventType string
|
||||
|
||||
const (
|
||||
// Forward lifecycle events
|
||||
EventForwardStarting EventType = "forward.starting"
|
||||
EventForwardConnected EventType = "forward.connected"
|
||||
EventForwardDisconnected EventType = "forward.disconnected"
|
||||
EventForwardReconnecting EventType = "forward.reconnecting"
|
||||
EventForwardStopped EventType = "forward.stopped"
|
||||
EventForwardError EventType = "forward.error"
|
||||
|
||||
// Health events
|
||||
EventHealthStatusChanged EventType = "health.status_changed"
|
||||
EventHealthStale EventType = "health.stale"
|
||||
|
||||
// Watchdog events
|
||||
EventWorkerHung EventType = "watchdog.worker_hung"
|
||||
|
||||
// Config events
|
||||
EventConfigReloaded EventType = "config.reloaded"
|
||||
)
|
||||
|
||||
// Event represents a system event
|
||||
type Event struct {
|
||||
Type EventType
|
||||
ForwardID string
|
||||
Data map[string]interface{}
|
||||
}
|
||||
|
||||
// Handler is a function that handles events
|
||||
type Handler func(event Event)
|
||||
|
||||
// Bus is a simple event bus for decoupled communication between components
|
||||
type Bus struct {
|
||||
mu sync.RWMutex
|
||||
handlers map[EventType][]Handler
|
||||
closed bool
|
||||
}
|
||||
|
||||
// NewBus creates a new event bus
|
||||
func NewBus() *Bus {
|
||||
return &Bus{
|
||||
handlers: make(map[EventType][]Handler),
|
||||
}
|
||||
}
|
||||
|
||||
// Subscribe registers a handler for a specific event type
|
||||
func (b *Bus) Subscribe(eventType EventType, handler Handler) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
|
||||
if b.closed {
|
||||
return
|
||||
}
|
||||
|
||||
b.handlers[eventType] = append(b.handlers[eventType], handler)
|
||||
}
|
||||
|
||||
// SubscribeAll registers a handler for all events
|
||||
func (b *Bus) SubscribeAll(handler Handler) {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
|
||||
if b.closed {
|
||||
return
|
||||
}
|
||||
|
||||
// Subscribe to all known event types
|
||||
eventTypes := []EventType{
|
||||
EventForwardStarting,
|
||||
EventForwardConnected,
|
||||
EventForwardDisconnected,
|
||||
EventForwardReconnecting,
|
||||
EventForwardStopped,
|
||||
EventForwardError,
|
||||
EventHealthStatusChanged,
|
||||
EventHealthStale,
|
||||
EventWorkerHung,
|
||||
EventConfigReloaded,
|
||||
}
|
||||
|
||||
for _, et := range eventTypes {
|
||||
b.handlers[et] = append(b.handlers[et], handler)
|
||||
}
|
||||
}
|
||||
|
||||
// Publish sends an event to all registered handlers
|
||||
// Handlers are called synchronously in the order they were registered
|
||||
func (b *Bus) Publish(event Event) {
|
||||
b.mu.RLock()
|
||||
if b.closed {
|
||||
b.mu.RUnlock()
|
||||
return
|
||||
}
|
||||
handlers := make([]Handler, len(b.handlers[event.Type]))
|
||||
copy(handlers, b.handlers[event.Type])
|
||||
b.mu.RUnlock()
|
||||
|
||||
for _, handler := range handlers {
|
||||
handler(event)
|
||||
}
|
||||
}
|
||||
|
||||
// PublishAsync sends an event to all registered handlers asynchronously
|
||||
func (b *Bus) PublishAsync(event Event) {
|
||||
b.mu.RLock()
|
||||
if b.closed {
|
||||
b.mu.RUnlock()
|
||||
return
|
||||
}
|
||||
handlers := make([]Handler, len(b.handlers[event.Type]))
|
||||
copy(handlers, b.handlers[event.Type])
|
||||
b.mu.RUnlock()
|
||||
|
||||
for _, handler := range handlers {
|
||||
go handler(event)
|
||||
}
|
||||
}
|
||||
|
||||
// Close stops the event bus and prevents new subscriptions/publications
|
||||
func (b *Bus) Close() {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
|
||||
b.closed = true
|
||||
b.handlers = make(map[EventType][]Handler)
|
||||
}
|
||||
|
||||
// Helper functions for creating common events
|
||||
|
||||
// NewForwardEvent creates a forward-related event
|
||||
func NewForwardEvent(eventType EventType, forwardID string, data map[string]interface{}) Event {
|
||||
return Event{
|
||||
Type: eventType,
|
||||
ForwardID: forwardID,
|
||||
Data: data,
|
||||
}
|
||||
}
|
||||
|
||||
// NewHealthEvent creates a health status change event
|
||||
func NewHealthEvent(forwardID string, status string, errorMsg string) Event {
|
||||
return Event{
|
||||
Type: EventHealthStatusChanged,
|
||||
ForwardID: forwardID,
|
||||
Data: map[string]interface{}{
|
||||
"status": status,
|
||||
"error_msg": errorMsg,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// NewStaleEvent creates a stale connection event
|
||||
func NewStaleEvent(forwardID string, reason string) Event {
|
||||
return Event{
|
||||
Type: EventHealthStale,
|
||||
ForwardID: forwardID,
|
||||
Data: map[string]interface{}{
|
||||
"reason": reason,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// NewWorkerHungEvent creates a hung worker event
|
||||
func NewWorkerHungEvent(forwardID string, timeSinceHeartbeat string) Event {
|
||||
return Event{
|
||||
Type: EventWorkerHung,
|
||||
ForwardID: forwardID,
|
||||
Data: map[string]interface{}{
|
||||
"time_since_heartbeat": timeSinceHeartbeat,
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
package events
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestBus_Subscribe(t *testing.T) {
|
||||
bus := NewBus()
|
||||
defer bus.Close()
|
||||
|
||||
var received bool
|
||||
bus.Subscribe(EventForwardStarting, func(e Event) {
|
||||
received = true
|
||||
})
|
||||
|
||||
bus.Publish(Event{Type: EventForwardStarting})
|
||||
assert.True(t, received)
|
||||
}
|
||||
|
||||
func TestBus_SubscribeMultipleHandlers(t *testing.T) {
|
||||
bus := NewBus()
|
||||
defer bus.Close()
|
||||
|
||||
var count int32
|
||||
handler := func(e Event) {
|
||||
atomic.AddInt32(&count, 1)
|
||||
}
|
||||
|
||||
bus.Subscribe(EventForwardStarting, handler)
|
||||
bus.Subscribe(EventForwardStarting, handler)
|
||||
bus.Subscribe(EventForwardStarting, handler)
|
||||
|
||||
bus.Publish(Event{Type: EventForwardStarting})
|
||||
assert.Equal(t, int32(3), atomic.LoadInt32(&count))
|
||||
}
|
||||
|
||||
func TestBus_SubscribeAll(t *testing.T) {
|
||||
bus := NewBus()
|
||||
defer bus.Close()
|
||||
|
||||
var count int32
|
||||
bus.SubscribeAll(func(e Event) {
|
||||
atomic.AddInt32(&count, 1)
|
||||
})
|
||||
|
||||
bus.Publish(Event{Type: EventForwardStarting})
|
||||
bus.Publish(Event{Type: EventForwardConnected})
|
||||
bus.Publish(Event{Type: EventHealthStatusChanged})
|
||||
|
||||
assert.Equal(t, int32(3), atomic.LoadInt32(&count))
|
||||
}
|
||||
|
||||
func TestBus_PublishWithData(t *testing.T) {
|
||||
bus := NewBus()
|
||||
defer bus.Close()
|
||||
|
||||
var receivedEvent Event
|
||||
bus.Subscribe(EventHealthStatusChanged, func(e Event) {
|
||||
receivedEvent = e
|
||||
})
|
||||
|
||||
bus.Publish(Event{
|
||||
Type: EventHealthStatusChanged,
|
||||
ForwardID: "test-forward",
|
||||
Data: map[string]interface{}{
|
||||
"status": "Active",
|
||||
},
|
||||
})
|
||||
|
||||
assert.Equal(t, EventHealthStatusChanged, receivedEvent.Type)
|
||||
assert.Equal(t, "test-forward", receivedEvent.ForwardID)
|
||||
assert.Equal(t, "Active", receivedEvent.Data["status"])
|
||||
}
|
||||
|
||||
func TestBus_PublishAsync(t *testing.T) {
|
||||
bus := NewBus()
|
||||
defer bus.Close()
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
|
||||
bus.Subscribe(EventForwardStarting, func(e Event) {
|
||||
wg.Done()
|
||||
})
|
||||
|
||||
bus.PublishAsync(Event{Type: EventForwardStarting})
|
||||
|
||||
// Wait for async handler with timeout
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-done:
|
||||
// Success
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Async handler not called within timeout")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBus_Close(t *testing.T) {
|
||||
bus := NewBus()
|
||||
|
||||
var received bool
|
||||
bus.Subscribe(EventForwardStarting, func(e Event) {
|
||||
received = true
|
||||
})
|
||||
|
||||
bus.Close()
|
||||
|
||||
// After close, publish should not call handlers
|
||||
bus.Publish(Event{Type: EventForwardStarting})
|
||||
assert.False(t, received)
|
||||
|
||||
// Subscribe after close should be a no-op
|
||||
bus.Subscribe(EventForwardConnected, func(e Event) {
|
||||
received = true
|
||||
})
|
||||
bus.Publish(Event{Type: EventForwardConnected})
|
||||
assert.False(t, received)
|
||||
}
|
||||
|
||||
func TestBus_ConcurrentAccess(t *testing.T) {
|
||||
bus := NewBus()
|
||||
defer bus.Close()
|
||||
|
||||
var count int64
|
||||
bus.Subscribe(EventForwardStarting, func(e Event) {
|
||||
atomic.AddInt64(&count, 1)
|
||||
})
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 100; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
bus.Publish(Event{Type: EventForwardStarting})
|
||||
}()
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
assert.Equal(t, int64(100), atomic.LoadInt64(&count))
|
||||
}
|
||||
|
||||
func TestNewForwardEvent(t *testing.T) {
|
||||
event := NewForwardEvent(EventForwardStarting, "test-id", map[string]interface{}{
|
||||
"pod": "my-pod",
|
||||
})
|
||||
|
||||
assert.Equal(t, EventForwardStarting, event.Type)
|
||||
assert.Equal(t, "test-id", event.ForwardID)
|
||||
assert.Equal(t, "my-pod", event.Data["pod"])
|
||||
}
|
||||
|
||||
func TestNewHealthEvent(t *testing.T) {
|
||||
event := NewHealthEvent("test-id", "Active", "")
|
||||
|
||||
assert.Equal(t, EventHealthStatusChanged, event.Type)
|
||||
assert.Equal(t, "test-id", event.ForwardID)
|
||||
assert.Equal(t, "Active", event.Data["status"])
|
||||
assert.Equal(t, "", event.Data["error_msg"])
|
||||
}
|
||||
|
||||
func TestNewStaleEvent(t *testing.T) {
|
||||
event := NewStaleEvent("test-id", "connection too old")
|
||||
|
||||
assert.Equal(t, EventHealthStale, event.Type)
|
||||
assert.Equal(t, "test-id", event.ForwardID)
|
||||
assert.Equal(t, "connection too old", event.Data["reason"])
|
||||
}
|
||||
|
||||
func TestNewWorkerHungEvent(t *testing.T) {
|
||||
event := NewWorkerHungEvent("test-id", "60s")
|
||||
|
||||
assert.Equal(t, EventWorkerHung, event.Type)
|
||||
assert.Equal(t, "test-id", event.ForwardID)
|
||||
assert.Equal(t, "60s", event.Data["time_since_heartbeat"])
|
||||
}
|
||||
Reference in New Issue
Block a user