storage

package
v0.0.0-...-12d1a89 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 4, 2026 License: MIT Imports: 21 Imported by: 0

Documentation

Overview

Package storage 为 tracing 采集记录提供本地分段存储、查询和健康投影。

连接采集循环

Engine 实现 tracing.Handler。通常由 Run 同时运行 tracing 的批量采集循环和本存储配置的 自动清理： 

engine, err := storage.Open(ctx, "data/tracing")
if err != nil {
	return err
}
defer engine.Close()

go func() {
	if err := engine.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
		reportError(err)
	}
}()

业务代码只依赖 tracing 并产生 Trace、Span、Event 和 Log，不需要依赖 storage。需要自行组合 消费者时，也可以把 Engine 直接传给 tracing.RunHandler。

打开与写入

Open 接收存储目录。一个目录同一时刻只能由一个 Engine 打开；Engine 的公开方法可以由多个 goroutine 并发调用： 

engine, err := storage.Open(
	ctx,
	"data/tracing",
	storage.WithCapacity(2<<30, 0.90, 0.75),
)
if err != nil {
	return err
}
defer engine.Close()

_, err = engine.Store(ctx, storage.Batch{
	Traces: []storage.Trace{{
		ID:        traceID,
		CreatedAt: time.Now(),
	}},
})

一次 Batch 被编码为一个带 CRC32C 校验的追加帧。Store 会先校验记录、父级和 Span 树，再写入 完整帧，因此同一个 Batch 内的 Trace、Span、Event、Log 和 SpanEnd 具有共同的追加边界。 同一类记录按 ID 使用“第一次成功写入生效”语义；在原始记录或对应清理标记仍保留期间， 重复写入不会重复更新健康统计。健康快照只保存自然日聚合和有限排查样本，不保存逐 Log ID 集合，因此其大小由服务数、保留天数和样本上限决定，而不是由全部 Log 数量决定。

索引与恢复

原始记录保存在只追加的分段文件中；Engine 使用 google/btree 和 ID 映射维护内存索引，查询 负载按记录引用从分段读取。MANIFEST 和健康快照通过临时文件、fsync 和 rename 原子替换。 Open 会重放 MANIFEST 中的分段；活动分段末尾未写完或校验失败的最后一帧会被截断，已封存 分段中的损坏则返回 ErrCorrupt，避免静默接受中间数据损坏。

默认 Store 不为每个 Batch 执行 fsync，以减少高频观测写入对业务 I/O 的影响。进程正常退出时 Close 会同步活动分段和健康快照；需要 Store 返回即具备更强落盘保证时，使用 WithSyncWrites(true)。该保证仍不等同于跨机器复制或业务事务。

容量与清理

Engine 同时受时间和物理容量约束。默认策略如下：

  • 原始 Trace 超过 7 天后整条删除；
  • 未达到高水位时，7 天内的原始数据不因级别而提前清理；
  • 达到高水位后，超过 3、4、5、6 天的 Debug、Info、Warn、Error Log 依次具备清理资格；
  • Critical Log 在原始 Trace 保留期内不按级别提前清理；
  • 超过硬容量时，先清理较旧 Log，仍不足时才删除最旧 Trace；
  • 服务健康投影默认独立保留 180 天，原始父记录删除后样本仍可查询，并返回 ParentsAvailable=false。

Cleanup 生成有界逻辑删除记录，查询结果立即反映清理。达到容量高水位或失效记录占比达到 阈值后，Engine 才把仍可查询的原始记录压缩到新分段，避免少量过期记录触发整库重写；也可 调用 Compact 主动压缩。替换 MANIFEST 前会先持久化健康快照，因此压缩不会依赖已被替换的 原始 Log 重新计算健康统计。WithRetention 的 health=0 表示不自动清理健康投影。

使用边界

Engine 用于允许过载丢失、按容量退化的非业务观测数据。它不提供外键、跨进程并发写、远程 复制、分布式一致性、备份恢复或业务事务语义。应用应在 Handler 停止并排空后调用 Close，且 不应直接修改存储目录中的文件。Observe 只发布成功提交后的轻量 ID 和健康自然日变化，供 查询适配层失效缓存或生成增量；它不保存 HTTP 连接、展示 DTO 或前端状态。

