簡単なOtel Collectorクローンを作りながら入門OTLP

この記事はOpenTelemetry Advent Calendar 202517日目の記事です。

はじめに

この記事では、OTLPについて触れながらの受信・送信ができるollectorクローンを作ってOTLPについて理解することを目指してます。

OpenTelemetryとは

公式サイトより、

OpenTelemetry is a collection of APIs, SDKs, and tools. Use it to instrument, generate, collect, and export telemetry data (metrics, logs, and traces) to help you analyze your software’s performance and behavior.

と記載されてます。
簡単にまとめると、メトリクス、ログ、トレースなどを取得するためのAPIやSDK、仕様のことを総称してOpenTelmetryと呼びます。

OTLP(OpenTelemetry Protocol)とは？

OTLP(OpenTelemetry Protocol)はテレメトリデータの送受信方法を定義したプロトコルです。
↓仕様はOpenTelmetryの公式サイトに載ってます。
https://opentelemetry.io/docs/specs/otlp/

今回はこの仕様に則って送受信ができるようにしていきます。

OpenTelemetry Collectorとは？

テレメトリーデータを受信、処理、エクスポートするためのソフトウェアです。
様々なベンダーが提供するディストリビューション（例えば、AWSが提供するADOT Collectorなど）やOpenTelemetry Collector Builderを使った自作ビルドのOtel Collectorを使ったりできます。

collectorクローンで学ぶOTLPとトレースの加工

Collectorクローンを作りつつ、OTLPについて調べていきます。
また、対象とするメトリクスはトレースとします。
ログはfluent-bit, メトリクスはCloudWatch Agentなどを使用することが多く、実際にOtel Collectorが使われるユースケースとして一番おおいのは、トレースだと考えてのことです。

受信の仕様確認

OTLPの受信部分、いわゆるreciver部分では、HTTPとgRPCの２種類があります。
今回はHTTPのみに対応します。
OTLP/HTTPの仕様はここに載ってます。
ドキュメントの記載から、reciverは

1. /v1/tracesのパスでPOSTリクエストを受け取る
   1. コンテンツ圧縮されてるかリクエストヘッダーで判断する
2. バイナリエンコードされたProtobuf形式またはJSONエンコードされたProtobuf形式のペイロードを取り出す
   1. エンコードされたprotobufはContent-Type: application/x-protobufをつけて送られてくる必要があるので、それの検査をする
3. Language Guide(proto3)にしたがってデコードする
4. テレメトリデータを受け入れたかレスポンスを返す
   1. 成功した場合: 200 okを返す。ボディはProtobuf でエンコードされた Export<signal>ServiceResponseメッセージ
   2. 失敗した場合: 400番台や500番台のステータスで返す
   3. リクエストが大量に来た場合: 429: Too Many Requestを返す

等の動作をすれば良さそうです。

受信部分の実装

受信部分を実装していきます。
大きな流れとしてはHTTPサーバーを用意、bodyをprotobufの仕様にしたがってデコード、表示です。
以下がコードです。

package main

import (
	"compress/gzip"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"os"
	"strings"
	"time"

	collectortracev1 "go.opentelemetry.io/proto/otlp/collector/trace/v1"
	"google.golang.org/genproto/googleapis/rpc/status"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

func main() {
	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
		Level: slog.LevelInfo,
	}))
	slog.SetDefault(logger)

	mux := http.NewServeMux()
	mux.HandleFunc("POST /v1/traces", otlpHTTPTracesHandler)

	srv := &http.Server{
		Addr:              ":4318",
		Handler:           mux,
		ReadHeaderTimeout: 5 * time.Second,
	}

	slog.Info("otlp http trace receiver started",
		"addr", srv.Addr,
		"endpoint", "http://0.0.0.0:4318/v1/traces",
	)
	if err := srv.ListenAndServe(); err != nil {
		slog.Error("server stopped", "err", err)
	}
}

