otl sdk使用

准备otl collector

docker run --rm -p 13133:13133 -p 14250:14250 -p 14268:14268 -p 55678-55679:55678-55679 -p 4317:4317 -p 8888:8888 -p 9411:9411 otel/opentelemetry-collector

http

准备client.go,内容如下

package main

import (
    "context"
    "fmt"
    "net/http"

    "google.golang.org/grpc"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
)

func initTracer() func() {
    ctx := context.Background()
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceNameKey.String("client"),
        ),
    )
    if err != nil {
        panic(err)
    }

    conn, err := grpc.DialContext(ctx, "localhost:4317", grpc.WithInsecure())
    if err != nil {
        panic(err)
    }

    traceExporter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithGRPCConn(conn))
    if err != nil {
        panic(err)
    }

    bsp := sdktrace.NewBatchSpanProcessor(traceExporter)
    tracerProvider := sdktrace.NewTracerProvider(
        sdktrace.WithSampler(sdktrace.AlwaysSample()),
        sdktrace.WithResource(res),
        sdktrace.WithSpanProcessor(bsp),
    )
    otel.SetTracerProvider(tracerProvider)
    otel.SetTextMapPropagator(propagation.TraceContext{})

    return func() {
        err := tracerProvider.Shutdown(ctx)
        if err != nil {
            panic(err)
        }
    }
}

func main() {
    cleanup := initTracer()
    defer cleanup()
    tracer := otel.Tracer("demo")

    ctx, span := tracer.Start(
        context.Background(),
        "dohttpreq")
    defer span.End()
    httpReq, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://127.0.0.1:8083", nil)
    if err != nil {
        panic(err)
    }
    otel.GetTextMapPropagator().Inject(ctx, propagation.HeaderCarrier(httpReq.Header))
    sc := span.SpanContext()
    fmt.Println(sc.SpanID(), sc.TraceID())
    httpResp, err := http.DefaultClient.Do(httpReq)
    if err != nil {
        panic(err)
    }
    defer httpResp.Body.Close()
}

准备server.go,内容如下

package main

import (
    "context"
    "fmt"
    "net/http"

    "google.golang.org/grpc"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
)

func initTracer() func() {
    ctx := context.Background()
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceNameKey.String("server"),
        ),
    )
    if err != nil {
        panic(err)
    }

    conn, err := grpc.DialContext(ctx, "localhost:4317", grpc.WithInsecure())
    if err != nil {
        panic(err)
    }

    traceExporter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithGRPCConn(conn))
    if err != nil {
        panic(err)
    }

    bsp := sdktrace.NewBatchSpanProcessor(traceExporter)
    tracerProvider := sdktrace.NewTracerProvider(
        sdktrace.WithSampler(sdktrace.AlwaysSample()),
        sdktrace.WithResource(res),
        sdktrace.WithSpanProcessor(bsp),
    )
    otel.SetTracerProvider(tracerProvider)
    otel.SetTextMapPropagator(propagation.TraceContext{})

    return func() {
        err := tracerProvider.Shutdown(ctx)
        if err != nil {
            panic(err)
        }
    }
}

func main() {
    cleanup := initTracer()
    defer cleanup()
    tracer := otel.Tracer("demo")
    http.HandleFunc("/", func(rw http.ResponseWriter, r *http.Request) {
        ctx := otel.GetTextMapPropagator().Extract(r.Context(), propagation.HeaderCarrier(r.Header))
        ctx, span := tracer.Start(
            ctx,
            "servehttpreq")
        defer span.End()
        sc := span.SpanContext()
        fmt.Println(sc.SpanID(), sc.TraceID())
    })
    http.ListenAndServe(":8083", nil)
}

试验

go run server.go
go run client.go
可以在server和client看到相同的traceid不同的spanid,如
server中bc190ca35b35f135 59f8040ec726a9e4580f0d0e5cbb54ce
client中3bba8bafbd326f5e 59f8040ec726a9e4580f0d0e5cbb54ce

grpc

假设mod name 为otl

准备helloworld.proto

syntax = "proto3";
option go_package = "otl/helloworld";

package helloworld;

