AWS Kinesis 上で OpenTelemetry トレースヘッダを伝搬する方法: パート3

PartitionKeyとExplicitHashKeyを使って統合する

2024年7月3日
翻訳: 竹洞 陽一郎

この記事は米Catchpoint Systems社のブログ記事「How to propagate OpenTelemetry trace headers over AWS Kinesis: Part 3」の翻訳です。
Spelldataは、Catchpointの日本代理店です。
この記事は、Catchpoint Systemsの許可を得て、翻訳しています。

AWS KinesisでのOpenTelemetryを使用したトレースに関するシリーズの続きとして、今回はPartitionKeyパラメータを使用した最初の試みで直面した課題を克服するための洗練されたアプローチを探ります。

2回目の試みと解決策: PartitionKeyを通じてトレースコンテキストを伝播し、ExplicitHashKeyパラメータでパーティションを設定する

最初の試みでは、PartitionKeyパラメータを使用してtraceparentを伝播することができましたが、その際にイベントの正しいパーティション割り当てが乱れてしまいました。
では、PartitionKeyパラメータでtraceparentを伝播する際に、ユーザが指定した元のPartitionKey値に従ってイベントが正しいシャードに割り当てられるようにするにはどうすれば良いでしょうか？

ここでの答えは、ExplicitHashKeyパラメータです。
しかし、解決策に入る前に、AWS Kinesisサービスにおけるキー空間のパーティショニングについて見てみましょう。

以下は、AWS Kinesis APIリファレンスドキュメントからの、レコードをシャードにパーティショニングする際のPartitionKeyの使用方法に関する記述です。

PartitionKeyは、Kinesis Data Streamsがデータをシャードに分散するために使用されます。
Kinesis Data Streamsは、各データレコードに関連付けられたPartitionKeyを使用して、ストリームに属するデータレコードを複数のシャードに分けます。

PartitionKeyはUnicode文字列であり、各キーの最大長は256文字です。
MD5ハッシュ関数が使用され、PartitionKeyを128ビットの整数値にマッピングし、関連するデータレコードをシャードのハッシュキー範囲にマッピングします。
ExplicitHashKeyパラメータを使用して、シャードを決定するためのハッシュ値を明示的に指定することで、PartitionKeyのハッシュ処理を上書きすることができます。

したがって、パーティショニングに使用されるハッシュキーは、次のようにPartitionKeyから計算されます。

HashKey = Int128(MD5())

これは、ハッシュキーの範囲（およびKinesisキー空間）が次のようになることを意味します。

[0, 2^128) => [0, 340282366920938463463374607431768211455]

Kinesisシャードの作成、分割、およびマージがキー範囲に与える影響

Kinesisシャードを作成、分割、マージすることで、シャードのキー範囲がどのように影響を受けるかを理解するために、少し試してみましょう。

#1 - eventsストリームを作成する

> aws kinesis create-stream --stream-name events --shard-count 1

#2 - eventsストリームが作成されたことを確認する

> aws kinesis describe-stream --stream-name events 

{ 
    "StreamDescription": { 
        "Shards": [ 
            { 
                "ShardId": "shardId-000000000000", 
                "HashKeyRange": { 
                    "StartingHashKey": "0", 
                    "EndingHashKey": "340282366920938463463374607431768211455" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646421264737784557324905820411013094940000525450477570" 
                } 
            } 
        ], 
        "StreamARN": "arn:aws:kinesis:us-west-2:************:stream/events", 
        "StreamName": "events", 
        "StreamStatus": "ACTIVE", 
        "RetentionPeriodHours": 24, 
        "EnhancedMonitoring": [ 
            { 
                "ShardLevelMetrics": [] 
            } 
        ], 
        "EncryptionType": "NONE", 
        "KeyId": null, 
        "StreamCreationTimestamp": "2023-11-14T12:31:26+03:00" 
    } 
}

ご覧のとおり、キー空間全体[0, 340282366920938463463374607431768211455]が単一のシャードshardId-000000000000 に割り当てられています。

#3 - シャード数を2に増やしてみましょう

> aws kinesis update-shard-count --stream-name events --target-shard-count 2 --scaling-type UNIFORM_SCALING 

{ 
    "StreamName": "events", 
    "CurrentShardCount": 1, 
    "TargetShardCount": 2, 
    "StreamARN": "arn:aws:kinesis:us-west-2:************:stream/events" 
} 

シャード数を増やすと、シャードshardId-000000000000の分割がトリガーされ、子シャード(shardId-000000000001とshardId-000000000002)に分割されます。

