rogee/atomctl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 增强 OpenTelemetry 提供者配置，添加连接安全、采样器和批处理选项

Browse Source
This commit is contained in:
Rogee
2025-09-11 18:23:14 +08:00
parent 9662d7d718
commit 9e7ce1b7d3
4 changed files with 500 additions and 262 deletions
Download Patch File Download Diff File Expand all files Collapse all files

453
templates/project/providers/grpc/options.md.raw
View File

@@ -12,8 +12,8 @@
> >
> ```go > ```go
> import ( > import (
> pgrpc "test/providers/grpc" // 本包 > pgrpc "test/providers/grpc" // 本包
> grpc "google.golang.org/grpc" // 避免命名冲突 > grpc "google.golang.org/grpc" // 避免命名冲突
> ) > )
> ``` > ```
@@ -23,8 +23,8 @@
```go ```go
pgrpc.UseOptions( pgrpc.UseOptions(
grpc.MaxRecvMsgSize(32<<20), // 32 MiB grpc.MaxRecvMsgSize(32<<20), // 32 MiB
grpc.MaxSendMsgSize(32<<20), // 32 MiB grpc.MaxSendMsgSize(32<<20), // 32 MiB
) )
``` ```
@@ -32,7 +32,7 @@ pgrpc.UseOptions(
```go ```go
pgrpc.UseOptions( pgrpc.UseOptions(
grpc.MaxConcurrentStreams(1024), grpc.MaxConcurrentStreams(1024),
) )
``` ```
@@ -40,22 +40,22 @@ Keepalive 参数（需要 keepalive 包）：
```go ```go
import ( import (
"time" "time"
"google.golang.org/grpc/keepalive" "google.golang.org/grpc/keepalive"
) )
pgrpc.UseOptions( pgrpc.UseOptions(
grpc.KeepaliveParams(keepalive.ServerParameters{ grpc.KeepaliveParams(keepalive.ServerParameters{
MaxConnectionIdle: 5 * time.Minute, MaxConnectionIdle: 5 * time.Minute,
MaxConnectionAge: 30 * time.Minute, MaxConnectionAge: 30 * time.Minute,
MaxConnectionAgeGrace: 5 * time.Minute, MaxConnectionAgeGrace: 5 * time.Minute,
Time: 2 * time.Minute, // ping 间隔 Time: 2 * time.Minute, // ping 间隔
Timeout: 20 * time.Second, Timeout: 20 * time.Second,
}), }),
grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{ grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{
MinTime: 1 * time.Minute, // 客户端 ping 最小间隔 MinTime: 1 * time.Minute, // 客户端 ping 最小间隔
PermitWithoutStream: true, PermitWithoutStream: true,
}), }),
) )
``` ```
@@ -65,31 +65,31 @@ pgrpc.UseOptions(
```go ```go
import ( import (
"context" "context"
"time" "time"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"google.golang.org/grpc" "google.golang.org/grpc"
) )
func LoggingUnaryInterceptor( func LoggingUnaryInterceptor(
ctx context.Context, ctx context.Context,
req any, req any,
info *grpc.UnaryServerInfo, info *grpc.UnaryServerInfo,
handler grpc.UnaryHandler, handler grpc.UnaryHandler,
) (any, error) { ) (any, error) {
start := time.Now() start := time.Now()
resp, err := handler(ctx, req) resp, err := handler(ctx, req)
dur := time.Since(start) dur := time.Since(start)
entry := log.WithFields(log.Fields{ entry := log.WithFields(log.Fields{
"grpc.method": info.FullMethod, "grpc.method": info.FullMethod,
"grpc.duration_ms": dur.Milliseconds(), "grpc.duration_ms": dur.Milliseconds(),
}) })
if err != nil { if err != nil {
entry.WithError(err).Warn("grpc unary request failed") entry.WithError(err).Warn("grpc unary request failed")
} else { } else {
entry.Info("grpc unary request finished") entry.Info("grpc unary request finished")
} }
return resp, err return resp, err
} }
// 注册 // 注册
@@ -100,46 +100,46 @@ pgrpc.UseUnaryInterceptors(LoggingUnaryInterceptor)
```go ```go
import ( import (
"context" "context"
"fmt" "fmt"
"runtime/debug" "runtime/debug"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"google.golang.org/grpc" "google.golang.org/grpc"
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
) )
func RecoveryUnaryInterceptor( func RecoveryUnaryInterceptor(
ctx context.Context, ctx context.Context,
req any, req any,
info *grpc.UnaryServerInfo, info *grpc.UnaryServerInfo,
handler grpc.UnaryHandler, handler grpc.UnaryHandler,
) (any, error) { ) (any, error) {
defer func() { defer func() {
if r := recover(); r != nil { if r := recover() ; r != nil {
log.WithField("grpc.method", info.FullMethod).Errorf("panic: %v\n%s", r, debug.Stack()) log.WithField("grpc.method", info.FullMethod).Errorf("panic: %v\n%s", r, debug.Stack())
} }
}() }()
return handler(ctx, req) return handler(ctx, req)
} }
// 或者向客户端返回内部错误： // 或者向客户端返回内部错误：
func RecoveryUnaryInterceptorWithError( func RecoveryUnaryInterceptorWithError(
ctx context.Context, ctx context.Context,
req any, req any,
info *grpc.UnaryServerInfo, info *grpc.UnaryServerInfo,
handler grpc.UnaryHandler, handler grpc.UnaryHandler,
) (any, error) { ) (any, error) {
defer func() { defer func() {
if r := recover(); r != nil { if r := recover() ; r != nil {
log.WithField("grpc.method", info.FullMethod).Errorf("panic: %v\n%s", r, debug.Stack()) log.WithField("grpc.method", info.FullMethod).Errorf("panic: %v\n%s", r, debug.Stack())
} }
}() }()
resp, err := handler(ctx, req) resp, err := handler(ctx, req)
if rec := recover(); rec != nil { if rec := recover() ; rec != nil {
return nil, status.Error(codes.Internal, fmt.Sprint(rec)) return nil, status.Error(codes.Internal, fmt.Sprint(rec))
} }
return resp, err return resp, err
} }
pgrpc.UseUnaryInterceptors(RecoveryUnaryInterceptor) pgrpc.UseUnaryInterceptors(RecoveryUnaryInterceptor)
@@ -157,32 +157,32 @@ pgrpc.UseUnaryInterceptors(LoggingUnaryInterceptor, RecoveryUnaryInterceptor)
```go ```go
import ( import (
"time" "time"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"google.golang.org/grpc" "google.golang.org/grpc"
) )
func LoggingStreamInterceptor( func LoggingStreamInterceptor(
srv any, srv any,
ss grpc.ServerStream, ss grpc.ServerStream,
info *grpc.StreamServerInfo, info *grpc.StreamServerInfo,
handler grpc.StreamHandler, handler grpc.StreamHandler,
) error { ) error {
start := time.Now() start := time.Now()
err := handler(srv, ss) err := handler(srv, ss)
dur := time.Since(start) dur := time.Since(start)
entry := log.WithFields(log.Fields{ entry := log.WithFields(log.Fields{
"grpc.method": info.FullMethod, "grpc.method": info.FullMethod,
"grpc.is_client_stream": info.IsClientStream, "grpc.is_client_stream": info.IsClientStream,
"grpc.is_server_stream": info.IsServerStream, "grpc.is_server_stream": info.IsServerStream,
"grpc.duration_ms": dur.Milliseconds(), "grpc.duration_ms": dur.Milliseconds(),
}) })
if err != nil { if err != nil {
entry.WithError(err).Warn("grpc stream request failed") entry.WithError(err).Warn("grpc stream request failed")
} else { } else {
entry.Info("grpc stream request finished") entry.Info("grpc stream request finished")
} }
return err return err
} }
pgrpc.UseStreamInterceptors(LoggingStreamInterceptor) pgrpc.UseStreamInterceptors(LoggingStreamInterceptor)
@@ -192,23 +192,23 @@ pgrpc.UseStreamInterceptors(LoggingStreamInterceptor)
```go ```go
import ( import (
"runtime/debug" "runtime/debug"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"google.golang.org/grpc" "google.golang.org/grpc"
) )
func RecoveryStreamInterceptor( func RecoveryStreamInterceptor(
srv any, srv any,
ss grpc.ServerStream, ss grpc.ServerStream,
info *grpc.StreamServerInfo, info *grpc.StreamServerInfo,
handler grpc.StreamHandler, handler grpc.StreamHandler,
) (err error) { ) (err error) {
defer func() { defer func() {
if r := recover(); r != nil { if r := recover() ; r != nil {
log.WithField("grpc.method", info.FullMethod).Errorf("panic: %v\n%s", r, debug.Stack()) log.WithField("grpc.method", info.FullMethod).Errorf("panic: %v\n%s", r, debug.Stack())
} }
}() }()
return handler(srv, ss) return handler(srv, ss)
} }
pgrpc.UseStreamInterceptors(RecoveryStreamInterceptor) pgrpc.UseStreamInterceptors(RecoveryStreamInterceptor)
@@ -219,8 +219,8 @@ pgrpc.UseStreamInterceptors(RecoveryStreamInterceptor)
- 可以多次调用 `UseOptions/UseUnaryInterceptors/UseStreamInterceptors`，最终会在服务构造时链式生效。 - 可以多次调用 `UseOptions/UseUnaryInterceptors/UseStreamInterceptors`，最终会在服务构造时链式生效。
- 单元测试中如需隔离，建议使用 `pgrpc.Reset()` 清理已注册的选项和拦截器。 - 单元测试中如需隔离，建议使用 `pgrpc.Reset()` 清理已注册的选项和拦截器。
- 若要启用健康检查或反射，请在配置中设置： - 若要启用健康检查或反射，请在配置中设置：
- `EnableHealth = true` - `EnableHealth = true`
- `EnableReflection = true` - `EnableReflection = true`
## 更多 ServerOption 示例 ## 更多 ServerOption 示例
@@ -228,14 +228,14 @@ TLS（服务端或 mTLS）：
```go ```go
import ( import (
"crypto/tls" "crypto/tls"
grpcCredentials "google.golang.org/grpc/credentials" grpcCredentials "google.golang.org/grpc/credentials"
) )
// 使用自定义 tls.Config（可配置 mTLS） // 使用自定义 tls.Config（可配置 mTLS）
var tlsConfig *tls.Config = &tls.Config{ /* ... */ } var tlsConfig *tls.Config = &tls.Config{ /* ... */ }
pgrpc.UseOptions( pgrpc.UseOptions(
grpc.Creds(grpcCredentials.NewTLS(tlsConfig)), grpc.Creds(grpcCredentials.NewTLS(tlsConfig)),
) )
// 或者从证书文件加载（仅服务端 TLS） // 或者从证书文件加载（仅服务端 TLS）
@@ -246,11 +246,11 @@ OpenTelemetry 统计/追踪（StatsHandler）：
```go ```go
import ( import (
otelgrpc "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" otelgrpc "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
) )
pgrpc.UseOptions( pgrpc.UseOptions(
grpc.StatsHandler(otelgrpc.NewServerHandler()), grpc.StatsHandler(otelgrpc.NewServerHandler()),
) )
``` ```
@@ -258,10 +258,10 @@ pgrpc.UseOptions(
```go ```go
pgrpc.UseOptions( pgrpc.UseOptions(
grpc.InitialWindowSize(1<<20), // 每个流初始窗口（字节） grpc.InitialWindowSize(1<<20), // 每个流初始窗口（字节）
grpc.InitialConnWindowSize(1<<21), // 连接级窗口 grpc.InitialConnWindowSize(1<<21), // 连接级窗口
grpc.ReadBufferSize(64<<10), // 读缓冲 64 KiB grpc.ReadBufferSize(64<<10), // 读缓冲 64 KiB
grpc.WriteBufferSize(64<<10), // 写缓冲 64 KiB grpc.WriteBufferSize(64<<10), // 写缓冲 64 KiB
) )
``` ```
@@ -269,17 +269,17 @@ pgrpc.UseOptions(
```go ```go
import ( import (
"context" "context"
"time" "time"
"google.golang.org/grpc/tap" "google.golang.org/grpc/tap"
) )
pgrpc.UseOptions( pgrpc.UseOptions(
grpc.ConnectionTimeout(5 * time.Second), grpc.ConnectionTimeout(5 * time.Second),
grpc.InTapHandle(func(ctx context.Context, info *tap.Info) (context.Context, error) { grpc.InTapHandle(func(ctx context.Context, info *tap.Info) (context.Context, error) {
// 在真正的 RPC 处理前进行快速拒绝（如黑名单、IP 检查等） // 在真正的 RPC 处理前进行快速拒绝（如黑名单、IP 检查等）
return ctx, nil return ctx, nil
}), }),
) )
``` ```
@@ -287,11 +287,11 @@ pgrpc.UseOptions(
```go ```go
pgrpc.UseOptions( pgrpc.UseOptions(
grpc.UnknownServiceHandler(func(srv any, stream grpc.ServerStream) error { grpc.UnknownServiceHandler(func(srv any, stream grpc.ServerStream) error {
// 统一记录未注册方法，或返回自定义错误 // 统一记录未注册方法，或返回自定义错误
return status.Error(codes.Unimplemented, "unknown service/method") return status.Error(codes.Unimplemented, "unknown service/method")
}), }),
grpc.NumStreamWorkers(8), // 针对 CPU 密集流处理的工作池 grpc.NumStreamWorkers(8), // 针对 CPU 密集流处理的工作池
) )
``` ```
@@ -301,24 +301,24 @@ pgrpc.UseOptions(
```go ```go
import ( import (
"context" "context"
"strings" "strings"
"google.golang.org/grpc/metadata" "google.golang.org/grpc/metadata"
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
) )
func AuthUnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) { func AuthUnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
md, _ := metadata.FromIncomingContext(ctx) md, _ := metadata.FromIncomingContext(ctx)
token := "" token := ""
if vals := md.Get("authorization"); len(vals) > 0 { if vals := md.Get("authorization"); len(vals) > 0 {
token = vals[0] token = vals[0]
} }
if token == "" || !strings.HasPrefix(strings.ToLower(token), "bearer ") { if token == "" || !strings.HasPrefix(strings.ToLower(token), "bearer ") {
return nil, status.Error(codes.Unauthenticated, "missing or invalid token") return nil, status.Error(codes.Unauthenticated, "missing or invalid token")
} }
// TODO: 验证 JWT / API-Key // TODO: 验证 JWT / API-Key
return handler(ctx, req) return handler(ctx, req)
} }
pgrpc.UseUnaryInterceptors(AuthUnaryInterceptor) pgrpc.UseUnaryInterceptors(AuthUnaryInterceptor)
@@ -328,32 +328,32 @@ pgrpc.UseUnaryInterceptors(AuthUnaryInterceptor)
```go ```go
import ( import (
"context" "context"
"sync" "sync"
"golang.org/x/time/rate" "golang.org/x/time/rate"
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
) )
var ( var (
rlmu sync.RWMutex rlmu sync.RWMutex
rlm = map[string]*rate.Limiter{} rlm = map[string]*rate.Limiter{}
) )
func limitFor(method string) *rate.Limiter { func limitFor(method string) *rate.Limiter {
rlmu.RLock(); l := rlm[method]; rlmu.RUnlock() rlmu.RLock() ; l := rlm[method]; rlmu.RUnlock()
if l != nil { return l } if l != nil { return l }
rlmu.Lock(); defer rlmu.Unlock() rlmu.Lock() ; defer rlmu.Unlock()
if rlm[method] == nil { rlm[method] = rate.NewLimiter(100, 200) } // 100 rps, burst 200 if rlm[method] == nil { rlm[method] = rate.NewLimiter(100, 200) } // 100 rps, burst 200
return rlm[method] return rlm[method]
} }
func RateLimitUnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) { func RateLimitUnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
l := limitFor(info.FullMethod) l := limitFor(info.FullMethod)
if !l.Allow() { if !l.Allow() {
return nil, status.Error(codes.ResourceExhausted, "rate limited") return nil, status.Error(codes.ResourceExhausted, "rate limited")
} }
return handler(ctx, req) return handler(ctx, req)
} }
pgrpc.UseUnaryInterceptors(RateLimitUnaryInterceptor) pgrpc.UseUnaryInterceptors(RateLimitUnaryInterceptor)
@@ -363,21 +363,21 @@ Request-ID 注入与日志关联：
```go ```go
import ( import (
"context" "context"
"github.com/google/uuid" "github.com/google/uuid"
"google.golang.org/grpc/metadata" "google.golang.org/grpc/metadata"
) )
type ctxKey string type ctxKey string
const requestIDKey ctxKey = "request_id" const requestIDKey ctxKey = "request_id"
func RequestIDUnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) { func RequestIDUnaryInterceptor(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
md, _ := metadata.FromIncomingContext(ctx) md, _ := metadata.FromIncomingContext(ctx)
var rid string var rid string
if v := md.Get("x-request-id"); len(v) > 0 { rid = v[0] } if v := md.Get("x-request-id"); len(v) > 0 { rid = v[0] }
if rid == "" { rid = uuid.New().String() } if rid == "" { rid = uuid.New().String() }
ctx = context.WithValue(ctx, requestIDKey, rid) ctx = context.WithValue(ctx, requestIDKey, rid)
return handler(ctx, req) return handler(ctx, req)
} }
pgrpc.UseUnaryInterceptors(RequestIDUnaryInterceptor) pgrpc.UseUnaryInterceptors(RequestIDUnaryInterceptor)
@@ -387,25 +387,25 @@ pgrpc.UseUnaryInterceptors(RequestIDUnaryInterceptor)
```go ```go
import ( import (
"context" "context"
"time" "time"
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
) )
func DeadlineUnaryInterceptor(max time.Duration) grpc.UnaryServerInterceptor { func DeadlineUnaryInterceptor(max time.Duration) grpc.UnaryServerInterceptor {
return func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) { return func(ctx context.Context, req any, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (any, error) {
if _, ok := ctx.Deadline(); !ok { // 未设置超时 if _, ok := ctx.Deadline() ; !ok { // 未设置超时
var cancel context.CancelFunc var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, max) ctx, cancel = context.WithTimeout(ctx, max)
defer cancel() defer cancel()
} }
resp, err := handler(ctx, req) resp, err := handler(ctx, req)
if err != nil && ctx.Err() == context.DeadlineExceeded { if err != nil && ctx.Err() == context.DeadlineExceeded {
return nil, status.Error(codes.DeadlineExceeded, "deadline exceeded") return nil, status.Error(codes.DeadlineExceeded, "deadline exceeded")
} }
return resp, err return resp, err
} }
} }
pgrpc.UseUnaryInterceptors(DeadlineUnaryInterceptor(5*time.Second)) pgrpc.UseUnaryInterceptors(DeadlineUnaryInterceptor(5*time.Second))
@@ -417,17 +417,17 @@ pgrpc.UseUnaryInterceptors(DeadlineUnaryInterceptor(5*time.Second))
```go ```go
import ( import (
"google.golang.org/grpc/metadata" "google.golang.org/grpc/metadata"
"google.golang.org/grpc/status" "google.golang.org/grpc/status"
"google.golang.org/grpc/codes" "google.golang.org/grpc/codes"
) )
func AuthStreamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { func AuthStreamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
md, _ := metadata.FromIncomingContext(ss.Context()) md, _ := metadata.FromIncomingContext(ss.Context())
if len(md.Get("authorization")) == 0 { if len(md.Get("authorization")) == 0 {
return status.Error(codes.Unauthenticated, "missing token") return status.Error(codes.Unauthenticated, "missing token")
} }
return handler(srv, ss) return handler(srv, ss)
} }
pgrpc.UseStreamInterceptors(AuthStreamInterceptor) pgrpc.UseStreamInterceptors(AuthStreamInterceptor)
@@ -437,11 +437,11 @@ pgrpc.UseStreamInterceptors(AuthStreamInterceptor)
```go ```go
func RateLimitStreamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { func RateLimitStreamInterceptor(srv any, ss grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
l := limitFor(info.FullMethod) l := limitFor(info.FullMethod)
if !l.Allow() { if !l.Allow() {
return status.Error(codes.ResourceExhausted, "rate limited") return status.Error(codes.ResourceExhausted, "rate limited")
} }
return handler(srv, ss) return handler(srv, ss)
} }
pgrpc.UseStreamInterceptors(RateLimitStreamInterceptor) pgrpc.UseStreamInterceptors(RateLimitStreamInterceptor)
@@ -453,8 +453,61 @@ pgrpc.UseStreamInterceptors(RateLimitStreamInterceptor)
```go ```go
import ( import (
_ "google.golang.org/grpc/encoding/gzip" // 注册 gzip 编解码器 _ "google.golang.org/grpc/encoding/gzip" // 注册 gzip 编解码器
) )
// 仅需 import 即可，无额外 ServerOption // 仅需 import 即可，无额外 ServerOption
``` ```
## OpenTelemetry 集成（推荐）
使用 StatsHandler（推荐，不与拦截器同时使用，避免重复埋点）：
```go
import (
otelgrpc "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
)
// 基本接入：使用全局 Tracer/Meter（由 OTEL Provider 初始化）
handler := otelgrpc.NewServerHandler(
otelgrpc.WithTraceEvents(), // 在 span 中记录消息事件
)
pgrpc.UseOptions(grpc.StatsHandler(handler))
// 忽略某些方法（如健康检查），避免噪声：
handler = otelgrpc.NewServerHandler(
otelgrpc.WithFilter(func(ctx context.Context, fullMethod string) bool {
return fullMethod != "/grpc.health.v1.Health/Check"
}),
)
pgrpc.UseOptions(grpc.StatsHandler(handler))
```
使用拦截器版本（如你更偏好 Interceptor 方案；与 StatsHandler 二选一）：
```go
import (
otelgrpc "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
)
pgrpc.UseUnaryInterceptors(otelgrpc.UnaryServerInterceptor())
pgrpc.UseStreamInterceptors(otelgrpc.StreamServerInterceptor())
```
> 注意：不要同时启用 StatsHandler 和拦截器，否则会重复生成 span/metrics。
## OpenTracing（Jaeger）集成
当使用 Tracing Provider（Jaeger + OpenTracing）时，可使用 opentracing 的 gRPC 拦截器：
```go
import (
opentracing "github.com/opentracing/opentracing-go"
otgrpc "github.com/grpc-ecosystem/grpc-opentracing/go/otgrpc"
)
pgrpc.UseUnaryInterceptors(otgrpc.OpenTracingServerInterceptor(opentracing.GlobalTracer()))
pgrpc.UseStreamInterceptors(otgrpc.OpenTracingStreamServerInterceptor(opentracing.GlobalTracer()))
```
> 与 OTEL 方案互斥：如果已启用 OTEL，请不要再开启 OpenTracing 拦截器，以免重复埋点。

