我无法将日志传给kafka #401

marksugar · 2022-09-16T09:28:15Z

marksugar
Sep 16, 2022

当我按照https://zhuanlan.zhihu.com/p/544081167进行操作，我并未能日志成功传递到kafka
我的环境和说明如下：

k8s: 1.19
kafka:3.2

在deployment中的nginx日志格式为json

        log_format log_json '{ "@timestamp": "$time_local", '
          '"remote_addr": "$remote_addr", '
          '"log_time": "$time_iso8601",'
          '"referer": "$http_referer", '
          '"request": "$request", '
          '"status": $status, '
          '"bytes": $body_bytes_sent, '
          '"agent": "$http_user_agent", '
          '"x_forwarded": "$http_x_forwarded_for", '
          '"up_addr": "$upstream_addr",'
          '"up_host": "$upstream_http_host",'
          '"up_resp_time": "$upstream_response_time",'
          '"request_time": "$request_time"'
          ' }';

得到的日志格式如下：

{ "@timestamp": "16/Sep/2022:17:24:03 +0800", "remote_addr": "130.130.0.0", "log_time": "2022-09-16T17:24:03+08:00","referer": "-", "request": "GET / HTTP/1.1", "status": 304, "bytes": 0, "agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36", "x_forwarded": "-", "up_addr": "-","up_host": "-","up_resp_time": "-","request_time": "0.000" }

ilogtail-user-cm文件如下：

apiVersion: v1
kind: ConfigMap
metadata:
  name: ilogtail-user-cm
  namespace: ilogtail
data:
  nginx_stdout.yaml: |
    enable: true
    inputs:
      - Type: service_docker_stdout
        Stderr: false
        Stdout: true
        IncludeK8sLabel:
          app: mark
    processors:
      - Type: processor_json
        SourceKey: content
        KeepSource: false
        ExpandDepth: 1
        ExpandConnector: ""
    flushers:
      - Type: flusher_kafka
        Brokers:
          - 172.16.15.136:9092
        Topic: mark-access-log
  nginx_stderr.yaml: |
    enable: true
    inputs:
      - Type: service_docker_stdout
        Stderr: true
        Stdout: false
        K8sNamespaceRegex: "^(default)$"
        K8sPodRegex: "^(mark-.*)$"
        K8sContainerRegex: "mark"
    flushers:
      - Type: flusher_kafka
        Brokers:
          - 172.16.15.136:9092
        Topic: mark-error-log
---

我在kafka中可以查看到topics

I have no name!@7b1e5f4c972b:/$ kafka-topics.sh --list --bootstrap-server localhost:9092
__consumer_offsets
access-log
error-log
mark-access-log

当我请求，并用kafka-console-consumer.sh 查看时，并没有任何数据

I have no name!@7b1e5f4c972b:/$ kafka-console-consumer.sh --topic mark-access-log --from-beginning --bootstrap-server localhost:9092

logstash也并未消费

请指正问题所在，感激不尽

谢谢

messixukejia · 2022-09-16T13:07:55Z

messixukejia
Sep 16, 2022
Maintainer

可以按照如下思路排查下：
1、nginx是打印到标准输出，不是文件？
2、登陆ilogtail容器，采集配置是否同步到user_yaml_config.d/
3、查看logtail_plugin.LOG有无异常打印https://ilogtail.gitbook.io/ilogtail-docs/configuration/logging#ilogtail-cha-jian-ri-zhi
4、将flusher_kafka替换成stdout flusher便于调试，先保证日志能正常采集
5、所采集的容器，是否跟上述的过滤条件所匹配？可以考虑临时去掉上述过滤条件，看看是否正常采集

另外，可以参考下https://zhuanlan.zhihu.com/p/544079650

3 replies

marksugar Sep 20, 2022
Author

@messixukejia 经过测试，我成功将日志发送到kafka

但又出现了新的问题

通过ilogtail配置发送到kafka后，logstash读取后发送到es时候发生了警告，日志内容不能发送到es

如下

"mapper [Contents.Value] of different type, current_type [date], merged_type [text]"}}}}

logstash日志如下