> aws kinesis describe-stream --stream-name events 

{ 
    "StreamDescription": { 
        "Shards": [ 
            { 
                "ShardId": "shardId-000000000000", 
                "HashKeyRange": { 
                    "StartingHashKey": "0", 
                    "EndingHashKey": "340282366920938463463374607431768211455" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646421264737784557324905820411013094940000525450477570", 
                    "EndingSequenceNumber": "49646421264748934929924171131980572028256847859579617282" 
                } 
            }, 
            { 
                "ShardId": "shardId-000000000001", 
                "ParentShardId": "shardId-000000000000", 
                "HashKeyRange": { 
                    "StartingHashKey": "0", 
                    "EndingHashKey": "170141183460469231731687303715884105727" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646421980524803194562316794282962492134597634174222354" 
                } 
            }, 
            { 
                "ShardId": "shardId-000000000002", 
                "ParentShardId": "shardId-000000000000", 
                "HashKeyRange": { 
                    "StartingHashKey": "170141183460469231731687303715884105728", 
                    "EndingHashKey": "340282366920938463463374607431768211455" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646421980547103939760847417424498210407245995680202786" 
                } 
            } 
        ], 
        "StreamARN": "arn:aws:kinesis:us-west-2:************:stream/events", 
        "StreamName": "events", 
        "StreamStatus": "ACTIVE", 
        "RetentionPeriodHours": 24, 
        "EnhancedMonitoring": [ 
            { 
                "ShardLevelMetrics": [] 
            } 
        ], 
        "EncryptionType": "NONE", 
        "KeyId": null, 
        "StreamCreationTimestamp": "2023-11-14T12:31:26+03:00" 
    } 
} 

応答からわかるように、シャードのハッシュキー範囲は不変です。
シャードが分割されると、親シャード (ここではshardId-000000000000)はCLOSED状態に切り替わり、その全ハッシュキー範囲は子シャードshardId-000000000001([0, 170141183460469231731687303715884105727])とshardId-000000000002([170141183460469231731687303715884105728, 340282366920938463463374607431768211455])に引き継がれ、均等に二分されます。

シャードのCLOSED状態は、そのシャードの説明にEndingSequenceNumberプロパティが含まれていることで識別できます。

#4 - それでは、シャード数を1に戻すとどうなるか見てみましょう

> aws kinesis update-shard-count --stream-name events --target-shard-count 1 --scaling-type UNIFORM_SCALING 

{ 
    "StreamName": "events", 
    "CurrentShardCount": 2, 
    "TargetShardCount": 1, 
    "StreamARN": "arn:aws:kinesis:us-west-2:************:stream/events" 
} 

シャード数を減らすと、shardId-000000000001とshardId-000000000002のシャードが新しい親シャード（shardId-000000000003）にマージされます:

> aws kinesis describe-stream --stream-name events  

{ 
    "StreamDescription": { 
        "Shards": [ 
            { 
                "ShardId": "shardId-000000000000", 
                "HashKeyRange": { 
                    "StartingHashKey": "0", 
                    "EndingHashKey": "340282366920938463463374607431768211455" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646421264737784557324905820411013094940000525450477570", 
                    "EndingSequenceNumber": "49646421264748934929924171131980572028256847859579617282" 
                } 
            }, 
            { 
                "ShardId": "shardId-000000000001", 
                "ParentShardId": "shardId-000000000000", 
                "HashKeyRange": { 
                    "StartingHashKey": "0", 
                    "EndingHashKey": "170141183460469231731687303715884105727" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646421980524803194562316794282962492134597634174222354", 
                    "EndingSequenceNumber": "49646421980535953567161582105852521425451370064073719826" 
                } 
            }, 
            { 
                "ShardId": "shardId-000000000002", 
                "ParentShardId": "shardId-000000000000", 
                "HashKeyRange": { 
                    "StartingHashKey": "170141183460469231731687303715884105728", 
                    "EndingHashKey": "340282366920938463463374607431768211455" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646421980547103939760847417424498210407245995680202786", 
                    "EndingSequenceNumber": "49646421980558254312360112728994057143724018425579700258" 
                } 
            }, 
            { 
                "ShardId": "shardId-000000000003", 
                "ParentShardId": "shardId-000000000001", 
                "AdjacentParentShardId": "shardId-000000000002", 
                "HashKeyRange": { 
                    "StartingHashKey": "0", 
                    "EndingHashKey": "340282366920938463463374607431768211455" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646422307409126314624190802913520932614343535875850290" 
                } 
            } 
        ], 
        "StreamARN": "arn:aws:kinesis:us-west-2:************:stream/events", 
        "StreamName": "events", 
        "StreamStatus": "ACTIVE", 
        "RetentionPeriodHours": 24, 
        "EnhancedMonitoring": [ 
            { 
                "ShardLevelMetrics": [] 
            } 
        ], 
        "EncryptionType": "NONE", 
        "KeyId": null, 
        "StreamCreationTimestamp": "2023-11-14T12:31:26+03:00" 
    } 
} 

