用 Go 写一个完整的指标操作库：定义、注册、查询与计算 · 志强vlog

从指标定义到查询结果，用 Go 标准库实现一个可运行、可测试的指标操作库

前面几篇聊了统计平台和指标引擎的设计思路。这一篇把代码写完整：一个不依赖第三方库、可以直接 go test、go run 的指标操作库。

实现目标：

• 指标可以注册、按名字请求；
• 直接指标支持计数和求和，自带时间字段和过滤条件；
• 间接指标声明依赖，按依赖顺序计算；
• 一次请求混合直接指标和间接指标；
• 后端可替换，测试不需要外部服务；
• 完整示例跑出真实结果。

目录结构：

metrics/
├── go.mod
├── metric.go       # 指标接口、取值方式
├── filter.go       # 条件与过滤
├── lookup.go       # 点号路径取值
├── registry.go     # 注册与依赖解析
├── backend.go      # 查询规格、内存后端
├── engine.go       # 时间桶合并、间接指标
├── numbers.go      # 比率与除法
├── filter_test.go
├── registry_test.go
├── engine_test.go
└── cmd/demo/
    └── main.go     # 完整示例

go.mod 只有模块声明，没有任何依赖：

go.mod

module github.com/chihqiang/metrics

go 1.22

---

一、指标定义

先定义指标本身。直接指标负责“取数”，间接指标负责“计算”。

metric.go

package metrics

// ValueKind 是直接指标的取值方式。
type ValueKind int

const (
	// Count 统计文档条数。
	Count ValueKind = iota
	// Sum 对数值字段求和。
	Sum
)

// Direct 是直接指标：时间字段 + 过滤条件 + 取值方式。
type Direct interface {
	Name() string
	Label() string
	TimeField() string
	Kind() ValueKind
	Field() string // Sum 时必填，Count 时忽略
	Filter() Filter
}

// Indirect 是间接指标：依赖若干指标，只做纯计算。
type Indirect interface {
	Name() string
	Label() string
	Deps() []string
	Calc(values map[string]float64) float64
}

两个关键设计：

1. 时间字段属于指标，不属于请求。这样“下单量按创建时间、支付量按支付时间”可以共存于一次查询。
2. 间接指标只做计算，不关心数据从哪来。比率、差值、求和、换算都可以用它表达。

---

二、过滤条件

条件支持等值、集合、存在、范围四种：

filter.go

package metrics

import "fmt"

// Op 是条件操作符。
type Op string

const (
	Eq     Op = "eq"
	In     Op = "in"
	Exists Op = "exists"
	Range  Op = "range"
)

// Cond 是一个过滤条件。
type Cond struct {
	Field  string
	Op     Op
	Values []any
}

// Filter 是若干条件的与关系。
type Filter []Cond

// Validate 校验条件是否合法；不认识的条件直接报错，避免静默放宽。
func (f Filter) Validate() error {
	for i, c := range f {
		if c.Field == "" {
			return fmt.Errorf("第 %d 个条件缺少字段", i)
		}
		switch c.Op {
		case Eq:
			if len(c.Values) != 1 {
				return fmt.Errorf("%s: eq 需要 1 个值", c.Field)
			}
		case In:
			if len(c.Values) == 0 {
				return fmt.Errorf("%s: in 至少需要 1 个值", c.Field)
			}
		case Exists:
			if len(c.Values) != 0 {
				return fmt.Errorf("%s: exists 不需要值", c.Field)
			}
		case Range:
			if len(c.Values) != 2 {
				return fmt.Errorf("%s: range 需要 2 个值", c.Field)
			}
		default:
			return fmt.Errorf("%s: 不支持的操作符 %q", c.Field, c.Op)
		}
	}
	return nil
}

