ApsaraMQ for Kafka への Prometheus データの配信 - Managed Service for Prometheus

前提条件

Prometheus インスタンスが接続されていることを確認してください。詳細については、次のトピックをご参照ください。
ApsaraMQ for Kafka インスタンスをデプロイし、トピックなどの必要なリソースを作成してください。詳細については、「概要」をご参照ください。
EventBridge がアクティブ化されていることを確認してください。詳細については、「EventBridge のアクティブ化と権限付与」をご参照ください。

制限

Prometheus インスタンスと宛先が異なる VPC にある場合は、ネットワーク接続の失敗を回避するために、宛先の vSwitch CIDR ブロックを Prometheus インスタンスの許可リストに追加する必要があります。

サポートされるデータソース：

インスタンスタイプ	説明
Prometheus for Cloud Services	名前が `cloud-product-prometheus` で始まる無料インスタンスは除きます。
Prometheus for Container Service	なし
Prometheus for Flink Serverless	なし
Prometheus for Kubernetes	なし
汎用	OpenTelemetry エンドポイントを介して報告されたデータは除きます。

タスク作成後に生成されたリアルタイムデータのみがシッピングされます。履歴データはサポートされていません。

ステップ 1：データ配信タスクの作成

にログインします。 Prometheusコンソールのマネージドサービス。
左側のナビゲーションペインで、Prometheusモニタリング > データ配布をクリックします。
データ配布 ページで、上部メニューから宛先リージョンを選択し、 新規タスク をクリックします。
ダイアログボックスで、 タスク名 と タスクの説明 を入力し、 OK をクリックします。

タスクの編集 ページで、データソースと宛先を設定します。

[+ Add Data Source] をクリックし、次のパラメーターを設定して OK をクリックします。

パラメーター	説明	例
Prometheus インスタンス	ソースの Prometheus インスタンスです。	c78cb8273c02*****
データフィルタリング	メトリクスラベルでデータをフィルタリングします。正規表現をサポートします。複数の条件は改行で区切ります (AND ロジック)。	`__name__=AliyunEcs_CPUUtilization\|AliyunEcs_memory_usedutilization regionId=cn-hangzhou id=i-2ze0mxp.*`
データラベリング	シッピングされるメトリクスデータに追加するラベルです。改行で区切ります。	`deliver_test_key1=ssss deliver_test_key2=yyyy`

ターゲットの追加 をクリックし、ターゲットタイプ を Alibaba Cloud ApsaraMQ for Kafka に設定し、残りのパラメーターを設定してから、確認をクリックします。

タスクの編集 ページで OK をクリックし、次に保存をクリックしてデータシッピングタスクを作成します。

ステップ 2：Prometheus モニタリングデータの表示

説明

ターゲットの負荷を軽減するため、サービスは Prometheus モニタリングデータを Snappy で圧縮した JSON 配列として Kafka に配信します。詳細については、「Snappy 圧縮形式」をご参照ください。

コンソールでの表示

ApsaraMQ for Kafka コンソールにログインします。
リソースの分布 セクションの概要ページで、管理する ApsaraMQ for Kafka インスタンスが存在するリージョンを選択します。
インスタンスリスト ページで、管理するインスタンスの名前をクリックします。
左側のナビゲーションペインで [Topics] をクリックします。ターゲットトピックを見つけ、詳細列の操作をクリックします。CloudMonitor または [Message Query] タブでインポートされたデータを表示できます。

クライアントでの表示

Kafka クライアントを初期化します。詳細については、「単一のコンシューマーを使用したメッセージのサブスクライブ」をご参照ください。

次のコードを KafkaConsumerDemo.java ファイルに追加します。このコードは、Kafka クライアントが初期化された後にデータを消費し、Snappy を使用して解凍します:

public static void main(String[] args) {

        // まず Kafka コンシューマーを初期化します。
        
        while (true){
            try {
                ConsumerRecords<String, byte[]> records = consumer.poll(1000);
                // 次の poll の前にレコードを消費する必要があり、合計時間は SESSION_TIMEOUT_MS_CONFIG を超えないようにしてください。
                // 専用のスレッドプールを使用してメッセージを消費し、非同期に結果を返すことを推奨します。
                for (ConsumerRecord<String, byte[]> record : records) {
                    byte[] compressedData = record.value();
                    byte[] data = Snappy.uncompress(compressedData);
                    
                    System.out.println(new String(data));
                }
            } catch (Exception e) {
                try {
                    Thread.sleep(1000);
                } catch (Throwable ignore) {

                }

                e.printStackTrace();
            }
        }
}

KafkaConsumerDemo.java ファイルをコンパイルして実行します。メトリックデータは、次の JSON 形式で出力されます。

[{
	"instance": "*****",
	"pod": "*****",
	"rejected": "false",
	"type": "validate",
	"pod_name": "*****",
	"endpoint": "http-metrics",
	"__name__": "apiserver_admission_controller_admission_duration_seconds_bucket",
	"service": "kubernetes",
	"name": "*****",
	"namespace": "default",
	"le": "2.5",
	"job": "apiserver",
	"operation": "UPDATE",
	"value": "675.0",
	"timestamp": "1698732988354"
}, {
	"instance": "*****",
	"pod": "*****",
	"rejected": "false",
	"type": "validate",
	"pod_name": "*****",
	"endpoint": "http-metrics",
	"__name__": "apiserver_admission_controller_admission_duration_seconds_bucket",
	"service": "kubernetes",
	"name": "*****",
	"namespace": "default",
	"le": "+Inf",
	"job": "apiserver",
	"operation": "UPDATE",
	"value": "675.0",
	"timestamp": "1698732988354"
}, {
	"instance": "*****",
	"pod": "*****",
	"rejected": "false",
	"type": "validate",
	"pod_name": "*****",
	"endpoint": "http-metrics",
	"__name__": "apiserver_admission_controller_admission_duration_seconds_bucket",
	"service": "kubernetes",
	"name": "*****",
	"namespace": "default",
	"le": "0.005",
	"job": "apiserver",
	"operation": "UPDATE",
	"value": "1037.0",
	"timestamp": "1698732988519"
}, {
	"instance": "*****",
	"pod": "*****",
	"rejected": "false",
	"type": "validate",
	"pod_name": "*****",
	"endpoint": "http-metrics",
	"__name__": "apiserver_admission_controller_admission_duration_seconds_bucket",
	"service": "kubernetes",
	"name": "*****",
	"namespace": "default",
	"le": "0.025",
	"job": "apiserver",
	"operation": "UPDATE",
	"value": "1037.0",
	"timestamp": "1698732988519"
}]