sub2api-cn-relay-manager/internal/routing/logwriter_metrics_test.go

package routing

import (
	"context"
	"errors"
	"testing"
	"time"
)

// mockLogSink 用于测试错误指标
type mockLogSink struct {
	appendDecisionError error
	appendFailoverError error
	appendStickyError   error
	closeCalled         bool
}

func (m *mockLogSink) AppendDecision(ctx context.Context, event RouteDecisionEvent) error {
	return m.appendDecisionError
}

func (m *mockLogSink) AppendFailover(ctx context.Context, event RouteFailoverEvent) error {
	return m.appendFailoverError
}

func (m *mockLogSink) AppendStickyAudit(ctx context.Context, event RouteStickyAuditEvent) error {
	return m.appendStickyError
}

func (m *mockLogSink) Close() error {
	m.closeCalled = true
	return nil
}

func TestErrorMetrics_RecordAndGet(t *testing.T) {
	var em ErrorMetrics

	// 初始值应为 0
	if em.GetFlushErrors() != 0 {
		t.Errorf("GetFlushErrors() = %d, want 0", em.GetFlushErrors())
	}
	if em.GetWriteErrors() != 0 {
		t.Errorf("GetWriteErrors() = %d, want 0", em.GetWriteErrors())
	}
	if em.GetDroppedEvents() != 0 {
		t.Errorf("GetDroppedEvents() = %d, want 0", em.GetDroppedEvents())
	}

	// 记录错误
	em.RecordFlushError()
	em.RecordFlushError()
	em.RecordWriteError()
	em.RecordDroppedEvent()
	em.RecordDroppedEvent()
	em.RecordDroppedEvent()

	// 验证计数
	if em.GetFlushErrors() != 2 {
		t.Errorf("GetFlushErrors() = %d, want 2", em.GetFlushErrors())
	}
	if em.GetWriteErrors() != 1 {
		t.Errorf("GetWriteErrors() = %d, want 1", em.GetWriteErrors())
	}
	if em.GetDroppedEvents() != 3 {
		t.Errorf("GetDroppedEvents() = %d, want 3", em.GetDroppedEvents())
	}
}

func TestErrorMetrics_ConcurrentAccess(t *testing.T) {
	var em ErrorMetrics

	// 并发记录错误
	done := make(chan bool, 3)

	go func() {
		for i := 0; i < 100; i++ {
			em.RecordFlushError()
		}
		done <- true
	}()

	go func() {
		for i := 0; i < 100; i++ {
			em.RecordWriteError()
		}
		done <- true
	}()

	go func() {
		for i := 0; i < 100; i++ {
			em.RecordDroppedEvent()
		}
		done <- true
	}()

	// 等待所有 goroutine 完成
	for i := 0; i < 3; i++ {
		<-done
	}

	// 验证计数正确
	if em.GetFlushErrors() != 100 {
		t.Errorf("GetFlushErrors() = %d, want 100", em.GetFlushErrors())
	}
	if em.GetWriteErrors() != 100 {
		t.Errorf("GetWriteErrors() = %d, want 100", em.GetWriteErrors())
	}
	if em.GetDroppedEvents() != 100 {
		t.Errorf("GetDroppedEvents() = %d, want 100", em.GetDroppedEvents())
	}
}

func TestAsyncLogWriter_Metrics(t *testing.T) {
	sink := &mockLogSink{
		appendDecisionError: errors.New("write error"),
	}

	writer := NewAsyncLogWriter(sink, AsyncLogWriterOptions{
		QueueSize:            10,
		FlushInterval:        time.Hour,
		MaxBatchSize:         2,
		FallbackWriteTimeout: time.Second,
	})
	defer writer.Close()

	// 触发写入错误
	_ = writer.AppendDecision(context.Background(), RouteDecisionEvent{
		RequestID:      "test-1",
		LogicalGroupID: "test-group",
	})

	// 等待 flush 完成
	time.Sleep(200 * time.Millisecond)

	// 强制执行 flush
	_ = writer.Flush(context.Background())

	// 验证指标被记录
	metrics := writer.Metrics()

	// 由于 batch flush 时会记录错误，应该有 flush error
	if metrics.GetFlushErrors() == 0 {
		t.Error("Expected FlushErrors > 0 after failed write")
	}
}

func TestAsyncLogWriter_ErrorHandler(t *testing.T) {
	var handledErrors []string
	errorHandler := func(ctx context.Context, err error, eventType string) {
		handledErrors = append(handledErrors, eventType+":"+err.Error())
	}

	sink := &mockLogSink{
		appendDecisionError: errors.New("decision error"),
	}

	writer := NewAsyncLogWriter(sink, AsyncLogWriterOptions{
		QueueSize:            10,
		FlushInterval:        time.Hour,
		MaxBatchSize:         1, // 立即触发 flush
		FallbackWriteTimeout: time.Second,
		OnError:              errorHandler,
	})
	defer writer.Close()

	// 触发写入
	_ = writer.AppendDecision(context.Background(), RouteDecisionEvent{
		RequestID:      "test-1",
		LogicalGroupID: "test-group",
	})

	// 等待处理
	time.Sleep(200 * time.Millisecond)
	_ = writer.Flush(context.Background())

	// 验证错误处理器被调用
	if len(handledErrors) == 0 {
		t.Error("Expected error handler to be called")
	}
}

func TestAsyncLogWriter_getEventType(t *testing.T) {
	writer := &AsyncLogWriter{}

	tests := []struct {
		name     string
		event    queuedLogEvent
		expected string
	}{
		{
			name:     "decision",
			event:    queuedLogEvent{decision: &RouteDecisionEvent{}},
			expected: "decision",
		},
		{
			name:     "failover",
			event:    queuedLogEvent{failover: &RouteFailoverEvent{}},
			expected: "failover",
		},
		{
			name:     "sticky_audit",
			event:    queuedLogEvent{sticky: &RouteStickyAuditEvent{}},
			expected: "sticky_audit",
		},
		{
			name:     "unknown",
			event:    queuedLogEvent{},
			expected: "unknown",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := writer.getEventType(tt.event)
			if got != tt.expected {
				t.Errorf("getEventType() = %q, want %q", got, tt.expected)
			}
		})
	}
}

func TestAsyncLogWriter_DroppedEventMetrics(t *testing.T) {
	sink := &mockLogSink{}

	// 创建只有 1 个缓冲区的 writer
	writer := NewAsyncLogWriter(sink, AsyncLogWriterOptions{
		QueueSize:            1,
		FlushInterval:        time.Hour, // 不自动 flush
		MaxBatchSize:         10,
		FallbackWriteTimeout: time.Second,
	})
	defer writer.Close()

	// 填满队列并触发丢弃
	for i := 0; i < 5; i++ {
		_ = writer.AppendDecision(context.Background(), RouteDecisionEvent{
			RequestID:      "test-" + string(rune('0'+i)),
			LogicalGroupID: "test-group",
		})
	}

	// 给 fallback 写入一点时间
	time.Sleep(100 * time.Millisecond)

	// 验证有事件被记录为丢弃
	metrics := writer.Metrics()
	if metrics.GetDroppedEvents() == 0 {
		t.Error("Expected DroppedEvents > 0 when queue is full")
	}
}