Docs

Torch Log Service

Data collection

Upload logs using the Kafka protocol

TLS supports uploading log data to the server via Kafka protocol. You can collect log data using the Kafka Producer SDK and upload it to TLS via Kafka protocol. This topic describes how to upload logs to TLS using Kafka protocol.

Background information

Kafka is a high-throughput message middleware that is widely used as a message pipeline in various self-hosted log collection scenarios. For example, you can use open-source collection tools on the log source server to collect logs or write log data directly with a producer, and ship them to downstream applications through a consumption pipeline. TLS supports uploading and consuming log data through the Kafka protocol; offers complete upstream and downstream data services based on the Kafka data pipeline.

If you use Kafka protocol for log uploading, you do not need to enable the feature manually or install data collection tools on the data source side. With simple configuration, a Kafka Producer will collect and upload log information to TLS. TLS provides sample projects in Java and Go language for your reference. For more details, see Example.

When collecting logs with Kafka protocol, TLS parses valid JSON format logs into key-value pairs. For invalid JSON formats, some fields may be parsed incorrectly. For other log formats, the full original log will be stored as a string in the __content__ field.

Notice

When parsing JSON format logs via the Kafka protocol, only one extension layer is supported at most. Log fields with multiple nested layers will be collected and stored as a single string.

Limits

Supported Kafka protocol versions: 0.11.x to 2.0.x.

Supported compression types: gzip, snappy, and lz4.

To ensure secure log transmission, you must use the SASL_SSL connection protocol. The corresponding username is your TLS project ID, and the password is the Volcanoengine account access key. For more details, see Example.

If the log topic contains multiple shards, data sequence can not be guaranteed. We recommend using load balancing mode to upload logs.

When using Kafka Producer Batch to send data, the size of each batch must not exceed 5 MiB. The maximum size for a single message is 5 MiB and each batch request can contain up to 10,000 messages. The server checks the length of log data written by each Producer request. If the request exceeds these limits, the entire request fails and no log data is written successfully.

Prerequisites

Log service has been enabled; a project and a topic are created; log data has been successfully collected. For more details, see Overview.

Ensure that your account has permission to enable log upload via Kafka protocol. That is to say, you need permission for Action PutLogs. For more information, see Authorizable operation.

Parameters description

When uploading logs using Kafka protocol, you need to configure the following parameters.

Parameters	Example value	Description
Connection type	SASL_SSL	To ensure secure log transmission, you must use the SASL_SSL connection protocol. The corresponding username is the TLS project ID, and the password is the Volcengine account key.
username	c8f20efe-405f-4d57-98cf-8c58d890****	Kafka SASL username. Set this to TLS project ID.
password	AKLTYmQzOWUzMWx*****#WVRnM05UWTRaVGhrTUdFNE5EazNZV0kyTjJRME******	Kafka SASL user password. Use Volcanoengine account key. The format is ${access-key-id}#${access-key-secret}, where: ${access-key-id} should be replaced with your AccessKey ID. ${access-key-secret} should be replaced with your AccessKey Secret. Notice We recommend using IAM user AK, and ensuring that the IAM user has Action PutLogs permission. For more details, see Authorizable operation.
hosts	Public network: tls-cn-beijing.volces.com:9094 Private network: tls-cn-beijing.ivolces.com:9094	The initial connection cluster address is formatted as service address:port number, for example tls-cn-beijing.ivolces.com:9094. In this format: The service address is the entry point for TLS in the current region. Select the correct service entry point based on region and network type. For details, see Service address. The port number is fixed at 9094. Notice There is no need to specify https:// in the service address part of hosts.
topic	20a50a35-304a-4c01-88d2-23349c30****	Configure the topic ID for TLS.

For better performance, we recommend adding the following Kafka producer parameters.

Parameters	Recommended value	Description
batch.size	With compression enabled: 262144 Without compression: 2621440	Controls the batch size for sending messages to a single shard. Unit: bytes. Notice When the total message size in a batch reaches batch.size or the producer’s waiting time for sending messages reaches linger.ms, the producer will immediately send the current batch.
max.request.size	With compression enabled: 1046576 No compression set: 10465760	Specifies the maximum size for a single request (a request may contain multiple shards). Unit: bytes.
linger.ms	100	Specifies the maximum wait time for the producer before sending the current batch of messages. The unit is milliseconds. If not configured, messages are sent immediately.
compression.type	lz4	Specifies the compression algorithm used before sending messages to Kafka.

Configuration steps

TLS can automatically generate parameter configurations for uploading log data using Kafka protocol. After you fill in the basic configuration and data source information for uploading logs as described below, the console page automatically generates a list of corresponding parameter configurations for your reference.

