11、gRPC与分布式链路追踪
尝鲜一下
Setup MicroDonuts
选择一个分布式追踪工具
官方示例 Tracing Demo
gRPC 集成 Jaeger
查看一下容器内的构成
tracing carrier
创建 GlobalTracer
gRPC Interceptor
gRPC Client 和 gRPC Server 中集成 Interceptor 和Tracer
运行一下
参考
Last updated
Last updated
git clone git@github.com:opentracing-contrib/java-opentracing-walkthrough.git
cd java-opentracing-walkthrough/microdonuts
mvn package exec:exectracer=jaeger
jaeger.reporter_host=localhost
jaeger.reporter_port=5775docker run -d -p 5775:5775/udp -p 16686:16686 jaegertracing/all-in-one:latestdocker run -d -p6831:6831/udp -p16686:16686 jaegertracing/all-in-one:latestversion: '2'
services:
jaeger-collector:
image: jaegertracing/jaeger-collector
command: ["--cassandra.keyspace=jaeger_v1_dc1", "--cassandra.servers=cassandra", "--collector.zipkin.http-port=9411"]
ports:
- "14269"
- "14268:14268"
- "14267"
- "14250"
- "9411:9411"
restart: on-failure
depends_on:
- cassandra-schema
jaeger-query:
image: jaegertracing/jaeger-query
command: ["--cassandra.keyspace=jaeger_v1_dc1", "--cassandra.servers=cassandra"]
ports:
- "16686:16686"
- "16687"
restart: on-failure
depends_on:
- cassandra-schema
jaeger-agent:
image: jaegertracing/jaeger-agent
command: ["--reporter.grpc.host-port=jaeger-collector:14250"]
ports:
- "5775:5775/udp"
- "6831:6831/udp"
- "6832:6832/udp"
- "5778:5778"
restart: on-failure
depends_on:
- jaeger-collector
cassandra:
image: cassandra:3.9
cassandra-schema:
image: jaegertracing/jaeger-cassandra-schema
depends_on:
- cassandra// TextMapCarrier allows the use of regular map[string]string
// as both TextMapWriter and TextMapReader.
type TextMapCarrier map[string]string
// HTTPHeadersCarrier satisfies both TextMapWriter and TextMapReader.
//
// Example usage for server side:
//
// carrier := opentracing.HTTPHeadersCarrier(httpReq.Header)
// clientContext, err := tracer.Extract(opentracing.HTTPHeaders, carrier)
//
// Example usage for client side:
//
// carrier := opentracing.HTTPHeadersCarrier(httpReq.Header)
// err := tracer.Inject(
// span.Context(),
// opentracing.HTTPHeaders,
// carrier)
//
type HTTPHeadersCarrier http.Header// TextMapWriter is the Inject() carrier for the TextMap builtin format. With
// it, the caller can encode a SpanContext for propagation as entries in a map
// of unicode strings.
type TextMapWriter interface {
// Set a key:value pair to the carrier. Multiple calls to Set() for the
// same key leads to undefined behavior.
//
// NOTE: The backing store for the TextMapWriter may contain data unrelated
// to SpanContext. As such, Inject() and Extract() implementations that
// call the TextMapWriter and TextMapReader interfaces must agree on a
// prefix or other convention to distinguish their own key:value pairs.
Set(key, val string)
}
// TextMapReader is the Extract() carrier for the TextMap builtin format. With it,
// the caller can decode a propagated SpanContext as entries in a map of
// unicode strings.
type TextMapReader interface {
// ForeachKey returns TextMap contents via repeated calls to the `handler`
// function. If any call to `handler` returns a non-nil error, ForeachKey
// terminates and returns that error.
//
// NOTE: The backing store for the TextMapReader may contain data unrelated
// to SpanContext. As such, Inject() and Extract() implementations that
// call the TextMapWriter and TextMapReader interfaces must agree on a
// prefix or other convention to distinguish their own key:value pairs.
//
// The "foreach" callback pattern reduces unnecessary copying in some cases
// and also allows implementations to hold locks while the map is read.
ForeachKey(handler func(key, val string) error) error
}// MDCarrier custome carrier
type MDCarrier struct {
metadata.MD
}
// ForeachKey conforms to the TextMapReader interface.
// 这里必须要实现这个 TextMapReader 这个接口
// TextMapReader is the Extract() carrier for the TextMap builtin format. With it,
// the caller can decode a propagated SpanContext as entries in a map of
// unicode strings.
//type TextMapReader interface {
// // ForeachKey returns TextMap contents via repeated calls to the `handler`
// // function. If any call to `handler` returns a non-nil error, ForeachKey
// // terminates and returns that error.
// //
// // NOTE: The backing store for the TextMapReader may contain data unrelated
// // to SpanContext. As such, Inject() and Extract() implementations that
// // call the TextMapWriter and TextMapReader interfaces must agree on a
// // prefix or other convention to distinguish their own key:value pairs.
// //
// // The "foreach" callback pattern reduces unnecessary copying in some cases
// // and also allows implementations to hold locks while the map is read.
// ForeachKey(handler func(key, val string) error) error
//}
func (m MDCarrier) ForeachKey(handler func(key, val string) error) error {
for k, strs := range m.MD {
for _, v := range strs {
if err := handler(k, v); err != nil {
return err
}
}
}
return nil
}
// Set implements Set() of opentracing.TextMapWriter
// 这里也必须要实现
// TextMapWriter is the Inject() carrier for the TextMap builtin format. With
// it, the caller can encode a SpanContext for propagation as entries in a map
// of unicode strings.
//type TextMapWriter interface {
// // Set a key:value pair to the carrier. Multiple calls to Set() for the
// // same key leads to undefined behavior.
// //
// // NOTE: The backing store for the TextMapWriter may contain data unrelated
// // to SpanContext. As such, Inject() and Extract() implementations that
// // call the TextMapWriter and TextMapReader interfaces must agree on a
// // prefix or other convention to distinguish their own key:value pairs.
// Set(key, val string)
//}
func (m MDCarrier) Set(key, val string) {
m.MD[key] = append(m.MD[key], val)
}// NewJaegerTracer NewJaegerTracer for current service
func NewJaegerTracer(serviceName string, jagentHost string) (tracer opentracing.Tracer, closer io.Closer, err error) {
cfg := jaegercfg.Configuration{
ServiceName: serviceName,
Sampler: &jaegercfg.SamplerConfig{
Type: jaeger.SamplerTypeConst,
Param: 1,
},
Reporter: &jaegercfg.ReporterConfig{
LogSpans: true,
BufferFlushInterval: 1 * time.Second,
LocalAgentHostPort: jagentHost,
},
}
// Example logger and metrics factory. Use github.com/uber/jaeger-client-go/log
// and github.com/uber/jaeger-lib/metrics respectively to bind to real logging and metrics
// frameworks.
jLogger := jaegerlog.StdLogger
jMetricsFactory := metrics.NullFactory
// Initialize tracer with a logger and a metrics factory
tracer, closer, err = cfg.NewTracer(
jaegercfg.Logger(jLogger),
jaegercfg.Metrics(jMetricsFactory))
opentracing.SetGlobalTracer(tracer)
if err != nil {
grpclog.Errorf("Could not initialize jaeger tracer: %s", err.Error())
return
}
return
}// ClientInterceptor 客户端拦截器
// https://godoc.org/google.golang.org/grpc#UnaryClientInterceptor
func ClientInterceptor(tracer opentracing.Tracer) grpc.UnaryClientInterceptor {
return func(ctx context.Context, method string, request, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
//一个RPC调用的服务端的span，和RPC服务客户端的span构成ChildOf关系
var parentCtx opentracing.SpanContext
parentSpan := opentracing.SpanFromContext(ctx)
if parentSpan != nil {
parentCtx = parentSpan.Context()
}
span := tracer.StartSpan(
method,
opentracing.ChildOf(parentCtx),
opentracing.Tag{Key: string(ext.Component), Value: "gRPC Client"},
ext.SpanKindRPCClient,
)
defer span.Finish()
md, ok := metadata.FromOutgoingContext(ctx)
if !ok {
md = metadata.New(nil)
} else {
md = md.Copy()
}
err := tracer.Inject(
span.Context(),
opentracing.TextMap,
MDCarrier{md}, // 自定义 carrier
)
if err != nil {
log.Errorf("inject span error :%v", err.Error())
}
newCtx := metadata.NewOutgoingContext(ctx, md)
err = invoker(newCtx, method, request, reply, cc, opts...)
if err != nil {
log.Errorf("call error : %v", err.Error())
}
return err
}
}
// ServerInterceptor Server 端的拦截器
func ServerInterceptor(tracer opentracing.Tracer) grpc.UnaryServerInterceptor {
return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
md, ok := metadata.FromIncomingContext(ctx)
if !ok {
md = metadata.New(nil)
}
spanContext, err := tracer.Extract(
opentracing.TextMap,
MDCarrier{md},
)
if err != nil && err != opentracing.ErrSpanContextNotFound {
grpclog.Errorf("extract from metadata err: %v", err)
} else {
span := tracer.StartSpan(
info.FullMethod,
ext.RPCServerOption(spanContext),
opentracing.Tag{Key: string(ext.Component), Value: "gRPC Server"},
ext.SpanKindRPCServer,
)
defer span.Finish()
ctx = opentracing.ContextWithSpan(ctx, span)
}
return handler(ctx, req)
}
}func main() {
consul.Init()
tracer, closer, err := intercepter.NewJaegerTracer(serviceName, jaegerAgent)
defer closer.Close()
if err != nil {
log.Printf("NewJaegerTracer err:", err.Error())
}
// Set up a connection to the server.
ctx, _ := context.WithTimeout(context.Background(), 5*time.Second)
conn, err := grpc.DialContext(ctx, consulService, grpc.WithInsecure(), grpc.WithBalancerName("round_robin"), grpc.WithUnaryInterceptor(intercepter.ClientInterceptor(tracer)))
if err != nil {
log.Fatalf("did not connect: %v", err)
}
defer conn.Close()
c := pb.NewGopherClient(conn)
........
}func main() {
tracer, closer, err := intercepter.NewJaegerTracer(serviceName, jaegerAgent)
defer closer.Close()
if err != nil {
log.Printf("NewJaegerTracer err: %v", err.Error())
}
lis, err := net.Listen("tcp", port)
if err != nil {
log.Fatalf("failed to listen: %v", err)
}
s := grpc.NewServer(grpc.UnaryInterceptor(intercepter.ServerInterceptor(tracer)))
pb.RegisterGopherServer(s, &server{})
grpc_health_v1.RegisterHealthServer(s, &HealthImpl{})
RegisterToConsul()
if err := s.Serve(lis); err != nil {
log.Fatalf("failed to serve: %v", err)
}
}