gRPC-Go+CloudTraceで分散トレーシング

はじめに

以前

以前にAkka gRPC（Scala）+CloudTraceで分散トレーシングなんて記事を書きましたが、Akka gRPCだけじゃなくてGo言語（gRPC-Go）でもサービスが動いてたんで追加してみました。

※CloudTraceおよびOpenTelemetryについての説明は割愛（上記記事または各ドキュメントを参照してください）。

※GCPドキュメント上ではGoでのOpenTelemetryの利用はβ扱いになっています。使用ライブラリであるOpenTelemetry-Goを見ると「Traces」はStableになっているのでトレースのみなら問題なさそうです。

Goで実装

導入は楽だよ

Akka gRPCの時もそうでしたが分散トレーシングを考えない場合、GoとOpenTelemetryを参考にすれば特に苦も無く実装できます。

ただ、Akka gRPC同様に分散トレーシングしたい場合はコンテキストをプロパゲーション（伝搬）する必要があります。

Akka gRPCではJavaとの兼ね合いでコンテキストのパーサーであるプロパーゲーター（伝搬屋）を自力実装したりInterceptorが無い等、結構面倒なことになってましたね。

それに比べてgRPC-Goは、W3C準拠のプロパゲーターがそのまま使えるし、Interceptorもあるしで楽勝です。

前準備

GoとOpenTelemetryの通り、必要なライブラリをインポートし、トレーサーを初期化します。

そこにプロパゲーターを指定する設定を追加します。←必須

// https://cloud.google.com/trace/docs/setup/go-otを一部改変
package main

import (
        "context"
        "log"
        "os"

        texporter "github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/trace"
        "go.opentelemetry.io/otel"
        sdktrace "go.opentelemetry.io/otel/sdk/trace"
)

func main() {
        ctx := context.Background()
        projectID := os.Getenv("GOOGLE_CLOUD_PROJECT")
        exporter, err := texporter.New(texporter.WithProjectID(projectID))
        if err != nil {
                log.Fatalf("texporter.NewExporter: %v", err)
        }

        tp := sdktrace.NewTracerProvider(sdktrace.WithBatcher(exporter))
        defer tp.ForceFlush(ctx) // flushes any pending spans
        otel.SetTracerProvider(tp)

        // ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓
        otel.SetTextMapPropagator(propagation.TraceContext{})
        // ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑
	
        tracer := otel.GetTracerProvider().Tracer("example.com/trace")
        err = func(ctx context.Context) error {
                ctx, span := tracer.Start(ctx, "foo")
                defer span.End()

                // Do some work.

                return nil
        }(ctx)
}

Interceptorを実装

gRPC-GoにはInterceptorがあるのでそこにトレーサーを組み込みましょう。

その前にキャリアを更新・削除用にプロパゲーターに引き渡すTextMapCarrierインタフェースに従った定義が必要になります。今回はgRPC-GoなのでMetadata（google.golang.org/grpc/metadata.MD）を拡張します。

package tracing

// importは省略

type MetadataCarrier metadata.MD

func (mc MetadataCarrier) Get(key string) string {
	md := metadata.MD(mc)

	v := md.Get(key)
	if len(v) == 0 {
		return ""
	}

	return v[0]
}

func (mc MetadataCarrier) Set(key string, value string) {
	md := metadata.MD(mc)

	md.Append(key, value)
}

func (mc MetadataCarrier) Keys() []string {
	md := metadata.MD(mc)

	keys := make([]string, 0, len(md))

	for k := range md {
		keys = append(keys, k)
	}

	return keys
}

上記を利用してInterceptorを定義します。

Interceptorは全サービスが通ってしまうので素朴なフィルタも考えてみました。

package tracing

// importは省略

var (
	IGNORE_METHOD = map[string]interface{}{
		"/grpc.health.v1.Health/Check": nil,
		"/grpc.health.v1.Health/Watch": nil,
	}
)

func UnaryServerInterceptor() grpc.UnaryServerInterceptor {
	return func(ctx context.Context,
		req interface{},
		info *grpc.UnaryServerInfo,
		handler grpc.UnaryHandler,
	) (interface{}, error) {
		if _, ignore := IGNORE_METHOD[info.FullMethod]; ignore {
			return handler(ctx, req)
		}

		md, ok := metadata.FromIncomingContext(ctx)
		if !ok {
			return handler(ctx, req)
		}

		copyMD := md.Copy()

		ctx = otel.GetTextMapPropagator().Extract(ctx, MetadataCarrier(copyMD))

		ctx, span := otel.GetTracerProvider().Tracer(TRACER_NAME).Start(
			ctx,
			info.FullMethod,
			trace.WithSpanKind(trace.SpanKindServer),
		)
		defer span.End()

		res, err := handler(ctx, req)
		if err != nil {
			span.SetStatus(codes.Error, err.Error())
		}

		return res, err
	}
}

func StreamServerInterceptor() grpc.StreamServerInterceptor {
	return func(
		srv interface{},
		ss grpc.ServerStream,
		info *grpc.StreamServerInfo,
		handler grpc.StreamHandler,
	) error {
		if _, ignore := IGNORE_METHOD[info.FullMethod]; ignore {
			return handler(srv, ss)
		}

		ctx := ss.Context()

		md, ok := metadata.FromIncomingContext(ctx)
		if !ok {
			return handler(srv, ss)
		}

		copyMD := md.Copy()

		ctx = otel.GetTextMapPropagator().Extract(ctx, MetadataCarrier(copyMD))

		tracer := otel.GetTracerProvider().Tracer(TRACER_NAME)

		_, span := tracer.Start(
			ctx,
			info.FullMethod,
			trace.WithSpanKind(trace.SpanKindServer),
		)
		defer span.End()

		err := handler(srv, ss)
		if err != nil {
			span.SetStatus(codes.Error, err.Error())
		}

		return err
	}
}

あとはこれらをサーバーの初期化時に噛ませるだけです。

// 初期化部分のみ抜粋
grpcServer := grpc.NewServer(
	grpc.StreamInterceptor(tracing.StreamServerInterceptor()),
	grpc.UnaryInterceptor(tracing.UnaryServerInterceptor()),
)

終わりに

まとめ

• gRPC-Go+CloudTraceは楽。