Select the region where log service is located from the top navigation bar.

On the left navigation bar, select Common functions > Log collection.

Then, on the Data import tab, click Create Kafka protocol write configuration.

Select the project and topic to store Kafka data.

Enter the data source configuration information.

Configuration	Description
Key	The Volcanoengine account key, including AccessKey ID and AccessKey Secret. You can refer to the page instructions to obtain the key. Notice We recommend using IAM user AK, and ensuring that the IAM user has Action PutLogs permission. For more details, see Authorizable operation. TLS does not save the above configurations. The above configuration will only be used to generate Kafka producer output configuration for you.
Network connection method	Select to connect to TLS and upload logs via private network or public network. If you choose private network, make sure the data source can access TLS through VolcEngine's private network.
Log compression method	Compression method for the original logs Currently supported compression methods: no compression, gzip, snappy, and lz4.
Kafka producer	Log data source type Currently, supports open source SDK and Logstash to upload logs using Kafka protocol.
Result preview	TLS automatically generates sample data based on these parameters for your reference. Kafka open-source SDK: copy each configuration value and fill them in the corresponding parameter of the Kafka open-source SDK. For a complete code example, see Upload logs using Kafka Java SDK. Logstash: TLS automatically generates the Kafka plugin configuration for Logstash and tests plugin connectivity. For more details, see Upload logs using Logstash.

Example preview:

Kafka open-source SDK

Logstash

Click Confirm. The page will display Copy parameter configuration.

The system does not directly create a writing task. Besides, writing tasks are not displayed in task list.

To create a writing task, paste the relevant configuration in the producer and check the index configuration of the target log topic.

Notice

To ensure that the log data collected on the server side is retrieved and analyzed, you need to set up index. For details of setting indexes, see Configuration Index.

Example

Upload logs using Logstash

Logstash has a built-in Kafka output plugin (logstash_output_kafka) that can collect data through Kafka protocol. In the scenario of building a log collection and analysis system with ELK, you only need to modify the Logstash configuration file to upload log data to TLS using the Kafka protocol.

If a Kafka plugin configuration example for Logstash automatically generates via TLS console, you can use this example to test the plugin connectivity. For details about parameters such as ${hosts}, see Result preview in Configuration steps.

Notice

TLS uses SASL_SSL connection protocol. Therefore, Kafka requires a SASL_SSL endpoint for message delivery and receipt.

Supported Logstash versions: 7.12–8.8.1. If you need other Logstash versions, you can contact technical support through Ticket system.

We recommend testing plugin connectivity with the configuration below during the testing phase. Remember to remove all stdout-related output configurations in production environments.

For logs in JSON format, set the output format to JSON (codec => json) in your configuration; otherwise, uploaded data may be incomplete. TLS automatically parses and structures log data in JSON format.

Upload logs using Kafka Java SDK

With simple parameter configuration, you can use various Kafka Producer SDKs to collect log data and upload it to TLS via the Kafka protocol. The dependencies and sample code for uploading logs using the Kafka Java SDK are as follows:

Add dependencies.

Add the relevant kafka-clients dependency in the pom file.

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

Upload logs.

The following code is an example of uploading logs via Kafka Java SDK.

Notice

Before running the example code, review Result Preview in Configuration steps and correctly fill in parameters such as userName.

package bytedance;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.config.SaslConfigs;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

// pom.xml dependency
// <dependency>
// <groupId>org.apache.kafka</groupId>
// <artifactId>kafka-clients</artifactId>
// <version>2.2.2</version>
// </dependency>

public class KafkaProducer {
public static void main(String[] args) throws InterruptedException, ExecutionException, TimeoutException {
Properties props = new Properties();

// TLS connection and authentication information configuration.
String userName = "${projectId}";
// Secret Access Key for the Volcanoengine account or IAM account with required permissions. STS temporary security tokens are not supported.
String passWord = "${AK}#${SK}";
props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "${hosts}");
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
props.put(SaslConfigs.SASL_JAAS_CONFIG,
"org.apache.kafka.common.security.plain.PlainLoginModule " +
"required username=\"" + userName + "\" password=\"" + passWord + "\";");

// producer parameter configuration.
props.put(ProducerConfig.ACKS_CONFIG, "1"); // NoResponse 0;WaitForLocal 1;WaitForAll -1
props.put(CommonClientConfigs.RETRIES_CONFIG, 5);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4"); // Configure the compression method.
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
props.put(ProducerConfig.LINGER_MS_CONFIG, "100");
props.put(ProducerConfig.BATCH_SIZE_CONFIG, "262144");
props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "1046576");
// 1. Create a producer object.
Producer<String, String> producer = new org.apache.kafka.clients.producer.KafkaProducer<String, String>(props);
// 2. Call the send method.
Future<RecordMetadata> meta = producer.send(new ProducerRecord<String, String>("${topicId}", "testMessageValue"));
RecordMetadata recordMetadata = meta.get(100, TimeUnit.SECONDS);
// The default returned offset is -1.
System.out.println("partition=" + recordMetadata.partition() + "offset = " + recordMetadata.offset());
// 3. Close the producer.
System.out.println("Kafka Producer sends data asynchronously. It is recommended to wait for 10 seconds before closing the process to ensure all data is successfully sent; otherwise, the uploaded data may be incomplete.");
Thread.sleep(10000);
// In the production environment, the producer should remain resident in the process. The close method is called to shut down the producer only when the process exits.
producer.close();
}
}