Index

Constants

View Source 
const (
	DefaultSegmentBytes     int64 = 64 << 20
	DefaultMaxBytes         int64 = 512 << 20
	DefaultHighWatermark          = 0.90
	DefaultLowWatermark           = 0.75
	DefaultRawRetention           = 7 * 24 * time.Hour
	DefaultHealthRetention        = 180 * 24 * time.Hour
	DefaultCleanupInterval        = time.Hour
	DefaultCleanupBatchSize       = 5000
	DefaultQueryLimit             = 100
	MaxQueryLimit                 = 500
)

Variables

View Source 
var (
	// ErrClosed 表示 Engine 已关闭。
	ErrClosed = errors.New("tracing storage: closed")
	// ErrLocked 表示存储目录正被另一个 Engine 实例占用。
	ErrLocked = errors.New("tracing storage: directory locked")
	// ErrCorrupt 表示持久化文件损坏且无法安全恢复。
	ErrCorrupt = errors.New("tracing storage: corrupt data")
	// ErrInvalidRecord 表示 Batch 包含缺少标识、时间或父记录的记录。
	ErrInvalidRecord = errors.New("tracing storage: invalid record")
	// ErrNotFound 表示指定记录不存在或已经被清理。
	ErrNotFound = errors.New("tracing storage: record not found")
	// ErrInvalidCursor 表示查询游标无效或不属于当前查询。
	ErrInvalidCursor = errors.New("tracing storage: invalid cursor")
	// ErrInvalidQuery 表示查询参数无效。
	ErrInvalidQuery = errors.New("tracing storage: invalid query")
)

Functions

This section is empty.

Types

type Attr 

type Attr struct {
	Key   string          `json:"k"`
	Kind  string          `json:"t"`
	Value json.RawMessage `json:"v"`
}

Attr 是已经完成类型校验的属性编码。

Value 必须保存一个合法 JSON 值。storage 不解释 Kind；上层负责按 Kind 恢复具体 Go 类型。

type Batch 

type Batch struct {
	Traces   []Trace   `json:"traces,omitempty"`
	Spans    []Span    `json:"spans,omitempty"`
	Events   []Event   `json:"events,omitempty"`
	Logs     []Log     `json:"logs,omitempty"`
	SpanEnds []SpanEnd `json:"span_ends,omitempty"`
}

Batch 是一次原子追加的记录集合。

func (Batch) Empty 

func (batch Batch) Empty() bool

Empty 报告 Batch 是否不含记录。

type Changes 

type Changes struct {
	TraceIDs []string `json:"trace_ids,omitempty"`
	SpanIDs  []string `json:"span_ids,omitempty"`
	EventIDs []string `json:"event_ids,omitempty"`
	LogIDs   []string `json:"log_ids,omitempty"`
	Services []string `json:"services,omitempty"`
}

Changes 描述一次 Store 实际新增或修改的记录。

func (Changes) Empty 

func (changes Changes) Empty() bool

Empty 报告本次写入是否没有改变可查询状态。

type CleanupResult 

type CleanupResult struct {
	PrunedLogs       int      `json:"pruned_logs"`
	RemovedTraces    int      `json:"removed_traces"`
	RemovedSpans     int      `json:"removed_spans"`
	RemovedEvents    int      `json:"removed_events"`
	RemovedLogs      int      `json:"removed_logs"`
	RemovedHealthDay int      `json:"removed_health_days"`
	CompactedBytes   int64    `json:"compacted_bytes"`
	BeforeBytes      int64    `json:"before_bytes"`
	AfterBytes       int64    `json:"after_bytes"`
	TraceIDs         []string `json:"trace_ids,omitempty"`
	HasMore          bool     `json:"has_more"`
}

CleanupResult 描述一次清理和压缩结果。

func (CleanupResult) Changed 

func (result CleanupResult) Changed() bool

Changed 报告清理是否改变了可查询数据或物理文件。

type Engine 

type Engine struct {
	// contains filtered or unexported fields
}

Engine 是单目录、单进程写入的 tracing 分段存储引擎。