// Match 判断文档是否满足全部条件。
func (f Filter) Match(doc map[string]any) bool {
	if f.Validate() != nil {
		return false
	}
	for _, c := range f {
		v, ok := lookup(doc, c.Field)
		switch c.Op {
		case Eq:
			if !ok || !equal(v, c.Values[0]) {
				return false
			}
		case In:
			if !ok || !contains(c.Values, v) {
				return false
			}
		case Exists:
			if !ok {
				return false
			}
		case Range:
			if !ok || !inRange(v, c.Values[0], c.Values[1]) {
				return false
			}
		}
	}
	return true
}

这里有一个重要原则：条件不认识时不放宽。

Match 在条件非法时直接返回 false，而不是跳过条件。统计查询宁可少返回，也不能因为条件写错而返回全量数据。

---

三、路径取值

过滤条件里的字段支持点号路径，比如 order.status：

lookup.go

package metrics

import (
	"fmt"
	"strings"
)

// lookup 按点号路径取值，例如 "order.status"。
func lookup(doc map[string]any, field string) (any, bool) {
	var cur any = doc
	for _, part := range strings.Split(field, ".") {
		m, ok := cur.(map[string]any)
		if !ok {
			return nil, false
		}
		if cur, ok = m[part]; !ok {
			return nil, false
		}
	}
	return cur, true
}

func equal(a, b any) bool {
	return fmt.Sprint(a) == fmt.Sprint(b)
}

func contains(values []any, v any) bool {
	for _, x := range values {
		if equal(x, v) {
			return true
		}
	}
	return false
}

func inRange(v, min, max any) bool {
	f, ok := toFloat(v)
	if !ok {
		return false
	}
	lo, ok := toFloat(min)
	if !ok {
		return false
	}
	hi, ok := toFloat(max)
	if !ok {
		return false
	}
	return f >= lo && f <= hi
}

func toFloat(v any) (float64, bool) {
	switch n := v.(type) {
	case float64:
		return n, true
	case float32:
		return float64(n), true
	case int:
		return float64(n), true
	case int32:
		return float64(n), true
	case int64:
		return float64(n), true
	default:
		return 0, false
	}
}

---

四、注册表与依赖解析

注册表保存所有指标，并负责把请求的指标名解析成“直接指标 + 按顺序排列的间接指标”。

registry.go

package metrics

import "fmt"

// Registry 保存指标定义，并负责依赖解析。
type Registry struct {
	directs   map[string]Direct
	indirects map[string]Indirect
}

func NewRegistry() *Registry {
	return &Registry{
		directs:   make(map[string]Direct),
		indirects: make(map[string]Indirect),
	}
}

// AddDirect 注册直接指标。
func (r *Registry) AddDirect(d Direct) error {
	if d.Name() == "" {
		return fmt.Errorf("指标名不能为空")
	}
	if d.TimeField() == "" {
		return fmt.Errorf("指标 %s 缺少时间字段", d.Name())
	}
	if d.Kind() == Sum && d.Field() == "" {
		return fmt.Errorf("求和指标 %s 缺少字段", d.Name())
	}
	if _, ok := r.directs[d.Name()]; ok {
		return fmt.Errorf("指标重复注册: %s", d.Name())
	}
	r.directs[d.Name()] = d
	return nil
}

// AddIndirect 注册间接指标。
func (r *Registry) AddIndirect(i Indirect) error {
	if i.Name() == "" {
		return fmt.Errorf("指标名不能为空")
	}
	if len(i.Deps()) == 0 {
		return fmt.Errorf("指标 %s 至少依赖一个指标", i.Name())
	}
	if _, ok := r.indirects[i.Name()]; ok {
		return fmt.Errorf("指标重复注册: %s", i.Name())
	}
	r.indirects[i.Name()] = i
	return nil
}

// Get 按名字返回指标。
func (r *Registry) Get(name string) (any, bool) {
	if d, ok := r.directs[name]; ok {
		return d, true
	}
	if i, ok := r.indirects[name]; ok {
		return i, true
	}
	return nil, false
}

