v0.13.1 folder restructure, test script added, format adapter async fix

2026-07-17 09:09:00 -04:00
parent a48514a2eb
commit e96ede32ae
49 changed files with 402 additions and 206 deletions
@@ -0,0 +1,164 @@
+package flow
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+
+	"logwisp/internal/config"
+	"logwisp/internal/core"
+	"logwisp/internal/filter"
+	"logwisp/internal/format"
+
+	"github.com/lixenwraith/log"
+)
+
+// Flow manages the complete processing pipeline for log entries:
+// LogEntry -> Rate Limiter -> Filters -> Formatter (with Sanitizer) -> TransportEvent
+type Flow struct {
+	rateLimiter *RateLimiter
+	filterChain *filter.Chain
+	formatter   format.Formatter
+	heartbeat   *HeartbeatGenerator
+	logger      *log.Logger
+
+	// Statistics
+	totalProcessed atomic.Uint64
+	totalDropped   atomic.Uint64
+	totalFormatted atomic.Uint64
+}
+
+// NewFlow creates a flow processor from configuration
+func NewFlow(cfg *config.FlowConfig, logger *log.Logger) (*Flow, error) {
+	if cfg == nil {
+		cfg = &config.FlowConfig{}
+	}
+
+	f := &Flow{
+		logger: logger,
+	}
+
+	// Create rate limiter if configured
+	if cfg.RateLimit != nil {
+		limiter, err := NewRateLimiter(*cfg.RateLimit, logger)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create rate limiter: %w", err)
+		}
+		f.rateLimiter = limiter
+	}
+
+	// Create filter chain if configured
+	if len(cfg.Filters) > 0 {
+		chain, err := filter.NewChain(cfg.Filters, logger)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create filter chain: %w", err)
+		}
+		f.filterChain = chain
+	}
+
+	// Create formatter with sanitizer integration
+	formatter, err := format.NewFormatter(cfg.Format)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create formatter: %w", err)
+	}
+	f.formatter = formatter
+
+	// Create heartbeat generator with the same formatter if configured
+	if cfg.Heartbeat != nil {
+		hb, err := NewHeartbeatGenerator(cfg.Heartbeat, formatter, logger)
+		if err != nil {
+			return nil, fmt.Errorf("heartbeat: %w", err)
+		}
+		f.heartbeat = hb
+	}
+
+	logger.Info("msg", "Flow processor created",
+		"component", "flow",
+		"rate_limiter", f.rateLimiter != nil,
+		"filter_chain", f.filterChain != nil,
+		"formatter", formatter.Name(),
+		"heartbeat", f.heartbeat != nil)
+
+	return f, nil
+}
+
+// Process applies all flow stages to a log entry
+// Returns TransportEvent and whether entry passed all stages
+func (f *Flow) Process(entry core.LogEntry) (core.TransportEvent, bool) {
+	f.totalProcessed.Add(1)
+
+	// Stage 1: Rate limiting
+	if f.rateLimiter != nil {
+		if !f.rateLimiter.Allow(entry) {
+			f.totalDropped.Add(1)
+			return core.TransportEvent{}, false
+		}
+	}
+
+	// Stage 2: Filtering
+	if f.filterChain != nil {
+		if !f.filterChain.Apply(entry) {
+			f.totalDropped.Add(1)
+			return core.TransportEvent{}, false
+		}
+	}
+
+	// Stage 3: Formatting
+	formatted, err := f.formatter.Format(entry)
+	if err != nil {
+		f.logger.Error("msg", "Failed to format log entry",
+			"component", "flow",
+			"error", err)
+		f.totalDropped.Add(1)
+		return core.TransportEvent{}, false
+	}
+
+	f.totalFormatted.Add(1)
+
+	// Create transport event
+	event := core.TransportEvent{
+		Time:    entry.Time,
+		Payload: formatted,
+		Entry:   entry, // Carry structured entry so chain sinks are format-independent
+	}
+
+	return event, true
+}
+
+// StartHeartbeat starts the heartbeat generator if configured
+// Returns channel that emits heartbeat events
+func (f *Flow) StartHeartbeat(ctx context.Context) <-chan core.TransportEvent {
+	if f.heartbeat == nil {
+		return nil
+	}
+	return f.heartbeat.Start(ctx)
+}
+
+// GetStats returns flow statistics
+func (f *Flow) GetStats() map[string]any {
+	stats := map[string]any{
+		"total_processed": f.totalProcessed.Load(),
+		"total_dropped":   f.totalDropped.Load(),
+		"total_formatted": f.totalFormatted.Load(),
+	}
+
+	if f.rateLimiter != nil {
+		stats["rate_limiter"] = f.rateLimiter.GetStats()
+	}
+
+	if f.filterChain != nil {
+		stats["filters"] = f.filterChain.GetStats()
+	}
+
+	if f.formatter != nil {
+		stats["formatter"] = f.formatter.Name()
+	}
+
+	if f.heartbeat != nil {
+		stats["heartbeat_enabled"] = true
+		stats["heartbeat_interval_ms"] = f.heartbeat.IntervalMS()
+	}
+
+	return stats
+}
+
@@ -0,0 +1,171 @@
+package flow
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"sync/atomic"
+	"time"
+
+	"logwisp/internal/config"
+	"logwisp/internal/core"
+	"logwisp/internal/format"
+
+	lconfig "github.com/lixenwraith/config"
+	"github.com/lixenwraith/log"
+	"github.com/lixenwraith/log/formatter"
+)
+
+const (
+	MinHeartbeatIntervalMS     = 100
+	DefaultHeartbeatIntervalMS = 1000
+	DefaultHeartbeatFormat     = "txt"
+)
+
+// HeartbeatGenerator produces periodic heartbeat events
+type HeartbeatGenerator struct {
+	config    *config.HeartbeatConfig
+	formatter format.Formatter // Use flow's formatter
+	logger    *log.Logger
+	beatCount atomic.Uint64
+	lastBeat  atomic.Value // time.Time
+}
+
+// NewHeartbeatGenerator creates a new heartbeat generator
+func NewHeartbeatGenerator(cfg *config.HeartbeatConfig, formatter format.Formatter, logger *log.Logger) (*HeartbeatGenerator, error) {
+	if cfg == nil || !cfg.Enabled {
+		return nil, nil
+	}
+
+	// Validate
+	if cfg.IntervalMS == 0 {
+		cfg.IntervalMS = DefaultHeartbeatIntervalMS
+	} else if cfg.IntervalMS < MinHeartbeatIntervalMS {
+		return nil, fmt.Errorf("interval_ms: must be >= %d, got %d", MinHeartbeatIntervalMS, cfg.IntervalMS)
+	}
+
+	validateFormat := lconfig.OneOf("txt", "json", "raw", "")
+	if err := validateFormat(cfg.Format); err != nil {
+		return nil, fmt.Errorf("format: %w", err)
+	}
+
+	// Defaults
+	if cfg.Format == "" {
+		cfg.Format = DefaultHeartbeatFormat
+	}
+
+	hg := &HeartbeatGenerator{
+		config:    cfg,
+		formatter: formatter,
+		logger:    logger,
+	}
+	hg.lastBeat.Store(time.Time{})
+	return hg, nil
+}
+
+// Start begins generating heartbeat events
+func (hg *HeartbeatGenerator) Start(ctx context.Context) <-chan core.TransportEvent {
+	ch := make(chan core.TransportEvent)
+
+	go func() {
+		defer close(ch)
+
+		ticker := time.NewTicker(time.Duration(hg.config.IntervalMS) * time.Millisecond)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case t := <-ticker.C:
+				event := hg.generateHeartbeat(t)
+				select {
+				case ch <- event:
+					hg.beatCount.Add(1)
+					hg.lastBeat.Store(t)
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}()
+
+	return ch
+}
+
+// generateHeartbeat creates a heartbeat transport event
+func (hg *HeartbeatGenerator) generateHeartbeat(t time.Time) core.TransportEvent {
+	// Create heartbeat as LogEntry for consistent formatting
+	entry := core.LogEntry{
+		Time:    t,
+		Source:  "heartbeat",
+		Level:   "INFO",
+		Message: "heartbeat",
+	}
+
+	// Add stats if configured
+	if hg.config.IncludeStats {
+		fields := map[string]any{
+			"type":       "heartbeat",
+			"beat_count": hg.beatCount.Load(),
+		}
+
+		if last, ok := hg.lastBeat.Load().(time.Time); ok && !last.IsZero() {
+			fields["interval_ms"] = t.Sub(last).Milliseconds()
+		}
+
+		fieldsJSON, _ := json.Marshal(fields)
+		entry.Fields = fieldsJSON
+	}
+
+	// Use formatter to generate payload
+	var payload []byte
+	var err error
+
+	// Check if we need special formatting for heartbeat
+	if hg.config.Format == "comment" {
+		// SSE comment format - bypass formatter for this special case
+		if hg.config.IncludeStats {
+			beatNum := hg.beatCount.Load()
+			payload = []byte(": heartbeat " + t.Format(time.RFC3339) +
+				" [#" + strconv.FormatUint(beatNum, 10) + "]\n")
+		} else {
+			payload = []byte(": heartbeat " + t.Format(time.RFC3339) + "\n")
+		}
+	} else {
+		// Use flow's formatter for consistent formatting
+		if adapter, ok := hg.formatter.(*format.FormatterAdapter); ok {
+			// Customize flags for heartbeat if needed
+			customFlags := int64(0)
+			if !hg.config.IncludeTimestamp {
+				// Remove timestamp flag if not wanted
+				customFlags = formatter.FlagShowLevel
+			} else {
+				customFlags = formatter.FlagDefault
+			}
+			payload, err = adapter.FormatWithFlags(entry, customFlags)
+		} else {
+			// Fallback to standard format
+			payload, err = hg.formatter.Format(entry)
+		}
+
+		if err != nil {
+			hg.logger.Error("msg", "Failed to format heartbeat",
+				"error", err)
+			// Fallback to simple text
+			payload = []byte("heartbeat: " + t.Format(time.RFC3339) + "\n")
+		}
+	}
+
+	return core.TransportEvent{
+		Time:    t,
+		Payload: payload,
+		Entry:   entry, // heartbeats traverse chain links as structured entries
+	}
+}
+
+// IntervalMS returns the heartbeat interval in milliseconds
+func (hg *HeartbeatGenerator) IntervalMS() int64 {
+	return hg.config.IntervalMS
+}
@@ -0,0 +1,133 @@
+package flow
+
+import (
+	"fmt"
+	"strings"
+	"sync/atomic"
+
+	"logwisp/internal/config"
+	"logwisp/internal/core"
+	"logwisp/internal/tokenbucket"
+
+	lconfig "github.com/lixenwraith/config"
+	"github.com/lixenwraith/log"
+)
+
+// RateLimiter enforces rate limits on log entries flowing through a pipeline
+type RateLimiter struct {
+	bucket *tokenbucket.TokenBucket
+	policy config.RateLimitPolicy
+	logger *log.Logger
+
+	// Statistics
+	maxEntrySizeBytes  int64
+	droppedBySizeCount atomic.Uint64
+	droppedCount       atomic.Uint64
+}
+
+// NewRateLimiter creates a new pipeline-level rate limiter from configuration
+func NewRateLimiter(cfg config.RateLimitConfig, logger *log.Logger) (*RateLimiter, error) {
+	// Rate <= 0 means disabled
+	if cfg.Rate <= 0 {
+		return nil, nil // No rate limit
+	}
+
+	// Validate
+	if err := lconfig.NonNegative(cfg.Rate); err != nil {
+		return nil, fmt.Errorf("rate: %w", err)
+	}
+	if err := lconfig.NonNegative(cfg.Burst); err != nil {
+		return nil, fmt.Errorf("burst: %w", err)
+	}
+	if err := lconfig.NonNegative(cfg.MaxEntrySizeBytes); err != nil {
+		return nil, fmt.Errorf("max_entry_size_bytes: %w", err)
+	}
+
+	// Defaults
+	burst := cfg.Burst
+	if burst <= 0 {
+		burst = cfg.Rate
+	}
+
+	var policy config.RateLimitPolicy
+	switch strings.ToLower(cfg.Policy) {
+	case "drop":
+		policy = config.PolicyDrop
+	case "pass", "":
+		policy = config.PolicyPass
+	default:
+		return nil, fmt.Errorf("policy: must be one of [drop, pass], got %s", cfg.Policy)
+	}
+
+	l := &RateLimiter{
+		bucket:            tokenbucket.New(burst, cfg.Rate),
+		policy:            policy,
+		logger:            logger,
+		maxEntrySizeBytes: cfg.MaxEntrySizeBytes,
+	}
+
+	return l, nil
+}
+
+// Allow checks if a log entry is permitted to pass based on the rate limit
+func (l *RateLimiter) Allow(entry core.LogEntry) bool {
+	if l == nil || l.policy == config.PolicyPass {
+		return true
+	}
+
+	// Check size limit first
+	if l.maxEntrySizeBytes > 0 && entry.RawSize > l.maxEntrySizeBytes {
+		l.droppedBySizeCount.Add(1)
+		return false
+	}
+
+	// Check rate limit if configured
+	if l.bucket != nil {
+		if l.bucket.Allow() {
+			return true
+		}
+		// Not enough tokens, drop the entry
+		l.droppedCount.Add(1)
+		return false
+	}
+
+	// No rate limit configured, size check passed
+	return true
+}
+
+// GetStats returns statistics for the rate limiter
+func (l *RateLimiter) GetStats() map[string]any {
+	if l == nil {
+		return map[string]any{
+			"enabled": false,
+		}
+	}
+
+	stats := map[string]any{
+		"enabled":               true,
+		"rate":                  l.bucket.Rate(),
+		"burst":                 l.bucket.Capacity(),
+		"dropped_total":         l.droppedCount.Load(),
+		"dropped_by_size_total": l.droppedBySizeCount.Load(),
+		"policy":                policyString(l.policy),
+		"max_entry_size_bytes":  l.maxEntrySizeBytes,
+	}
+
+	if l.bucket != nil {
+		stats["available_tokens"] = l.bucket.Tokens()
+	}
+
+	return stats
+}
+
+// policyString returns the string representation of a rate limit policy
+func policyString(p config.RateLimitPolicy) string {
+	switch p {
+	case config.PolicyDrop:
+		return "drop"
+	case config.PolicyPass:
+		return "pass"
+	default:
+		return "unknown"
+	}
+}