gistfile1.txt

topology: ops_hangout_dimg

inputs:
    - Kafka:
        codec: json
        topic: logstash-opslinuxlog-nginx-dimg
        spoutname: kafkaspout
        consumer_settings:
            group.id: logstash
            zookeeper.connect: "10.8.84.74:2181"
            auto.commit.interval.ms: "1000"


filters:
    - Grok:
        if:
          - '<#if path!="/opt/app/nginx/logs/error">true</#if>'
        src: message
        match:
            - '%{IPORHOST:remote_addr}%{SPACE}-%{SPACE}-%{SPACE}\[%{HTTPDATE:time_local}\]%{SPACE}"%{WORD:method}%{SPACE}%{NOTSPACE:request_other}%{SPACE}%{NOTSPACE}"%{SPACE}%{INT:status_code}%{SPACE}%{INT:body_bytes_sent}%{SPACE}"%{NOTSPACE:http_referer}"%{SPACE}"%{DATA:http_user_agent}"%{SPACE}"%{DATA:http_x_forwarded_for}"%{SPACE}"%{NOTSPACE:domain}"%{SPACE}"%{NOTSPACE:req_time_f}"'
        remove_fields: ['message']
    - Add:
        if:
          - '<#if request_other??>true</#if>'
        fields:
          request: '${domain}${request_other}'
    - Add:
        if:
          - '<#if url?? && url?contains("images4.c-ctrip.com/target/")>true</#if>'
        fields:
          target: '<#assign a=url?split("/")>${a[2]}'
    - Date:
        src: time_local
        formats:
            - 'dd/MMM/YYYY:HH:mm:ss Z'

outputs:
    - Elasticsearch:
        cluster: toolsES
        hosts:
          - 10.8.84.65
          - 10.8.84.66
          - 10.8.84.67
          - 10.8.84.68
        index: 'dimg-%{+YYYY.MM}-test'
        index_type: "logs"
        bulk_actions: 20000
        bulk_size: 20
        flush_interval: 30
        concurrent_requests: 0