// Resolve 返回请求指标及其依赖；间接指标按依赖顺序排列。
func (r *Registry) Resolve(names []string) ([]Direct, []Indirect, error) {
	seen := make(map[string]bool)
	var directs []Direct
	var indirects []Indirect

	var visit func(string, map[string]bool) error
	visit = func(name string, stack map[string]bool) error {
		if seen[name] {
			return nil
		}
		if stack[name] {
			return fmt.Errorf("指标依赖成环: %s", name)
		}
		if d, ok := r.directs[name]; ok {
			seen[name] = true
			directs = append(directs, d)
			return nil
		}
		ind, ok := r.indirects[name]
		if !ok {
			return fmt.Errorf("未知指标: %s", name)
		}
		stack[name] = true
		for _, dep := range ind.Deps() {
			if err := visit(dep, stack); err != nil {
				return err
			}
		}
		delete(stack, name)
		seen[name] = true
		indirects = append(indirects, ind)
		return nil
	}

	for _, name := range names {
		if err := visit(name, map[string]bool{}); err != nil {
			return nil, nil, err
		}
	}
	return directs, indirects, nil
}

请求 客单价 时，会先解析出 营收、支付量，再把 客单价 放到最后计算；如果依赖成环，直接报错。

---

五、查询规格与后端

引擎不直接拼查询语句，而是把指标翻译成查询规格，交给后端执行：

backend.go

package metrics

import (
	"context"
	"sort"
	"time"
)

// Spec 是交给后端执行的单指标查询规格。
type Spec struct {
	Name      string
	TimeField string
	Kind      ValueKind
	Field     string
	Filter    Filter
}

// Bucket 是一个时间桶的结果。
type Bucket struct {
	Key   string
	Value float64
}

// Backend 执行查询规格；生产环境可以替换成搜索集群、数据库或缓存。
type Backend interface {
	Query(ctx context.Context, specs []Spec, start, end int64, step string, global Filter) (map[string][]Bucket, error)
}

// MemoryBackend 是内存实现，用于示例和测试。
type MemoryBackend struct {
	Docs []map[string]any
}

func (m *MemoryBackend) Query(ctx context.Context, specs []Spec, start, end int64, step string, global Filter) (map[string][]Bucket, error) {
	out := make(map[string][]Bucket, len(specs))
	for _, spec := range specs {
		values := make(map[string]float64)
		for _, doc := range m.Docs {
			if err := ctx.Err(); err != nil {
				return nil, err
			}
			if !global.Match(doc) || !spec.Filter.Match(doc) {
				continue
			}
			ts, ok := doc[spec.TimeField].(int64)
			if !ok || ts < start || ts >= end {
				continue
			}
			key := BucketKey(ts, step)
			if spec.Kind == Sum {
				if v, ok := toFloat(doc[spec.Field]); ok {
					values[key] += v
				}
				continue
			}
			values[key]++
		}
		out[spec.Name] = toBuckets(values)
	}
	return out, nil
}

// BucketKey 把时间戳截断到指定粒度。
func BucketKey(ts int64, step string) string {
	t := time.Unix(ts, 0).In(time.Local)
	switch step {
	case "hour":
		return t.Format("2006-01-02 15:00")
	case "month":
		return t.Format("2006-01")
	default:
		return t.Format("2006-01-02")
	}
}

func toBuckets(values map[string]float64) []Bucket {
	list := make([]Bucket, 0, len(values))
	for k, v := range values {
		list = append(list, Bucket{Key: k, Value: v})
	}
	sort.Slice(list, func(i, j int) bool { return list[i].Key < list[j].Key })
	return list
}

生产环境换后端时，只需要把 Spec 翻译成真实的查询，返回同样的 map[string][]Bucket。指标定义和引擎不用改。

---

六、引擎：合并时间桶，再算间接指标

Engine.Evaluate 是整个库的核心，顺序不能乱：

1. 校验请求；
2. 解析指标依赖；
3. 把直接指标翻译成查询规格；
4. 调后端拿每个指标的桶；
5. 按时间 key 合并；
6. 在合并后的值上计算间接指标。

