Managed Service for Prometheus:Prometheus インスタンスから Apsara MQ for Kafka にモニタリングデータを配信する

左側のナビゲーションウィンドウで、[トピック] をクリックします。管理するトピックを見つけ、[アクション] 列の [詳細] をクリックします。[CloudMonitor] タブまたは [メッセージクエリ] タブをクリックして、インポートしたモニタリングデータを表示します。

Kafka コンシューマーを初期化します。詳細については、「インスタンスエンドポイントを使用してメッセージを送受信する」トピックの「単一のコンシューマーを使用してメッセージをサブスクライブする」セクションをご参照ください。

KafkaConsumerDemo.java ファイルに次の内容を追加します。次のサンプルコードでは、初期化された Kafka コンシューマーは Snappy を使用してモニタリングデータを解凍して消費します：

public static void main(String[] args) {

        // Kafka コンシューマーを最初に初期化します。
        // 初期化処理は省略

        while (true){
            try {
                ConsumerRecords<String, byte[]> records = consumer.poll(1000);
                // 次のポーリングサイクルが開始される前に、すべてのメッセージが消費されている必要があります。合計期間は、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"
}]