ลองใช้ OpenTelemetry บน Go ในรูปแบบต่างๆ

• Overview
• รับส่งด้วย REST
• รับส่งด้วย gRPC
• รับส่งด้วย PubSub
• Baggage
• Attribute

พอดีวันนี้จะเปลี่ยนโปรเจคเก่าๆที่ใช้ OpenTracing และ OpenCensus มาใช้ OpenTelemetry แทน ก็เลยอยากจะลองใช้มันก่อน ดูว่าติดตรงไหนมั้ย

จุดประสงค์หลักๆในการเขียนโพสนี้ ก็เพื่อจะทดลองใช้ OpenTelemetry ในแบบต่างๆนะครับ จะไม่อธิบายว่ามันคืออะไร แล้วใช้เพื่ออะไรนะครับ ถ้าสงสัยลองหาจากโพสอื่นๆดูก่อน

Overview

โดยในโพสนี้จะเริ่มจากสร้างโปรเจคเล็กๆ จำลองเป็น 2 service คุยกันนะครับ มีฟังก์ชันการทำงานอยู่ฟังก์ชันเดียวคือสั่งอาหารจาก platform -> restaurant -> platform
แต่ว่าการส่ง request ระหว่างกันจะส่งได้ 3 แบบคือ HTTP, gRPC, PubSub บน NATS ในแต่ละแบบก็จะมีโค้ดตัวอย่างทั้งการรับและส่ง

trace ที่จะทำคือการสั่งอาหาร เกิดจาก platform ไปหา restaurant แล้วพอร้านทำอาหารเสร็จ ก็จะส่ง event food_ready กลับมาที่ platform อีกที ตามรูปนะครับ

โค้ดโปรเจคนี้อยู่ที่นี่นะครับ
https://github.com/atthavit/myblog/tree/master/try-opentelemetry/delivery

ถ้าจะลองรันขึ้นมาเล่นดูก็ใช้ docker-compose ได้เลยครับ

docker-compose up -d

จะ bind port ไว้ตามนี้
localhost:16686 - jaeger web ui
localhost:8001 - platform api
localhost:4222 - nats

ลองยิง request ไปที่ platform api (method มี grpc, rest, pubsub

curl -v 'http://localhost:8001/order?food=egg&name=John&address=Home&method=grpc'

ก็จะมี trace ใหม่ขึ้นมาใน jaeger

Code

ก่อนที่จะเริ่มส่งข้อมูล trace ได้เราก็จะต้อง setup ตัว OpenTelemetry ก่อน

exp, err := jaeger.NewRawExporter(jaeger.WithCollectorEndpoint(jaeger.WithEndpoint("http://localhost:14268/api/traces")))
if err != nil {
    log.Fatal(err)
}
bsp := sdktrace.NewBatchSpanProcessor(exp)
tp := sdktrace.NewTracerProvider(
    sdktrace.WithSampler(sdktrace.AlwaysSample()),
    sdktrace.WithResource(resource.NewWithAttributes(
        semconv.ServiceNameKey.String(serviceName),
    )),
    sdktrace.WithSpanProcessor(bsp),
)
otel.SetTracerProvider(tp)
otel.SetTextMapPropagator(
    propagation.NewCompositeTextMapPropagator(
        propagation.TraceContext{},
        propagation.Baggage{},
    ),
)

span exporter จะใช้เป็น jaeger นะครับ เท่าที่เห็นตอนนี้มี jaeger กับ zipkin ลองดูเพิ่มได้ที่นี่

ตรง SetTextMapPropagator นี้ จะเป็นตัวบอกว่าเราอยากให้มันส่งข้อมูลอะไรต่อบ้าง ในตัวอย่างข้างบนก็จะให้มันส่ง TraceContext และ Baggage

TraceContext เป็นข้อมูล trace ถ้าไม่ส่งไปจะทำให้ span ของแต่ละ service แยกออกเป็นหลายๆ trace ไม่ต่อกัน

REST

วิธีนี้จะเป็นวิธีที่ง่ายที่สุด มีfunctionให้ใช้ง่ายๆเลย หรือถ้าใครใช้ Gin, Echo ก็สามารถใช้ middleware ได้เลย ลองดูเพิ่มจากที่นี่ แต่ในโพสนี้จะลองใช้แค่ http.Handler นะครับ

ส่ง

การส่ง request ที่มี trace อยู่ไปหา service อื่น เราก็จะต้องแนบข้อมูลของ trace ไปด้วย ซึ่งเราสามารถใช้ transport ที่ OpenTelemetry มีมาให้ได้เลย มันจะแนบข้อมูล trace ติดไปกับ header ของ request ให้

client := http.Client{Transport: otelhttp.NewTransport(http.DefaultTransport)}
resp, err := client.Do(req)

รับ

ใช้ otelhttp.NewHandler มาครอบ http.Handler ไว้

if err := http.ListenAndServe(":8001", otelhttp.NewHandler(r, "order")); err != nil {
    log.Fatal(err)
}

ที่นี่ request context ของเราก็จะมีข้อมูล trace แล้ว สามารถสร้าง span เพิ่มได้จาก context เลย

span := trace.SpanFromContext(ctx)
span.SetAttributes(attribute.String("order.type", "rest"))

ctx, span := p.Tracer.Start(ctx, "process food order")
defer span.End()
doSomething()

ต้องไม่ลืมเรียก span.End() นะครับ ถ้าลืม span จะหายไปเลย

gRPC

การใช้ OpenTelemetry กับ gRPC ก็ง่ายพอๆกับ http เลย แค่เพิ่ม interceptor เท่านั้น

ส่ง

ตอนส่งก็คล้ายๆกัน เพิ่ม interceptor ไปตอนเรียก grpc.Dial

conn, err := grpc.Dial(
    target,
    grpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor()),
    grpc.WithStreamInterceptor(otelgrpc.StreamClientInterceptor()),
    grpc.WithInsecure(),
)