engine.go

package metrics

import (
	"context"
	"fmt"
	"sort"
)

// Request 是一次指标查询请求。
type Request struct {
	Start      int64
	End        int64
	Step       string
	Indicators []string
	Filters    Filter
}

func (r Request) validate() error {
	if len(r.Indicators) == 0 {
		return fmt.Errorf("指标不能为空")
	}
	if r.End <= r.Start {
		return fmt.Errorf("时间范围不合法")
	}
	switch r.Step {
	case "", "hour", "day", "month":
	default:
		return fmt.Errorf("不支持的粒度: %s", r.Step)
	}
	return r.Filters.Validate()
}

// Point 是某个时间桶上的指标值。
type Point struct {
	Key   string
	Value float64
}

// Result 按指标名返回时间序列。
type Result map[string][]Point

// Engine 负责解析指标、调用后端、合并时间桶并计算间接指标。
type Engine struct {
	registry *Registry
	backend  Backend
}

func NewEngine(registry *Registry, backend Backend) *Engine {
	return &Engine{registry: registry, backend: backend}
}

// Evaluate 执行一次指标查询。
func (e *Engine) Evaluate(ctx context.Context, req Request) (Result, error) {
	if err := req.validate(); err != nil {
		return nil, err
	}
	if req.Step == "" {
		req.Step = "day"
	}

	directs, indirects, err := e.registry.Resolve(req.Indicators)
	if err != nil {
		return nil, err
	}

	specs := make([]Spec, 0, len(directs))
	for _, d := range directs {
		specs = append(specs, Spec{
			Name:      d.Name(),
			TimeField: d.TimeField(),
			Kind:      d.Kind(),
			Field:     d.Field(),
			Filter:    d.Filter(),
		})
	}

	raw, err := e.backend.Query(ctx, specs, req.Start, req.End, req.Step, req.Filters)
	if err != nil {
		return nil, err
	}

	requested := make(map[string]bool, len(req.Indicators))
	for _, name := range req.Indicators {
		requested[name] = true
	}

	directAt := make(map[string]map[string]float64, len(directs))
	keySet := make(map[string]bool)
	for _, d := range directs {
		at := make(map[string]float64)
		for _, b := range raw[d.Name()] {
			at[b.Key] = b.Value
			keySet[b.Key] = true
		}
		directAt[d.Name()] = at
	}

	keys := make([]string, 0, len(keySet))
	for k := range keySet {
		keys = append(keys, k)
	}
	sort.Strings(keys)

	result := make(Result, len(req.Indicators))
	for _, k := range keys {
		values := make(map[string]float64, len(directs)+len(indirects))
		for _, d := range directs {
			values[d.Name()] = directAt[d.Name()][k]
			if requested[d.Name()] {
				result[d.Name()] = append(result[d.Name()], Point{Key: k, Value: values[d.Name()]})
			}
		}
		for _, ind := range indirects {
			values[ind.Name()] = ind.Calc(values)
			if requested[ind.Name()] {
				result[ind.Name()] = append(result[ind.Name()], Point{Key: k, Value: values[ind.Name()]})
			}
		}
	}
	return result, nil
}

最容易写错的是顺序：间接指标必须在跨时间桶合并之后计算。

支付率这类指标，分子分母的时间字段可能不同。如果先在每个桶里除、再把比率平均，结果一定偏；正确做法是先合并分子分母，再做除法。

---

七、数值工具

比率和除法都要处理分母为零：

numbers.go

package metrics

// Percent 返回 a / b * 100；分母为 0 时返回 0。
func Percent(a, b float64) float64 {
	if b == 0 {
		return 0
	}
	return a / b * 100
}

// SafeDiv 返回 a / b；分母为 0 时返回 0。
func SafeDiv(a, b float64) float64 {
	if b == 0 {
		return 0
	}
	return a / b
}