func otlpHTTPTracesHandler(w http.ResponseWriter, r *http.Request) {
	ct := r.Header.Get("Content-Type")
	if !strings.HasPrefix(ct, "application/x-protobuf") {
		writeOTLPError(w, http.StatusUnsupportedMediaType, "only Content-Type: application/x-protobuf is supported")
		slog.Warn("rejected request", "reason", "unsupported content-type", "content_type", ct, "path", r.URL.Path)
		return
	}

	body, err := readMaybeGzip(r)
	if err != nil {
		writeOTLPError(w, http.StatusBadRequest, "failed to read body: "+err.Error())
		slog.Warn("bad request", "reason", "read body failed", "err", err, "path", r.URL.Path)
		return
	}

	var req collectortracev1.ExportTraceServiceRequest
	if err := proto.Unmarshal(body, &req); err != nil {
		writeOTLPError(w, http.StatusBadRequest, "failed to unmarshal ExportTraceServiceRequest: "+err.Error())
		slog.Warn("bad request", "reason", "unmarshal failed", "err", err, "bytes", len(body), "path", r.URL.Path)
		return
	}

	j, err := protojson.MarshalOptions{
		Multiline:     true,
		Indent:        "  ",
		UseProtoNames: true,
	}.Marshal(&req)
	if err != nil {
		writeOTLPError(w, http.StatusInternalServerError, "failed to marshal protojson: "+err.Error())
		slog.Error("internal error", "reason", "protojson marshal failed", "err", err)
		return
	}

	slog.Info("received traces",
		"remote_addr", r.RemoteAddr,
		"bytes", len(body),
		"content_encoding", r.Header.Get("Content-Encoding"),
	)

	fmt.Println("----- otlp ExportTraceServiceRequest (protojson) BEGIN -----")
	fmt.Println(string(j))
	fmt.Println("----- otlp ExportTraceServiceRequest (protojson) END -----")

	// 成功レスポンス（ExportTraceServiceResponse）
	resp := &collectortracev1.ExportTraceServiceResponse{}
	out, _ := proto.Marshal(resp)

	w.Header().Set("Content-Type", "application/x-protobuf")
	w.WriteHeader(http.StatusOK)
	_, _ = w.Write(out)
}

var ErrBodyTooLarge = errors.New("request body too large")

func readMaybeGzip(r *http.Request) ([]byte, error) {
	defer r.Body.Close()

	var reader io.Reader = r.Body
	if strings.EqualFold(r.Header.Get("Content-Encoding"), "gzip") {
		gz, err := gzip.NewReader(r.Body)
		if err != nil {
			return nil, err
		}
		defer gz.Close()
		reader = gz
	}

	const maxBody = 16 << 20 // 16MiB
	body, err := io.ReadAll(io.LimitReader(reader, maxBody))
	if err != nil {
		return nil, err
	}

	if int64(len(body)) > maxBody {
		return nil, fmt.Errorf("%w: limit=%d bytes", ErrBodyTooLarge, maxBody)
	}

	return body, nil
}

func writeOTLPError(w http.ResponseWriter, httpStatus int, msg string) {
	st := &status.Status{
		Code:    int32(httpStatus), // 簡易: HTTPステータスをそのまま入れる
		Message: msg,
	}
	b, _ := proto.Marshal(st)

	w.Header().Set("Content-Type", "application/x-protobuf")
	w.WriteHeader(httpStatus)
	_, _ = w.Write(b)
}

これを動かして、実際にトレースを送ってみます。
↓トレース送信コマンド

export OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://localhost:4318/v1/traces
otel-cli span --service test-svc --name "hello-span" --attrs foo=bar,env=dev

画像のようになりました。
jsonにデコードして表示できており、受信できてそうです。
ちなみにPOST以外を送るとGoのWebサーバーの仕様で弾けるので、仕様にも準拠してます。

送信部分の実装

今度は受信コードに送信処理もくっつけてみます。
一度展開したので、バイト列をまた圧縮して送信するようにします。
Status CodeはgRPCの形式に従うように変換します。

以下は全体のコードです。

package main

import (
	"bytes"
	"compress/gzip"
	"context"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net/http"
	"os"
	"strings"
	"time"

	collectortracev1 "go.opentelemetry.io/proto/otlp/collector/trace/v1"
	"google.golang.org/genproto/googleapis/rpc/status"
	"google.golang.org/grpc/codes"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"
)