シャードの分割と同様に、shardId-000000000001とshardId-000000000002のシャードはCLOSED状態で依然として存在しています。
これらのシャードはEndingSequenceNumberプロパティを持っているため、閉じられていることが示されています。

新しく作成されたシャードshardId-000000000003は、親シャードshardId-000000000001とshardId-000000000002からその履歴を継承しています。 このシャードは、shardId-000000000001をParentShardIdとして、shardId-000000000002をAdjacentParentShardIDとして指し示しています。

分割とは対照的に、分割後に新しく作成されたシャードは、親シャード (shardId-000000000001とshardId-000000000002)のハッシュキー範囲([0, 170141183460469231731687303715884105727]と[170141183460469231731687303715884105728, 340282366920938463463374607431768211455])をマージして、自身のハッシュキー範囲[0, 340282366920938463463374607431768211455]を取得します。

ExplicitHashKeyパラメータの使用方法

さて、元の問題に戻りましょう。

私たちはPartitionKeyパラメータを使用してトレースヘッダーを伝播していますが、今度は元のPartitionKeyパラメータ（ユーザによって与えられたもの）に基づいてハッシュキーを直接指定する方法を見つけなければなりません。
そうすれば、AWSはイベントを正しいシャードに配置します。

このセクションの最初に述べたように、ExplicitHashKeyパラメータを使用してハッシュキーを明示的に指定できます。
PartitionKeyに基づいたハッシュキーの計算方法を思い出しましょう。

HashKey = Int128(MD5())

そして、その実装は以下の通りです。

private String toHashKey(String partitionKey) throws Exception { 
     MessageDigest md = MessageDigest.getInstance("MD5"); 
     md.update(partitionKey.getBytes()); 
     byte[] digest = md.digest(); 
     BigInteger bi = new BigInteger(1, digest); 
     return bi.toString(); 
} 

それでは、カスタムハッシュキー計算アルゴリズムをテストして検証しましょう。
ExplicitHashKeyパラメータを通じてカスタム生成されたハッシュキーでパーティションが指定された場合に、AWS Kinesisがレコードを正しいシャードに配置するかどうかを確認します。

#1 - まず、5つのシャードを持つKinesisストリームを作成します

> aws kinesis create-stream --stream-name events --shard-count 5

#2 - 次に、ハッシュキーがシャード間でどのように分割されるか見てみましょう

> aws kinesis describe-stream --stream-name events  

{     
    "StreamDescription": { 
        "Shards": [ 
            { 
                "ShardId": "shardId-000000000000", 
                "HashKeyRange": { 
                    "StartingHashKey": "0", 
                    "EndingHashKey": "68056473384187692692674921486353642290" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646447835808079664180011482520906336196071965037953026" 
                } 
            }, 
            { 
                "ShardId": "shardId-000000000001", 
                "HashKeyRange": { 
                    "StartingHashKey": "68056473384187692692674921486353642291", 
                    "EndingHashKey": "136112946768375385385349842972707284581" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646447835830380409378542105662442054468720326543933458" 
                } 
            }, 
            { 
                "ShardId": "shardId-000000000002", 
                "HashKeyRange": { 
                    "StartingHashKey": "136112946768375385385349842972707284582", 
                    "EndingHashKey": "204169420152563078078024764459060926872" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646447835852681154577072728803977772741368688049913890" 
                } 
            }, 
            { 
                "ShardId": "shardId-000000000003", 
                "HashKeyRange": { 
                    "StartingHashKey": "204169420152563078078024764459060926873", 
                    "EndingHashKey": "272225893536750770770699685945414569163" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646447835874981899775603351945513491014017049555894322" 
                } 
            }, 
            { 
                "ShardId": "shardId-000000000004", 
                "HashKeyRange": { 
                    "StartingHashKey": "272225893536750770770699685945414569164", 
                    "EndingHashKey": "340282366920938463463374607431768211455" 
                }, 
                "SequenceNumberRange": { 
                    "StartingSequenceNumber": "49646447835897282644974133975087049209286665411061874754" 
                } 
            } 
        ], 
        "StreamARN": "arn:aws:kinesis:us-west-2:************:stream/test", 
        "StreamName": "test", 
        "StreamStatus": "ACTIVE", 
        "RetentionPeriodHours": 24, 
        "EnhancedMonitoring": [ 
            { 
                "ShardLevelMetrics": [] 
            } 
        ], 
        "EncryptionType": "NONE", 
        "KeyId": null, 
        "StreamCreationTimestamp": "2023-11-15T09:12:34+03:00" 
    } 
}