---

八、完整示例

下面这个 main.go 注册五个指标：下单量、支付量、营收、支付率、客单价。

cmd/demo/main.go

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/chihqiang/metrics"
)

type countIndicator struct {
	name   string
	time   string
	filter metrics.Filter
}

func (c countIndicator) Name() string            { return c.name }
func (c countIndicator) Label() string           { return c.name }
func (c countIndicator) TimeField() string       { return c.time }
func (c countIndicator) Kind() metrics.ValueKind { return metrics.Count }
func (c countIndicator) Field() string           { return "" }
func (c countIndicator) Filter() metrics.Filter  { return c.filter }

type sumIndicator struct {
	name   string
	time   string
	field  string
	filter metrics.Filter
}

func (s sumIndicator) Name() string            { return s.name }
func (s sumIndicator) Label() string           { return s.name }
func (s sumIndicator) TimeField() string       { return s.time }
func (s sumIndicator) Kind() metrics.ValueKind { return metrics.Sum }
func (s sumIndicator) Field() string           { return s.field }
func (s sumIndicator) Filter() metrics.Filter  { return s.filter }

type ratioIndicator struct {
	name string
	deps []string
}

func (r ratioIndicator) Name() string   { return r.name }
func (r ratioIndicator) Label() string  { return r.name }
func (r ratioIndicator) Deps() []string { return r.deps }
func (r ratioIndicator) Calc(v map[string]float64) float64 {
	return metrics.Percent(v[r.deps[0]], v[r.deps[1]])
}

type divIndicator struct {
	name string
	deps []string
}

func (d divIndicator) Name() string   { return d.name }
func (d divIndicator) Label() string  { return d.name }
func (d divIndicator) Deps() []string { return d.deps }
func (d divIndicator) Calc(v map[string]float64) float64 {
	return metrics.SafeDiv(v[d.deps[0]], v[d.deps[1]])
}

func day(d int) int64 {
	return time.Date(2026, 11, d, 10, 0, 0, 0, time.Local).Unix()
}

func main() {
	backend := &metrics.MemoryBackend{Docs: []map[string]any{
		{"create_time": day(1), "pay_time": day(1), "amount": 100.0, "status": "paid"},
		{"create_time": day(1), "status": "cancel"},
		{"create_time": day(2), "pay_time": day(2), "amount": 200.0, "status": "paid"},
	}}

	paid := metrics.Filter{{Field: "status", Op: metrics.Eq, Values: []any{"paid"}}}

	reg := metrics.NewRegistry()
	if err := reg.AddDirect(countIndicator{name: "下单量", time: "create_time"}); err != nil {
		log.Fatal(err)
	}
	if err := reg.AddDirect(countIndicator{name: "支付量", time: "pay_time", filter: paid}); err != nil {
		log.Fatal(err)
	}
	if err := reg.AddDirect(sumIndicator{name: "营收", time: "pay_time", field: "amount", filter: paid}); err != nil {
		log.Fatal(err)
	}
	if err := reg.AddIndirect(ratioIndicator{name: "支付率", deps: []string{"支付量", "下单量"}}); err != nil {
		log.Fatal(err)
	}
	if err := reg.AddIndirect(divIndicator{name: "客单价", deps: []string{"营收", "支付量"}}); err != nil {
		log.Fatal(err)
	}

	engine := metrics.NewEngine(reg, backend)
	res, err := engine.Evaluate(context.Background(), metrics.Request{
		Start: day(1), End: day(3), Step: "day",
		Indicators: []string{"下单量", "支付量", "营收", "支付率", "客单价"},
	})
	if err != nil {
		log.Fatal(err)
	}

	for _, name := range []string{"下单量", "支付量", "营收", "支付率", "客单价"} {
		fmt.Println(name)
		for _, p := range res[name] {
			fmt.Printf("  %s  %.2f\n", p.Key, p.Value)
		}
	}
}

输出：

