我无法将日志传给kafka

[marksugar]

当我按照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]

可以按照如下思路排查下：
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

[marksugar]

@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]

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

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

[lavyw]

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