service Greeter {
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

message HelloRequest {
  string name = 1;
}

message HelloReply {
  string message = 1;
}

准备client.go,内容如下

package main

import (
    "context"
    "fmt"
    "log"
    pb "otl/helloworld"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
    "google.golang.org/grpc"
    "google.golang.org/grpc/metadata"
)

func main() {
    cleanup := initTracer()
    defer cleanup()
    tracer := otel.Tracer("demo")
    ctx := context.Background()
    m := metadata.New(nil)
    ctx, span := tracer.Start(
        ctx,
        "dogrpcreq")
    defer span.End()
    otel.GetTextMapPropagator().Inject(ctx, &header{m})
    ctx = metadata.NewOutgoingContext(ctx, m)
    sc := span.SpanContext()
    fmt.Println(sc.SpanID(), sc.TraceID())
    conn, err := grpc.Dial("127.0.0.1:8083", grpc.WithInsecure())
    if err != nil {
        log.Fatalf("did not connect: %v", err)
    }
    defer conn.Close()
    c := pb.NewGreeterClient(conn)

    if _, err := c.SayHello(ctx, &pb.HelloRequest{Name: "demo"}); err != nil {
        panic(err)
    }
}

type header struct {
    m metadata.MD
}

func (h *header) Get(key string) string {
    if len(h.m) == 0 {
        return ""
    }
    vals := h.m[key]
    if len(vals) == 0 {
        return ""
    }
    return vals[0]
}

func (h *header) Set(key, val string) {
    h.m.Set(key, val)
}

func (h *header) Keys() []string {
    keys := make([]string, 0, len(h.m))
    for k := range h.m {
        keys = append(keys, k)
    }
    return keys
}

func initTracer() func() {
    ctx := context.Background()
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceNameKey.String("server"),
        ),
    )
    if err != nil {
        panic(err)
    }
    conn, err := grpc.DialContext(ctx, "localhost:4317", grpc.WithInsecure())
    if err != nil {
        panic(err)
    }

    traceExporter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithGRPCConn(conn))
    if err != nil {
        panic(err)
    }

    bsp := sdktrace.NewBatchSpanProcessor(traceExporter)
    tracerProvider := sdktrace.NewTracerProvider(
        sdktrace.WithSampler(sdktrace.AlwaysSample()),
        sdktrace.WithResource(res),
        sdktrace.WithSpanProcessor(bsp),
    )
    otel.SetTracerProvider(tracerProvider)

    otel.SetTextMapPropagator(propagation.TraceContext{})

    return func() {
        err := tracerProvider.Shutdown(ctx)
        if err != nil {
            panic(err)
        }
    }
}

准备server.go,内容如下

package main

import (
    "net"

    "context"
    "fmt"

    pb "otl/helloworld"

    "google.golang.org/grpc"
    "google.golang.org/grpc/metadata"

    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/sdk/resource"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.7.0"
)

type svc struct {
    pb.UnimplementedGreeterServer
}

func (s *svc) SayHello(ctx context.Context, in *pb.HelloRequest) (*pb.HelloReply, error) {
    m, ok := metadata.FromIncomingContext(ctx)
    if !ok {
        m = metadata.New(nil)
    }
    ctx = otel.GetTextMapPropagator().Extract(ctx, &header{m})
    tracer := otel.Tracer("demo")
    ctx, span := tracer.Start(
        ctx,
        "servegrpcreq")

    sc := span.SpanContext()
    fmt.Println(sc.SpanID(), sc.TraceID())
    return &pb.HelloReply{Message: in.Name}, nil
}

func main() {
    cleanup := initTracer()
    defer cleanup()
    lis, err := net.Listen("tcp", ":8083")
    if err != nil {
        panic(err)
    }
    s := grpc.NewServer()
    pb.RegisterGreeterServer(s, &svc{})
    if err := s.Serve(lis); err != nil {
        panic(err)
    }
}

type header struct {
    m metadata.MD
}

func (h *header) Get(key string) string {
    if len(h.m) == 0 {
        return ""
    }
    vals := h.m[key]
    if len(vals) == 0 {
        return ""
    }
    return vals[0]
}

func (h *header) Set(key, val string) {
    h.m.Set(key, val)
}

func (h *header) Keys() []string {
    keys := make([]string, 0, len(h.m))
    for k := range h.m {
        keys = append(keys, k)
    }
    return keys
}

func initTracer() func() {
    ctx := context.Background()
    res, err := resource.New(ctx,
        resource.WithAttributes(
            semconv.ServiceNameKey.String("server"),
        ),
    )
    if err != nil {
        panic(err)
    }
    conn, err := grpc.DialContext(ctx, "localhost:4317", grpc.WithInsecure())
    if err != nil {
        panic(err)
    }

    traceExporter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithGRPCConn(conn))
    if err != nil {
        panic(err)
    }

    bsp := sdktrace.NewBatchSpanProcessor(traceExporter)
    tracerProvider := sdktrace.NewTracerProvider(
        sdktrace.WithSampler(sdktrace.AlwaysSample()),
        sdktrace.WithResource(res),
        sdktrace.WithSpanProcessor(bsp),
    )
    otel.SetTracerProvider(tracerProvider)

    otel.SetTextMapPropagator(propagation.TraceContext{})

    return func() {
        err := tracerProvider.Shutdown(ctx)
        if err != nil {
            panic(err)
        }
    }
}

试验

go run server.go
go run client.go
可以在server和client看到相同的traceid不同的spanid,如
server中cd8d5e56ca26be42 177270fd4d8cd0954113b8f4b3bdb391
client中cd8d5e56ca26be42 177270fd4d8cd0954113b8f4b3bdb391

遇到的问题

由于http2中headerkey是全小写的,而go的http.header的set操作是会把headerkey转成首字母大写的,所以使用如下方式注入header是错误的

otel.GetTextMapPropagator().Inject(ctx, propagation.HeaderCarrier(m))

报错信息如下

rpc error: code = Internal desc = stream terminated by RST_STREAM with error code: PROTOCOL_ERROR

同样的server端也不可以,因为收到的headerkey是全小写的,http.header的get操作会对传入的key进行首字母大小写再去取