日志存储分析

Logstash

1. 下载并安装 Logstash Doris Output 插件，支持以下两种方式：

   • 直接下载：点此下载。

   • 从源码编译，并运行下方命令安装：

     ./bin/logstash-plugin install logstash-output-doris-1.2.0.gem

2. 配置 Logstash。需配置以下参数：

   • logstash.yml：配置 Logstash 批处理日志的条数和时间，用于提升数据写入性能。

     pipeline.batch.size: 1000000  
     pipeline.batch.delay: 10000

   • logstash_demo.conf：配置所采集日志的具体输入路径和输出到 SelectDB 的设置。

     input {  
         file {  
         path => "/path/to/your/log"  
       }  
     }  
     
     output {  
       doris {  
         http_hosts => [ "<http://fehost1:http_port>", "<http://fehost2:http_port>", "<http://fehost3:http_port">]  
         user => "your_username"  
         password => "your_password"  
         db => "your_db"  
         table => "your_table"  
         
         # doris stream load http headers  
         headers => {  
         "format" => "json"  
         "read_json_by_line" => "true"  
         "load_to_single_tablet" => "true"  
         }  
         
         # field mapping: doris fileld name => logstash field name  
         # %{} to get a logstash field, [] for nested field such as [host][name] for host.name  
         mapping => {  
         "ts" => "%{@timestamp}"  
         "host" => "%{[host][name]}"  
         "path" => "%{[log][file][path]}"  
         "message" => "%{message}"  
         }  
         log_request => true  
         log_speed_interval => 10  
       }  
     }

3. 按照下方命令运行 Logstash，采集日志并输出至 SelectDB。

   ./bin/logstash -f logstash_demo.conf

更多关于 Logstash 配置和使用的说明，可参考 Logstash Doris Output Plugin。

Filebeat

1. 获取支持输出至 SelectDB 的 Filebeat 二进制文件:下载地址。

2. 配置 Filebeat。需配置以下参数：

   • filebeat_demo.yml：配置所采集日志的具体输入路径和输出到 SelectDB 的设置。

     # input
     filebeat.inputs:
     - type: log
       enabled: true
       paths:
         - /path/to/your/log
       # multiline 可以将跨行的日志（比如 Java stacktrace）拼接起来
       multiline:
         type: pattern
         # 效果：以 yyyy-mm-dd HH:MM:SS 开头的行认为是一条新的日志，其他都拼接到上一条日志
         pattern: '^[0-9]{4}-[0-9]{2}-[0-9]{2} [0-9]{2}:[0-9]{2}:[0-9]{2}'
         negate: true
         match: after
         skip_newline: true
     
     processors:
     # 用 js script 插件将日志中的 \t 替换成空格，避免 JSON 解析报错
     - script:
         lang: javascript
         source: >
             function process(event) {
                 var msg = event.Get("message");
                 msg = msg.replace(/\t/g, "  ");
                 event.Put("message", msg);
             }
     # 用 dissect 插件做简单的日志解析
     - dissect:
         # 2024-06-08 18:26:25,481 INFO (report-thread|199) [ReportHandler.cpuReport():617] begin to handle
         tokenizer: "%{day} %{time} %{log_level} (%{thread}) [%{position}] %{content}"
         target_prefix: ""
         ignore_failure: true
         overwrite_keys: true
     
     # queue and batch
     queue.mem:
       events: 1000000
       flush.min_events: 100000
       flush.timeout: 10s
     
     # output
     output.doris:
       fenodes: [ "http://fehost1:http_port", "http://fehost2:http_port", "http://fehost3:http_port" ]
       user: "your_username"
       password: "your_password"
       database: "your_db"
       table: "your_table"
       # output string format
       ## %{[agent][hostname]} %{[log][file][path]} 是filebeat自带的metadata
       ## 常用的 filebeat metadata 还是有采集时间戳 %{[@timestamp]}
       ## %{[day]} %{[time]} 是上面 dissect 解析得到字段
       codec_format_string: '{"ts": "%{[day]} %{[time]}", "host": "%{[agent][hostname]}", "path": "%{[log][file][path]}", "message": "%{[message]}"}'
       headers:
         format: "json"
         read_json_by_line: "true"
         load_to_single_tablet: "true"

3. 按照下方命令运行 Filebeat，采集日志并输出至 SelectDB。

   chmod +x filebeat-doris-2.1.1  
   ./filebeat-doris-2.1.1 -c filebeat_demo.yml

更多关于 Filebeat 配置和使用的说明，可参考 Beats Doris Output Plugin。

Kafka

将 JSON 格式的日志写入 Kafka 的消息队列，创建 Kafka Routine Load，即可让 SelectDB 从 Kafka 主动拉取数据。

可参考如下示例。其中，property.× 是 Librdkafka 客户端相关配置，根据实际 Kafka 集群情况配置。

-- 准备好 kafka 集群和 topic log__topic_  
-- 创建 routine load，从 kafka log__topic_将数据导入 log_table 表  
CREATE ROUTINE LOAD load_log_kafka ON log_db.log_table  
COLUMNS(ts, clientip, request, status, size)  
PROPERTIES (
"max_batch_interval" = "60",
"max_batch_rows" = "20000000",
"max_batch_size" = "1073741824", 
"load_to_single_tablet" = "true",
"format" = "json"
)
FROM KAFKA (  
"kafka_broker_list" = "host:port",  
"kafka_topic" = "log__topic_",  
"property.group.id" = "your_group_id",  
"property.security.protocol"="SASL_PLAINTEXT",  
"property.sasl.mechanism"="GSSAPI",  
"property.sasl.kerberos.service.name"="kafka",  
"property.sasl.kerberos.keytab"="/path/to/xxx.keytab",  
"property.sasl.kerberos.principal"="<xxx@yyy.com>"  
);  
-- 查看 routine 的状态  
SHOW ROUTINE LOAD;

更多关于 Kafka 配置和使用的说明，可参考Routine Load。

自定义程序采集日志

可通过 HTTP API Stream Load 自定义程序将日志数据导入 SelectDB。示例代码：

curl   
--location-trusted   
-u username:password   
-H "format:json"   
-H "read_json_by_line:true"   
-H "load_to_single_tablet:true"   
-H "timeout:600"   
-T logfile.json   
http://fe_host:fe_http_port/api/log_db/log_table/_stream_load

在使用自定义程序时，需注意以下关键点：

• 使用 Basic Auth 进行 HTTP 鉴权，用命令 echo -n 'username:password' | base64 进行计算。

• 设置 HTTP header "format:json"，指定数据格式为 JSON。

• 设置 HTTP header "read_json_by_line:true"，指定每行一个 JSON。

• 设置 HTTP header "load_to_single_tablet:true"，指定一次导入写入一个分桶减少导入的小文件。

• 建议写入客户端单个 Batch 大小为 100MB～1GB。Apache Doris 2.1 及更高版本需通过服务端 Group Commit 功能降低客户端 Batch 大小。