Da ich keine Lösung zum Ausschließen von Datensatzschlüssel mit leerem Wert finden kann, verwende ich die umgekehrte Lösung. Ich verwende grep, um den angegebenen Schlüsselwert zu speichern. Siehe unten meine Fluentd-Konfiguration.
Fluentd an jedem WSO2-Knoten.
#############################################################################################
# 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 Sammler (Daten aus allen fluentd auf jedem Knoten zu sammeln):
#############################################################################################
# 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>
HINWEIS: Ich benutze frontd bis zum Schwanz log von WSO2 und fügen Sie anschließend in Oracle Database.
PLATFORM: RedHat 7, Rubin 2.4.1p111, fluentd 0.12.40, active-oracle_enhanced-Adapter (1.8.2), Rubin-oci8 (2.2.5), fließend-plugin-SQL (0.6.1).
UPDATE Ich habe alle auf https://github.com/oemdaro/fluent-oracle-example
GitHub Detail Konfiguration und Installation veröffentlichen