以下は、5つのシャードそれぞれのハッシュキー範囲です。

#3 - それでは、トレースコンテキストの伝播なしに、グループIDをパーティションキーとして使用し、異なるイベントグループID（group-1、…、group-20）を持つ20個のイベントをproducerアプリケーションから送信しましょう

以下は、コンシューマ関数のログから収集したパーティション割り当て結果です。

#4 - 次に、パーティションキー（ここではイベントグループID）に基づくハッシュキー生成アルゴリズムを検証するために、各イベントグループID（group-1、…、group-20）についてハッシュキーを生成するハッシュキー生成器を実行します

表1に示されている5つのシャードのハッシュキー範囲に従って、表3に示されているカスタム生成されたハッシュキーを使用すると、イベントは次のシャードに割り当てられます。

ご覧の通り、グループIDに基づくイベントのKinesisシャード割り当ては、標準のPartitionKeyメソッドとカスタム生成されたハッシュキーメソッドの両方で同じです。
これにより、ExplicitHashKeyパラメータ用にカスタムハッシュキーを生成することで、正確なシャード割り当てを維持できることが確認されました。 これにより、PartitionKeyパラメータを効果的に使用してトレースコンテキストを伝播することが可能になります。

PartitionKeyを通じてトレースコンテキストを伝播し、ExplicitHashKeyパラメータでパーティションを設定する

次の実装手順で新しいアプローチを実際に見てみましょう。

private PutRecordRequest injectTraceHeader(PutRecordRequest request) throws Exception { 
    if (!TRACE_CONTEXT_PROPAGATION_ENABLED) { 
        return request; 
    } 
    Span currentSpan = Span.current(); 
    if (currentSpan == null) { 
        return request; 
    } 
    SpanContext currentSpanContext = currentSpan.getSpanContext(); 
    if (currentSpanContext == null) { 
        return request; 
    }  

    String partitionKey = request.partitionKey(); 
    PutRecordRequest.Builder requestBuilder = request.toBuilder(); 
    String traceParent = String.format("00-%s-%s-%s", 
            currentSpanContext.getTraceId(), 
            currentSpanContext.getSpanId(), 
            currentSpanContext.getTraceFlags().asHex()); 
    requestBuilder.partitionKey(traceParent); 
    if (partitionKey != null) { 
        requestBuilder.explicitHashKey(toHashKey(partitionKey)); 
    } 

    return requestBuilder.build(); 
} 

private String toHashKey(String partitionKey) throws Exception { 
    MessageDigest md = MessageDigest.getInstance("MD5"); 
    md.update(partitionKey.getBytes()); 
    byte[] digest = md.digest(); 
    BigInteger bi = new BigInteger(1, digest); 
    return bi.toString(); 
} 

テストシナリオを再実行すると、event-producerアプリケーションとevent-consumerアプリケーションが同じトレースに含まれていることが確認できます。
これは、PutRecordリクエストのPartitionKeyパラメータを介してトレースIDが伝播され、ExplicitHashKeyパラメータでハッシュキーを設定することにより、イベントのシャード割り当てが正しく維持されたためです。

ログから確認できるように、イベントはイベントグループIDに基づいて適切にパーティショニングされており、最初の試みでやったようなPartitionKeyパラメータを使用してランダムにトレースIDに基づいてパーティショニングされたわけではありません。

結論

この3部構成のシリーズでは、AWS Kinesisを介してOTELトレースコンテキストを伝播する実践的なアプローチを示しました。
レコードデータを変更することなく、PartitionKeyおよびExplicitHashKeyパラメータを活用することで、クラウドネイティブ環境でのトレースコンテキストの伝播の複雑さをアプリケーション内のカスタムコードを統合することによって克服しました。

まだ読んでいない方は、このシリーズのパート1とパート2もご覧ください。