v0.2.0 restructured to pipeline architecture, dirty

2025-07-11 04:52:41 -04:00
parent 5936f82970
commit b503816de3
51 changed files with 4132 additions and 5936 deletions
--- a/src/internal/sink/http.go
+++ b/src/internal/sink/http.go
@ -0,0 +1,506 @@
+// FILE: src/internal/sink/http.go
+package sink
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"logwisp/src/internal/config"
+	"logwisp/src/internal/ratelimit"
+	"logwisp/src/internal/source"
+	"logwisp/src/internal/version"
+
+	"github.com/lixenwraith/log"
+	"github.com/lixenwraith/log/compat"
+	"github.com/valyala/fasthttp"
+)
+
+// HTTPSink streams log entries via Server-Sent Events
+type HTTPSink struct {
+	input         chan source.LogEntry
+	config        HTTPConfig
+	server        *fasthttp.Server
+	activeClients atomic.Int32
+	mu            sync.RWMutex
+	startTime     time.Time
+	done          chan struct{}
+	wg            sync.WaitGroup
+	logger        *log.Logger
+
+	// Path configuration
+	streamPath string
+	statusPath string
+
+	// For router integration
+	standalone bool
+
+	// Rate limiting
+	rateLimiter *ratelimit.Limiter
+
+	// Statistics
+	totalProcessed atomic.Uint64
+	lastProcessed  atomic.Value // time.Time
+}
+
+// HTTPConfig holds HTTP sink configuration
+type HTTPConfig struct {
+	Port       int
+	BufferSize int
+	StreamPath string
+	StatusPath string
+	Heartbeat  config.HeartbeatConfig
+	SSL        *config.SSLConfig
+	RateLimit  *config.RateLimitConfig
+}
+
+// NewHTTPSink creates a new HTTP streaming sink
+func NewHTTPSink(options map[string]any, logger *log.Logger) (*HTTPSink, error) {
+	cfg := HTTPConfig{
+		Port:       8080,
+		BufferSize: 1000,
+		StreamPath: "/transport",
+		StatusPath: "/status",
+	}
+
+	// Extract configuration from options
+	if port, ok := toInt(options["port"]); ok {
+		cfg.Port = port
+	}
+	if bufSize, ok := toInt(options["buffer_size"]); ok {
+		cfg.BufferSize = bufSize
+	}
+	if path, ok := options["stream_path"].(string); ok {
+		cfg.StreamPath = path
+	}
+	if path, ok := options["status_path"].(string); ok {
+		cfg.StatusPath = path
+	}
+
+	// Extract heartbeat config
+	if hb, ok := options["heartbeat"].(map[string]any); ok {
+		cfg.Heartbeat.Enabled, _ = hb["enabled"].(bool)
+		if interval, ok := toInt(hb["interval_seconds"]); ok {
+			cfg.Heartbeat.IntervalSeconds = interval
+		}
+		cfg.Heartbeat.IncludeTimestamp, _ = hb["include_timestamp"].(bool)
+		cfg.Heartbeat.IncludeStats, _ = hb["include_stats"].(bool)
+		if format, ok := hb["format"].(string); ok {
+			cfg.Heartbeat.Format = format
+		}
+	}
+
+	// Extract rate limit config
+	if rl, ok := options["rate_limit"].(map[string]any); ok {
+		cfg.RateLimit = &config.RateLimitConfig{}
+		cfg.RateLimit.Enabled, _ = rl["enabled"].(bool)
+		if rps, ok := toFloat(rl["requests_per_second"]); ok {
+			cfg.RateLimit.RequestsPerSecond = rps
+		}
+		if burst, ok := toInt(rl["burst_size"]); ok {
+			cfg.RateLimit.BurstSize = burst
+		}
+		if limitBy, ok := rl["limit_by"].(string); ok {
+			cfg.RateLimit.LimitBy = limitBy
+		}
+		if respCode, ok := toInt(rl["response_code"]); ok {
+			cfg.RateLimit.ResponseCode = respCode
+		}
+		if msg, ok := rl["response_message"].(string); ok {
+			cfg.RateLimit.ResponseMessage = msg
+		}
+		if maxPerIP, ok := toInt(rl["max_connections_per_ip"]); ok {
+			cfg.RateLimit.MaxConnectionsPerIP = maxPerIP
+		}
+		if maxTotal, ok := toInt(rl["max_total_connections"]); ok {
+			cfg.RateLimit.MaxTotalConnections = maxTotal
+		}
+	}
+
+	h := &HTTPSink{
+		input:      make(chan source.LogEntry, cfg.BufferSize),
+		config:     cfg,
+		startTime:  time.Now(),
+		done:       make(chan struct{}),
+		streamPath: cfg.StreamPath,
+		statusPath: cfg.StatusPath,
+		standalone: true,
+		logger:     logger,
+	}
+	h.lastProcessed.Store(time.Time{})
+
+	// Initialize rate limiter if configured
+	if cfg.RateLimit != nil && cfg.RateLimit.Enabled {
+		h.rateLimiter = ratelimit.New(*cfg.RateLimit)
+	}
+
+	return h, nil
+}
+
+func (h *HTTPSink) Input() chan<- source.LogEntry {
+	return h.input
+}
+
+func (h *HTTPSink) Start(ctx context.Context) error {
+	if !h.standalone {
+		// In router mode, don't start our own server
+		h.logger.Debug("msg", "HTTP sink in router mode, skipping server start",
+			"component", "http_sink")
+		return nil
+	}
+
+	// Create fasthttp adapter for logging
+	fasthttpLogger := compat.NewFastHTTPAdapter(h.logger)
+
+	h.server = &fasthttp.Server{
+		Handler:           h.requestHandler,
+		DisableKeepalive:  false,
+		StreamRequestBody: true,
+		Logger:            fasthttpLogger,
+	}
+
+	addr := fmt.Sprintf(":%d", h.config.Port)
+
+	// Run server in separate goroutine to avoid blocking
+	errChan := make(chan error, 1)
+	go func() {
+		h.logger.Info("msg", "HTTP server started",
+			"component", "http_sink",
+			"port", h.config.Port,
+			"stream_path", h.streamPath,
+			"status_path", h.statusPath)
+		err := h.server.ListenAndServe(addr)
+		if err != nil {
+			errChan <- err
+		}
+	}()
+
+	// Check if server started successfully
+	select {
+	case err := <-errChan:
+		return err
+	case <-time.After(100 * time.Millisecond):
+		// Server started successfully
+		return nil
+	}
+}
+
+func (h *HTTPSink) Stop() {
+	h.logger.Info("msg", "Stopping HTTP sink")
+
+	// Signal all client handlers to stop
+	close(h.done)
+
+	// Shutdown HTTP server if in standalone mode
+	if h.standalone && h.server != nil {
+		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+		defer cancel()
+		h.server.ShutdownWithContext(ctx)
+	}
+
+	// Wait for all active client handlers to finish
+	h.wg.Wait()
+
+	h.logger.Info("msg", "HTTP sink stopped")
+}
+
+func (h *HTTPSink) GetStats() SinkStats {
+	lastProc, _ := h.lastProcessed.Load().(time.Time)
+
+	var rateLimitStats map[string]any
+	if h.rateLimiter != nil {
+		rateLimitStats = h.rateLimiter.GetStats()
+	}
+
+	return SinkStats{
+		Type:              "http",
+		TotalProcessed:    h.totalProcessed.Load(),
+		ActiveConnections: h.activeClients.Load(),
+		StartTime:         h.startTime,
+		LastProcessed:     lastProc,
+		Details: map[string]any{
+			"port":        h.config.Port,
+			"buffer_size": h.config.BufferSize,
+			"endpoints": map[string]string{
+				"stream": h.streamPath,
+				"status": h.statusPath,
+			},
+			"rate_limit": rateLimitStats,
+		},
+	}
+}
+
+// SetRouterMode configures the sink for use with a router
+func (h *HTTPSink) SetRouterMode() {
+	h.standalone = false
+	h.logger.Debug("msg", "HTTP sink set to router mode",
+		"component", "http_sink")
+}
+
+// RouteRequest handles a request from the router
+func (h *HTTPSink) RouteRequest(ctx *fasthttp.RequestCtx) {
+	h.requestHandler(ctx)
+}
+
+func (h *HTTPSink) requestHandler(ctx *fasthttp.RequestCtx) {
+	// Check rate limit first
+	remoteAddr := ctx.RemoteAddr().String()
+	if allowed, statusCode, message := h.rateLimiter.CheckHTTP(remoteAddr); !allowed {
+		ctx.SetStatusCode(statusCode)
+		ctx.SetContentType("application/json")
+		json.NewEncoder(ctx).Encode(map[string]any{
+			"error":       message,
+			"retry_after": "60", // seconds
+		})
+		return
+	}
+
+	path := string(ctx.Path())
+
+	switch path {
+	case h.streamPath:
+		h.handleStream(ctx)
+	case h.statusPath:
+		h.handleStatus(ctx)
+	default:
+		ctx.SetStatusCode(fasthttp.StatusNotFound)
+		ctx.SetContentType("application/json")
+		json.NewEncoder(ctx).Encode(map[string]any{
+			"error": "Not Found",
+			"message": fmt.Sprintf("Available endpoints: %s (SSE transport), %s (status)",
+				h.streamPath, h.statusPath),
+		})
+	}
+}
+
+func (h *HTTPSink) handleStream(ctx *fasthttp.RequestCtx) {
+	// Track connection for rate limiting
+	remoteAddr := ctx.RemoteAddr().String()
+	if h.rateLimiter != nil {
+		h.rateLimiter.AddConnection(remoteAddr)
+		defer h.rateLimiter.RemoveConnection(remoteAddr)
+	}
+
+	// Set SSE headers
+	ctx.Response.Header.Set("Content-Type", "text/event-transport")
+	ctx.Response.Header.Set("Cache-Control", "no-cache")
+	ctx.Response.Header.Set("Connection", "keep-alive")
+	ctx.Response.Header.Set("Access-Control-Allow-Origin", "*")
+	ctx.Response.Header.Set("X-Accel-Buffering", "no")
+
+	// Create subscription for this client
+	clientChan := make(chan source.LogEntry, h.config.BufferSize)
+	clientDone := make(chan struct{})
+
+	// Subscribe to input channel
+	go func() {
+		defer close(clientChan)
+		for {
+			select {
+			case entry, ok := <-h.input:
+				if !ok {
+					return
+				}
+				h.totalProcessed.Add(1)
+				h.lastProcessed.Store(time.Now())
+
+				select {
+				case clientChan <- entry:
+				case <-clientDone:
+					return
+				case <-h.done:
+					return
+				default:
+					// Drop if client buffer full
+					h.logger.Debug("msg", "Dropped entry for slow client",
+						"component", "http_sink",
+						"remote_addr", remoteAddr)
+				}
+			case <-clientDone:
+				return
+			case <-h.done:
+				return
+			}
+		}
+	}()
+
+	// Define the transport writer function
+	streamFunc := func(w *bufio.Writer) {
+		newCount := h.activeClients.Add(1)
+		h.logger.Debug("msg", "HTTP client connected",
+			"remote_addr", remoteAddr,
+			"active_clients", newCount)
+
+		h.wg.Add(1)
+		defer func() {
+			close(clientDone)
+			newCount := h.activeClients.Add(-1)
+			h.logger.Debug("msg", "HTTP client disconnected",
+				"remote_addr", remoteAddr,
+				"active_clients", newCount)
+			h.wg.Done()
+		}()
+
+		// Send initial connected event
+		clientID := fmt.Sprintf("%d", time.Now().UnixNano())
+		connectionInfo := map[string]any{
+			"client_id":   clientID,
+			"stream_path": h.streamPath,
+			"status_path": h.statusPath,
+			"buffer_size": h.config.BufferSize,
+		}
+		data, _ := json.Marshal(connectionInfo)
+		fmt.Fprintf(w, "event: connected\ndata: %s\n\n", data)
+		w.Flush()
+
+		var ticker *time.Ticker
+		var tickerChan <-chan time.Time
+
+		if h.config.Heartbeat.Enabled {
+			ticker = time.NewTicker(time.Duration(h.config.Heartbeat.IntervalSeconds) * time.Second)
+			tickerChan = ticker.C
+			defer ticker.Stop()
+		}
+
+		for {
+			select {
+			case entry, ok := <-clientChan:
+				if !ok {
+					return
+				}
+
+				data, err := json.Marshal(entry)
+				if err != nil {
+					h.logger.Error("msg", "Failed to marshal log entry",
+						"component", "http_sink",
+						"error", err,
+						"entry_source", entry.Source)
+					continue
+				}
+
+				fmt.Fprintf(w, "data: %s\n\n", data)
+				if err := w.Flush(); err != nil {
+					return
+				}
+
+			case <-tickerChan:
+				if heartbeat := h.formatHeartbeat(); heartbeat != "" {
+					fmt.Fprint(w, heartbeat)
+					if err := w.Flush(); err != nil {
+						return
+					}
+				}
+
+			case <-h.done:
+				// Send final disconnect event
+				fmt.Fprintf(w, "event: disconnect\ndata: {\"reason\":\"server_shutdown\"}\n\n")
+				w.Flush()
+				return
+			}
+		}
+	}
+
+	ctx.SetBodyStreamWriter(streamFunc)
+}
+
+func (h *HTTPSink) formatHeartbeat() string {
+	if !h.config.Heartbeat.Enabled {
+		return ""
+	}
+
+	if h.config.Heartbeat.Format == "json" {
+		data := make(map[string]any)
+		data["type"] = "heartbeat"
+
+		if h.config.Heartbeat.IncludeTimestamp {
+			data["timestamp"] = time.Now().UTC().Format(time.RFC3339)
+		}
+
+		if h.config.Heartbeat.IncludeStats {
+			data["active_clients"] = h.activeClients.Load()
+			data["uptime_seconds"] = int(time.Since(h.startTime).Seconds())
+		}
+
+		jsonData, _ := json.Marshal(data)
+		return fmt.Sprintf("data: %s\n\n", jsonData)
+	}
+
+	// Default comment format
+	var parts []string
+	parts = append(parts, "heartbeat")
+
+	if h.config.Heartbeat.IncludeTimestamp {
+		parts = append(parts, time.Now().UTC().Format(time.RFC3339))
+	}
+
+	if h.config.Heartbeat.IncludeStats {
+		parts = append(parts, fmt.Sprintf("clients=%d", h.activeClients.Load()))
+		parts = append(parts, fmt.Sprintf("uptime=%ds", int(time.Since(h.startTime).Seconds())))
+	}
+
+	return fmt.Sprintf(": %s\n\n", strings.Join(parts, " "))
+}
+
+func (h *HTTPSink) handleStatus(ctx *fasthttp.RequestCtx) {
+	ctx.SetContentType("application/json")
+
+	var rateLimitStats any
+	if h.rateLimiter != nil {
+		rateLimitStats = h.rateLimiter.GetStats()
+	} else {
+		rateLimitStats = map[string]any{
+			"enabled": false,
+		}
+	}
+
+	status := map[string]any{
+		"service": "LogWisp",
+		"version": version.Short(),
+		"server": map[string]any{
+			"type":           "http",
+			"port":           h.config.Port,
+			"active_clients": h.activeClients.Load(),
+			"buffer_size":    h.config.BufferSize,
+			"uptime_seconds": int(time.Since(h.startTime).Seconds()),
+			"mode":           map[string]bool{"standalone": h.standalone, "router": !h.standalone},
+		},
+		"endpoints": map[string]string{
+			"transport": h.streamPath,
+			"status":    h.statusPath,
+		},
+		"features": map[string]any{
+			"heartbeat": map[string]any{
+				"enabled":  h.config.Heartbeat.Enabled,
+				"interval": h.config.Heartbeat.IntervalSeconds,
+				"format":   h.config.Heartbeat.Format,
+			},
+			"ssl": map[string]bool{
+				"enabled": h.config.SSL != nil && h.config.SSL.Enabled,
+			},
+			"rate_limit": rateLimitStats,
+		},
+	}
+
+	data, _ := json.Marshal(status)
+	ctx.SetBody(data)
+}
+
+// GetActiveConnections returns the current number of active clients
+func (h *HTTPSink) GetActiveConnections() int32 {
+	return h.activeClients.Load()
+}
+
+// GetStreamPath returns the configured transport endpoint path
+func (h *HTTPSink) GetStreamPath() string {
+	return h.streamPath
+}
+
+// GetStatusPath returns the configured status endpoint path
+func (h *HTTPSink) GetStatusPath() string {
+	return h.statusPath
+}