Simple Log Service:Kafka プロトコルを用いたログのアップロード

依存関係

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.1.0</version>
</dependency>

サンプルコード

package org.example;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProduceExample {

    public static void main(String[] args) {
        // 構成情報。
        Properties props = new Properties();
        String project = "etl-shanghai-b";
        String logstore = "testlog";
        // プロデューサーの内容を JSON ログとして解析する場合は、このパラメーターを true に設定します。
        boolean parseJson = false;
        // Alibaba Cloud アカウントはすべての API オペレーションに対して完全な権限を持ちます。これは高いセキュリティリスクを伴います。API オペレーションの呼び出しや日常的な運用・保守作業を行う際には、RAM ユーザーの作成および利用を推奨します。RAM ユーザーを作成するには、RAM コンソールにログインしてください。
        // このセクションでは、AccessKey ID および AccessKey Secret を環境変数に保存する例を示します。必要に応じて、構成ファイルに保存することも可能です。
        // セキュリティリスクを回避するため、コード内に AccessKey ID および AccessKey Secret を保存しないことを推奨します。
        String accessKeyID = System.getenv("SLS_ACCESS_KEY_ID");
        String accessKeySecret = System.getenv("SLS_ACCESS_KEY_SECRET");
        String endpoint = "cn-shanghai.log.aliyuncs.com"; // このパラメーターは、プロジェクトのエンドポイントに基づいて設定してください。
        String port = "10012"; // インターネットではポート 10012、内部ネットワークではポート 10011 を使用します。

        String hosts = project + "." + endpoint + ":" + port;
        String topic = logstore;
        if(parseJson) {
            topic = topic + ".json";
        }

        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, hosts);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put("security.protocol", "sasl_ssl");
        props.put("sasl.mechanism", "PLAIN");
        props.put("sasl.jaas.config",
                "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"" +
                        project + "\" password=\"" + accessKeyID + "#" + accessKeySecret + "\";");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "false");

        // プロデューサーインスタンスを作成します。
        KafkaProducer<String,String> producer = new KafkaProducer<>(props);

        // レコードを送信します。
        for(int i=0;i<1;i++){
            String content = "{\"msg\": \"Hello World\"}";
            // 必要に応じて、以下のメソッドを使用してメッセージのタイムスタンプを設定できます。
            // long timestamp = System.currentTimeMillis();
            // ProducerRecord<String, String> record = new ProducerRecord<>(topic, null, timestamp, null, content);

            ProducerRecord<String, String> record = new ProducerRecord<>(topic, content);
            producer.send(record, (metadata, exception) -> {
                if (exception != null) {
                    System.err.println("ERROR: メッセージの送信に失敗しました：" + exception.getMessage());
                    exception.printStackTrace();
                } else {
                    System.out.println("メッセージが正常に送信されました。トピック：" + metadata.topic() +
                                     "、パーティション：" + metadata.partition() +
                                     "、オフセット：" + metadata.offset() +
                                     "、タイムスタンプ：" + metadata.timestamp());
                }
            });
        }

        producer.close();
    }
}

依存関係

pip install confluent-kafka

サンプルコード

#!/bin/env python3
import time
import os
from confluent_kafka import Producer

def delivery_report(err, msg):
    """ 各メッセージの配信結果を示すために、メッセージごとに 1 回呼び出されます。
        poll() または flush() によってトリガーされます。 """
    if err is not None:
        print('メッセージの配信に失敗しました：{}'.format(err))
    else:
        print('メッセージが {} [{}] へ配信されました。オフセット：{}'.format(msg.topic(), msg.partition(), msg.offset()))

def main():
    project = "etl-shanghai-b"
    logstore = "testlog"
    parse_json = False

    # 環境変数から認証情報を取得
    access_key_id = os.getenv("SLS_ACCESS_KEY_ID")
    access_key_secret = os.getenv("SLS_ACCESS_KEY_SECRET")
    endpoint = "cn-shanghai.log.aliyuncs.com"
    port = "10012" # インターネットではポート 10012、内部ネットワークではポート 10011 を使用します。

    hosts = f"{project}.{endpoint}:{port}"
    topic = logstore
    if parse_json:
        topic = topic + ".json"

    # Kafka プロデューサーを構成
    conf = {
        'bootstrap.servers': hosts,
        'security.protocol': 'sasl_ssl',
        'sasl.mechanisms': 'PLAIN',
        'sasl.username': project,
        'sasl.password': f"{access_key_id}#{access_key_secret}",
        'enable.idempotence': False,
    }

    # プロデューサーインスタンスを作成
    producer = Producer(conf)

    # メッセージを送信
    content = "{\"msg\": \"Hello World\"}"
    producer.produce(topic=topic,
                     value=content.encode('utf-8'),
                     #timestamp=int(time.time() * 1000),  # (オプション) レコードのタイムスタンプ（ミリ秒単位）を設定します。
                     callback=delivery_report)

    # 未送信のメッセージがすべて配信され、配信レポートのコールバックが呼び出されるのを待機します。
    producer.flush()

if __name__ == '__main__':
    main()

依存関係

go get github.com/confluentinc/confluent-kafka-go/kafka

サンプルコード

package main

import (
	"fmt"
	"log"
	"os"
//	"time"

	"github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
	project := "etl-shanghai-b"
	logstore := "testlog"
	parseJson := false

	// 環境変数から認証情報を取得
	accessKeyID := os.Getenv("SLS_ACCESS_KEY_ID")
	accessKeySecret := os.Getenv("SLS_ACCESS_KEY_SECRET")
	endpoint := "cn-shanghai.log.aliyuncs.com"
	port := "10012" // インターネットではポート 10012、内部ネットワークではポート 10011 を使用します。

	hosts := fmt.Sprintf("%s.%s:%s", project, endpoint, port)
	topic := logstore
	if parseJson {
		topic = topic + ".json"
	}

	// Kafka プロデューサーを構成
	config := &kafka.ConfigMap{
		"bootstrap.servers":  hosts,
		"security.protocol":  "sasl_ssl",
		"sasl.mechanisms":    "PLAIN",
		"sasl.username":      project,
		"sasl.password":      accessKeyID + "#" + accessKeySecret,
		"enable.idempotence": false,
	}

	// プロデューサーインスタンスを作成
	producer, err := kafka.NewProducer(config)
	if err != nil {
		log.Fatalf("プロデューサーの作成に失敗しました：%v", err)
	}
	defer producer.Close()

	// バッチでメッセージを送信
	messages := []string{
		"{\"msg\": \"Hello World 1\"}",
		"{\"msg\": \"Hello World 2\"}",
		"{\"msg\": \"Hello World 3\"}",
	}

	for _, content := range messages {
		err := producer.Produce(&kafka.Message{
			TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
			Value:          []byte(content),
			//Timestamp:      time.Now(), // 必要に応じて時間を設定
		}, nil)

		if err != nil {
			log.Printf("メッセージの送信に失敗しました：%v", err)
		}
	}

	// プロデューサーがメッセージを正常に送信したかどうかを監視するゴルーチンを有効化
	go func() {
		for e := range producer.Events() {
			switch ev := e.(type) {
			case *kafka.Message:
				if ev.TopicPartition.Error != nil {
					fmt.Printf("配信に失敗しました：%v\n", ev.TopicPartition.Error)
				} else {
					fmt.Printf("トピック %s [%d] へメッセージを配信しました。オフセット：%v\n",
						*ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)
				}
			}
		}
	}()

	producer.Flush(5 * 1000)
}