Log Shipping Agent

mkdir log-shipping-agent
cd log-shipping-agent
go mod init github.com/yourorg/log-shipping-agent
# no external dependencies — uses stdlib only

// Run: go run main.go /path/to/app.log http://localhost:3100/loki/api/v1/push
package main

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

// LogEntry represents a parsed log line from log/slog JSON output.
type LogEntry struct {
	Timestamp time.Time              `json:"time"`
	Level     string                 `json:"level"`
	Message   string                 `json:"msg"`
	Host      string                 `json:"host,omitempty"`
	Fields    map[string]interface{} `json:"fields,omitempty"`
}

// Tailer follows a file and emits new lines on the Lines() channel.
// Note: does not handle log rotation — see lesson for rotation-aware version.
type Tailer struct {
	path  string
	lines chan string
}

// NewTailer creates a Tailer for the file at path with a 1000-line buffer.
// Call Run to start tailing; read lines from Lines().
func NewTailer(path string) *Tailer {
	return &Tailer{path: path, lines: make(chan string, 1000)}
}

func (t *Tailer) Lines() <-chan string { return t.lines }

// Run seeks to the end of the file and streams new lines to Lines() until ctx is cancelled.
// Closes the lines channel on return.
func (t *Tailer) Run(ctx context.Context) {
	defer close(t.lines)

	f, err := os.Open(t.path)
	if err != nil {
		fmt.Fprintf(os.Stderr, "open %s: %v\n", t.path, err)
		return
	}
	defer f.Close()

	f.Seek(0, io.SeekEnd)

	scanner := bufio.NewScanner(f)
	for {
		select {
		case <-ctx.Done():
			return
		default:
			if scanner.Scan() {
				t.lines <- scanner.Text()
			} else {
				time.Sleep(100 * time.Millisecond)
			}
		}
	}
}

// ParseJSON parses a JSON log line (as produced by log/slog) into a LogEntry.
func ParseJSON(line string) (*LogEntry, error) {
	var entry LogEntry
	if err := json.Unmarshal([]byte(line), &entry); err != nil {
		return nil, err
	}
	return &entry, nil
}

// Enrich adds hostname metadata to every entry.
func Enrich(entry *LogEntry) {
	if entry.Host == "" {
		entry.Host, _ = os.Hostname()
	}
}

// AlertRule defines a threshold-based alert with a sliding window.
type AlertRule struct {
	Name      string
	// Level matches log/slog output: "DEBUG", "INFO", "WARN", "ERROR"
	Level     string
	Threshold int
	Window    time.Duration
	OnFire    func(count int)
}

// AlertEngine evaluates log entries against rules.
type AlertEngine struct {
	rules  []AlertRule
	counts map[string][]time.Time
	mu     sync.Mutex
}

// NewAlertEngine creates an engine that evaluates log entries against the given rules.
func NewAlertEngine(rules []AlertRule) *AlertEngine {
	return &AlertEngine{
		rules:  rules,
		counts: make(map[string][]time.Time),
	}
}

// Evaluate checks entry against all rules and fires OnFire if the threshold
// is exceeded within the rule's time window.
func (ae *AlertEngine) Evaluate(entry *LogEntry) {
	ae.mu.Lock()
	defer ae.mu.Unlock()

	for _, rule := range ae.rules {
		if entry.Level != rule.Level {
			continue
		}
		now := time.Now()
		ae.counts[rule.Name] = append(ae.counts[rule.Name], now)

		// Evict timestamps outside the window
		cutoff := now.Add(-rule.Window)
		kept := ae.counts[rule.Name][:0]
		for _, t := range ae.counts[rule.Name] {
			if t.After(cutoff) {
				kept = append(kept, t)
			}
		}
		ae.counts[rule.Name] = kept

		if len(ae.counts[rule.Name]) >= rule.Threshold {
			rule.OnFire(len(ae.counts[rule.Name]))
			ae.counts[rule.Name] = nil // reset after firing to avoid alert storm
		}
	}
}

// Forwarder batches log entries and ships them via HTTP POST.
type Forwarder struct {
	endpoint  string
	batchSize int
	interval  time.Duration
	batch     []*LogEntry
	mu        sync.Mutex
}

// NewForwarder creates a Forwarder that ships entries to endpoint in batches of
// batchSize or every interval, whichever comes first.
func NewForwarder(endpoint string, batchSize int, interval time.Duration) *Forwarder {
	return &Forwarder{
		endpoint:  endpoint,
		batchSize: batchSize,
		interval:  interval,
	}
}

