《MMO 服务器开发》6.2 埋点与数据管道

客户端 / 服务端
      │  CmdTelemetryBatch 0x1301
      ▼
[采集 collect]   gateway 收到帧，原样投递，不做业务判断
      │
      ▼
[缓冲 buffer]    NATS subject: aegis.telemetry.events.>  削峰 + 解耦
      │
      ▼
[清洗 clean]     丢弃畸形事件、按 EventID 去重
      │
      ▼
[落库 store]     按事件名聚合，写入数仓（ClickHouse / Doris）

// file: internal/analytics/telemetry.go
// tier: L2
// verify: go test ./internal/analytics/ -run TestTelemetry
//
// This file is the whole telemetry path in miniature: the frame codec (shared
// with gameplay traffic), the event model, and the four pipeline stages --
// collect, buffer, clean, store. Buffering is NATS; the other three are here.
package analytics

import (
	"encoding/binary"
	"encoding/json"
	"fmt"
	"sync"
	"time"

	"github.com/nats-io/nats.go"
)

// SubjectPrefix is the NATS subject namespace for chapter six's pipeline. Every
// test publishes under it, so cleanup is a single unsubscribe.
const SubjectPrefix = "aegis.test.ch06."

const (
	// HeaderSize is length(4) + cmd(2) + seq(4) + flags(1).
	HeaderSize = 11
	// MaxPayload bounds one frame so a buggy client cannot exhaust memory.
	MaxPayload = 1 << 20
	// CmdTelemetryBatch carries a batch of client/server events upstream.
	CmdTelemetryBatch uint16 = 0x1301
	// CmdTelemetryAck acknowledges a batch, echoing the accepted count.
	CmdTelemetryAck uint16 = 0x1302
)

// Frame is the aegis wire frame from volume one: a fixed 11-byte header plus a
// payload. Telemetry reuses it unchanged, so the same decoder handles gameplay
// traffic and analytics traffic.
type Frame struct {
	Cmd     uint16
	Seq     uint32
	Flags   uint8
	Payload []byte
}

// Encode serialises a frame: length = 7 + N, total = 11 + N.
func Encode(f Frame) ([]byte, error) {
	if len(f.Payload) > MaxPayload {
		return nil, fmt.Errorf("payload %d exceeds MaxPayload %d", len(f.Payload), MaxPayload)
	}
	buf := make([]byte, HeaderSize+len(f.Payload))
	binary.BigEndian.PutUint32(buf[0:4], uint32(7+len(f.Payload)))
	binary.BigEndian.PutUint16(buf[4:6], f.Cmd)
	binary.BigEndian.PutUint32(buf[6:10], f.Seq)
	buf[10] = f.Flags
	copy(buf[HeaderSize:], f.Payload)
	return buf, nil
}

// Decode parses a frame and checks that the length field agrees with the buffer.
func Decode(buf []byte) (Frame, error) {
	if len(buf) < HeaderSize {
		return Frame{}, fmt.Errorf("frame too short: %d", len(buf))
	}
	if length := binary.BigEndian.Uint32(buf[0:4]); int(length) != len(buf)-4 {
		return Frame{}, fmt.Errorf("length %d does not match buffer %d", length, len(buf)-4)
	}
	payload := make([]byte, len(buf)-HeaderSize)
	copy(payload, buf[HeaderSize:])
	return Frame{binary.BigEndian.Uint16(buf[4:6]), binary.BigEndian.Uint32(buf[6:10]), buf[10], payload}, nil
}

// Source marks where an event was produced: the client reports what the player
// did, the server reports what the world did in response.
type Source string

const (
	SourceClient Source = "client"
	SourceServer Source = "server"
)

// Event is one tracked action. EventID is client-generated and is the dedup
// key: the same event re-sent after a reconnect must not be counted twice.
type Event struct {
	EventID  string            `json:"event_id"`
	Name     string            `json:"name"`
	PlayerID uint64            `json:"player_id"`
	ServerID uint16            `json:"server_id"`
	TsMs     int64             `json:"ts_ms"`
	Source   Source            `json:"source"`
	Props    map[string]string `json:"props,omitempty"`
}

// TelemetryBatch is the payload of a CmdTelemetryBatch frame.
type TelemetryBatch struct {
	BatchID string  `json:"batch_id"`
	Events  []Event `json:"events"`
}

// Encode wraps a batch in a CmdTelemetryBatch frame.
func (b TelemetryBatch) Encode() ([]byte, error) {
	payload, err := json.Marshal(b)
	if err != nil {
		return nil, err
	}
	return Encode(Frame{Cmd: CmdTelemetryBatch, Seq: 1, Payload: payload})
}

