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 操作を呼び出したり、日常の O&M を実行したりするには、RAM ユーザーを作成して使用することをお勧めします。RAM ユーザーを作成するには、RAM コンソールにログインします。
        // このセクションでは、AccessKey ID と AccessKey シークレットを環境変数に保存する方法の例を示します。必要に応じて、AccessKey ID と AccessKey シークレットを構成ファイルに保存することもできます。
        // セキュリティリスクを防ぐため、コードに AccessKey ID と AccessKey シークレットを保存しないことをお勧めします。
        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: Failed to send message: " + exception.getMessage());
                    exception.printStackTrace();
                } else {
                    System.out.println("Message sent successfully to topic: " + metadata.topic() +
                                     ", partition: " + metadata.partition() +
                                     ", offset: " + metadata.offset() +
                                     ", timestamp: " + 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):
    """ 生成された各メッセージに対して一度呼び出され、配信結果を示します。
        poll() または flush() によってトリガーされます。"""
    if err is not None:
        print('Message delivery failed: {}'.format(err))
    else:
        print('Message delivered to {} [{}] at offset {}'.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("Failed to create producer: %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("Failed to produce message: %v", err)
		}
	}

	// プロデューサーがメッセージを正常に送信したかどうかをリッスンするゴルーチンを有効にします。
	go func() {
		for e := range producer.Events() {
			switch ev := e.(type) {
			case *kafka.Message:
				if ev.TopicPartition.Error != nil {
					fmt.Printf("Delivery failed: %v\n", ev.TopicPartition.Error)
				} else {
					fmt.Printf("Delivered message to topic %s [%d] at offset %v\n",
						*ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)
				}
			}
		}
	}()

	producer.Flush(5 * 1000)
}