// Send adds entry to the batch and flushes immediately if the batch is full.
func (fw *Forwarder) Send(entry *LogEntry) {
	fw.mu.Lock()
	fw.batch = append(fw.batch, entry)
	shouldFlush := len(fw.batch) >= fw.batchSize
	fw.mu.Unlock()

	if shouldFlush {
		fw.flush()
	}
}

// RunFlusher periodically flushes the batch on interval ticks and performs a
// final flush when ctx is cancelled so no entries are lost on shutdown.
func (fw *Forwarder) RunFlusher(ctx context.Context) {
	ticker := time.NewTicker(fw.interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			fw.flush()
			return
		case <-ticker.C:
			fw.flush()
		}
	}
}

func (fw *Forwarder) flush() {
	fw.mu.Lock()
	if len(fw.batch) == 0 {
		fw.mu.Unlock()
		return
	}
	batch := fw.batch
	fw.batch = nil
	fw.mu.Unlock()

	data, err := json.Marshal(batch)
	if err != nil {
		fmt.Fprintf(os.Stderr, "marshal: %v\n", err)
		return
	}

	resp, err := http.Post(fw.endpoint, "application/json", bytes.NewReader(data))
	if err != nil {
		fmt.Fprintf(os.Stderr, "forward error: %v — re-enqueueing %d entries\n", err, len(batch))
		fw.mu.Lock()
		fw.batch = append(batch, fw.batch...)
		fw.mu.Unlock()
		return
	}
	resp.Body.Close()
	fmt.Printf("shipped %d entries → %s (HTTP %d)\n", len(batch), fw.endpoint, resp.StatusCode)
}

func main() {
	if len(os.Args) < 3 {
		fmt.Fprintln(os.Stderr, "usage: log-agent <log-file> <loki-endpoint>")
		fmt.Fprintln(os.Stderr, "example: log-agent /var/log/app.log http://localhost:3100/loki/api/v1/push")
		os.Exit(1)
	}
	logFile := os.Args[1]
	lokiEndpoint := os.Args[2]

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	sigCh := make(chan os.Signal, 1)
	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
	go func() {
		<-sigCh
		fmt.Println("\nshutting down...")
		cancel()
	}()

	// log/slog outputs uppercase level strings: "DEBUG", "INFO", "WARN", "ERROR"
	rules := []AlertRule{
		{
			Name:      "high-error-rate",
			Level:     "ERROR",
			Threshold: 10,
			Window:    time.Minute,
			OnFire: func(count int) {
				fmt.Printf("[ALERT] high-error-rate: %d errors in last minute\n", count)
			},
		},
	}

	tail := NewTailer(logFile)
	fwd := NewForwarder(lokiEndpoint, 100, 5*time.Second)
	alerts := NewAlertEngine(rules)

	go fwd.RunFlusher(ctx)
	go tail.Run(ctx)

	fmt.Printf("tailing %s → %s\n", logFile, lokiEndpoint)

	for line := range tail.Lines() {
		entry, err := ParseJSON(line)
		if err != nil {
			continue // skip non-JSON lines
		}
		Enrich(entry)
		alerts.Evaluate(entry)
		fwd.Send(entry)
	}

	fmt.Println("done.")
}

# terminal 1 — run the agent (prints to stdout if no real Loki)
go run main.go /tmp/test.log http://localhost:3100/loki/api/v1/push

# terminal 2 — write log lines to the file
echo '{"time":"2026-08-11T10:00:00Z","level":"INFO","msg":"request completed","status":200}' >> /tmp/test.log
echo '{"time":"2026-08-11T10:00:01Z","level":"ERROR","msg":"database timeout","error":"context deadline exceeded"}' >> /tmp/test.log

tailing /tmp/test.log → http://localhost:3100/loki/api/v1/push
forward error: Post "http://localhost:3100/...": dial tcp: connection refused — re-enqueueing 1 entries

docker run -d -p 3100:3100 grafana/loki:latest
go run main.go /tmp/test.log http://localhost:3100/loki/api/v1/push

for i in $(seq 1 11); do
  echo '{"time":"2026-08-11T10:00:00Z","level":"ERROR","msg":"something failed"}' >> /tmp/test.log
done

[ALERT] high-error-rate: 11 errors in last minute