package logging import ( "context" "encoding/json" "io" "log/slog" "net/http" "net/http/httptest" "strings" "sync" "sync/atomic" "testing" "time" ) func TestNewShipper(t *testing.T) { tests := []struct { name string cfg ShipperConfig wantErr bool errMsg string }{ { name: "empty backend returns nil", cfg: ShipperConfig{Backend: ""}, wantErr: false, }, { name: "victoria backend requires URL", cfg: ShipperConfig{Backend: "victoria", URL: ""}, wantErr: true, errMsg: "URL is required", }, { name: "victoria backend with URL succeeds", cfg: ShipperConfig{Backend: "victoria", URL: "http://localhost:9428"}, wantErr: false, }, { name: "unknown backend returns error", cfg: ShipperConfig{Backend: "unknown"}, wantErr: true, errMsg: "unknown log shipper backend", }, { name: "opensearch not implemented", cfg: ShipperConfig{Backend: "opensearch"}, wantErr: true, errMsg: "not yet implemented", }, { name: "loki not implemented", cfg: ShipperConfig{Backend: "loki"}, wantErr: true, errMsg: "not yet implemented", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { shipper, err := NewShipper(tt.cfg) if tt.wantErr { if err == nil { t.Errorf("NewShipper() expected error, got nil") } else if tt.errMsg != "" && !strings.Contains(err.Error(), tt.errMsg) { t.Errorf("NewShipper() error = %v, want error containing %q", err, tt.errMsg) } return } if err != nil { t.Errorf("NewShipper() unexpected error: %v", err) } if tt.cfg.Backend == "" && shipper != nil { t.Error("NewShipper() with empty backend should return nil shipper") } if shipper != nil { shipper.Close() } }) } } func TestVictoriaShipper_Ship(t *testing.T) { var receivedLogs []map[string]any var mu sync.Mutex server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { t.Errorf("expected POST, got %s", r.Method) } if !strings.Contains(r.URL.Path, "/insert/jsonline") { t.Errorf("expected /insert/jsonline path, got %s", r.URL.Path) } if r.Header.Get("Content-Type") != "application/stream+json" { t.Errorf("expected application/stream+json content type, got %s", r.Header.Get("Content-Type")) } body, _ := io.ReadAll(r.Body) lines := strings.Split(strings.TrimSpace(string(body)), "\n") mu.Lock() for _, line := range lines { var doc map[string]any if err := json.Unmarshal([]byte(line), &doc); err != nil { t.Errorf("failed to unmarshal log line: %v", err) } receivedLogs = append(receivedLogs, doc) } mu.Unlock() w.WriteHeader(http.StatusOK) })) defer server.Close() shipper, err := NewVictoriaShipper(ShipperConfig{ URL: server.URL, Service: "test-service", }) if err != nil { t.Fatalf("NewVictoriaShipper() error: %v", err) } defer shipper.Close() entries := []LogEntry{ { Time: time.Date(2024, 1, 8, 12, 0, 0, 0, time.UTC), Level: slog.LevelInfo, Message: "test message 1", Source: "test.go:42", Attrs: map[string]any{"key1": "value1"}, }, { Time: time.Date(2024, 1, 8, 12, 0, 1, 0, time.UTC), Level: slog.LevelError, Message: "test message 2", Source: "test.go:43", Attrs: map[string]any{"key2": 123}, }, } ctx := context.Background() if err := shipper.Ship(ctx, entries); err != nil { t.Fatalf("Ship() error: %v", err) } mu.Lock() defer mu.Unlock() if len(receivedLogs) != 2 { t.Errorf("expected 2 logs, got %d", len(receivedLogs)) } // Check first log if receivedLogs[0]["_msg"] != "test message 1" { t.Errorf("expected _msg 'test message 1', got %v", receivedLogs[0]["_msg"]) } if receivedLogs[0]["level"] != "INFO" { t.Errorf("expected level 'INFO', got %v", receivedLogs[0]["level"]) } if receivedLogs[0]["service"] != "test-service" { t.Errorf("expected service 'test-service', got %v", receivedLogs[0]["service"]) } if receivedLogs[0]["key1"] != "value1" { t.Errorf("expected key1 'value1', got %v", receivedLogs[0]["key1"]) } // Check second log if receivedLogs[1]["level"] != "ERROR" { t.Errorf("expected level 'ERROR', got %v", receivedLogs[1]["level"]) } } func TestVictoriaShipper_BasicAuth(t *testing.T) { var authHeader string server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { authHeader = r.Header.Get("Authorization") w.WriteHeader(http.StatusOK) })) defer server.Close() shipper, err := NewVictoriaShipper(ShipperConfig{ URL: server.URL, Username: "testuser", Password: "testpass", }) if err != nil { t.Fatalf("NewVictoriaShipper() error: %v", err) } defer shipper.Close() entries := []LogEntry{{Time: time.Now(), Level: slog.LevelInfo, Message: "test"}} if err := shipper.Ship(context.Background(), entries); err != nil { t.Fatalf("Ship() error: %v", err) } if authHeader == "" { t.Error("expected Authorization header to be set") } if !strings.HasPrefix(authHeader, "Basic ") { t.Errorf("expected Basic auth, got: %s", authHeader) } } func TestVictoriaShipper_EmptyBatch(t *testing.T) { var called bool server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { called = true w.WriteHeader(http.StatusOK) })) defer server.Close() shipper, _ := NewVictoriaShipper(ShipperConfig{URL: server.URL}) defer shipper.Close() if err := shipper.Ship(context.Background(), nil); err != nil { t.Errorf("Ship() with nil entries should not error: %v", err) } if err := shipper.Ship(context.Background(), []LogEntry{}); err != nil { t.Errorf("Ship() with empty entries should not error: %v", err) } if called { t.Error("Ship() should not make HTTP request for empty batch") } } func TestVictoriaShipper_ServerError(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("internal error")) })) defer server.Close() shipper, _ := NewVictoriaShipper(ShipperConfig{URL: server.URL}) defer shipper.Close() entries := []LogEntry{{Time: time.Now(), Level: slog.LevelInfo, Message: "test"}} err := shipper.Ship(context.Background(), entries) if err == nil { t.Error("Ship() should return error on server error") } if !strings.Contains(err.Error(), "500") { t.Errorf("error should contain status code, got: %v", err) } } func TestAsyncHandler_Batching(t *testing.T) { var shipCount atomic.Int32 var totalEntries atomic.Int32 mockShipper := &mockShipper{ shipFunc: func(ctx context.Context, entries []LogEntry) error { shipCount.Add(1) totalEntries.Add(int32(len(entries))) return nil }, } cfg := ShipperConfig{ BatchSize: 5, FlushInterval: 100 * time.Millisecond, } stdoutHandler := slog.NewTextHandler(io.Discard, nil) handler := NewAsyncHandler(stdoutHandler, mockShipper, cfg, nil) // Log 12 entries - should trigger 2 batch flushes (at 5 and 10) plus 2 remaining for range 12 { record := slog.NewRecord(time.Now(), slog.LevelInfo, "test message", 0) handler.Handle(context.Background(), record) } // Wait for flush interval to trigger time.Sleep(200 * time.Millisecond) handler.Shutdown() if totalEntries.Load() != 12 { t.Errorf("expected 12 total entries shipped, got %d", totalEntries.Load()) } } func TestAsyncHandler_Shutdown(t *testing.T) { var shipped []LogEntry var mu sync.Mutex mockShipper := &mockShipper{ shipFunc: func(ctx context.Context, entries []LogEntry) error { mu.Lock() shipped = append(shipped, entries...) mu.Unlock() return nil }, } cfg := ShipperConfig{ BatchSize: 100, // Large batch size so nothing flushes immediately FlushInterval: 10 * time.Second, } stdoutHandler := slog.NewTextHandler(io.Discard, nil) handler := NewAsyncHandler(stdoutHandler, mockShipper, cfg, nil) // Log a few entries for range 3 { record := slog.NewRecord(time.Now(), slog.LevelInfo, "test", 0) handler.Handle(context.Background(), record) } // Entries should be pending (not shipped yet due to large batch size) mu.Lock() if len(shipped) != 0 { t.Errorf("expected 0 shipped before shutdown, got %d", len(shipped)) } mu.Unlock() // Shutdown should flush pending entries handler.Shutdown() mu.Lock() if len(shipped) != 3 { t.Errorf("expected 3 shipped after shutdown, got %d", len(shipped)) } mu.Unlock() } func TestAsyncHandler_NoShipper(t *testing.T) { cfg := ShipperConfig{} stdoutHandler := slog.NewTextHandler(io.Discard, nil) handler := NewAsyncHandler(stdoutHandler, nil, cfg, nil) // Should not panic with nil shipper record := slog.NewRecord(time.Now(), slog.LevelInfo, "test", 0) if err := handler.Handle(context.Background(), record); err != nil { t.Errorf("Handle() with nil shipper should not error: %v", err) } // Shutdown should not panic handler.Shutdown() } func TestResolveAttrValue(t *testing.T) { tests := []struct { name string value slog.Value expected any }{ {"string", slog.StringValue("test"), "test"}, {"int64", slog.Int64Value(42), int64(42)}, {"uint64", slog.Uint64Value(42), uint64(42)}, {"float64", slog.Float64Value(3.14), 3.14}, {"bool", slog.BoolValue(true), true}, {"duration", slog.DurationValue(5 * time.Second), "5s"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { got := resolveAttrValue(tt.value) if got != tt.expected { t.Errorf("resolveAttrValue() = %v, want %v", got, tt.expected) } }) } } // mockShipper implements Shipper for testing type mockShipper struct { shipFunc func(ctx context.Context, entries []LogEntry) error closeFunc func() error } func (m *mockShipper) Ship(ctx context.Context, entries []LogEntry) error { if m.shipFunc != nil { return m.shipFunc(ctx, entries) } return nil } func (m *mockShipper) Close() error { if m.closeFunc != nil { return m.closeFunc() } return nil }