// main.go
package main

import (
    "bufio"
    "bytes"
    "compress/gzip"
    "context"
    "encoding/json"
    "flag"
    "fmt"
    "io"
    "net/http"
    "os"
    "os/signal"
    "path/filepath"
    "sync"
    "syscall"
    "time"

    "github.com/hpcloud/tail"
)

// LogRecord 定義發(fā)送到服務(wù)器的 JSON 結(jié)構(gòu)
type LogRecord struct {
    Timestamp time.Time `json:"timestamp"`
    Host      string    `json:"host"`
    Path      string    `json:"path"`
    Line      string    `json:"line"`
}

// Config
var (
    globPattern = flag.String("paths", "/var/log/*.log", "日志文件路徑，支持通配符")
    endpoint    = flag.String("endpoint", "http://127.0.0.1:8080/ingest", "日志收集服務(wù)地址")
    batchSize   = flag.Int("batch", 200, "每次發(fā)送最大條數(shù)")
    batchWait   = flag.Duration("wait", 2*time.Second, "批量發(fā)送最大等待時間")
    workers     = flag.Int("workers", 4, "并發(fā)發(fā)送 worker 數(shù)")
    maxQueue    = flag.Int("queue", 2000, "本地隊列最大條數(shù)，超出丟棄最老")
)

func main() {
    flag.Parse()

    host, _ := os.Hostname()

    paths, err := filepath.Glob(*globPattern)
    if err != nil {
        fmt.Fprintf(os.Stderr, "invalid pattern: %v\n", err)
        os.Exit(1)
    }
    if len(paths) == 0 {
        fmt.Fprintf(os.Stderr, "no logs matched pattern: %s\n", *globPattern)
        os.Exit(1)
    }

    // context for graceful shutdown
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // signal handling
    sigc := make(chan os.Signal, 1)
    signal.Notify(sigc, syscall.SIGINT, syscall.SIGTERM)
    go func() {
        <-sigc
        fmt.Println("received shutdown signal, stopping...")
        cancel()
    }()

    // central channel for lines
    lineCh := make(chan LogRecord, *maxQueue)

    var wg sync.WaitGroup

    // start tailers
    for _, p := range paths {
        wg.Add(1)
        go func(path string) {
            defer wg.Done()
            if err := tailFile(ctx, path, host, lineCh); err != nil {
                fmt.Fprintf(os.Stderr, "tail %s error: %v\n", path, err)
            }
        }(p)
    }

    // start sender workers
    senderWg := &sync.WaitGroup{}
    for i := 0; i < *workers; i++ {
        senderWg.Add(1)
        go func(id int) {
            defer senderWg.Done()
            runSender(ctx, id, lineCh, *endpoint, *batchSize, *batchWait)
        }(i)
    }

    // wait for tailers to finish (on ctx cancel they will exit)
    wg.Wait()
    // close channel to signal senders to flush and exit
    close(lineCh)
    // wait for senders to finish
    senderWg.Wait()

    fmt.Println("agent stopped")
}

// tailFile 使用 hpcloud/tail 跟蹤文件
func tailFile(ctx context.Context, path, host string, out chan<- LogRecord) error {
    cfg := tail.Config{
        Follow:    true,
        ReOpen:    true, // 支持日志切割后重新打開
        MustExist: false,
        Poll:      true,
        Logger:    tail.DiscardingLogger,
    }
    t, err := tail.TailFile(path, cfg)
    if err != nil {
        return err
    }
    defer t.Cleanup()

    for {
        select {
        case <-ctx.Done():
            t.Cleanup()
            return nil
        case line, ok := <-t.Lines:
            if !ok {
                // channel closed; end
                return nil
            }
            if line == nil {
                continue
            }
            rec := LogRecord{
                Timestamp: time.Now().UTC(),
                Host:      host,
                Path:      path,
                Line:      line.Text,
            }
            // non-blocking send to avoid blocking tail; drop oldest if full
            select {
            case out <- rec:
            default:
                // drop one and push new (simple policy)
                select {
                case <-out:
                default:
                }
                select {
                case out <- rec:
                default:
                    // give up if still full
                }
            }
        }
    }
}

// runSender 聚合并發(fā)送日志，帶簡單重試
func runSender(ctx context.Context, id int, in <-chan LogRecord, endpoint string, batchSize int, batchWait time.Duration) {
    httpClient := &http.Client{
        Timeout: 10 * time.Second,
    }
    buf := make([]LogRecord, 0, batchSize)

    sendBatch := func(batch []LogRecord) error {
        if len(batch) == 0 {
            return nil
        }
        // marshal
        data, err := json.Marshal(batch)
        if err != nil {
            return err
        }
        // gzip body
        var b bytes.Buffer
        gw := gzip.NewWriter(&b)
        if _, err := gw.Write(data); err != nil {
            _ = gw.Close()
            return err
        }
        _ = gw.Close()

        req, _ := http.NewRequest("POST", endpoint, &b)
        req.Header.Set("Content-Encoding", "gzip")
        req.Header.Set("Content-Type", "application/json")
        // retry with exponential backoff
        var attempt int
        for {
            attempt++
            resp, err := httpClient.Do(req)
            if err == nil {
                io.Copy(io.Discard, resp.Body)
                resp.Body.Close()
                if resp.StatusCode >= 200 && resp.StatusCode < 300 {
                    return nil
                }
                err = fmt.Errorf("bad status: %s", resp.Status)
            }
            // on ctx done, abort immediately
            select {
            case <-ctx.Done():
                return fmt.Errorf("context canceled")
            default:
            }
            if attempt >= 5 {
                return err
            }
            // backoff
            sleep := time.Duration(500*(1<<uint(attempt-1))) * time.Millisecond
            if sleep > 10*time.Second {
                sleep = 10 * time.Second
            }
            time.Sleep(sleep)
        }
    }

    timer := time.NewTimer(batchWait)
    defer timer.Stop()

    for {
        select {
        case <-ctx.Done():
            // flush remaining
            _ = sendBatch(buf)
            return
        case rec, ok := <-in:
            if !ok {
                // channel closed -> flush and exit
                _ = sendBatch(buf)
                return
            }
            buf = append(buf, rec)
            if len(buf) >= batchSize {
                _ = sendBatch(buf)
                buf = buf[:0]
                if !timer.Stop() {
                    select {
                    case <-timer.C:
                    default:
                    }
                }
                timer.Reset(batchWait)
            }
        case <-timer.C:
            if len(buf) > 0 {
                _ = sendBatch(buf)
                buf = buf[:0]
            }
            timer.Reset(batchWait)
        }
    }
}

go mod init example.com/log-agent
go get github.com/hpcloud/tail
go build -o log-agent main.go

./log-agent -paths "/var/log/myapp/*.log" -endpoint "http://log-collector:8080/ingest" -batch 100 -workers 4