下单量
  2026-11-01  2.00
  2026-11-02  1.00
支付量
  2026-11-01  1.00
  2026-11-02  1.00
营收
  2026-11-01  100.00
  2026-11-02  200.00
支付率
  2026-11-01  50.00
  2026-11-02  100.00
客单价
  2026-11-01  100.00
  2026-11-02  200.00

第一天两单只付一单，支付率 50%，客单价 100；第二天一单一付，支付率 100%，客单价 200。

---

九、测试

后端是接口，测试不需要启动任何外部服务。下面是引擎的关键用例：

engine_test.go

func TestIndirectRatioAcrossTimeFields(t *testing.T) {
	backend := &MemoryBackend{Docs: []map[string]any{
		{"create_time": day(2026, 11, 1), "pay_time": day(2026, 11, 1), "status": "paid"},
		{"create_time": day(2026, 11, 1), "status": "cancel"},
		{"create_time": day(2026, 11, 2), "pay_time": day(2026, 11, 2), "status": "paid"},
	}}

	reg := NewRegistry()
	_ = reg.AddDirect(countIndicator{name: "下单量", time: "create_time"})
	_ = reg.AddDirect(countIndicator{
		name: "支付量", time: "pay_time",
		filter: Filter{{Field: "status", Op: Eq, Values: []any{"paid"}}},
	})
	_ = reg.AddIndirect(ratioIndicator{name: "支付率", deps: []string{"支付量", "下单量"}})

	res, err := NewEngine(reg, backend).Evaluate(context.Background(), Request{
		Start: day(2026, 11, 1), End: day(2026, 11, 3),
		Indicators: []string{"支付率"},
	})
	if err != nil {
		t.Fatal(err)
	}
	if got := res["支付率"]; len(got) != 2 || got[0].Value != 50 || got[1].Value != 100 {
		t.Fatalf("unexpected: %+v", got)
	}
}

完整测试覆盖：

用例	验证点
TestDirectCountWithFilter	计数指标按过滤条件生效
TestDirectSum	求和指标只累加匹配的金额
TestIndirectRatioAcrossTimeFields	分子分母时间字段不同也能正确相除
TestEmptyData	没有数据时返回空序列
TestZeroDenominator	分母为零返回 0
TestInvalidRequest	时间范围、粒度、指标为空都报错
TestResolveOrdersDependencies	间接指标排在被依赖者之后
TestResolveUnknown	未知指标报错
TestResolveCycle	依赖成环报错
TestFilterMatchNestedField	点号路径过滤
TestFilterValidateRejectsUnknownOp	非法操作符不放宽
TestAddDirectValidation	缺时间字段、重名都报错

$ go test ./...
ok  	github.com/chihqiang/metrics

---

十、离生产化还差什么

这个库是完整可运行的骨架，上生产还需要补几块：

查询后端。 内存后端只做演示，真实后端要把 Spec 翻译成搜索请求：全局条件、指标过滤、时间范围、日期直方图、聚合。

多时间字段。 真实查询里，不同时间字段要分开建桶，再在结果层合并；引擎已经按指标时间字段分开处理。

缓存。 只缓存纯聚合结果，缓存键包含指标、时间、过滤、粒度；并发相同请求要合并。

并发。 多个指标可以并行查询，但要控制并发度，避免压垮后端。

可观测性。 记录每次查询的指标数、时间字段数、聚合规模、耗时，慢查询要能定位到具体指标。

错误处理。 后端超时、部分指标失败、数据格式异常，都要有明确的降级策略，而不是整页报错。

---

十一、小结

一个指标操作库，核心就四件事：

指标定义 → 依赖解析 → 查询后端 → 结果合并与计算

把时间字段放进指标、把计算放到合并之后、把后端做成接口，口径就有了唯一的落点，测试也不再依赖真实集群。

代码不复杂，难的是约束：什么时候该拒绝、什么时候该返回零、什么时候必须报错。这些约束，才是统计平台“可信”的来源。