Engine 的公开方法可以由多个 goroutine 并发调用。一个目录同一时刻只能被一个 Engine 打开；Close 会同步活动分段和健康快照并释放目录锁。

func Open 

func Open(ctx context.Context, path string, opts ...Option) (*Engine, error)

Open 打开或创建一个存储目录，并恢复完整追加帧。

func (*Engine) Cleanup 

func (engine *Engine) Cleanup(ctx context.Context) (CleanupResult, error)

Cleanup 按当前时间执行一次有界分级清理，并按需回收失效分段空间。

func (*Engine) CleanupAt 

func (engine *Engine) CleanupAt(ctx context.Context, now time.Time) (CleanupResult, error)

CleanupAt 按指定时间执行清理。该方法主要用于确定性的维护任务和测试。

func (*Engine) Close 

func (engine *Engine) Close() error

Close 同步并关闭全部分段。重复调用返回 ErrClosed。

func (*Engine) Compact 

func (engine *Engine) Compact(ctx context.Context) error

Compact 强制把当前可查询原始记录重写为新的分段集合。

func (*Engine) Handle 

func (engine *Engine) Handle(ctx context.Context, batch tracing.Batch) error

Handle 将 tracing 采集批次转换为存储记录并原子追加。

func (*Engine) NeedsCleanup 

func (engine *Engine) NeedsCleanup() bool

NeedsCleanup 报告物理大小是否达到高水位。

func (*Engine) Observe 

func (engine *Engine) Observe(observer Observer) (
	revision uint64,
	unsubscribe func(),
	err error,
)

Observe 注册一个存储变化观察者。

revision 是注册完成时的最新修订号；unsubscribe 可安全重复调用。observer 不能为 nil。

func (*Engine) QueryEvent 

func (engine *Engine) QueryEvent(ctx context.Context, eventID string) (*Event, error)

QueryEvent 按 Event ID 返回仍保留的 Event。

func (*Engine) QueryServiceSamples 

func (engine *Engine) QueryServiceSamples(
	ctx context.Context,
	service string,
	day string,
) ([]ServiceSample, error)

QueryServiceSamples 返回服务指定自然日内的有限样本。

func (*Engine) QueryServices 

func (engine *Engine) QueryServices(ctx context.Context, query ServiceQuery) ([]ServiceView, error)

QueryServices 返回服务健康投影。原始 Trace 清理不会减少历史计数。

func (*Engine) QuerySpans 

func (engine *Engine) QuerySpans(
	ctx context.Context,
	traceID string,
	query SpanQuery,
) (SpanPage, error)

QuerySpans 返回一条 Trace 的 Span，顺序为 started_at、span_id 正序。

func (*Engine) QueryTimeline 

func (engine *Engine) QueryTimeline(
	ctx context.Context,
	traceID string,
	query TimelineQuery,
) (TimelinePage, error)

QueryTimeline 返回一条 Trace 的 Event 和当前仍保留的 Log，顺序为时间倒序。

func (*Engine) QueryTrace 

func (engine *Engine) QueryTrace(ctx context.Context, traceID string) (*TraceView, error)

QueryTrace 返回一条仍保留原始数据的 Trace。

func (*Engine) QueryTraces 

func (engine *Engine) QueryTraces(ctx context.Context, query TraceQuery) (TracePage, error)

QueryTraces 返回按 last seen、Trace ID 倒序排列的一页。

func (*Engine) Run 

func (engine *Engine) Run(ctx context.Context) error

Run 运行 tracing 采集循环，并按配置执行自动清理。

同一进程同一时间只能运行一个 tracing 采集循环。ctx 取消后，Run 会等待采集队列尽量排空 并停止自动清理，然后返回 ctx.Err()。

func (*Engine) RunWithReady 

func (engine *Engine) RunWithReady(ctx context.Context, ready chan<- struct{}) error

RunWithReady 与 Engine.Run 相同，但会在 tracing 采集循环开始接收记录后通知 ready。

ready 不能为 nil。该方法主要供应用生命周期编排使用；普通调用方应继续使用 Run。

func (*Engine) Stats 

func (engine *Engine) Stats() (Stats, error)

Stats 返回当前索引和分段大小，不读取记录负载。

func (*Engine) Store 

