Fluentd过滤器排除具有空值的密钥

Question

我想要排除一行有serive_name空"service_name":""。 这里是我的fluentd的conf

## match tag=debug.** and dump to console 
<match debug.**> 
    @type stdout 
</match> 
<source> 
    @type tail 
    path /opt/wso2esb-4.9.0-wkr-1/repository/logs/wso2carbon.log 
    pos_file /var/log/td-agent/tmp/wso2carbon.log.pos 
    tag debug.wso2.esb 
    format /^([TID:]* [^ ]* [^ ]* \[(?<time>[^\]]*)\]) ([^ ]* (?<level>[^ ]*))([^***]*[^=]*[^ ]*(?<service_name>[^,]*)[^=]*[^ ]*(?<step>[^,]*)[^ ]*[^=]*[^ ]*(?<message_id>[^,]*))/ 
    time_format %Y-%m-%d %H:%M:%S 
# keep_time_key true 
</source> 

这里是注销，把

2017-08-21 09:57:10 +0700 debug.wso2.esb: {"level":"INFO","service_name":" SA_VasGWLogSeq","step":" before vasgwInsertlog","message_id":" urn:uuid:2046f0ed-690d-47b1-aa86-d4a71c021a74"} 
2017-08-21 09:57:10 +0700 debug.wso2.esb: {"level":"INFO","service_name":"","step":"","message_id":""} 
2017-08-21 09:57:10 +0700 debug.wso2.esb: {"level":"INFO","service_name":" SA_VasGWLogSeq","step":" after vasgwInsertlog","message_id":" urn:uuid:2046f0ed-690d-47b1-aa86-d4a71c021a74"} 
2017-08-21 10:16:10 +0700 debug.wso2.esb: {"level":"INFO","service_name":" SERVICE_NAME","step":" Before - SA_ServiceApiDSEp","message_id":" urn:uuid:39e0ecc1-dda5-4cd9-91fc-90e7ed4f5233"} 

我要排除下面一行。怎么做？

{"level":"INFO","service_name":"","step":"","message_id":""} 

第二个问题是，为什么我有值"service_name":" SERVICE_NAME"前的空间，当我尝试上Fluentular I get a nice output without space。

我已经通过给正则表达式添加空间来解决了第二个任务。例如变更

[^=]*[^ ]*(?<service_name>[^,]*)至[^=]*[^ ]* (?<service_name>[^,]*)。

---

但我不知道如何编写一个过滤器由具有像"service_name":""空值KEY_NAME排除的记录。

Answer 1

因为我找不到解决方案来排除键值为空的记录，我使用相反的解决方案。我使用grep来保存指定键值的记录。请参阅下面我的Fluentd配置。

Fluentd在每个WSO2节点上。

############################################################################################# 
# Fluentd Configuration File                # 
#                       # 
# In v1 configuration, type and id are @ prefix parameters.         # 
# @type and @id are recommended. type and id are still available for backward compatibility # 
############################################################################################# 

################################ 
#   Source   # 
################################ 
## built-in TCP input 
## $ echo <json> | fluent-cat <tag> 
<source> 
    @type forward 
    @id forward_input 

    port 24224 
</source> 

# Listen DRb for debug 
<source> 
    @type debug_agent 
    @id debug_agent_input 

    bind 127.0.0.1 
    port 24230 
</source> 

# HTTP input 
# http://localhost:8888/<tag>?json=<json> 
#<source> 
# @type http 
# @id http_input 

# port 8888 
#</source> 

# Listen HTTP for monitoring 
# http://localhost:24220/api/plugins 
# http://localhost:24220/api/plugins?type=TYPE 
# http://localhost:24220/api/plugins?tag=MYTAG 
<source> 
    @type monitor_agent 
    @id monitor_agent_input 

    port 24220 
</source> 

<source> 
    @type tail 

    path /opt/wso2esb-4.9.0-wkr-1/repository/logs/wso2carbon.log 
    pos_file /cellcard/fluent/wso2carbon.log.pos 
    tag wso2.esb.service.test 
    format /^([TID:]+ [^ ]+ [^ ]+ \[(?<time>[^\]]+)\]) ([^***]+[^=]+[^ ]+(?<transaction_id>[^,]*)[^=]+[^ ]+(?<service_name>[^,]*)[^=]+[^ ]+(?<data>[^,]*))/ 
    time_format %Y-%m-%d %H:%M:%S 
    keep_time_key true 
</source> 

