feat: 实现完整可观测性架构与火山v3适配器重构
重构整体架构： 1. 新增telemetry包实现零依赖的Prometheus指标系统 2. 新增metrics包集中管理业务埋点指标 3. 重构火山v3适配器，拆分client/request/response等模块 4. 替换旧的service/stats统计系统为标准指标埋点 5. 新增/metrics观测端点与完整仪表盘支持 功能更新： - 实现基于IP的限流与并发限制，添加指标埋点 - 重构TTS控制器，支持多格式输出与完整错误分类 - 更新.env.example配置示例，新增多项可选参数 - 替换旧的volcano适配器实现，支持完整的v3 API特性 - 清理冗余代码，移除service/stats与旧adapter实现
This commit is contained in:
@@ -0,0 +1,90 @@
|
||||
package telemetry
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"sort"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
// Counter 单调递增的累计指标(整数语义,内部用 float64 位以 atomic 操作)。
|
||||
type Counter struct {
|
||||
metricName string
|
||||
help string
|
||||
labelNames []string
|
||||
|
||||
mu sync.RWMutex
|
||||
values map[string]*counterChild // key = labelKey(...)
|
||||
}
|
||||
|
||||
type counterChild struct {
|
||||
labels Labels
|
||||
bits atomic.Uint64 // float64
|
||||
}
|
||||
|
||||
func newCounter(name, help string, labelNames []string) *Counter {
|
||||
return &Counter{
|
||||
metricName: name,
|
||||
help: help,
|
||||
labelNames: append([]string(nil), labelNames...),
|
||||
values: make(map[string]*counterChild),
|
||||
}
|
||||
}
|
||||
|
||||
// Inc 计数 +1。
|
||||
func (c *Counter) Inc(labels Labels) { c.Add(1, labels) }
|
||||
|
||||
// Add 累加 v(v 必须 >= 0)。
|
||||
func (c *Counter) Add(v float64, labels Labels) {
|
||||
if v < 0 {
|
||||
return
|
||||
}
|
||||
child := c.getOrCreate(labels)
|
||||
for {
|
||||
bits := child.bits.Load()
|
||||
cur := float64frombits(bits)
|
||||
next := float64bits(cur + v)
|
||||
if child.bits.CompareAndSwap(bits, next) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Counter) getOrCreate(labels Labels) *counterChild {
|
||||
key := labelKey(c.labelNames, labels)
|
||||
c.mu.RLock()
|
||||
if child, ok := c.values[key]; ok {
|
||||
c.mu.RUnlock()
|
||||
return child
|
||||
}
|
||||
c.mu.RUnlock()
|
||||
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if child, ok := c.values[key]; ok {
|
||||
return child
|
||||
}
|
||||
child := &counterChild{labels: copyLabels(labels, c.labelNames)}
|
||||
c.values[key] = child
|
||||
return child
|
||||
}
|
||||
|
||||
func (c *Counter) collect(w io.Writer) {
|
||||
fmt.Fprintf(w, "# HELP %s %s\n", c.metricName, c.help)
|
||||
fmt.Fprintf(w, "# TYPE %s counter\n", c.metricName)
|
||||
|
||||
c.mu.RLock()
|
||||
keys := make([]string, 0, len(c.values))
|
||||
for k := range c.values {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
defer c.mu.RUnlock()
|
||||
|
||||
for _, k := range keys {
|
||||
child := c.values[k]
|
||||
val := float64frombits(child.bits.Load())
|
||||
writeMetricLine(w, c.metricName, child.labels, val)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
package telemetry
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// copyLabels 返回只包含 labelNames 中声明的 key 的副本,缺失补空串。
|
||||
// 这样序列化时输出顺序和数量固定。
|
||||
func copyLabels(labels Labels, names []string) Labels {
|
||||
if len(names) == 0 {
|
||||
return Labels{}
|
||||
}
|
||||
out := make(Labels, len(names))
|
||||
for _, n := range names {
|
||||
out[n] = labels[n]
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func mergeLabels(a, b Labels) Labels {
|
||||
out := make(Labels, len(a)+len(b))
|
||||
for k, v := range a {
|
||||
out[k] = v
|
||||
}
|
||||
for k, v := range b {
|
||||
out[k] = v
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// formatLabels 序列化为 `{k1="v1",k2="v2"}`;空集合返回空字符串。
|
||||
// value 内的 `\`, `"`, 换行会按 Prometheus 规范转义。
|
||||
func formatLabels(labels Labels) string {
|
||||
if len(labels) == 0 {
|
||||
return ""
|
||||
}
|
||||
keys := sortedKeys(labels)
|
||||
var sb strings.Builder
|
||||
sb.WriteByte('{')
|
||||
for i, k := range keys {
|
||||
if i > 0 {
|
||||
sb.WriteByte(',')
|
||||
}
|
||||
sb.WriteString(k)
|
||||
sb.WriteString(`="`)
|
||||
sb.WriteString(escapeLabelValue(labels[k]))
|
||||
sb.WriteByte('"')
|
||||
}
|
||||
sb.WriteByte('}')
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
func escapeLabelValue(v string) string {
|
||||
if !strings.ContainsAny(v, "\\\"\n") {
|
||||
return v
|
||||
}
|
||||
var sb strings.Builder
|
||||
sb.Grow(len(v) + 2)
|
||||
for i := 0; i < len(v); i++ {
|
||||
switch v[i] {
|
||||
case '\\':
|
||||
sb.WriteString(`\\`)
|
||||
case '"':
|
||||
sb.WriteString(`\"`)
|
||||
case '\n':
|
||||
sb.WriteString(`\n`)
|
||||
default:
|
||||
sb.WriteByte(v[i])
|
||||
}
|
||||
}
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
func writeMetricLine(w io.Writer, name string, labels Labels, value float64) {
|
||||
fmt.Fprintf(w, "%s%s %s\n", name, formatLabels(labels), formatFloat(value))
|
||||
}
|
||||
|
||||
func formatFloat(f float64) string {
|
||||
return strconv.FormatFloat(f, 'g', -1, 64)
|
||||
}
|
||||
|
||||
// float64 bits 互转,封装到独立文件避免重复。
|
||||
func float64bits(f float64) uint64 { return math.Float64bits(f) }
|
||||
func float64frombits(b uint64) float64 { return math.Float64frombits(b) }
|
||||
@@ -0,0 +1,96 @@
|
||||
package telemetry
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"sort"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
// Gauge 可增可减的瞬时值。
|
||||
type Gauge struct {
|
||||
metricName string
|
||||
help string
|
||||
labelNames []string
|
||||
|
||||
mu sync.RWMutex
|
||||
values map[string]*gaugeChild
|
||||
}
|
||||
|
||||
type gaugeChild struct {
|
||||
labels Labels
|
||||
bits atomic.Uint64
|
||||
}
|
||||
|
||||
func newGauge(name, help string, labelNames []string) *Gauge {
|
||||
return &Gauge{
|
||||
metricName: name,
|
||||
help: help,
|
||||
labelNames: append([]string(nil), labelNames...),
|
||||
values: make(map[string]*gaugeChild),
|
||||
}
|
||||
}
|
||||
|
||||
// Set 直接设置当前值。
|
||||
func (g *Gauge) Set(v float64, labels Labels) {
|
||||
child := g.getOrCreate(labels)
|
||||
child.bits.Store(float64bits(v))
|
||||
}
|
||||
|
||||
// Inc +1。
|
||||
func (g *Gauge) Inc(labels Labels) { g.Add(1, labels) }
|
||||
|
||||
// Dec -1。
|
||||
func (g *Gauge) Dec(labels Labels) { g.Add(-1, labels) }
|
||||
|
||||
// Add 累加 v(可负)。
|
||||
func (g *Gauge) Add(v float64, labels Labels) {
|
||||
child := g.getOrCreate(labels)
|
||||
for {
|
||||
bits := child.bits.Load()
|
||||
cur := float64frombits(bits)
|
||||
next := float64bits(cur + v)
|
||||
if child.bits.CompareAndSwap(bits, next) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (g *Gauge) getOrCreate(labels Labels) *gaugeChild {
|
||||
key := labelKey(g.labelNames, labels)
|
||||
g.mu.RLock()
|
||||
if c, ok := g.values[key]; ok {
|
||||
g.mu.RUnlock()
|
||||
return c
|
||||
}
|
||||
g.mu.RUnlock()
|
||||
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if c, ok := g.values[key]; ok {
|
||||
return c
|
||||
}
|
||||
c := &gaugeChild{labels: copyLabels(labels, g.labelNames)}
|
||||
g.values[key] = c
|
||||
return c
|
||||
}
|
||||
|
||||
func (g *Gauge) collect(w io.Writer) {
|
||||
fmt.Fprintf(w, "# HELP %s %s\n", g.metricName, g.help)
|
||||
fmt.Fprintf(w, "# TYPE %s gauge\n", g.metricName)
|
||||
|
||||
g.mu.RLock()
|
||||
keys := make([]string, 0, len(g.values))
|
||||
for k := range g.values {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
defer g.mu.RUnlock()
|
||||
|
||||
for _, k := range keys {
|
||||
child := g.values[k]
|
||||
val := float64frombits(child.bits.Load())
|
||||
writeMetricLine(w, g.metricName, child.labels, val)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
package telemetry
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"sort"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
// DefaultLatencyBuckets 适合 HTTP/TTS 场景的默认桶(秒)。
|
||||
var DefaultLatencyBuckets = []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30}
|
||||
|
||||
// Histogram 累计分布型指标,记录观测值的分布。
|
||||
//
|
||||
// 内部为每个 child 维护:
|
||||
// - buckets[i] 累计计数(<= le_i 的观测数,不含 +Inf 桶)
|
||||
// - count 全部观测计数
|
||||
// - sum 全部观测值之和
|
||||
type Histogram struct {
|
||||
metricName string
|
||||
help string
|
||||
labelNames []string
|
||||
buckets []float64 // 用户声明的上界,不含 +Inf
|
||||
|
||||
mu sync.RWMutex
|
||||
values map[string]*histChild
|
||||
}
|
||||
|
||||
type histChild struct {
|
||||
labels Labels
|
||||
buckets []atomic.Uint64 // 累计计数
|
||||
count atomic.Uint64
|
||||
sumBits atomic.Uint64 // float64
|
||||
}
|
||||
|
||||
func newHistogram(name, help string, buckets []float64, labelNames []string) *Histogram {
|
||||
bs := append([]float64(nil), buckets...)
|
||||
sort.Float64s(bs)
|
||||
return &Histogram{
|
||||
metricName: name,
|
||||
help: help,
|
||||
labelNames: append([]string(nil), labelNames...),
|
||||
buckets: bs,
|
||||
values: make(map[string]*histChild),
|
||||
}
|
||||
}
|
||||
|
||||
// Observe 记录一个观测值。
|
||||
func (h *Histogram) Observe(v float64, labels Labels) {
|
||||
child := h.getOrCreate(labels)
|
||||
for {
|
||||
bits := child.sumBits.Load()
|
||||
cur := float64frombits(bits)
|
||||
next := float64bits(cur + v)
|
||||
if child.sumBits.CompareAndSwap(bits, next) {
|
||||
break
|
||||
}
|
||||
}
|
||||
child.count.Add(1)
|
||||
for i, le := range h.buckets {
|
||||
if v <= le {
|
||||
child.buckets[i].Add(1)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Histogram) getOrCreate(labels Labels) *histChild {
|
||||
key := labelKey(h.labelNames, labels)
|
||||
h.mu.RLock()
|
||||
if c, ok := h.values[key]; ok {
|
||||
h.mu.RUnlock()
|
||||
return c
|
||||
}
|
||||
h.mu.RUnlock()
|
||||
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if c, ok := h.values[key]; ok {
|
||||
return c
|
||||
}
|
||||
c := &histChild{
|
||||
labels: copyLabels(labels, h.labelNames),
|
||||
buckets: make([]atomic.Uint64, len(h.buckets)),
|
||||
}
|
||||
h.values[key] = c
|
||||
return c
|
||||
}
|
||||
|
||||
func (h *Histogram) collect(w io.Writer) {
|
||||
fmt.Fprintf(w, "# HELP %s %s\n", h.metricName, h.help)
|
||||
fmt.Fprintf(w, "# TYPE %s histogram\n", h.metricName)
|
||||
|
||||
h.mu.RLock()
|
||||
keys := make([]string, 0, len(h.values))
|
||||
for k := range h.values {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
defer h.mu.RUnlock()
|
||||
|
||||
for _, k := range keys {
|
||||
child := h.values[k]
|
||||
for i, le := range h.buckets {
|
||||
merged := mergeLabels(child.labels, Labels{"le": formatFloat(le)})
|
||||
fmt.Fprintf(w, "%s_bucket%s %d\n", h.metricName, formatLabels(merged), child.buckets[i].Load())
|
||||
}
|
||||
merged := mergeLabels(child.labels, Labels{"le": "+Inf"})
|
||||
fmt.Fprintf(w, "%s_bucket%s %d\n", h.metricName, formatLabels(merged), child.count.Load())
|
||||
sum := float64frombits(child.sumBits.Load())
|
||||
fmt.Fprintf(w, "%s_sum%s %s\n", h.metricName, formatLabels(child.labels), formatFloat(sum))
|
||||
fmt.Fprintf(w, "%s_count%s %d\n", h.metricName, formatLabels(child.labels), child.count.Load())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
// Package telemetry 提供进程内可观测能力:Counter / Gauge / Histogram,
|
||||
// 以及 Prometheus 文本格式导出。
|
||||
//
|
||||
// 设计原则:
|
||||
// - 零外部依赖,只使用标准库;
|
||||
// - label key 在指标注册时锁定,运行期不可新增(避免 cardinality 爆炸);
|
||||
// - 所有并发安全由实现保证,调用方无需加锁;
|
||||
// - Meter 是高层入口,NoopMeter 用于测试。
|
||||
package telemetry
|
||||
|
||||
import "sort"
|
||||
|
||||
// Labels 是指标附加的标签集合。Value 在序列化时会按 Prometheus 规范转义。
|
||||
type Labels map[string]string
|
||||
|
||||
// labelKey 计算一组标签的稳定 key,用于在内部 map 中唯一定位 child。
|
||||
// 缺失或多余的 label 一律视为空串,以保证 child 数量与 label 名集合一致。
|
||||
func labelKey(names []string, labels Labels) string {
|
||||
if len(names) == 0 {
|
||||
return ""
|
||||
}
|
||||
parts := make([]string, 0, len(names)*2)
|
||||
for _, n := range names {
|
||||
parts = append(parts, n, labels[n])
|
||||
}
|
||||
return joinLabelParts(parts)
|
||||
}
|
||||
|
||||
func joinLabelParts(parts []string) string {
|
||||
out := make([]byte, 0, 16*len(parts))
|
||||
for i, p := range parts {
|
||||
if i > 0 {
|
||||
out = append(out, 0)
|
||||
}
|
||||
out = append(out, p...)
|
||||
}
|
||||
return string(out)
|
||||
}
|
||||
|
||||
// sortedKeys 返回按字典序排列的 key,用于导出时输出稳定顺序。
|
||||
func sortedKeys(m map[string]string) []string {
|
||||
keys := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
return keys
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package telemetry
|
||||
|
||||
import "net/http"
|
||||
|
||||
// Meter 是 telemetry 的高层入口。所有业务埋点都通过它创建指标;
|
||||
// 测试时可替换为 NoopMeter 或带缓冲的自定义实现。
|
||||
type Meter struct {
|
||||
reg *Registry
|
||||
}
|
||||
|
||||
// NewMeter 创建默认实现。
|
||||
func NewMeter() *Meter {
|
||||
return &Meter{reg: newRegistry()}
|
||||
}
|
||||
|
||||
// Handler 返回 /metrics 端点的 http.Handler。
|
||||
func (m *Meter) Handler() http.Handler { return m.reg.Handler() }
|
||||
|
||||
// Registry 暴露给特殊用例(如测试断言),生产代码不应使用。
|
||||
func (m *Meter) Registry() *Registry { return m.reg }
|
||||
|
||||
// NewCounter 注册并返回一个 Counter。
|
||||
// - name 指标名(Prometheus 风格,如 "tts_request_total")
|
||||
// - help 帮助文本
|
||||
// - labelNames 注册时锁定的 label key 集合,运行期不可变
|
||||
func (m *Meter) NewCounter(name, help string, labelNames ...string) *Counter {
|
||||
c := newCounter(name, help, labelNames)
|
||||
if err := m.reg.register(name, c); err != nil {
|
||||
// 注册重名是启动期 bug,直接 panic 让问题在启动时暴露。
|
||||
panic(err)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
// NewGauge 同 NewCounter。
|
||||
func (m *Meter) NewGauge(name, help string, labelNames ...string) *Gauge {
|
||||
g := newGauge(name, help, labelNames)
|
||||
if err := m.reg.register(name, g); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return g
|
||||
}
|
||||
|
||||
// NewHistogram 同 NewCounter,额外接受桶上界(不含 +Inf,+Inf 由实现自动追加)。
|
||||
func (m *Meter) NewHistogram(name, help string, buckets []float64, labelNames ...string) *Histogram {
|
||||
h := newHistogram(name, help, buckets, labelNames)
|
||||
if err := m.reg.register(name, h); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
return h
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
package telemetry
|
||||
|
||||
import "net/http"
|
||||
|
||||
// NoopMeter 是一个不采集、不输出的 Meter,用于单元测试或禁用观测的场景。
|
||||
// 返回的 Counter / Gauge / Histogram 实例不会被注册到任何 Registry,
|
||||
// 它们的 Inc/Add/Observe 调用在本进程内没有可见效果(每次返回新的空实例)。
|
||||
type NoopMeter struct{}
|
||||
|
||||
func (NoopMeter) NewCounter(string, string, ...string) *Counter {
|
||||
return newCounter("", "", nil)
|
||||
}
|
||||
func (NoopMeter) NewGauge(string, string, ...string) *Gauge {
|
||||
return newGauge("", "", nil)
|
||||
}
|
||||
func (NoopMeter) NewHistogram(string, string, []float64, ...string) *Histogram {
|
||||
return newHistogram("", "", nil, nil)
|
||||
}
|
||||
func (NoopMeter) Handler() http.Handler { return http.NotFoundHandler() }
|
||||
func (NoopMeter) Registry() *Registry { return nil }
|
||||
@@ -0,0 +1,66 @@
|
||||
package telemetry
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"sort"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// collector 是 Counter / Gauge / Histogram 共同实现的内部接口。
|
||||
type collector interface {
|
||||
collect(w io.Writer)
|
||||
}
|
||||
|
||||
// Registry 持有已注册的全部指标,提供 Prometheus 文本格式导出。
|
||||
type Registry struct {
|
||||
mu sync.RWMutex
|
||||
entries map[string]collector
|
||||
order []string // 保留注册顺序,使输出可预测
|
||||
}
|
||||
|
||||
func newRegistry() *Registry {
|
||||
return &Registry{
|
||||
entries: make(map[string]collector),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Registry) register(name string, c collector) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if _, exists := r.entries[name]; exists {
|
||||
return fmt.Errorf("metric %q already registered", name)
|
||||
}
|
||||
r.entries[name] = c
|
||||
r.order = append(r.order, name)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Gather 把所有指标按注册顺序写入 w,文本格式遵循 Prometheus 0.0.4。
|
||||
func (r *Registry) Gather(w io.Writer) error {
|
||||
r.mu.RLock()
|
||||
order := append([]string(nil), r.order...)
|
||||
defer r.mu.RUnlock()
|
||||
for _, name := range order {
|
||||
r.entries[name].collect(w)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Handler 返回标准 Prometheus 抓取端点。
|
||||
func (r *Registry) Handler() http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
|
||||
_ = r.Gather(w)
|
||||
})
|
||||
}
|
||||
|
||||
// 注册顺序的辅助,用于测试断言。
|
||||
func (r *Registry) names() []string {
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
out := append([]string(nil), r.order...)
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
Reference in New Issue
Block a user