func (engine *Engine) Store(ctx context.Context, batch Batch) (Changes, error)

Store 原子追加 Batch 中尚未出现的记录。

同一 ID 第一次成功写入的值生效，后续重复记录被忽略。Event 和 Log 的 Trace、Span 必须 已存在或包含在同一 Batch 中。Store 返回前不会修改调用方提供的切片。

func (*Engine) Sync 

func (engine *Engine) Sync(ctx context.Context) error

Sync 将活动分段和健康快照同步到稳定存储。

type Event 

type Event struct {
	ID        string    `json:"id"`
	TraceID   string    `json:"trace_id"`
	SpanID    string    `json:"span_id"`
	Name      string    `json:"name"`
	Timestamp time.Time `json:"timestamp"`
	Attrs     []Attr    `json:"attrs,omitempty"`
}

Event 是 Span 内发生的一项结构化事实。

type HealthDay 

type HealthDay struct {
	Service string
	Day     string
}

HealthDay 标识一次提交改变的服务健康自然日。

type Level 

type Level uint8

Level 表示 Log 的严重程度。数值顺序同时是严重程度顺序。 

const (
	Debug Level = iota
	Info
	Warn
	Error
	Critical
)

func (Level) String 

func (level Level) String() string

String 返回 Level 的稳定小写文本。

func (Level) Valid 

func (level Level) Valid() bool

Valid 报告 Level 是否属于已定义范围。

type LevelCounts 

type LevelCounts [levelCount]uint64

LevelCounts 保存五种级别的计数。

func (LevelCounts) Get 

func (counts LevelCounts) Get(level Level) uint64

Get 返回指定级别的计数；无效级别返回 0。

func (LevelCounts) Total 

func (counts LevelCounts) Total() uint64

Total 返回全部级别计数之和。

type Log 

type Log struct {
	ID        string    `json:"id"`
	TraceID   string    `json:"trace_id"`
	SpanID    string    `json:"span_id"`
	Service   string    `json:"service,omitempty"`
	Level     Level     `json:"level"`
	Timestamp time.Time `json:"timestamp"`
	What      string    `json:"what"`
	Detail    string    `json:"detail"`
	Attrs     []Attr    `json:"attrs,omitempty"`
}

Log 是 Span 内的一条业务观测记录。

type Observer 

type Observer func(Update)

Observer 接收 Engine 成功提交后的轻量变化通知。

Observer 在 Store、Cleanup 或 Close 的调用 goroutine 中执行，必须快速返回，不能执行 I/O 或长时间阻塞。需要查询或聚合时，应只把通知转交给自己的队列。

type Option 

type Option func(*options)

Option 配置 Engine。

func WithCapacity 

func WithCapacity(maxBytes int64, highWatermark, lowWatermark float64) Option

WithCapacity 设置物理容量、高水位和压缩目标水位。

func WithCleanup 

func WithCleanup(interval time.Duration, batchSize int) Option

WithCleanup 设置 Run 的自动清理间隔和单次处理上限。

interval 小于等于 0 时关闭自动清理。batchSize 必须大于 0。

func WithCleanupBatchSize 

func WithCleanupBatchSize(size int) Option

WithCleanupBatchSize 设置一次 Cleanup 最多生成的清理操作数。

func WithLevelAges 

func WithLevelAges(debug, info, warn, errorAge time.Duration) Option

WithLevelAges 设置各级 Log 进入可清理状态的年龄。

四个值必须非递减且不能超过原始 Trace 保留时间。Critical 跟随原始 Trace 保留时间， 不会按级别提前清理。最终约束在全部 Option 应用后统一校验，因此 Option 顺序不影响结果。

func WithLocation 

func WithLocation(location *time.Location) Option

WithLocation 设置健康自然日使用的时区。

func WithRetention 

func WithRetention(raw, health time.Duration) Option

WithRetention 设置原始 Trace 和健康投影的保留时间。

raw 必须大于 0。health 小于 0 时无效，等于 0 时不自动清理健康投影。

func WithSegmentBytes 

func WithSegmentBytes(size int64) Option

WithSegmentBytes 设置单个活动分段的目标大小。

func WithSyncWrites 

