Upgrade to Pro — share decks privately, control downloads, hide ads and more …

OpenCensusでcustom context propagationとexporterを...

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

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

takashabe

June 27, 2019
Tweet

More Decks by takashabe

Other Decks in Technology

Transcript

  1. whoami • Takashi Abe (@takashabe) • גࣜձࣾαΠόʔΤʔδΣϯτ ΞυςΫελδΦ • GoͰ޿ࠂ഑৴ϓϩμΫτΛ࡞͍ͬͯ·͢

    • KubernetesͰMicroservicesతͳ΋ͷ • ਪ͠ΩʔϘʔυ͸irisͰ͢ • Corne Cherry࡞੒த…
  2. 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༻ }
  3. 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) }
  4. 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
  5. 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
  6. 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͸ҙࣝ͠ͳͯ͘ྑ͍
  7. Cloud Pub/Sub ʹద༻ͨ͠ • GCPͷϚωʔδυPub/Sub • B3 PropagationΛϕʔεʹͯ͠ɺϝοηʔδͷ AttributesʹσʔλΛ٧Ίͯ఻ൖͤ͞Δ •

    https://github.com/takashabe/oc-propagation-demo • ΄΅ಉ͡΋ͷΛϓϩμΫγϣϯʹಋೖࡁΈ͕ͩɺ؀ڥ ґଘ͕͋ΔͷͰαϯϓϧ࣮૷Λ༻ҙ • αʔό༻ҙ͢Δͷ໘౗ͰCloud Run༻ʹHTTPαʔό ͱ࣮ͯ͠૷͍ͯ͠ΔͷͰएׯݟͮΒ͍
  8. 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 }
  9. 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, } }
  10. ར༻ଆ // 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)
  11. exporterͷ࣮૷(Go) type Exporter interface { ExportSpan(s *SpanData) } • ExporterΠϯλϑΣʔεΛ࣮૷͢Δ͚ͩͰྑ͍

    • ೚ҙͷόοΫΤϯυαʔϏεAPIͳͲΛୟ͍ͯσʔλΛ ૹ͍ͬͯΔ • ࣮ࡍʹ͸ύϑΥʔϚϯεͰ͔ͳΓؾΛ࢖͍ͬͯΔ(͸ͣ)
  12. Tail LatencyΛิ଍͢ΔͨΊͷexporter • Tail Latency • 99ύʔηϯλΠϧ஋ͷΑ͏ͳ΋ͷ • ௨ৗɺτϥϑΟοΫͷଟ͍αʔϏεͩͱαϯϓϦϯά ͷඞཁ͕͋Δ͕ɺͦ͏͢ΔͱTail

    Latency͕શ͘ه࿥͞ Εͳ͍ • Tail LatencyͰ೰ΜͰ͍Δͱ͖͸मਖ਼ɺܭଌͷαΠΫ ϧΛճ͍͕ͨ͠ɺ؍ଌ͢ΔͨΊʹ਺࣌ؒҎ্͔͔Δ ͜ͱ΋͋Δ…
  13. edge exporter • https://github.com/takashabe/edge-exporter • ϓϩμΫγϣϯ౤ೖग़དྷͯͳ͍ͷͰࢀߟ࣮૷ͱͯ͠… • ύϑΥʔϚϯε͸վળͷ༨஍͕େ͍ʹ࢒ͬͯΔ • ߏ૝Λ࿅͍ͬͯΔ࣌ʹϓϩδΣΫτҠಈʹͳͬͨ

    • ଞexporterͷલஈͰproxyͱͯ͠ಈ࡞ͤ͞Δ • શ݅αϯϓϦϯάͭͭ͠ɺҰఆ࣌ؒ͝ͱʹ࠷΋ϨΠςϯ γͷߴ͍τϨʔε͚ͩΛ࠾༻͢Δ • Φʔόʔϔου͸͋Δ
  14. spanϥΠϑαΠΫϧ • ࠓճͷΑ͏ʹTail LatencyΛั·͑ΔͨΊʹ͸શ݅αϯ ϓϦϯά͢Δඞཁ͕ग़ͯ͘Δ • αϯϓϦϯάର৅Ͱ͸ͳ͍spanʹର͢Δૢ࡞͸ૣظϦ λʔϯ͢ΔΑ͏ʹͳ͍ͬͯΔ • αϯϓϦϯάର৅ʹͳ͍ͬͯΔ͚ͩͰएׯͷΦʔόʔ

    ϔου͕͋Δ _, span := trace.StartSpan(ctx, name) // αϯϓϦϯάܾఆ. SpanDatasੜ੒ span.AddAttributes(...) // αϯϓϦϯάର৅֎ͳΒεΩοϓ ... span.End() // ExportSpan͕ݺ͹ΕΔ
  15. 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ߏ଄ମ
  16. 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 }
  17. 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
  18. 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ʹྲྀ͢