func main() {
	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
		Level: slog.LevelInfo,
	}))
	slog.SetDefault(logger)

	mux := http.NewServeMux()
	mux.HandleFunc("POST /v1/traces", otlpHTTPTracesHandler)

	srv := &http.Server{
		Addr:              ":4318",
		Handler:           mux,
		ReadHeaderTimeout: 5 * time.Second,
	}

	slog.Info("otlp http trace receiver started",
		"addr", srv.Addr,
		"endpoint", "http://0.0.0.0:4318/v1/traces",
	)

	if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { // CHANGED
		slog.Error("server stopped", "err", err)
	}
}

func otlpHTTPTracesHandler(w http.ResponseWriter, r *http.Request) {
	ct := r.Header.Get("Content-Type")
	if !strings.HasPrefix(ct, "application/x-protobuf") {
		writeOTLPError(w, http.StatusUnsupportedMediaType, "only Content-Type: application/x-protobuf is supported")
		slog.Warn("rejected request", "reason", "unsupported content-type", "content_type", ct, "path", r.URL.Path)
		return
	}

	body, err := readMaybeGzip(r)
	if err != nil {
		writeOTLPError(w, http.StatusBadRequest, "failed to read body: "+err.Error())
		slog.Warn("bad request", "reason", "read body failed", "err", err, "path", r.URL.Path)
		return
	}

	var req collectortracev1.ExportTraceServiceRequest
	if err := proto.Unmarshal(body, &req); err != nil {
		writeOTLPError(w, http.StatusBadRequest, "failed to unmarshal ExportTraceServiceRequest: "+err.Error())
		slog.Warn("bad request", "reason", "unmarshal failed", "err", err, "bytes", len(body), "path", r.URL.Path)
		return
	}

	j, err := protojson.MarshalOptions{
		Multiline:     true,
		Indent:        "  ",
		UseProtoNames: true,
	}.Marshal(&req)
	if err != nil {
		writeOTLPError(w, http.StatusInternalServerError, "failed to marshal protojson: "+err.Error())
		slog.Error("internal error", "reason", "protojson marshal failed", "err", err)
		return
	}

	slog.Info("received traces",
		"remote_addr", r.RemoteAddr,
		"bytes", len(body),
		"content_encoding", r.Header.Get("Content-Encoding"),
	)

	fmt.Println("----- otlp ExportTraceServiceRequest (protojson) BEGIN -----")
	fmt.Println(string(j))
	fmt.Println("----- otlp ExportTraceServiceRequest (protojson) END -----")

	if err := exportTrace(r.Context(), body); err != nil {
		slog.Error("export failed", "err", err)
	} else {
		slog.Info("export traces")
	}

	// 成功レスポンス（ExportTraceServiceResponse）
	resp := &collectortracev1.ExportTraceServiceResponse{}
	out, _ := proto.Marshal(resp)

	w.Header().Set("Content-Type", "application/x-protobuf")
	w.WriteHeader(http.StatusOK)
	_, _ = w.Write(out)
}

var (
	ExportEndPoint = "http://localhost:1418/v1/traces"
	httpClient     = &http.Client{Timeout: 10 * time.Second}
)

func exportTrace(ctx context.Context, body []byte) error {
	var bodyReader io.Reader

	var buf bytes.Buffer
	gw := gzip.NewWriter(&buf)
	if _, err := gw.Write(body); err != nil {
		_ = gw.Close()
		return err
	}

	if err := gw.Close(); err != nil {
		return err
	}

	payload := buf.Bytes()
	contentEncoding := "gzip"
	retry := 5
	var lastErr error
	for i := 0; i < retry; i++ {
		bodyReader = bytes.NewReader(payload)
		req, err := http.NewRequestWithContext(ctx, http.MethodPost, ExportEndPoint, bodyReader)
		if err != nil {
			return err
		}

		req.Header.Set("Content-Type", "application/x-protobuf")
		req.Header.Set("Content-Encoding", contentEncoding)
		resp, err := httpClient.Do(req)
		if err != nil {
			lastErr = err
			time.Sleep(1 * time.Second)
			continue
		}

		respBody, _ := io.ReadAll(resp.Body)
		_ = resp.Body.Close()

		if resp.StatusCode == http.StatusOK {
			var otlpResp collectortracev1.ExportTraceServiceResponse
			if err := proto.Unmarshal(respBody, &otlpResp); err != nil {
				return fmt.Errorf("forward: 200 but failed to unmarshal ExportTraceServiceResponse: %w", err)
			}
			if ps := otlpResp.GetPartialSuccess(); ps != nil && ps.GetRejectedSpans() > 0 {
				slog.Warn("forward: partial success",
					"rejected_spans", ps.GetRejectedSpans(),
					"error_message", ps.GetErrorMessage(),
				)
			}
			return nil
		}

		var st status.Status
		if err := proto.Unmarshal(respBody, &st); err == nil && (st.Message != "" || st.Code != 0) {
			lastErr = fmt.Errorf("forward failed: http=%d status_code=%d msg=%q", resp.StatusCode, st.Code, st.Message)
		} else {
			lastErr = fmt.Errorf("forward failed: http=%d body_bytes=%d", resp.StatusCode, len(respBody))
		}

		time.Sleep(3 * time.Second)
	}

	slog.Error("retry count over.")
	return fmt.Errorf("forward failed after retries: %w", lastErr)
}