func WithSyncWrites(enabled bool) Option

WithSyncWrites 控制 Store 返回前是否对活动分段执行 fsync。

type ServiceCurrent 

type ServiceCurrent struct {
	Service     string        `json:"service"`
	Status      ServiceStatus `json:"status"`
	FirstSeenAt time.Time     `json:"first_seen_at"`
	LastSeenAt  time.Time     `json:"last_seen_at"`
	LastLogID   string        `json:"last_log_id"`
	LastTraceID string        `json:"last_trace_id"`
	LastSpanID  string        `json:"last_span_id"`
	LastLevel   Level         `json:"last_level"`
	LastWhat    string        `json:"last_what"`
	LastDetail  string        `json:"last_detail"`
	UpdatedAt   time.Time     `json:"updated_at"`
}

ServiceCurrent 是服务最近一条 Log 派生的状态。

type ServiceDay 

type ServiceDay struct {
	Day         string         `json:"day"`
	StartedAt   time.Time      `json:"started_at"`
	EndedAt     time.Time      `json:"ended_at"`
	Status      ServiceStatus  `json:"status"`
	Counts      LevelCounts    `json:"counts"`
	FirstSeenAt *time.Time     `json:"first_seen_at,omitempty"`
	LastSeenAt  *time.Time     `json:"last_seen_at,omitempty"`
	LastLogID   string         `json:"last_log_id,omitempty"`
	LastLevel   *Level         `json:"last_level,omitempty"`
	Preview     *ServiceSample `json:"preview,omitempty"`
}

ServiceDay 是一个服务自然日内的精确级别计数。

func (ServiceDay) LogCount 

func (day ServiceDay) LogCount() uint64

LogCount 返回该日全部 Log 数量。

type ServiceQuery 

type ServiceQuery struct {
	Services []string
	Days     int
	After    string
	Limit    int
}

ServiceQuery 描述服务健康查询。

type ServiceSample 

type ServiceSample struct {
	LogID            string    `json:"log_id"`
	Service          string    `json:"service"`
	Day              string    `json:"day"`
	Level            Level     `json:"level"`
	TraceID          string    `json:"trace_id"`
	SpanID           string    `json:"span_id"`
	What             string    `json:"what"`
	Detail           string    `json:"detail"`
	Attrs            []Attr    `json:"attrs,omitempty"`
	SeenAt           time.Time `json:"seen_at"`
	ParentsAvailable bool      `json:"parents_available"`
}

ServiceSample 是独立于原始 Trace 保留的一条排查样本。

type ServiceStatus 

type ServiceStatus string

ServiceStatus 是服务健康展示状态。 

const (
	ServiceUnknown  ServiceStatus = "unknown"
	ServiceOK       ServiceStatus = "ok"
	ServiceWarn     ServiceStatus = "warn"
	ServiceError    ServiceStatus = "error"
	ServiceCritical ServiceStatus = "critical"
)

type ServiceView 

type ServiceView struct {
	Current         ServiceCurrent `json:"current"`
	Availability    float64        `json:"availability"`
	ErrorRate       float64        `json:"error_rate"`
	HistoryLogCount uint64         `json:"history_log_count"`
	History         []ServiceDay   `json:"history"`
}

ServiceView 是服务当前状态和连续自然日历史。

type Span 

type Span struct {
	ID        string     `json:"id"`
	TraceID   string     `json:"trace_id"`
	ParentID  string     `json:"parent_id,omitempty"`
	Name      string     `json:"name"`
	StartedAt time.Time  `json:"started_at"`
	EndedAt   *time.Time `json:"ended_at,omitempty"`
	Attrs     []Attr     `json:"attrs,omitempty"`
}

Span 是 Span 创建时的不可变记录。EndedAt 由 SpanEnd 首次补齐。

type SpanEnd 

type SpanEnd struct {
	TraceID string    `json:"trace_id"`
	SpanID  string    `json:"span_id"`
	EndedAt time.Time `json:"ended_at"`
}

SpanEnd 记录 Span 第一次结束的时间。

type SpanPage 

type SpanPage struct {
	Spans      []Span `json:"spans"`
	HasMore    bool   `json:"has_more"`
	NextCursor string `json:"next_cursor,omitempty"`
}

