Simple Log Service:Use the Kafka protocol to upload logs

Dependencies

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

Sample code

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) {
        // The configurations.
        Properties props = new Properties();
        String project = "etl-shanghai-b";
        String logstore = "testlog";
        // Set this parameter to true if you want the content of the producer to be parsed as a JSON log.
        boolean parseJson = false;
        // An Alibaba Cloud account has permissions on all API operations, which poses high security risks. We recommend creating and using a RAM user to call API operations or perform routine O&M. To create a RAM user, log on to the RAM console.
        // The following example stores the AccessKey ID and AccessKey secret in environment variables. Alternatively, store them in a configuration file.
        // Do not hard-code the AccessKey ID and AccessKey secret in source code to avoid credential leaks.
        String accessKeyID = System.getenv("SLS_ACCESS_KEY_ID");
        String accessKeySecret = System.getenv("SLS_ACCESS_KEY_SECRET");
        String endpoint = "cn-shanghai.log.aliyuncs.com"; // Configure this parameter based on the endpoint of the project.
        String port = "10012"; // Use port 10012 for the Internet and port 10011 for the internal network.

        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");

        //Create a producer instance.
        KafkaProducer<String,String> producer = new KafkaProducer<>(props);

        //Send records.
        for(int i=0;i<1;i++){
            String content = "{\"msg\": \"Hello World\"}";
            // If needed, use the following method to set the timestamp of the message.
            // 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();
    }
}

Dependencies

pip install confluent-kafka

Sample code

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

def delivery_report(err, msg):
    """ Called once for each message produced to indicate delivery result.
        Triggered by poll() or 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

    # Get credentials from environment variables
    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" # Use port 10012 for the Internet and port 10011 for the internal network.

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

    # Configure Kafka producer
    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,
    }

    # Create producer instance
    producer = Producer(conf)

    # Send message
    content = "{\"msg\": \"Hello World\"}"
    producer.produce(topic=topic,
                     value=content.encode('utf-8'),
                     #timestamp=int(time.time() * 1000),  # (Optional) Set the timestamp of the record in milliseconds.
                     callback=delivery_report)

    # Wait for any outstanding messages to be delivered and delivery report
    # callbacks to be triggered.
    producer.flush()

if __name__ == '__main__':
    main()

Dependencies

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

Sample code

package main

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

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

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

	// Get credentials from environment variables
	accessKeyID := os.Getenv("SLS_ACCESS_KEY_ID")
	accessKeySecret := os.Getenv("SLS_ACCESS_KEY_SECRET")
	endpoint := "cn-shanghai.log.aliyuncs.com"
	port := "10012" // Use port 10012 for the Internet and port 10011 for the internal network.

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

	// Configure Kafka producer
	config := &kafka.ConfigMap{
		"bootstrap.servers":  hosts,
		"security.protocol":  "sasl_ssl",
		"sasl.mechanisms":    "PLAIN",
		"sasl.username":      project,
		"sasl.password":      accessKeyID + "#" + accessKeySecret,
		"enable.idempotence": false,
	}

	// Create producer instance
	producer, err := kafka.NewProducer(config)
	if err != nil {
		log.Fatalf("Failed to create producer: %v", err)
	}
	defer producer.Close()

	// Send messages in batches.
	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(), // Set the time if needed.
		}, nil)

		if err != nil {
			log.Printf("Failed to produce message: %v", err)
		}
	}

	// Enable a goroutine to listen for whether the producer successfully sends the message.
	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)
}