OpenTelemetry 0.7.0 メッセージを解析する方法 - Amazon CloudWatch

OpenTelemetry 0.7.0 メッセージを解析する方法

このセクションでは、OpenTelemetry 0.7.0 の解析を開始する際に役立つ情報を提供します。

まず、言語固有のバインディングを取得する必要があります。これにより、OpenTelemetry 0.7.0 のメッセージを適宜の言語で解析できます。

言語固有のバインディングを取得するには

  • 手順は、使用する言語によって異なります。

    • Java を使用するには、次の Maven 依存関係を Java プロジェクトに追加します。OpenTelemetry Java >> 0.14.1。

    • 他の言語を使用するには、次の手順を実行します。

      1. 「Generating Your Classes」のリストをチェックして、言語がサポートされていることを確認します。

      2. 「Download Protocol Buffers」の手順に従って、Protobuf コンパイラをインストールします。

      3. 「v0.7.0 release」で OpenTelemetry 0.7.0 ProtoBuf 定義をダウンロードします。

      4. ダウンロードした OpenTelemetry 0.7.0 ProtoBuf 定義のルートフォルダにいることを確認します。次に、src フォルダを作成し、言語固有のバインディングを生成するコマンドを実行します。詳細は、「Generating Your Classes」を参照してください。

        以下は、Javascript バインディングを生成する方法の例です。

        protoc --proto_path=./ --js_out=import_style=commonjs,binary:src \
opentelemetry/proto/common/v1/common.proto \
opentelemetry/proto/resource/v1/resource.proto \
opentelemetry/proto/metrics/v1/metrics.proto \
opentelemetry/proto/collector/metrics/v1/metrics_service.proto

次のセクションでは、前の手順を使用して構築できる言語固有のバインディングの使用例について説明します。

Java

package com.example;

import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceRequest;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;

public class MyOpenTelemetryParser {

    public List<ExportMetricsServiceRequest> parse(InputStream inputStream) throws IOException {
        List<ExportMetricsServiceRequest> result = new ArrayList<>();

        ExportMetricsServiceRequest request;
        /* A Kinesis record can contain multiple `ExportMetricsServiceRequest`
           records, each of them starting with a header with an
           UnsignedVarInt32 indicating the record length in bytes:
            ------ --------------------------- ------ -----------------------
           |UINT32|ExportMetricsServiceRequest|UINT32|ExportMetricsService...
            ------ --------------------------- ------ -----------------------
         */
        while ((request = ExportMetricsServiceRequest.parseDelimitedFrom(inputStream)) != null) {
            // Do whatever we want with the parsed message
            result.add(request);
        }

        return result;
    }
}

Javascript

この例では、生成されたバインディングを含むルートフォルダが ./ であるとしています。

関数 parseRecord の data 引数には、次のいずれかの型を指定できます。

  • Uint8Array これが最適です

  • Buffer ノードの下では最適です

  • Array.number 8 ビット整数

const pb = require('google-protobuf')
const pbMetrics =
    require('./opentelemetry/proto/collector/metrics/v1/metrics_service_pb')

function parseRecord(data) {
    const result = []

    // Loop until we've read all the data from the buffer
    while (data.length) {
        /* A Kinesis record can contain multiple `ExportMetricsServiceRequest`
           records, each of them starting with a header with an
           UnsignedVarInt32 indicating the record length in bytes:
            ------ --------------------------- ------ -----------------------
           |UINT32|ExportMetricsServiceRequest|UINT32|ExportMetricsService...
            ------ --------------------------- ------ -----------------------
         */
        const reader = new pb.BinaryReader(data)
        const messageLength = reader.decoder_.readUnsignedVarint32()
        const messageFrom = reader.decoder_.cursor_
        const messageTo = messageFrom + messageLength

        // Extract the current `ExportMetricsServiceRequest` message to parse
        const message = data.subarray(messageFrom, messageTo)

        // Parse the current message using the ProtoBuf library
        const parsed =
            pbMetrics.ExportMetricsServiceRequest.deserializeBinary(message)

        // Do whatever we want with the parsed message
        result.push(parsed.toObject())

        // Shrink the remaining buffer, removing the already parsed data
        data = data.subarray(messageTo)
    }

    return result
}

Python

var-int デリミタは自分で読み取るか、内部メソッド _VarintBytes(size) と _DecodeVarint32(buffer, position) を使用する必要があります。これらは、バッファ内でのサイズバイトの直後の位置を返します。読み取り側は、メッセージからそのバイトのみを読み取る新しいバッファを構築します。

size = my_metric.ByteSize()
f.write(_VarintBytes(size))
f.write(my_metric.SerializeToString())
msg_len, new_pos = _DecodeVarint32(buf, 0)
msg_buf = buf[new_pos:new_pos+msg_len]
request = metrics_service_pb.ExportMetricsServiceRequest()
request.ParseFromString(msg_buf)

Go

Buffer.DecodeMessage() を使用します。

C#

CodedInputStream を使用します。このクラスは、サイズ区切りのメッセージを読み取ることができます。

C++

google/protobuf/util/delimited_message_util.h に記述されている関数は、サイズ区切りのメッセージを読み取ることができます。

その他の言語

その他の言語については、「Download Protocol Buffers」を参照してください。

パーサーを実装するときは、Kinesis レコードに複数の ExportMetricsServiceRequest Protocol Buffers メッセージを含めることができることを考慮してください。各メッセージは、レコードの長さをバイト単位で示す UnsignedVarInt32 のヘッダーで始まります。