Log Shipping Agent

// Run: go run main.go /var/log/myapp/app.log
package main

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

// LogEntry represents a parsed log line
type LogEntry struct {
	Timestamp time.Time              `json:"timestamp"`
	Level     string                 `json:"level"`
	Message   string                 `json:"message"`
	Service   string                 `json:"service"`
	Host      string                 `json:"host"`
	Fields    map[string]interface{} `json:"fields,omitempty"`
}

// Tailer follows a file and emits new lines
type Tailer struct {
	path  string
	lines chan string
}

func NewTailer(path string) *Tailer {
	return &Tailer{path: path, lines: make(chan string, 1000)}
}

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

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: %v\n", err)
		return
	}
	defer f.Close()

	// Seek to end to only tail new lines
	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) // poll for new data
			}
		}
	}
}

// ParseJSON parses a JSON log line 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 host and service metadata
func Enrich(entry *LogEntry) {
	if entry.Host == "" {
		entry.Host, _ = os.Hostname()
	}
	if entry.Service == "" {
		entry.Service = "myapp"
	}
}

// AlertRule defines a threshold-based alert
type AlertRule struct {
	Name      string
	Level     string
	Threshold int
	Window    time.Duration
}

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

func NewAlertEngine(rules []AlertRule) *AlertEngine {
	return &AlertEngine{
		rules:  rules,
		counts: make(map[string][]time.Time),
	}
}

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()
		key := rule.Name
		ae.counts[key] = append(ae.counts[key], now)

		// Trim entries outside the window
		cutoff := now.Add(-rule.Window)
		filtered := ae.counts[key][:0]
		for _, t := range ae.counts[key] {
			if t.After(cutoff) {
				filtered = append(filtered, t)
			}
		}
		ae.counts[key] = filtered

		if len(ae.counts[key]) >= rule.Threshold {
			fmt.Printf("[ALERT] %s: %d events in %v\n", rule.Name, len(ae.counts[key]), rule.Window)
		}
	}
}

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

func NewForwarder(endpoint string, batchSize int, interval time.Duration) *Forwarder {
	return &Forwarder{
		endpoint:  endpoint,
		batchSize: batchSize,
		interval:  interval,
	}
}

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()
	}
}

func (fw *Forwarder) RunFlusher(ctx context.Context) {
	ticker := time.NewTicker(fw.interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			fw.flush() // final 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: %v\n", err)
		// Re-enqueue on failure
		fw.mu.Lock()
		fw.batch = append(batch, fw.batch...)
		fw.mu.Unlock()
		return
	}
	resp.Body.Close()
	fmt.Printf("shipped %d entries (status %d)\n", len(batch), resp.StatusCode)
}

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

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

	// Alert rules
	rules := []AlertRule{
		{Name: "high-error-rate", Level: "error", Threshold: 10, Window: time.Minute},
		{Name: "warn-spike", Level: "warn", Threshold: 50, Window: 5 * time.Minute},
	}

	// Initialize components
	logFile := "/var/log/myapp/app.log"
	if len(os.Args) > 1 {
		logFile = os.Args[1]
	}

	tail := NewTailer(logFile)
	fwd := NewForwarder("http://localhost:3100/loki/api/v1/push", 100, 5*time.Second)
	alerts := NewAlertEngine(rules)

	// Start background workers
	go fwd.RunFlusher(ctx)
	go tail.Run(ctx)

	// Process pipeline: tail → parse → enrich → alert → forward
	for line := range tail.Lines() {
		entry, err := ParseJSON(line)
		if err != nil {
			continue
		}
		Enrich(entry)
		alerts.Evaluate(entry)
		fwd.Send(entry)
	}
}