<source> 
    @type tail 

    path /opt/wso2esb-4.9.0-wkr-1/repository/logs/wso2carbon.log 
    pos_file /cellcard/fluent/wso2carbon.log.pos 
    tag wso2.esb.ne.surepay 
    format /^([TID:]+ [^ ]+ [^ ]+ \[(?<time>[^\]]+)\]) ([^***]+[^=]+[^ ]+(?<service_name>[^,]*)[^=]+[^ ]+(?<transaction_id>[^,]*)[^<?]+(?<payload>[^{]*))/ 
    time_format %Y-%m-%d %H:%M:%S 
    keep_time_key true 
</source> 

<source> 
    @type tail 

    path /opt/wso2esb-4.9.0-wkr-1/repository/logs/wso2carbon.log 
    pos_file /cellcard/fluent/wso2carbon.log.pos 
    tag wso2.esb.surepay.trigger 
    format /^([TID:]+ [^ ]+ [^ ]+ \[(?<time>[^\]]+)\]) ([^*]+[^=]+[^ ]+(?<client_ip>[^,]*)[^=]+[^ ]+(?<service_name>[^,]*)[^=]+[^ ]+(?<req_id>[^,]*)[^=]+[^ ]+(?<content_massage>[^,]*)[^=]+[^ ]+)/ 
    time_format %Y-%m-%d %H:%M:%S 
    keep_time_key true 
</source> 


########################### 
#  Filter   # 
########################### 
<filter wso2.esb.service.**> 
    @type grep 

    <regexp> 
    key  service_name 
    pattern ^\sNew 
    </regexp> 
</filter> 

<filter wso2.esb.service.**> 
    @type record_transformer 
    enable_ruby 

    <record> 
    data ${record["data"].strip.split(";").each_slice(2).to_h.to_json} 
    </record> 
</filter> 

<filter wso2.esb.service.**> 
    @type parser 

    format json 
    key_name data 
</filter> 

<filter wso2.esb.ne.surepay> 
    @type grep 

    <regexp> 
    key  service_name 
    pattern ^\sNE_SurePay 
    </regexp> 
</filter> 

<filter wso2.esb.ne.surepay> 
    @type record_transformer 
    enable_ruby 

    <record> 
    service_name ${record["service_name"].strip!} 
    transaction_id ${record["transaction_id"].strip!} 
    payload ${record["payload"].strip!} 
    </record> 
</filter> 

<filter wso2.esb.surepay.trigger> 
    @type grep 

    <regexp> 
    key  service_name 
    pattern ^\sSurePayPassiveTrigger 
    </regexp> 
</filter> 

<filter wso2.esb.surepay.trigger> 
    @type record_transformer 
    enable_ruby 

    <record> 
    client_ip ${record["client_ip"].strip!} 
    service_name ${record["service_name"].strip!} 
    req_id ${record["req_id"].strip!} 
    content_massage ${record["content_massage"].strip!} 
    </record> 
</filter> 


########################### 
#  Output   # 
########################### 
## Debug 
## match tag=debug.** and dump to console 
<match debug.**> 
    @type stdout 
    @id stdout_output 
</match> 

## ESB Service Log 
## match tag=wso2.esb.**. Forward to Fluentd Collector (, stdout for debug) and write to file 
<match wso2.esb.**> 
    @type copy 

    <store> 
    @type forward 
    @id forward_output 
    buffer_path /cellcard/fluent/buffer/fluentd.forward 
    buffer_type file 
    flush_interval 10 
    send_timeout 60 
    heartbeat_type tcp 
    heartbeat_interval 20 

    <server> 
     host 172.16.100.243 
     port 24224 
    </server> 
    ## If have sencondary fluentd server for fail-over, enable <secondary> block 
    # <secondary> 
    # <server> 
    #  host 192.168.0.12 
    # </server> 
    # </secondary> 
    </store> 

    <store> 
    @type file 
    @id file_output 

    path /cellcard/fluent/log/wso2 
    time_slice_format %Y%m%d%H 
    time_slice_wait 10m 
    time_format  %Y-%m-%d %H:%M:%S%z 
    </store> 

    <store> 
    @type stdout 
    </store> 
</match> 

Fluentd收集器（从所有fluentd每个节点上收集的数据）：

############################################################################################# 
# Fluentd Server Configuration File                # 
#                       # 
# In v1 configuration, type and id are @ prefix parameters.         # 
# @type and @id are recommended. type and id are still available for backward compatibility # 
############################################################################################# 

################################ 
#   Source   # 
################################ 
## built-in TCP input 
## $ echo <json> | fluent-cat <tag> 
<source> 
    @type forward 
    @id forward_input 

    port 24224 