Upload logs using Kafka Go SDK

Run the following command to install Sarama.

go get github.com/Shopify/sarama@v1.38.1

Import the required dependencies such as Sarama, and upload logs.

Refer to the following sample code to upload logs using the Kafka Go SDK.

Notice

Before running the sample code below, refer to the result preview in Configuration stepsand correctly fill in parameters such as User.

package kafka_produce_example

import (
"fmt"
"time"

"github.com/Shopify/sarama"
)

func ExampleProduce() {
config := sarama.NewConfig()
config.Version = sarama.V2_0_0_0 // Specify the appropriate version.
config.ApiVersionsRequest = true
config.Net.SASL.Mechanism = "PLAIN"
config.Producer.Return.Successes = true
config.Net.SASL.Version = int16(0)
config.Net.SASL.Enable = true
// Project ID.
config.Net.SASL.User = "${projectId}"
// Secret Access Key for the Volcanoengine account or IAM account with required permissions. STS temporary security tokens are not supported.
config.Net.SASL.Password = "${access-key-id}#${access-key-secret}"
config.Producer.RequiredAcks = sarama.WaitForLocal
config.Producer.Compression = sarama.CompressionLZ4
config.Producer.Flush.Bytes = 262144
config.Producer.Flush.Frequency = 100 * time.Millisecond

config.Net.TLS.Enable = true

// hosts is the address for production. For detailed information, see the parameter descriptions in this document.
asyncProducer, err := sarama.NewAsyncProducer([]string{"${hosts}"}, config)
if err != nil {
fmt.Println("new producer error:" + err.Error())
panic(err)
}

msg := &sarama.ProducerMessage{
// ${topicID} refers to the log topic ID in TLS, for example: 0fdaa6b6-3c9f-424c-8664-fc0d222c****.
Topic: "${topicID}",
Value: sarama.StringEncoder("test"),
}
asyncProducer.Input() <- msg
go func() {
for {
select {
case msg := <-asyncProducer.Successes():
if msg != nil {
// The default value returned for offset is -1.
fmt.Printf("send response; partition:%d, offset:%d\n", msg.Partition, msg.Offset)
}
case msg := <-asyncProducer.Errors():
if msg != nil {
fmt.Println("producer send error:" + msg.Error())
}
}
}
}()

time.Sleep(time.Minute)
_ = asyncProducer.Close()
}

Error message

If uploading logs using Kafka protocol fails, error messages will be returned based on Kafka error code. For more information, see Kafka error list.

In addition, TLS also encapsulates authentication and configuration parameter error messages in the Java Kafka error code SASLAuthenticationException. Details are described as below:

Error message	Description
invalid SASL/PLAIN request: expected 3 tokens	User or password is not configured. You can refer to Result overview in Configuration steps to correctly fill in configurations.
invalid SASL/PLAIN request: empty projectId	The user field is not configured. You can refer to Result overview in Configuration steps to correctly fill in configurations.
invalid SASL/PLAIN request: Password format wrong	Password field configuration error. You can refer to Result overview in Configuration steps to correctly fill in configurations.
invalid SASL/PLAIN request: empty AccessKey	AK is not configured. You can refer to Result overview in Configuration steps to correctly fill in configurations.
invalid SASL/PLAIN request: empty SecretKey	SK is not configured. You can refer to Result overview in Configuration steps to correctly fill in configurations.
invalid SASL/PLAIN request: Invalid projectId	The specified Project ID does not exist. Check if the Project ID is entered correctly.
Access Denied. You are not authorized to perform this operation.	The current user does not have permission for this action. We recommend checking user permissions

Last updated: 2025.12.18 10:28:37