รับ

เพิ่ม unary และ stream interceptor ไปตอนสร้าง grpc server

grpcServer := grpc.NewServer(
    grpc.UnaryInterceptor(otelgrpc.UnaryServerInterceptor()),
    grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor()),
)

ถ้ามี interceptor เดิมอยู่แล้ว ก็ใช้ grpc.ChainUnaryInterceptor, grpc.ChainStreamInterceptor เพื่อต่อ interceptor หลายๆอันได้

Publish/Subscribe (NATS)

ในรูปแบบสุดท้ายนี้ คือจะสมมติว่าถ้าวิธีการส่ง request ของเราเนี่ย มันไม่มี lib support อาจจะไม่ได้ส่งด้วย rest api หรือ grpc เราเลยจะลองใช้วิธีแบบแมนนวลหน่อยๆ คือ inject/extract ข้อมูลของ trace จาก
message ตรงๆ

เท่าที่ลองหาตอนนี้ OpenTelemetry จะมี Propagator แบบ TextMap เท่านั้น ยังไม่มีแบบ Binary เหมือน OpenTracing แต่ในอนาคตน่าจะมีเหมือนกัน #437 และจะใช้ Carrier เป็น HeaderCarrier ซึ่งเป็น type alias ของ http.Header หรือก็คือ map[string][]string

ส่ง

การส่งด้วย pubsub แบบที่ไม่มี lib support ก็จะลำบากหน่อย ต้องเอา context มาแล้ว inject เข้าไปใน carrier แต่ก็ยังดีที่ NATS message ใส่ header ได้ แล้วเป็น type เดียวกับ http.Header ด้วย เลยใช้ propagation.HeaderCarrier ได้เลย

func (p *Platform) OrderByPubSub(ctx context.Context, order FoodOrder) error {
    span := trace.SpanFromContext(ctx)
    span.SetAttributes(attribute.String("order.type", "pubsub"))
    ctx = p.order(ctx, order)
    bs, err := json.Marshal(order)
    if err != nil {
        return err
    }
    msg := nats.NewMsg("order")
    msg.Data = bs
    propagator := otel.GetTextMapPropagator()
    propagator.Inject(ctx, propagation.HeaderCarrier(msg.Header))
    return p.PubSub.PublishMsg(msg)
}

รับ

การรับก็จะคล้ายๆกับส่ง แค่เปลี่ยนจาก Inject มาใช้ Extract แทน ก็จะได้ context ที่มีข้อมูล trace มาเลย

r.PubSub.Subscribe("order", func(msg *nats.Msg) {
    propagator := otel.GetTextMapPropagator()
    ctx := propagator.Extract(context.Background(), propagation.HeaderCarrier(msg.Header))
    ctx, span := r.Tracer.Start(ctx, "receive food order")
    defer span.End()
    var order FoodOrder
    if err := json.Unmarshal(msg.Data, &order); err != nil {
        return
    }
    r.cook(ctx, order)
})

แต่ถ้าการ publish message ไม่ได้รองรับ header ก็อาจจะต้องสร้าง http.Header แล้ว Inject ใส่มัน แล้วเอาไปติดใน body ของ message ที่จะส่งไปเลย แล้วตัวรับก็ Unmarshal ออกมา แล้วค่อยเอาไปเข้า propagator.Extract()

อ่านเขียน Baggage

Baggage เป็นข้อมูล key/value ที่เราสามารถส่งข้ามไปแต่ละ service ได้ อ่านเพิ่มเติมที่นี่

วิธีเขียนใช้ baggage.Value

baggage.ContextWithValues(ctx, attribute.String("food", order.Food))

วิธีอ่านก็อ่านจาก context มาได้เลย

import "go.opentelemetry.io/otel/baggage"
...
v := baggage.Value(ctx, "food")
log.Printf("received baggage food with value: %s", v.AsString())

Attribute

attribute จะเป็นข้อมูลรูปแบบ key/value คล้ายๆกับ baggage แต่จะไว้ให้คนดูมากกว่า เท่าที่เข้าใจคือ baggage ไว้ส่งข้อมูลข้าม service แต่ attribute จะไปโผล่บน jaeger ไว้ให้คนดูข้อมูลมากกว่า

ตัวอย่าง set attribute ให้กับ span

span.SetAttributes(attribute.String("address", order.Address))

ใน jaeger จะไปขึ้นตรง tag ของ span

แต่ถ้าเพิ่ม event ของ span พร้อมกับ attribute

span.AddEvent("delivered", trace.WithAttributes(
    attribute.String("food", order.Food),
    attribute.String("customer", order.CustomerName),
    attribute.String("address", order.Address),
))

บน jaeger จะขึ้นใน log