Logging-System/sink/sink_test.go

package sink

import (
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"logging-system/logsys"
)

func entry(i int) logsys.Entry {
	return logsys.Entry{Time: time.Unix(0, 0).UTC(), Level: "info", Message: "m", Fields: map[string]any{"i": i}}
}

func TestFileRotateAndConsole(t *testing.T) {
	dir := t.TempDir()
	path := filepath.Join(dir, "app.jsonl")
	f := &File{Path: path, Format: "json", MaxBytes: 80, Keep: 2}
	for i := 0; i < 5; i++ {
		if err := f.Write(entry(i)); err != nil {
			t.Fatal(err)
		}
	}
	if err := f.Sync(); err != nil {
		t.Fatal(err)
	}
	if err := f.Close(); err != nil {
		t.Fatal(err)
	}
	matches, err := filepath.Glob(filepath.Join(dir, "app.jsonl*"))
	if err != nil {
		t.Fatal(err)
	}
	if len(matches) < 2 {
		t.Fatalf("rotated files: %v", matches)
	}
	var lines int
	for _, name := range matches {
		b, err := os.ReadFile(name)
		if err != nil {
			t.Fatal(err)
		}
		for _, c := range b {
			if c == '\n' {
				lines++
			}
		}
	}
	if lines != 3 {
		t.Fatalf("kept lines %d, want current+2 backups, files %v", lines, matches)
	}

	var buf safeBuf
	c := &Console{Out: &buf, Format: "human"}
	if err := c.Write(logsys.Entry{Time: time.Unix(0, 0).UTC(), Level: "info", Message: "hi"}); err != nil {
		t.Fatal(err)
	}
	if buf.String() == "" {
		t.Fatal("empty console")
	}
}

type safeBuf struct {
	mu sync.Mutex
	b  []byte
}

func (s *safeBuf) Write(p []byte) (int, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.b = append(s.b, p...)
	return len(p), nil
}
func (s *safeBuf) String() string {
	s.mu.Lock()
	defer s.mu.Unlock()
	return string(s.b)
}

func TestAsyncDropAndDrain(t *testing.T) {
	hold := &lockSink{entered: make(chan struct{})}
	hold.mu.Lock()
	a := &Async{Inner: hold, Queue: 2, FlushTimeout: 2 * time.Second}
	go func() { _ = a.Write(entry(0)) }()
	select {
	case <-hold.entered:
	case <-time.After(2 * time.Second):
		t.Fatal("worker did not enter write")
	}
	for i := 0; i < a.Queue; i++ {
		if err := a.Write(entry(i + 1)); err != nil {
			t.Fatal(err)
		}
	}
	const extra = 15
	for i := 0; i < extra; i++ {
		if err := a.Write(entry(100 + i)); err != nil {
			t.Fatal(err)
		}
	}
	if a.Dropped() != extra {
		t.Fatalf("dropped %d", a.Dropped())
	}
	hold.mu.Unlock()
	if err := a.Close(); err != nil {
		t.Fatal(err)
	}
	if hold.written.Load() != int64(1+a.Queue) {
		t.Fatalf("written %d accepted %d", hold.written.Load(), a.Accepted())
	}
}

type lockSink struct {
	mu      sync.Mutex
	entered chan struct{}
	written atomic.Int64
}

func (s *lockSink) Write(e logsys.Entry) error {
	select {
	case <-s.entered:
	default:
		close(s.entered)
	}
	s.mu.Lock()
	s.mu.Unlock()
	s.written.Add(1)
	return nil
}
func (s *lockSink) Sync() error  { return nil }
func (s *lockSink) Close() error { return nil }

func TestAsyncSyncDrains(t *testing.T) {
	mem := &Memory{}
	a := &Async{Inner: mem, Queue: 64}
	for i := 0; i < 30; i++ {
		if err := a.Write(entry(i)); err != nil {
			t.Fatal(err)
		}
	}
	if err := a.Sync(); err != nil {
		t.Fatal(err)
	}
	if len(mem.Snapshot()) != 30 {
		t.Fatalf("synced %d", len(mem.Snapshot()))
	}
	if err := a.Close(); err != nil {
		t.Fatal(err)
	}
}

func TestFanoutAndHTTPBatch(t *testing.T) {
	a, b := &Memory{}, &Memory{}
	f := Fanout{Sinks: []logsys.Sink{
		MinLevel{Min: logsys.LevelWarn, Next: a},
		b,
	}}
	_ = f.Write(logsys.Entry{Level: "info", Message: "low"})
	_ = f.Write(logsys.Entry{Level: "error", Message: "high"})
	if len(a.Snapshot()) != 1 || len(b.Snapshot()) != 2 {
		t.Fatalf("a=%d b=%d", len(a.Snapshot()), len(b.Snapshot()))
	}

	var got atomic.Int32
	var calls atomic.Int32
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		calls.Add(1)
		if r.Header.Get("Authorization") != "Bearer secret" {
			http.Error(w, "unauthorized", http.StatusUnauthorized)
			return
		}
		if calls.Load() == 1 {
			http.Error(w, "busy", http.StatusServiceUnavailable)
			return
		}
		var body struct {
			Entries []logsys.Entry `json:"entries"`
		}
		if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
			t.Error(err)
			return
		}
		got.Add(int32(len(body.Entries)))
	}))
	defer srv.Close()
	h := &HTTPBatch{
		URL: srv.URL, Token: "secret", Async: true,
		MaxEntries: 2, FlushInterval: time.Hour, MaxRetries: 3,
	}
	if err := h.Write(entry(1)); err != nil {
		t.Fatal(err)
	}
	if err := h.Write(entry(2)); err != nil {
		t.Fatal(err)
	}
	if err := h.Write(entry(3)); err != nil {
		t.Fatal(err)
	}
	if err := h.Close(); err != nil {
		t.Fatal(err)
	}
	if got.Load() != 3 {
		t.Fatalf("posted %d dropped %d", got.Load(), h.Dropped())
	}
}