121
templates/project/providers/otel/README.md.raw Normal file
View File

@@ -0,0 +1,121 @@
OpenTelemetry Provider (OTLP Traces + Metrics)
该 Provider 基于 OpenTelemetry Go SDK，初始化全局 Tracer 与 Meter，支持 OTLP(gRPC/HTTP) 导出，并收集运行时指标。
配置（config.toml）
```
[OTEL]
ServiceName = "my-service"
Version = "1.0.0"
Env = "dev"
# 导出端点（二选一）
EndpointGRPC = "otel-collector:4317"
EndpointHTTP = "otel-collector:4318"
# 认证（可选）
Token = "Bearer <your-token>" # 也可只填纯 token，Provider 会自动补齐 Bearer 前缀
# 安全（可选）
InsecureGRPC = true # gRPC 导出是否使用 insecure
InsecureHTTP = true # HTTP 导出是否使用 insecure
# 采样（可选）
Sampler = "always" # always|ratio
SamplerRatio = 0.1 # Sampler=ratio 时生效，0..1
# 批处理（可选，毫秒）
BatchTimeoutMs = 5000
ExportTimeoutMs = 10000
MaxQueueSize = 2048
MaxExportBatchSize = 512
# 指标（可选，毫秒）
MetricReaderIntervalMs = 10000 # 指标导出周期
RuntimeReadMemStatsIntervalMs = 5000 # 运行时指标读取周期
```
启用
```
import "test/providers/otel"
func providers() container.Providers {
return container.Providers{
otel.DefaultProvider(),
}
}
```
使用
- Traces: 通过 `go.opentelemetry.io/otel` 获取全局 Tracer，或使用仓库提供的 `providers/otel/funcs.go` 包装。
```
ctx, span := otel.Tracer("my-service").Start(ctx, "my-op")
// ...
span.End()
```
- Metrics: 通过 `otel.Meter("my-service")` 创建仪表，或使用 `providers/otel/funcs.go` 的便捷函数。
与 Tracing Provider 的区别与场景建议
- Tracing Provider（Jaeger + OpenTracing）只做链路，适合已有 OpenTracing 项目；
- OTEL Provider（OpenTelemetry）统一 Traces+Metrics，对接 OTLP 生态，适合新项目或希望统一可观测性；
- 可先混用：保留 Jaeger 链路，同时启用 OTEL 运行时指标，逐步迁移。
快速启动（本地 Collector）
最小化 docker-compose：
```
services:
otel-collector:
image: otel/opentelemetry-collector:0.104.0
command: ["--config=/etc/otelcol-config.yml"]
volumes:
- ./otelcol-config.yml:/etc/otelcol-config.yml:ro
ports:
- "4317:4317" # OTLP gRPC
- "4318:4318" # OTLP HTTP
```
示例 otelcol-config.yml：
```
receivers:
otlp:
protocols:
grpc:
http:
exporters:
debug:
verbosity: detailed
processors:
batch:
service:
pipelines:
traces:
receivers: [otlp]
processors: [batch]
exporters: [debug]
metrics:
receivers: [otlp]
processors: [batch]
exporters: [debug]
```
应用端：
```
[OTEL]
EndpointGRPC = "127.0.0.1:4317"
InsecureGRPC = true
```
故障与降级
- Collector/网络异常：OTEL SDK 异步批处理，不阻塞业务；可能丢点/丢指标；
- 启动失败：初始化报错会阻止启动；如需“不可达也不影响启动”，可加开关降级为 no-op（可按需补充）。