{
          "tags" => [
        [0] "_dateparsefailure"
    ],
    "@timestamp" => 2022-09-20T21:26:38.229Z,
          "Time" => 1663680396,
      "Contents" => [
        [0] {
            "Value" => "130.130.0.0 - - [20/Sep/2022:21:26:35 +0800] \"GET / HTTP/1.1\" 304 0 \"-\" \"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36\" \"-\"",
              "Key" => "content"
        },
        [1] {
            "Value" => "2022-09-20T13:26:35.315717332Z",
              "Key" => "_time_"
        },
        [2] {
            "Value" => "stdout",
              "Key" => "_source_"
        },
        [3] {
            "Value" => "130.130.1.239",
              "Key" => "_container_ip_"
        },
        [4] {
            "Value" => "nginx:1.21.1-alpine",
              "Key" => "_image_name_"
        },
        [5] {
            "Value" => "mark-test",
              "Key" => "_container_name_"
        },
        [6] {
            "Value" => "mark-test-784db854b5-dtghm",
              "Key" => "_pod_name_"
        },
        [7] {
            "Value" => "default",
              "Key" => "_namespace_"
        },
        [8] {
            "Value" => "9bd7ea9d-8952-40a7-8928-6b9b2e6e8c04",
              "Key" => "_pod_uid_"
        }
    ],
          "type" => "elk-nginx-access-log",
      "@version" => "1"
}
[2022-09-20T13:26:38,768][WARN ][logstash.outputs.elasticsearch] Could not index event to Elasticsearch. {:status=>400, :action=>["index", {:_id=>nil, :_index=>"elk-nginx-access-log-2022.09.20", :_type=>"_doc", :routing=>nil}, #<LogStash::Event:0x2ad63192>], :response=>{"index"=>{"_index"=>"elk-nginx-access-log-2022.09.20", "_type"=>"_doc", "_id"=>"Z8UUW4MBaUpkBYrURf-m", "status"=>400, "error"=>{"type"=>"illegal_argument_exception", "reason"=>"mapper [Contents.Value] of different type, current_type [date], merged_type [text]"}}}}

elasticsearch7

"stacktrace": ["java.lang.IllegalArgumentException: mapper [Contents.Value] of different type, current_type [date], merged_type [text]",
"at org.elasticsearch.index.mapper.FieldMapper.doMerge(FieldMapper.java:330) ~[elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.mapper.DateFieldMapper.doMerge(DateFieldMapper.java:563) ~[elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.mapper.FieldMapper.merge(FieldMapper.java:317) ~[elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.mapper.FieldMapper.merge(FieldMapper.java:52) ~[elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.mapper.DocumentParser.createDynamicUpdate(DocumentParser.java:237) ~[elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.mapper.DocumentParser.parseDocument(DocumentParser.java:83) ~[elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.mapper.DocumentMapper.parse(DocumentMapper.java:267) ~[elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.shard.IndexShard.prepareIndex(IndexShard.java:764) ~[elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.shard.IndexShard.applyIndexOperation(IndexShard.java:741) ~[elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.shard.IndexShard.applyIndexOperationOnPrimary(IndexShard.java:713) ~[elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.bulk.TransportShardBulkAction.executeBulkItemRequest(TransportShardBulkAction.java:256) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.bulk.TransportShardBulkAction$2.doRun(TransportShardBulkAction.java:159) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.bulk.TransportShardBulkAction.performOnPrimary(TransportShardBulkAction.java:191) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.bulk.TransportShardBulkAction.shardOperationOnPrimary(TransportShardBulkAction.java:116) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.bulk.TransportShardBulkAction.shardOperationOnPrimary(TransportShardBulkAction.java:77) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.support.replication.TransportReplicationAction$PrimaryShardReference.perform(TransportReplicationAction.java:927) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.support.replication.ReplicationOperation.execute(ReplicationOperation.java:108) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.support.replication.TransportReplicationAction$AsyncPrimaryAction.runWithPrimaryShardReference(TransportReplicationAction.java:398) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.support.replication.TransportReplicationAction$AsyncPrimaryAction.lambda$doRun$0(TransportReplicationAction.java:316) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.ActionListener$1.onResponse(ActionListener.java:62) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.shard.IndexShard.lambda$wrapPrimaryOperationPermitListener$14(IndexShard.java:2525) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.ActionListener$3.onResponse(ActionListener.java:112) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.shard.IndexShardOperationPermits.acquire(IndexShardOperationPermits.java:269) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.shard.IndexShardOperationPermits.acquire(IndexShardOperationPermits.java:236) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.index.shard.IndexShard.acquirePrimaryOperationPermit(IndexShard.java:2499) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.support.replication.TransportReplicationAction.acquirePrimaryOperationPermit(TransportReplicationAction.java:864) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.support.replication.TransportReplicationAction$AsyncPrimaryAction.doRun(TransportReplicationAction.java:312) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.action.support.replication.TransportReplicationAction.handlePrimaryRequest(TransportReplicationAction.java:275) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor$ProfileSecuredRequestHandler$1.doRun(SecurityServerTransportInterceptor.java:250) [x-pack-security-7.2.0.jar:7.2.0]",
"at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.xpack.security.transport.SecurityServerTransportInterceptor$ProfileSecuredRequestHandler.messageReceived(SecurityServerTransportInterceptor.java:308) [x-pack-security-7.2.0.jar:7.2.0]",
"at org.elasticsearch.transport.RequestHandlerRegistry.processMessageReceived(RequestHandlerRegistry.java:63) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.transport.TransportService$7.doRun(TransportService.java:703) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingAbstractRunnable.doRun(ThreadContext.java:758) [elasticsearch-7.2.0.jar:7.2.0]",
"at org.elasticsearch.common.util.concurrent.AbstractRunnable.run(AbstractRunnable.java:37) [elasticsearch-7.2.0.jar:7.2.0]",
"at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]",
"at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]",
"at java.lang.Thread.run(Thread.java:835) [?:?]"] }

我将日志复制到文件内，通过Lostash读取日志文件，则不会出现错误

你能解释下我还需要怎么配置吗

thanks

messixukejia Sep 21, 2022
Maintainer

看描述感觉跟ilogtail没有直接的关联。

建议讲logstash的输出设成文件，对比看下差异吧。

lavyw Sep 23, 2022

遇到了跟你一样的问题，把配置文件和日志都转移到容器里面， content 还是数组，好像容器环境就不行，单独在机器上部署没问题的，还在测试中

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

我无法将日志传给kafka #401

{{title}}

Replies: 1 comment 3 replies

{{title}}

{{editor}}'s edit

{{editor}}'s edit

{{title}}

{{title}}

{{title}}

Select a reply

我无法将日志传给kafka #401

marksugar Sep 16, 2022

Replies: 1 comment · 3 replies

messixukejia Sep 16, 2022 Maintainer

marksugar Sep 20, 2022 Author

messixukejia Sep 21, 2022 Maintainer

lavyw Sep 23, 2022

marksugar
Sep 16, 2022

Replies: 1 comment 3 replies

messixukejia
Sep 16, 2022
Maintainer

marksugar Sep 20, 2022
Author

messixukejia Sep 21, 2022
Maintainer