《游戏服务端编程实践》15.2 埋点与日志系统

{"name":"purchase_success","v":3,"ts":"2026-11-04T10:34:40+08:00","user_id":"u_1024","device_id":"d_9f3c","common":{"app_version":"1.8.2","server_id":"s1001","platform":"ios"},"props":{"sku":"gem_60","amount":6,"currency":"CNY","order_id":"o_7781"}}

// file: ch15/tracker/tracker.go
// tier: L1
// verify: go test ./ch15/tracker/

// Package tracker 实现一个带缓冲、批量发送和失败重试的埋点上报器。
//
// 设计目标是“进程内不丢已接受的埋点”：Track 只在缓冲区满或
// 已关闭时返回错误；Flush 发送失败的批次会按原顺序放回缓冲区，
// 等待下一次 Flush 或 Close 重试。采样发生在进入缓冲区之前，
// 被采样丢弃的事件不占用缓冲区。
package tracker

import (
	"context"
	"errors"
	"fmt"
	"math/rand"
	"sync"
	"time"
)

var (
	// ErrClosed 表示 Tracker 已关闭。
	ErrClosed = errors.New("tracker: closed")
	// ErrBufferFull 表示缓冲区已满，调用方需要降级或丢弃。
	ErrBufferFull = errors.New("tracker: buffer full")
)

// Event 是一条埋点。公共属性与自定义属性由调用方合并在 Props 中。
type Event struct {
	Name   string
	UserID string
	Time   time.Time
	Props  map[string]any
}

// Sink 负责把一批事件送到远端；返回错误表示整批发送失败。
type Sink interface {
	Send(ctx context.Context, events []Event) error
}

// Options 控制缓冲区、批量、采样与重试。
//
// SampleRate 小于等于 0 表示关闭采样、全部保留；大于等于 1 也
// 全部保留；只有 (0,1) 之间的值才按比例随机保留。
type Options struct {
	BufferSize    int
	MaxBatch      int
	FlushInterval time.Duration
	SampleRate    float64
	MaxAttempts   int
	Backoff       time.Duration
	Rand          *rand.Rand
}

func (o Options) withDefaults() Options {
	if o.BufferSize <= 0 {
		o.BufferSize = 1024
	}
	if o.MaxBatch <= 0 {
		o.MaxBatch = 100
	}
	if o.MaxBatch > o.BufferSize {
		o.MaxBatch = o.BufferSize
	}
	if o.SampleRate > 1 {
		o.SampleRate = 1
	}
	if o.MaxAttempts <= 0 {
		o.MaxAttempts = 3
	}
	if o.Backoff <= 0 {
		o.Backoff = 50 * time.Millisecond
	}
	if o.Rand == nil {
		o.Rand = rand.New(rand.NewSource(time.Now().UnixNano()))
	}
	return o
}

// Tracker 是一个并发安全的批量埋点缓冲器。
type Tracker struct {
	mu      sync.Mutex
	flushMu sync.Mutex
	opts    Options
	sink    Sink
	rng     *rand.Rand
	buf     []Event
	closed  bool
	stop    chan struct{}
	done    chan struct{}
}

// New 创建 Tracker；FlushInterval 大于 0 时启动后台定时发送。
func New(sink Sink, opts Options) *Tracker {
	opts = opts.withDefaults()
	t := &Tracker{
		opts: opts,
		sink: sink,
		rng:  opts.Rand,
		stop: make(chan struct{}),
		done: make(chan struct{}),
	}
	if opts.FlushInterval > 0 {
		go t.loop()
	} else {
		close(t.done)
	}
	return t
}

// Track 尝试记录一条事件。
//
// 返回 nil 表示事件已进入缓冲区或被采样丢弃；返回 ErrBufferFull
// 表示调用方需要自行降级，例如写本地磁盘或记录丢弃计数。
func (t *Tracker) Track(ev Event) error {
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.closed {
		return ErrClosed
	}
	if t.opts.SampleRate > 0 && t.opts.SampleRate < 1 {
		if t.rng.Float64() >= t.opts.SampleRate {
			return nil
		}
	}
	if len(t.buf) >= t.opts.BufferSize {
		return ErrBufferFull
	}
	if ev.Time.IsZero() {
		ev.Time = time.Now()
	}
	t.buf = append(t.buf, ev)
	return nil
}

// Flush 把当前缓冲区按 MaxBatch 切分发送。
//
// 某一个批次重试耗尽后，该批次及其后尚未发送的事件会按原顺序
// 放回缓冲区，已成功的批次不会重复发送。放回后缓冲区可能短暂
// 超过 BufferSize，此时新的 Track 会得到 ErrBufferFull，直到
// 下一次 Flush 排空。
func (t *Tracker) Flush(ctx context.Context) error {
	t.flushMu.Lock()
	defer t.flushMu.Unlock()

	t.mu.Lock()
	pending := t.buf
	t.buf = nil
	t.mu.Unlock()

	for start := 0; start < len(pending); start += t.opts.MaxBatch {
		end := start + t.opts.MaxBatch
		if end > len(pending) {
			end = len(pending)
		}
		batch := append([]Event(nil), pending[start:end]...)
		if err := t.send(ctx, batch); err != nil {
			t.requeue(pending[start:])
			return err
		}
	}
	return nil
}

// send 对单个批次做有限次重试；ctx 取消时立即返回。
func (t *Tracker) send(ctx context.Context, batch []Event) error {
	var lastErr error
	for attempt := 1; attempt <= t.opts.MaxAttempts; attempt++ {
		if err := ctx.Err(); err != nil {
			return err
		}
		lastErr = t.sink.Send(ctx, batch)
		if lastErr == nil {
			return nil
		}
		if attempt == t.opts.MaxAttempts {
			break
		}
		timer := time.NewTimer(t.opts.Backoff)
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		}
	}
	return fmt.Errorf("tracker: send batch after %d attempts: %w", t.opts.MaxAttempts, lastErr)
}

// requeue 把失败批次放回缓冲区头部，保持事件原始顺序。
func (t *Tracker) requeue(events []Event) {
	t.mu.Lock()
	defer t.mu.Unlock()
	combined := make([]Event, 0, len(events)+len(t.buf))
	combined = append(combined, events...)
	combined = append(combined, t.buf...)
	t.buf = combined
}

// loop 按 FlushInterval 周期性发送；单次失败留给下一次重试。
func (t *Tracker) loop() {
	defer close(t.done)
	ticker := time.NewTicker(t.opts.FlushInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			_ = t.Flush(context.Background())
		case <-t.stop:
			return
		}
	}
}

// Close 停止定时发送，并把剩余事件做最后一次 Flush。
//
// 关闭后 Track 返回 ErrClosed；Flush 仍可调用，便于调用方在进程
// 退出前用带超时的 ctx 完成收尾。
func (t *Tracker) Close(ctx context.Context) error {
	t.mu.Lock()
	if t.closed {
		t.mu.Unlock()
		<-t.done
		return nil
	}
	t.closed = true
	t.mu.Unlock()
	close(t.stop)
	<-t.done
	return t.Flush(ctx)
}

// Len 返回当前缓冲区中的事件数，主要用于观测和测试。
func (t *Tracker) Len() int {
	t.mu.Lock()
	defer t.mu.Unlock()
	return len(t.buf)
}