</source> 

# Listen DRb for debug 
<source> 
    @type debug_agent 
    @id debug_agent_input 

    bind 127.0.0.1 
    port 24230 
</source> 

# HTTP input 
# http://localhost:8888/<tag>?json=<json> 
#<source> 
# @type http 
# @id http_input 

# port 8888 
#</source> 

# Listen HTTP for monitoring 
# http://localhost:24220/api/plugins 
# http://localhost:24220/api/plugins?type=TYPE 
# http://localhost:24220/api/plugins?tag=MYTAG 
<source> 
    @type monitor_agent 
    @id monitor_agent_input 

    port 24220 
</source> 


########################### 
#  Filter   # 
########################### 
# <filter wso2.esb.service.**> 
# @type grep 

# <regexp> 
#  key  service_name 
#  pattern ^New 
# </regexp> 
# </filter> 

# <filter wso2.esb.ne.surepay> 
# @type grep 

# <regexp> 
#  key  service_name 
#  pattern ^NE_SurePay 
# </regexp> 
# </filter> 

# <filter wso2.esb.ne.surepay> 
# @type grep 

# <regexp> 
#  key  service_name 
#  pattern ^SurePayPassiveTrigger 
# </regexp> 
# </filter> 


########################### 
#  Output   # 
########################### 
## Debug 
## match tag=debug.** and dump to console 
<match debug.**> 
    @type stdout 
    @id stdout_output 
</match> 

## ESB Service Log 
## match tag=wso2.esb.service.** and insert into database (, stdout for debug) and write to file 
<match wso2.esb.**> 
    @type copy 

    <store> 
    @type sql 
    buffer_path /cellcard/fluent/buffer/fluentd.sql 
    buffer_type file 
    flush_interval 10 

    host {ORACLE_HOST} 
    port 1521 
    database {ORACLE_DATABASE} 
    adapter oracle_enhanced 
    username {ORACLE_USERNAME} 
    password {ORACLE_PADDWORD} 

    <table> 
     table {TABLE_NAME} 
     column_mapping 'insert_date:insert_date,transaction_id:transaction_id,service_name:service_name,process_step:process_step,msisdn:msisdn,command:command,transaction_type:transaction_type,action:action,service_price:service_price,subcriber_type:subcriber_type,transaction_status:transaction_status,notification:notification,remark:remark,vas_error_code:vas_error_code,client_username:client_username,client_ip:client_ip,api_url:api_url,api_method:api_method,nei_name:nei_name,nei_error_code:nei_error_code,server_host:server_host' 
     # This is the default table because it has no "pattern" argument in <table> 
     # The logic is such that if all non-default <table> blocks 
     # do not match, the default one is chosen. 
     # The default table is required. 
    </table> 

    <table wso2.esb.service.test> 
     table {TABLE_NAME} 
     column_mapping 'insert_date:insert_date,transaction_id:transaction_id,service_name:service_name,process_step:process_step,msisdn:msisdn,command:command,transaction_type:transaction_type,action:action,service_price:service_price,subcriber_type:subcriber_type,transaction_status:transaction_status,notification:notification,remark:remark,vas_error_code:vas_error_code,client_username:client_username,client_ip:client_ip,api_url:api_url,api_method:api_method,nei_name:nei_name,nei_error_code:nei_error_code,server_host:server_host' 
    </table> 

    <table wso2.esb.ne.surepay> 
     table {TABLE_NAME} 
     column_mapping 'time:insert_date,transaction_id:transaction_id,service_name:service_name,payload:payload' 
    </table> 

    <table wso2.esb.surepay.trigger> 
     table {TABLE_NAME} 
     column_mapping 'time:insert_date,client_ip:client_ip,service_name:service_name,req_id:req_id,content_massage:content_massage' 
    </table> 
    </store> 

    <store> 
    @type file 
    path /cellcard/fluent/log/service 
    time_slice_format %Y%m%d%H 
    time_slice_wait 10m 
    time_format  %Y-%m-%d %H:%M:%S%z 
    </store> 

    <store> 
    @type stdout 
    </store> 
</match> 

注：我使用frontd从WSO2尾部日志，然后插入到Oracle数据库。

PLATFORM：的RedHat 7，红宝石2.4.1p111，fluentd 0.12.40，ActiveRecord的-oracle_enhanced适配器（1.8.2），红宝石OCI8（2.2.5），流利-插件-SQL（0.6.1 ）。

UPDATE 我发布在GitHub上 https://github.com/oemdaro/fluent-oracle-example

所有的配置和安装细节