通过 fluentd 将接收到的数据解析为 json答案

【问题标题】：Parse received data as json by fluentd通过 fluentd 将接收到的数据解析为 json
【发布时间】：2015-12-10 17:34:01
【问题描述】：

我正在尝试通过 fluentd 从外部系统接收数据，如下所示：数据={“版本”：“0.0”；“秘密”：null}

响应是： 400 错误请求需要“json”或“msgpack”参数

如果我用“json”而不是“data”发送（不能更改真实来源）相同的字符串（如 json={"version":"0.0";"secret":null}），一切正常。我如何配置 fluentd 以同样的方式接受它？谢谢。

fluent.conf 示例：

<source>                                                  
  @type http                                              
  port 24224                                              
  bind 0.0.0.0          

  # accept "{"key":"value"} input                                    
  format json   

  # accept "json={"key":"value"} input                                    
  #format default
</source>                                               
<match **>                                              
  @type file                                            
  @id   output1                                         
  path         /fluentd/log/data.*.log                  
  symlink_path /fluentd/log/data.log                   
  format json                                           
  append       true                                     
  time_slice_format %Y%m%d                              
  time_slice_wait   10m                                 
  time_format       %Y%m%dT%H%M%S%z                     
</match>

我尝试过使用正则表达式或通过 nginx 修改数据。由于编码和复杂的数据，正则表达式是不可能的，并且没有找到如何使用 nginx 修改 POST 数据的方法（这也是不好的方法）。

【问题讨论】：

标签： fluentd

【解决方案1】：

我会自己回答。在尝试了很多配置（以及数小时阅读 fluentd/nginx 和博客的官方文档）之后，我决定创建插件（http://docs.fluentd.org/articles/plugin-development#parser-plugins）。我已经结束了这个解决方案：

解析器插件

module Fluent
  class TextParser
    class CMXParser < Parser
      # Register this parser
      Plugin.register_parser("parser_CMX", self)

      config_param :format_hash, :string, :default => "data" #  delimiter is configurable with " " as default

      def configure(conf)
        super
      end

      # This is the main method. The input "text" is the unit of data to be parsed.
      def parse(text)
        text = WEBrick::HTTPUtils.parse_query(text)
        record = JSON.parse(text[@format_hash])
        yield nil, record
      end
    end
  end
end

Fluentd 的配置

<source>                                
  @type http                            
  port 24224                            
  bind 0.0.0.0                          
  body_size_limit 32m                   
  keepalive_timeout 5s                  
  format parser_CMX
</source>                     

<match **>                            
  @type file                          
  @id   output1                       
  path         /fluentd/log/data.*.log
  symlink_path /fluentd/log/data.log  
  format json                         
  append       true                   
  time_slice_format %Y%m%d            
  time_slice_wait   10m               
  time_format       %Y%m%dT%H%M%S%z   
</match>

我认为核心代码有空间实现这一点，因为基本 in_http 脚本做同样的事情，除了它只使用硬编码字符串“params['json']”。它可以使用诸如“format_hash”/“format_map”之类的新变量来包含地图。

【讨论】：

【解决方案2】：

http://docs.fluentd.org/articles/in_http

本文展示了可接受的格式。

我如何配置 fluentd 以同样的方式接受它？

这意味着你想用format json 解析data={"k":"v"} 吗？如果是这样，它不能。

【讨论】：