SpanPage 是按 started_at、span_id 正序排列的 Span 页。

type SpanQuery 

type SpanQuery struct {
	Cursor string
	Limit  int
}

SpanQuery 描述一条 Trace 的 Span 游标查询。

type Stats 

type Stats struct {
	PhysicalBytes int64 `json:"physical_bytes"`
	LiveBytes     int64 `json:"live_bytes"`
	DeadBytes     int64 `json:"dead_bytes"`
	Segments      int   `json:"segments"`
	Traces        int   `json:"traces"`
	Spans         int   `json:"spans"`
	Events        int   `json:"events"`
	Logs          int   `json:"logs"`
	Services      int   `json:"services"`
}

Stats 是 Engine 的轻量运行状态。

type TimelineItem 

type TimelineItem struct {
	Kind  TimelineKind `json:"kind"`
	Event *Event       `json:"event,omitempty"`
	Log   *Log         `json:"log,omitempty"`
}

TimelineItem 是统一时间线中的一项。

type TimelineKind 

type TimelineKind uint8

TimelineKind 区分 Event 与 Log。 

const (
	TimelineEvent TimelineKind = iota + 1
	TimelineLog
)

type TimelinePage 

type TimelinePage struct {
	Items      []TimelineItem `json:"items"`
	HasMore    bool           `json:"has_more"`
	NextCursor string         `json:"next_cursor,omitempty"`
}

TimelinePage 是时间线游标页。

type TimelineQuery 

type TimelineQuery struct {
	Cursor string
	Limit  int
	SpanID string
}

TimelineQuery 描述按时间倒序的 Event/Log 查询。

type Trace 

type Trace struct {
	ID             string    `json:"id"`
	TriggerEventID string    `json:"trigger_event_id,omitempty"`
	CreatedAt      time.Time `json:"created_at"`
	Attrs          []Attr    `json:"attrs,omitempty"`
}

Trace 是 Trace 创建时的不可变记录。

type TracePage 

type TracePage struct {
	Traces     []TraceSummary `json:"traces"`
	HasMore    bool           `json:"has_more"`
	NextCursor string         `json:"next_cursor,omitempty"`
}

TracePage 是 Trace 游标页。

type TraceQuery 

type TraceQuery struct {
	Cursor         string
	Limit          int
	From           time.Time
	To             time.Time
	MinLevel       *Level
	TriggerEventID string
	Service        string
	Search         string
}

TraceQuery 描述按 last seen 倒序的 Trace 游标查询。

type TraceSummary 

type TraceSummary struct {
	TraceID          string      `json:"trace_id"`
	TriggerEventID   string      `json:"trigger_event_id,omitempty"`
	CreatedAt        time.Time   `json:"created_at"`
	FirstSeenAt      time.Time   `json:"first_seen_at"`
	LastSeenAt       time.Time   `json:"last_seen_at"`
	HighestLevel     *Level      `json:"highest_level,omitempty"`
	SpanCount        int         `json:"span_count"`
	EventCount       int         `json:"event_count"`
	LogCount         int         `json:"log_count"`
	ObservedLogCount int         `json:"observed_log_count"`
	ObservedLevels   LevelCounts `json:"observed_levels"`
	StoredLevels     LevelCounts `json:"stored_levels"`
	RootSpanName     string      `json:"root_span_name,omitempty"`
	Pruned           bool        `json:"pruned"`
}

TraceSummary 是不读取属性负载即可返回的 Trace 列表项。

type TraceView 

type TraceView struct {
	TraceSummary
	Attrs []Attr `json:"attrs,omitempty"`
}

TraceView 是带属性的单条 Trace。

type Update 

type Update struct {
	Revision       uint64
	Changes        Changes
	HealthDays     []HealthDay
	ReloadTraces   bool
	ReloadServices bool
	Closed         bool
}

Update 描述 Engine 已经提交的可查询状态变化。

Update 只携带记录标识和健康自然日，不包含 HTTP 展示模型或连接状态。ReloadTraces 和 ReloadServices 表示清理改变了已有数据，读取方应重新查询对应快照。Closed 表示 Engine 已关闭。

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.
⌂ 换网址