OpenCensusでcustom context propagationとexporterを書いた話 / OpenCensus with custom context propagation and exporter

OpenCensus/OpenTelemetry meetup vol.2 の発表資料です。
https://opencensus.connpass.com/event/132588/

Transcript

1. OpenCensus/OpenTelemetry meetup vol.2 @takashabe OpenCensusͰcustom context propagationͱexporterΛॻ͍ ͨ࿩
2. whoami • Takashi Abe (@takashabe) • גࣜձࣾαΠόʔΤʔδΣϯτ ΞυςΫελδΦ • GoͰ޿ࠂ഑৴ϓϩμΫτΛ࡞͍ͬͯ·͢ • KubernetesͰMicroservicesతͳ΋ͷ • ਪ͠ΩʔϘʔυ͸irisͰ͢ • Corne Cherry࡞੒த…
3. ͸͡Ίʹ • Go + OpenCensus + Stackdriver Λલఏͱ͍ͯ͠·͢
4. ໨࣍ • ෼ࢄτϨʔγϯάͷ֓ཁ • context propagation • exporter • ·ͱΊ
5. ෼ࢄτϨʔγϯά
6. ෼ࢄτϨʔγϯά https://github.com/census-instrumentation/opencensus-go • 1 traceʹରͯ͠ɺෳ਺ͷspan͕ඥͮ͘ • span͸ಉҰΞϓϦέʔγϣϯ಺Ͱෳ਺ൃߦͯ͠΋ྑ͍͠ɺҟͳΔΞϓϦέʔγϣϯͰ΋ τϨʔεID͕ಉҰͰ͋Ε͹ඥͮ͘
7. context propagation
8. context propagation is …
9. context propagation is … • τϨʔεIDΛ఻ൖ͢ΔͨΊͷϓϩτίϧతͳ΋ͷ • opencensus-goͰ͸ocgrpc΍ochttpͰ࣮૷͞Ε͍ͯΔ • SpanContextΛհͯ͠σʔλΛ఻ൖͤ͞Δ • TraceID, SpanID, TraceOptions(αϯϓϦϯάର৅͔Ͳ ͏͔) type SpanContext struct { TraceID TraceID // [16]byte SpanID SpanID // [8]byte TraceOptions TraceOptions // uint32 Tracestate *tracestate.Tracestate // Ұ෦ͷpropagation༻ }
10. ocgrpc, ochttp ocgrpc ochttp
11. HTTPFormatͷ࣮૷ • ocgrpc͸࣮૷͕1ͭ • ochttpͰ͸HTTPFormatΛ࣮૷ͨ͠b3ͱtracecontextͱ͍͏2 ͭͷ࣮૷͕ଘࡏ͢Δ • b3: ZipkinͰఆٛ͞Εͨpropagation(σϑΥϧτ) • https://github.com/openzipkin/b3-propagation • tracecontext: w3cͰఆٛ͞Εͨpropagation • https://github.com/w3c/trace-context type HTTPFormat interface { SpanContextFromRequest(req *http.Request) (sc trace.SpanContext, ok bool) SpanContextToRequest(sc trace.SpanContext, req *http.Request) }
12. B3 Propagation https://github.com/openzipkin/b3-propagation
13. B3 Propagation https://github.com/openzipkin/b3-propagation 4QBO$POUFYU'SPN3FRVFTU 4QBO$POUFYU5P3FRVFTU
14. OpenCensusͰͷ࣮૷(ToRequest) const ( TraceIDHeader = "X-B3-TraceId" SpanIDHeader = "X-B3-SpanId" SampledHeader = "X-B3-Sampled" ) func (f *HTTPFormat) SpanContextToRequest(sc trace.SpanContext, req *http.Request) { req.Header.Set(TraceIDHeader, hex.EncodeToString(sc.TraceID[:])) req.Header.Set(SpanIDHeader, hex.EncodeToString(sc.SpanID[:])) var sampled string if sc.IsSampled() { sampled = "1" } else { sampled = "0" } req.Header.Set(SampledHeader, sampled) } https://github.com/census-instrumentation/opencensus-go/blob/master/plugin/ochttp/propagation/b3/b3.go
15. OpenCensusͰͷ࣮૷(FromRequest) const ( TraceIDHeader = "X-B3-TraceId" SpanIDHeader = "X-B3-SpanId" SampledHeader = "X-B3-Sampled" ) func (f *HTTPFormat) SpanContextFromRequest(req *http.Request) (sc trace.SpanContext, ok bool) { tid, ok := ParseTraceID(req.Header.Get(TraceIDHeader)) if !ok { return trace.SpanContext{}, false } sid, ok := ParseSpanID(req.Header.Get(SpanIDHeader)) if !ok { return trace.SpanContext{}, false } sampled, _ := ParseSampled(req.Header.Get(SampledHeader)) return trace.SpanContext{ TraceID: tid, SpanID: sid, TraceOptions: sampled, }, true } https://github.com/census-instrumentation/opencensus-go/blob/master/plugin/ochttp/propagation/b3/b3.go
16. ochttpΛར༻͢Δ // import "go.opencensus.io/plugin/ochttp" mux := http.NewServeMux() mux.Handle("/users", ochttp.WithRouteTag(usersHandler, "/users")) log.Fatal(http.ListenAndServe("localhost:8080", &ochttp.Handler{ Handler: mux, Propagation: &b3.HTTPFormat{}, })) https://github.com/census-instrumentation/opencensus-go/blob/master/plugin/ochttp/example_test.go • ochttp.Handler͕ϛυϧ΢ΣΞͱͯ͠τϨʔγϯάपΓ Λ໘౗ݟͯ͘ΕΔ • context propagation͸ҙࣝ͠ͳͯ͘ྑ͍
17. custom context propagation
18. Cloud Pub/Sub ʹద༻ͨ͠ • GCPͷϚωʔδυPub/Sub • B3 PropagationΛϕʔεʹͯ͠ɺϝοηʔδͷ AttributesʹσʔλΛ٧Ίͯ఻ൖͤ͞Δ • https://github.com/takashabe/oc-propagation-demo • ΄΅ಉ͡΋ͷΛϓϩμΫγϣϯʹಋೖࡁΈ͕ͩɺ؀ڥ ґଘ͕͋ΔͷͰαϯϓϧ࣮૷Λ༻ҙ • αʔό༻ҙ͢Δͷ໘౗ͰCloud Run༻ʹHTTPαʔό ͱ࣮ͯ͠૷͍ͯ͠ΔͷͰएׯݟͮΒ͍
19. Pub/Sub൛ͷ࣮૷(ToRequest) const ( TraceIDField = "X-Pubsub-TraceId" SpanIDField = "X-Pubsub-SpanId" SampledField = "X-Pubsub-Sampled" ) func WrapMessage(ctx context.Context, m *pubsub.Message) *pubsub.Message { if m.Attributes != nil { if m.Attributes[TraceIDField] != "" || m.Attributes[SpanIDField] != "" { return m } } else { m.Attributes = make(map[string]string, 3) } sc := trace.FromContext(ctx).SpanContext() m.Attributes[TraceIDField] = sc.TraceID.String() m.Attributes[SpanIDField] = sc.SpanID.String() m.Attributes[SampledField] = fmt.Sprintf("%t", sc.IsSampled()) return m }
20. Pub/Sub൛ͷ࣮૷(FromRequest) const ( TraceIDField = "X-Pubsub-TraceId" SpanIDField = "X-Pubsub-SpanId" SampledField = "X-Pubsub-Sampled" ) func SpanContextFromMessage(m *pubsub.Message) trace.SpanContext { if m.Attributes == nil { return trace.SpanContext{} } tid, ok := parseTraceID(m.Attributes[TraceIDField]) if !ok { return trace.SpanContext{} } sid, ok := parseSpanID(m.Attributes[SpanIDField]) if !ok { return trace.SpanContext{} } sampled := parseSampled(m.Attributes[SampledField]) return trace.SpanContext{ TraceID: tid, SpanID: sid, TraceOptions: sampled, } }
21. ར༻ଆ // import “github.com/takashabe/oc-propagation-demo/internal/propagation" ... ctx, span := trace.StartSpan(context.Background(), "publish") defer span.End() msg := &pubsub.Message{ Data: []byte("foo"), } topic.Publish(ctx, propagation.WrapMessage(ctx, msg)).Get(ctx) // import “github.com/takashabe/oc-propagation-demo/internal/propagation" ... subscription.Receive(ctx, func(ctx context.Context, msg *pubsub.Message) { sc := propagation.SpanContextFromMessage(msg) _, span := trace.StartSpanWithRemoteParent(ctx, "receive", sc) defer span.End() publisher(client) subscriber(server)
22. Stackdriver Trace্ͷදࣔ publisher subscriber Pub/Sub
23. exporter
24. exporter is …
25. exporter is … • Stackdriver΍DatadogͳͲͷόοΫΤϯυαʔϏεʹ τϨʔεɺ͋Δ͍͸ϝτϦΫεΛૹΔͨΊͷ࢓૊Έ • σʔλऩू෦෼ͱexporterʹΑΔόοΫΤϯυαʔϏε ͕෼཭͞Ε͍ͯΔͷ͕OpenCensusͷ1ͭͷಛ௃
26. exporterͷ࣮૷(Go) type Exporter interface { ExportSpan(s *SpanData) } • ExporterΠϯλϑΣʔεΛ࣮૷͢Δ͚ͩͰྑ͍ • ೚ҙͷόοΫΤϯυαʔϏεAPIͳͲΛୟ͍ͯσʔλΛ ૹ͍ͬͯΔ • ࣮ࡍʹ͸ύϑΥʔϚϯεͰ͔ͳΓؾΛ࢖͍ͬͯΔ(͸ͣ)
27. custom exporter
28. Tail LatencyΛิ଍͢ΔͨΊͷexporter • Tail Latency • 99ύʔηϯλΠϧ஋ͷΑ͏ͳ΋ͷ • ௨ৗɺτϥϑΟοΫͷଟ͍αʔϏεͩͱαϯϓϦϯά ͷඞཁ͕͋Δ͕ɺͦ͏͢ΔͱTail Latency͕શ͘ه࿥͞ Εͳ͍ • Tail LatencyͰ೰ΜͰ͍Δͱ͖͸मਖ਼ɺܭଌͷαΠΫ ϧΛճ͍͕ͨ͠ɺ؍ଌ͢ΔͨΊʹ਺࣌ؒҎ্͔͔Δ ͜ͱ΋͋Δ…
29. edge exporter • https://github.com/takashabe/edge-exporter • ϓϩμΫγϣϯ౤ೖग़དྷͯͳ͍ͷͰࢀߟ࣮૷ͱͯ͠… • ύϑΥʔϚϯε͸վળͷ༨஍͕େ͍ʹ࢒ͬͯΔ • ߏ૝Λ࿅͍ͬͯΔ࣌ʹϓϩδΣΫτҠಈʹͳͬͨ • ଞexporterͷલஈͰproxyͱͯ͠ಈ࡞ͤ͞Δ • શ݅αϯϓϦϯάͭͭ͠ɺҰఆ࣌ؒ͝ͱʹ࠷΋ϨΠςϯ γͷߴ͍τϨʔε͚ͩΛ࠾༻͢Δ • Φʔόʔϔου͸͋Δ
30. spanϥΠϑαΠΫϧ • ࠓճͷΑ͏ʹTail LatencyΛั·͑ΔͨΊʹ͸શ݅αϯ ϓϦϯά͢Δඞཁ͕ग़ͯ͘Δ • αϯϓϦϯάର৅Ͱ͸ͳ͍spanʹର͢Δૢ࡞͸ૣظϦ λʔϯ͢ΔΑ͏ʹͳ͍ͬͯΔ • αϯϓϦϯάର৅ʹͳ͍ͬͯΔ͚ͩͰएׯͷΦʔόʔ ϔου͕͋Δ _, span := trace.StartSpan(ctx, name) // αϯϓϦϯάܾఆ. SpanDatasੜ੒ span.AddAttributes(...) // αϯϓϦϯάର৅֎ͳΒεΩοϓ ... span.End() // ExportSpan͕ݺ͹ΕΔ
31. edge exporter࣮૷ type EdgeExporter struct { // Stackdriver exporterͳͲͷଞexporterΛอ࣋͢Δ exporters exportersList // Tail LatencyͷܭଌִؒͱRateLimit interval time.Duration limiter *rate.Limiter // interval͝ͱͷTail LatencyΛอ࣋͢Δ tail *trace.SpanData tailLatency time.Duration tailMu sync.Mutex } • EdgeExporterߏ଄ମ
32. edge exporter࣮૷ • ExportSpan() func (e *EdgeExporter) ExportSpan(sd *trace.SpanData) { e.storeTailLatencySpan(sd) if !e.limiter.Allow() { return } e.tailMu.Lock() defer e.tailMu.Unlock() if e.tail == nil { return } for _, exp := range e.exporters.Load() { exp.ExportSpan(e.tail) } e.tail = nil e.tailLatency = 0 }
33. example func main() { sd, _ := stackdriver.NewExporter(stackdriver.Options{ ProjectID: os.Getenv("PROJECT_ID"), }) edge := edgeexporter.New(edgeexporter.WithExportInterval(10*time.Second)) edge.RegisterExporter(sd) trace.RegisterExporter(edge) trace.ApplyConfig(trace.Config{ DefaultSampler: trace.AlwaysSample(), }) http.HandleFunc("/", handler) http.ListenAndServe(":8080", nil) } var cnt int64 func handler(w http.ResponseWriter, req *http.Request) { _, span := trace.StartSpan(context.Background(), "handler") c := atomic.LoadInt64(&cnt) if c%5 == 0 { time.Sleep(50 * time.Millisecond) } span.End() atomic.AddInt64(&cnt, 1) } https://github.com/takashabe/edge-exporter/blob/master/example/main.go
34. example func main() { sd, _ := stackdriver.NewExporter(stackdriver.Options{ ProjectID: os.Getenv("PROJECT_ID"), }) edge := edgeexporter.New(edgeexporter.WithExportInterval(10*time.Second)) edge.RegisterExporter(sd) trace.RegisterExporter(edge) trace.ApplyConfig(trace.Config{ DefaultSampler: trace.AlwaysSample(), }) http.HandleFunc("/", handler) http.ListenAndServe(":8080", nil) } var cnt int64 func handler(w http.ResponseWriter, req *http.Request) { _, span := trace.StartSpan(context.Background(), "handler") c := atomic.LoadInt64(&cnt) if c%5 == 0 { time.Sleep(50 * time.Millisecond) } span.End() atomic.AddInt64(&cnt, 1) } https://github.com/takashabe/edge-exporter/blob/master/example/main.go 10ඵ͝ͱʹTail Latency SpanΛ Stackdriver exporterʹྲྀ͢
35. Stackdriver Trace্ͷදࣔ $ echo "GET https://…” | vegeta attack -rate=50 -duration=1m
36. custom exporterͷεεϝ • ࠓճ͸Tail LatencyΛର৅ͱ͕ͨ͠ɺྫ͑͹Τϥʔ͕ൃ ੜͨ͠ͱ͖͚ͩɺಛఆͷϦΫΤετ͚ͩͳͲԠ༻͸ޮ ͖ͦ͏ • ܭଌ։࢝࣌఺(StartSpan)Ͱ൑அग़དྷͳ͍΋ͷΛܭଌ͢ ΔͨΊʹ͸ࠓճͷΑ͏ʹύϑΥʔϚϯεͱͷτϨʔυ ΦϑʹͳΓͦ͏
37. Conclusion
38. Conclusion • Cloud Pub/Sub༻ͷcontext propagationͱɺTail Latency༻ͷexporterΛ࡞ͬͯΈͨ • ಠ࣮ࣗ૷ͷϋʔυϧ͸ͦΕ΄Ͳߴ͘ͳ͍ • ؍ଌ͍ͨ͠΋ͷΛ؍ଌग़དྷΔΑ͏ʹ͠Α͏