Application Real-Time Monitoring Service: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"
}]