example.yml

inputs:
    - NewKafka:
        topic:
            nginx: 1
        codec: json
        consumer_settings:
            bootstrap.servers: 192.168.0.1:9092
            value.deserializer: org.apache.kafka.common.serialization.StringDeserializer
            key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
            group.id: hangout
    - Kafka:
        codec: plain
        encoding: UTF8 # defaut UTF8
        topic: 
          app: 2
        consumer_settings:
          group.id: hangout
          zookeeper.connect: 192.168.1.200:2181
          auto.commit.interval.ms: "1000"
    - Kafka:
        codec: json
        topic: 
          web: 1
        consumer_settings:
          group.id: hangout
          zookeeper.connect: 192.168.1.201:2181
          auto.commit.interval.ms: "5000"

filters:
    - Filters:
        if:
          - '<#if message??>true</#if>'
          - '<#if message?contains("liu")>true<#elseif message?contains("warn")>true</#if>'
        filters:
            - Grok:
                match:
                  - '^(?<logtime>\S+) (?<user>.+) (-|(?<level>\w+)) %{DATA:msg}$'
                remove_fields: ['message']
            - Add:
                fields:
                    test: 'abcd'
            - Date:
                src: logtime
                formats:
                    - 'ISO8601'
                remove_fields: ['logtime']
    - Grok:
        match:
          - '^(?<logtime>\S+) (?<user>.+) (-|(?<level>\w+)) %{DATA:msg}$'
        remove_fields: ['message']
    - Add:
        fields:
            test: 'abcd'
        if:
          - '<#if message??>true</#if>'
          - '<#if message?contains("liu")>true<#elseif message?contains("warn")>true</#if>'
    - Date:
        src: logtime
        formats:
            - 'ISO8601'
        remove_fields: ['logtime']
    - Lowercase:
        fields: ['user']
    - Add:
        fields:
          me: 'I am ${user}'
    - Remove:
        fields:
          - logtime
    - Trim:
        fields:
          - user
    - Rename:
        fields:
          me: he
          user: she
    - Gsub:
        fields:
          she: ['c','CCC']
          he: ['(^\w+)|(\w+$)','XXX']
    - Translate:
        source: user
        target: nick
        dictionary_path: /tmp/app.dic
    - KV:
        source: msg
        target: kv
        field_split: ' '
        value_split: '='
        trim: '\t\"'
        trimkey: '\"'
        include_keys: ["a","b","xyz","12"]
        exclude_keys: ["b","c"] # b in excluded
        tag_on_failure: "KVfail"
        remove_fields: ['msg']
    - Convert:
        fields:
            cs_bytes: 
                to: integer
                remove_if_fail: true # default false
            time_taken: 
                to: float
                setto_if_fail: 0.0 # default NOT set anything; should set remove_if_fail to false
    - URLDecode:
        fields: ["query1","query2"]
    - Json:
        field: message # required
        remove_fields: ['a', 'b']
    - GeoIP2:
        source: message # required
        target: geoip # default geoip
        database: '/tmp/GeoLite2-City.mmdb'
        country_code: false # default true
        country_name: false # default true
        country_isocode: false # default true
        subdivision_name: false # default true
        city_name: false # default true
        latitude: false # default true
        longitude: false # default true
        location: false # default true
    - UA:
        source: ua # required

outputs:
    - Stdout:
        if:
            - '<#if user=="childe">true</#if>'
    - Elasticsearch:
        cluster: hangoutcluster
        hosts:
          - 192.168.1.200
        index: 'hangout-%{user}-%{+YYYY.MM.dd}'
        index_type: logs # default logs
        document_id: ${id} # defautt null, generated by es
        bulk_actions: 20000 #default 20000
        bulk_size: 15 # default 15 MB
        flush_interval: 10 # default 10 seconds
        concurrent_requests: 0 # default 0, concurrent_requests设置成大于0的数, 意思着多线程处理, 以我应用的经验,还有是一定OOM风险的,强烈建议设置为0
        timezone: "Asia/Shanghai" # defaut UTC 时区. 只用于生成索引名字的字符串格式化
        sniff: false #default true
    - Kafka:
        broker_list: 192.168.1.200:9092
        topic: test2