func httpStatusTogRPCCode(httpStatus int) codes.Code {
	switch httpStatus {
	case http.StatusBadRequest:
		return codes.InvalidArgument
	case http.StatusUnsupportedMediaType:
		return codes.InvalidArgument
	case http.StatusRequestEntityTooLarge:
		return codes.ResourceExhausted
	case http.StatusTooManyRequests:
		return codes.ResourceExhausted
	case http.StatusNotFound:
		return codes.Unimplemented
	default:
		if httpStatus >= 500 {
			return codes.Internal
		}
		return codes.Unknown
	}
}

var ErrBodyTooLarge = errors.New("request body too large")

func readMaybeGzip(r *http.Request) ([]byte, error) {
	defer r.Body.Close()

	var reader io.Reader = r.Body
	if strings.EqualFold(r.Header.Get("Content-Encoding"), "gzip") {
		gz, err := gzip.NewReader(r.Body)
		if err != nil {
			return nil, err
		}
		defer gz.Close()
		reader = gz
	}

	const maxBody = 16 << 20 // 16MiB
	body, err := io.ReadAll(io.LimitReader(reader, maxBody+1))
	if err != nil {
		return nil, err
	}

	if int64(len(body)) > maxBody {
		return nil, fmt.Errorf("%w: limit=%d bytes", ErrBodyTooLarge, maxBody)
	}

	return body, nil
}

func writeOTLPError(w http.ResponseWriter, httpStatus int, msg string) {
	st := &status.Status{
		Code:    int32(httpStatusTogRPCCode(httpStatus)), // 簡易: HTTPステータスをそのまま入れる
		Message: msg,
	}
	b, _ := proto.Marshal(st)

	w.Header().Set("Content-Type", "application/x-protobuf")
	w.WriteHeader(httpStatus)
	_, _ = w.Write(b)
}

では、otel-cli→自作Collector→jeagerでトレースの送受信ができるか試してみます。
Collectorはgo run で起動しておいてください。
jeagerは以下のcompose.yamlで動かします。

services:
  jaeger:
    image: jaegertracing/all-in-one:1.76.0
    container_name: jaeger
    ports:
      - "16686:16686"   # Jaeger UI
      - "1418:4318"     # OTLP/HTTP receiver (host:1418 -> container:4318)
      - "4317:4317"     # OTLP/gRPC receiver (任意)
    restart: unless-stopped

otel-cliからのトレースの受信、jeagerへの送信ともにできてそうです！

まとめ

OTLPを受信・送信できるOtel Collectorクローンを作りました。

• 受信はPOSTで/v1/tracesに送る・bodyをExportTraceServiceRequest型にUnmarshalする・protobufメッセージをJSONにシリアライズする
• 送信は逆にgzipして送る
  これだけ理解していただければOKです！
  なんかよくわからないけどSDKとか使っておくれば受信できるから一歩すすんで、jsonのメッセージの受け渡しであることが理解できたかと思います。
  OTLPの理解のきかっけになれば幸いです。
  今回作ったように送受信ができれば中身の加工などは自由自在に変更できそうです。
  今回は間に合わず実装できなかったですが、tail samplingなどの再現もしてみたいです。
  ここまで読んでいただきありがとうございました！

参考文献

• https://opentelemetry.io/docs/specs/otlp/