// DecodeBatch parses a frame and requires it to be a telemetry batch.
func DecodeBatch(buf []byte) (TelemetryBatch, error) {
	f, err := Decode(buf)
	if err != nil {
		return TelemetryBatch{}, err
	}
	if f.Cmd != CmdTelemetryBatch {
		return TelemetryBatch{}, fmt.Errorf("unexpected cmd 0x%04X", f.Cmd)
	}
	var b TelemetryBatch
	if err := json.Unmarshal(f.Payload, &b); err != nil {
		return TelemetryBatch{}, err
	}
	return b, nil
}

// Sink is the "落库" stage: an in-memory aggregation of cleaned events. In
// production it would be ClickHouse or Doris; the clean rules are the same.
type Sink struct {
	mu      sync.Mutex
	Counts  map[string]int  // event name -> cleaned count
	Seen    map[string]bool // dedup keys
	DropDup int             // duplicate EventIDs dropped
	DropBad int             // malformed events dropped
}

// NewSink returns an empty sink.
func NewSink() *Sink {
	return &Sink{Counts: map[string]int{}, Seen: map[string]bool{}}
}

// Clean applies the clean rules -- drop malformed events and duplicate
// EventIDs -- and aggregates the survivors by name. It is the "清洗 -> 落库"
// hop and returns how many events it accepted and dropped.
func (s *Sink) Clean(b TelemetryBatch) (accepted, dropped int) {
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, e := range b.Events {
		if e.Name == "" || e.PlayerID == 0 {
			s.DropBad++
			dropped++
			continue
		}
		if s.Seen[e.EventID] {
			s.DropDup++
			dropped++
			continue
		}
		s.Seen[e.EventID] = true
		s.Counts[e.Name]++
		accepted++
	}
	return accepted, dropped
}

// Publisher is the "采集" stage: it encodes a batch into a CmdTelemetryBatch
// frame before publishing, so the bytes on NATS are what a gateway would send.
type Publisher struct{ nc *nats.Conn }

// NewPublisher wraps a NATS connection.
func NewPublisher(nc *nats.Conn) *Publisher { return &Publisher{nc: nc} }

// Publish encodes and sends a batch to subject.
func (p *Publisher) Publish(subject string, b TelemetryBatch) error {
	frame, err := b.Encode()
	if err != nil {
		return err
	}
	return p.nc.Publish(subject, frame)
}

// Consumer is the "缓冲 -> 清洗 -> 落库" stage: it drains a NATS subscription,
// decodes each frame, and feeds the batch to a Sink.
type Consumer struct {
	sub  *nats.Subscription
	sink *Sink
}

// Subscribe creates a synchronous subscription and flushes it, so a publish
// issued after this call is guaranteed to be seen by this consumer.
func Subscribe(nc *nats.Conn, subject string, sink *Sink) (*Consumer, error) {
	sub, err := nc.SubscribeSync(subject)
	if err != nil {
		return nil, err
	}
	if err := nc.Flush(); err != nil {
		return nil, err
	}
	return &Consumer{sub: sub, sink: sink}, nil
}

// Run drains n messages within timeout, decoding and cleaning each.
func (c *Consumer) Run(n int, timeout time.Duration) (accepted, dropped int, err error) {
	for i := 0; i < n; i++ {
		msg, err := c.sub.NextMsg(timeout)
		if err != nil {
			return accepted, dropped, fmt.Errorf("waiting for message %d: %w", i, err)
		}
		b, err := DecodeBatch(msg.Data)
		if err != nil {
			return accepted, dropped, fmt.Errorf("decode batch %d: %w", i, err)
		}
		a, d := c.sink.Clean(b)
		accepted += a
		dropped += d
	}
	return accepted, dropped, nil
}

// Close unsubscribes the consumer.
func (c *Consumer) Close() error { return c.sub.Unsubscribe() }

// snippet: internal/analytics/telemetry_test.go (节选)
	accepted, dropped, err := consumer.Run(2, 3*time.Second)
	if err != nil {
		t.Fatalf("consume: %v", err)
	}
	if accepted != 4 || dropped != 2 {
		t.Fatalf("accepted=%d dropped=%d, want 4 and 2", accepted, dropped)
	}
	if sink.DropDup != 1 || sink.DropBad != 1 {
		t.Fatalf("dropped(dup=%d, bad=%d), want (1, 1)", sink.DropDup, sink.DropBad)
	}
	if sink.Counts["login"] != 2 || sink.Counts["level_up"] != 1 || sink.Counts["purchase"] != 1 {
		t.Fatalf("counts = %v, want login:2 level_up:1 purchase:1", sink.Counts)
	}

=== RUN   TestTelemetryFrameRoundTrip
--- PASS: TestTelemetryFrameRoundTrip (0.00s)
=== RUN   TestTelemetryPipelineOverNATS
--- PASS: TestTelemetryPipelineOverNATS (0.04s)
PASS
ok  	aegis/internal/analytics	1.488s