31
templates/project/providers/otel/config.go.tpl
View File

@@ -21,13 +21,32 @@ func DefaultProvider() container.ProviderContainer {
} }
type Config struct { type Config struct {
ServiceName string ServiceName string
Version string Version string
Env string Env string
EndpointGRPC string EndpointGRPC string
EndpointHTTP string EndpointHTTP string
Token string Token string
// Connection security
InsecureGRPC bool // if true, use grpc insecure for OTLP gRPC
InsecureHTTP bool // if true, use http insecure for OTLP HTTP
// Tracing sampler
// Sampler: "always" (default) or "ratio"
Sampler string
SamplerRatio float64 // used when Sampler == "ratio"; 0..1
// Tracing batcher options (milliseconds)
BatchTimeoutMs uint
ExportTimeoutMs uint
MaxQueueSize int
MaxExportBatchSize int
// Metrics options (milliseconds)
MetricReaderIntervalMs uint // export interval for PeriodicReader
RuntimeReadMemStatsIntervalMs uint // runtime metrics min read interval
} }
func (c *Config) format() { func (c *Config) format() {

157
templates/project/providers/otel/provider.go.tpl
View File

@@ -26,6 +26,17 @@ import (
"google.golang.org/grpc/encoding/gzip" "google.golang.org/grpc/encoding/gzip"
) )
// formatAuth formats token into an Authorization header value.
func formatAuth(token string) string {
if token == "" {
return ""
}
if len(token) > 7 && (token[:7] == "Bearer " || token[:7] == "bearer ") {
return token
}
return "Bearer " + token
}
func Provide(opts ...opt.Option) error { func Provide(opts ...opt.Option) error {
o := opt.New(opts...) o := opt.New(opts...)
var config Config var config Config
@@ -82,7 +93,7 @@ func (o *builder) initResource(ctx context.Context) (err error) {
semconv.HostNameKey.String(hostName), // 主机名 semconv.HostNameKey.String(hostName), // 主机名
), ),
) )
return return err
} }
func (o *builder) initMeterProvider(ctx context.Context) (err error) { func (o *builder) initMeterProvider(ctx context.Context) (err error) {
@@ -92,34 +103,35 @@ func (o *builder) initMeterProvider(ctx context.Context) (err error) {
otlpmetricgrpc.WithCompressor(gzip.Name), otlpmetricgrpc.WithCompressor(gzip.Name),
} }
if o.config.Token != "" { if h := formatAuth(o.config.Token); h != "" {
headers := map[string]string{"Authentication": o.config.Token} opts = append(opts, otlpmetricgrpc.WithHeaders(map[string]string{"Authorization": h}))
opts = append(opts, otlpmetricgrpc.WithHeaders(headers))
} }
exporter, err := otlpmetricgrpc.New(ctx, opts...) exporter, err := otlpmetricgrpc.New(ctx, opts...)
if err != nil { if err != nil {
return nil, err return nil, err
} }
return exporter, nil return exporter, nil
} }
exporterHttpFunc := func(ctx context.Context) (sdkmetric.Exporter, error) { exporterHttpFunc := func(ctx context.Context) (sdkmetric.Exporter, error) {
opts := []otlpmetrichttp.Option{ opts := []otlpmetrichttp.Option{
otlpmetrichttp.WithEndpoint(o.config.EndpointHTTP), otlpmetrichttp.WithEndpoint(o.config.EndpointHTTP),
otlpmetrichttp.WithCompression(1), otlpmetrichttp.WithCompression(1),
} }
if o.config.InsecureHTTP {
opts = append(opts, otlpmetrichttp.WithInsecure())
}
if h := formatAuth(o.config.Token); h != "" {
opts = append(opts, otlpmetrichttp.WithHeaders(map[string]string{"Authorization": h}))
}
if o.config.Token != "" { exporter, err := otlpmetrichttp.New(ctx, opts...)
opts = append(opts, otlpmetrichttp.WithURLPath(o.config.Token)) if err != nil {
} return nil, err
}
exporter, err := otlpmetrichttp.New(ctx, opts...) return exporter, nil
if err != nil { }
return nil, err
}
return exporter, nil
}
var exporter sdkmetric.Exporter var exporter sdkmetric.Exporter
if o.config.EndpointHTTP != "" { if o.config.EndpointHTTP != "" {
@@ -129,18 +141,27 @@ func (o *builder) initMeterProvider(ctx context.Context) (err error) {
} }
if err != nil { if err != nil {
return return err
} }
meterProvider := sdkmetric.NewMeterProvider( // periodic reader with optional custom interval
sdkmetric.WithReader( var readerOpts []sdkmetric.PeriodicReaderOption
sdkmetric.NewPeriodicReader(exporter), if o.config.MetricReaderIntervalMs > 0 {
), readerOpts = append(readerOpts, sdkmetric.WithInterval(time.Duration(o.config.MetricReaderIntervalMs)*time.Millisecond))
sdkmetric.WithResource(o.resource), }
) meterProvider := sdkmetric.NewMeterProvider(
sdkmetric.WithReader(
sdkmetric.NewPeriodicReader(exporter, readerOpts...),
),
sdkmetric.WithResource(o.resource),
)
otel.SetMeterProvider(meterProvider) otel.SetMeterProvider(meterProvider)
err = runtime.Start(runtime.WithMinimumReadMemStatsInterval(time.Second * 5)) interval := 5 * time.Second
if o.config.RuntimeReadMemStatsIntervalMs > 0 {
interval = time.Duration(o.config.RuntimeReadMemStatsIntervalMs) * time.Millisecond
}
err = runtime.Start(runtime.WithMinimumReadMemStatsInterval(interval))
if err != nil { if err != nil {
return errors.Wrapf(err, "Failed to start runtime metrics") return errors.Wrapf(err, "Failed to start runtime metrics")
} }
@@ -151,23 +172,21 @@ func (o *builder) initMeterProvider(ctx context.Context) (err error) {
} }
}) })
return return err
} }
func (o *builder) initTracerProvider(ctx context.Context) error { func (o *builder) initTracerProvider(ctx context.Context) error {
exporterGrpcFunc := func(ctx context.Context) (*otlptrace.Exporter, error) { exporterGrpcFunc := func(ctx context.Context) (*otlptrace.Exporter, error) {
opts := []otlptracegrpc.Option{ opts := []otlptracegrpc.Option{
otlptracegrpc.WithCompressor(gzip.Name), otlptracegrpc.WithCompressor(gzip.Name),
otlptracegrpc.WithEndpoint(o.config.EndpointGRPC), otlptracegrpc.WithEndpoint(o.config.EndpointGRPC),
otlptracegrpc.WithInsecure(), // 添加不安全连接选项 }
} if o.config.InsecureGRPC {
opts = append(opts, otlptracegrpc.WithInsecure())
}
if o.config.Token != "" { if h := formatAuth(o.config.Token); h != "" {
headers := map[string]string{ opts = append(opts, otlptracegrpc.WithHeaders(map[string]string{"Authorization": h}))
"Authentication": o.config.Token,
"authorization": o.config.Token, // 添加标准认证头
}
opts = append(opts, otlptracegrpc.WithHeaders(headers))
} }
log.Debugf("Creating GRPC trace exporter with endpoint: %s", o.config.EndpointGRPC) log.Debugf("Creating GRPC trace exporter with endpoint: %s", o.config.EndpointGRPC)
@@ -188,18 +207,16 @@ func (o *builder) initTracerProvider(ctx context.Context) error {
} }
exporterHttpFunc := func(ctx context.Context) (*otlptrace.Exporter, error) { exporterHttpFunc := func(ctx context.Context) (*otlptrace.Exporter, error) {
opts := []otlptracehttp.Option{ opts := []otlptracehttp.Option{
otlptracehttp.WithInsecure(), otlptracehttp.WithCompression(1),
otlptracehttp.WithCompression(1), otlptracehttp.WithEndpoint(o.config.EndpointHTTP),
otlptracehttp.WithEndpoint(o.config.EndpointHTTP), }
} if o.config.InsecureHTTP {
opts = append(opts, otlptracehttp.WithInsecure())
}
if o.config.Token != "" { if h := formatAuth(o.config.Token); h != "" {
opts = append(opts, opts = append(opts, otlptracehttp.WithHeaders(map[string]string{"Authorization": h}))
otlptracehttp.WithHeaders(map[string]string{
"Authentication": o.config.Token,
}),
)
} }
log.Debugf("Creating HTTP trace exporter with endpoint: %s", o.config.EndpointHTTP) log.Debugf("Creating HTTP trace exporter with endpoint: %s", o.config.EndpointHTTP)
@@ -225,11 +242,39 @@ func (o *builder) initTracerProvider(ctx context.Context) error {
return err return err
} }
traceProvider := sdktrace.NewTracerProvider( // Sampler
sdktrace.WithSampler(sdktrace.AlwaysSample()), sampler := sdktrace.AlwaysSample()
sdktrace.WithResource(o.resource), if o.config.Sampler == "ratio" {
sdktrace.WithBatcher(exporter), ratio := o.config.SamplerRatio
) if ratio <= 0 {
ratio = 0
}
if ratio > 1 {
ratio = 1
}
sampler = sdktrace.ParentBased(sdktrace.TraceIDRatioBased(ratio))
}
// Batcher options
var batchOpts []sdktrace.BatchSpanProcessorOption
if o.config.BatchTimeoutMs > 0 {
batchOpts = append(batchOpts, sdktrace.WithBatchTimeout(time.Duration(o.config.BatchTimeoutMs)*time.Millisecond))
}
if o.config.ExportTimeoutMs > 0 {
batchOpts = append(batchOpts, sdktrace.WithExportTimeout(time.Duration(o.config.ExportTimeoutMs)*time.Millisecond))
}
if o.config.MaxQueueSize > 0 {
batchOpts = append(batchOpts, sdktrace.WithMaxQueueSize(o.config.MaxQueueSize))
}
if o.config.MaxExportBatchSize > 0 {
batchOpts = append(batchOpts, sdktrace.WithMaxExportBatchSize(o.config.MaxExportBatchSize))
}
traceProvider := sdktrace.NewTracerProvider(
sdktrace.WithSampler(sampler),
sdktrace.WithResource(o.resource),
sdktrace.WithBatcher(exporter, batchOpts...),
)
container.AddCloseAble(func() { container.AddCloseAble(func() {
log.Error("shut down") log.Error("shut down")
if err := traceProvider.Shutdown(ctx); err != nil { if err := traceProvider.Shutdown(ctx); err != nil {