[
    {
        "source":"/foo/bar.log",
        "offset":702,
        "FileStateOS":{"inode":535737,"device":64529},
        "timestamp":"2019-04-29T16:32:59.514463586+08:00",
        "ttl":-2
    },
    {
        "source":"/foo/foo_bar.log",
        "offset":190,
        "FileStateOS":{"inode":540909,"device":64529},
        "timestamp":"2019-04-29T16:32:59.514466675+08:00",
        "ttl":-2
    }
]

      # Filebeat全局配置
# spool_size:后台事件计数阈值，超过后强制发送，默认2048
filebeat.spool_size: 2048
# idle_timeout:后台刷新超时时间，超过定义时间后强制发送，不管spool_size是否达到，默认5秒
filebeat.idle_timeout: 5s

# shutdown_timeout:filebeat关闭前最长等待publisher发送事件成功时间
filebeat.shutdown_timeout: 5s
filebeat.config_dir: /etc/filebeat/conf.d

# 日志路径
logging.to_files: true
logging.level: info
logging.files:
  path: /data/filebeat/logs/

output:
    logstash:
      hosts: ["10.10.15.100:5044", "10.10.16.101:5044"]
      # loadbalance事实上filebeat5.5只会选hosts中一个logstash连接,而不是测量所有logstash状态来平均发送,只有当前连接中的logstash连接不上,才会重新尝试hosts里面其他logstash
      loadbalance: true

    redis:
      hosts: ["redis_host_1","redis_host_2"]
      port: "redis_port"
      db: "7"
      password: "password"
      # key:这里%{[log_group]}我们是取log_group变量的值,这个值具体会在每个日志收集配置里面设置，通过log_group变量区分redis上的key
      key: '%{[log_group]}'
      data_type : "list"
      # 启用线程数量
      threads : 10
      # 返回的事件数量,此属性仅在list模式下起作用
      batch_count : 2500

    kafka:
      hosts: ["100.200.200.100:9093","100.200.200.101:9093","100.200.200.102:9093"]
      topic: '%{[kafka_key]}'
      version: 0.10.0.0
      compression: gzip
      worker: 1
      required_acks: 1
      max_retries: 3
      max_message_bytes: 1000000
      partition.round_robin:
        reachable_only: false
      keep_alive: 86400

      timeout: 60
      broker_timeout: 60

      # 如果输出到的Kafka开启了安全认证,就需要配置下面这些参数
      # username: "kafka_username"
      # password: "kafka_password"
      # ssl.enabled: true
      # ssl.certificate_authorities: ["/path_to/ca-cert"]
      # ssl.verification_mode: none

      # Filebeat每个日志配置
filebeat.prospectors:
  - input_type: log
    paths:
      - /foo/bar.log
    # enabled:每个prospectors的开关，默认true
    enabled: true
    # scan_frequency:prospect指定的目录下面检测文件更新
    # 如果你发现filebeat占用CPU过高，可以调低scan_frequency扫描频率
    scan_frequency: 10s
    # close_inactive:如果在指定时间没有被读取，将关闭文件句柄
    # Solve Too many open files handlers problem
    close_inactive: 5m
    # close_renamed:如果文件被重命名和移动，关闭文件的处理读取
    close_renamed: true
    # close_removed:文件被删除时，关闭文件的处理读取
    # Solve file rotation problem
    close_removed: true
    # ignore_older:忽略指定时间段以外修改的日志内容
    ignore_older: 24h
    # clean_inactive: >=ignore_older+scan_frequency
    # 以确保在文件仍在收集时没有删除任何状态
    clean_inactive: 48h
    # clean_removed:如果文件在磁盘上找不到，将从注册表中清除记录
    # 若关闭close_removed 必须关闭clean_removed
    clean_removed: true
    fields:
        kafka_key: foo_bar_kafka_key
        log_group: foo_bar_group
        type: foo_bar_type
    fields_under_root: true

      input{
    beats {
        port => 5044
    }
    file {
        path => ["/path_to/logstash-plain.log"]
        add_field => { "log_group"=>"logstash-log" }
    }
    redis {
        host => "redis_host"
        port => "redis_port"
        db => "redis_db"
        password => "password"
        key => "list_name"
        data_type => "list"
        threads => 1 #启用线程数量
        batch_count => 500 #返回的事件数量，此属性仅在list模式下起作用
    }
    kafka {
        bootstrap_servers => "100.200.200.100:9093","100.200.200.101:9093","100.200.200.102:9093"
        topics => ["foo_bar_kafka_key","topic_2"]
        group_id => "GoupID"

        consumer_threads => 3
        # decorate_events:把Kafka一些元数据(比如topic,partition,offset等)加入事件中
        decorate_events => true
        # codec:输入数据以json格式解析
        codec => json
        # type:设置事件的"type"字段值，如果原本有"type"字段，则不会覆盖原本type里面的值
        type => "access"

        # poll_timeout_ms:kafka等待从topic拉取数据的时间
        poll_timeout_ms => 10000
        # fetch_max_wait_ms:拉取数据最大等待时间,该值应该小于等于poll_timeout_ms
        fetch_max_wait_ms => "10000"
        session_timeout_ms => "150000"
        # request_timeout_ms:请求超时时间,应该大于session_timeout_ms + fetch_max_wait_ms
        request_timeout_ms => "300000"

        max_poll_records => "500"
        auto_commit_interval_ms => "5000"

        # fetch_max_bytes(3*1024k=3145728 advice from alibaba) is a new option start from logstash6.0
        # fetch_max_bytes => "3145728"
        # logstash5.5 doesn't have fetch_max_bytes option,use max_partition_fetch_bytes(3*1024k/24 = 131072 advice from alibaba) instead
        max_partition_fetch_bytes => "131072"

        # 如果输出到的Kafka开启了安全认证,就需要配置下面这些参数
        # security_protocol => "SASL_SSL"
        # sasl_mechanism => "PLAIN"
        # jaas_path => "/path_to/jaas.conf"
        # ssl_truststore_password => "KafkaOnsClient"
        # ssl_truststore_location => "/path_to/kafka.client.truststore.jks"
    }
}

      filter{
  if [log_group] == "foo_bar_group" {
    if [type] == "for_bar_type" {
        grok {
            keep_empty_captures => true
            match =>{
                "message" => "^\"(?<ymd>%{NUMBER}|null|)\",\"%{DATA:network}\",\"(?<active_time>%{NUMBER}|null|)\",\"%{DATA:ip}\""
            }
            # remove_field:去除被解析过的message字段,减少相同信息的传输,减少传输体积
            remove_field=>["message"]
        }

        mutate {
            convert => ["ymd","integer","active_time","integer"]
        }

        uuid {
              target => 'log_uuid'
        }

        fingerprint {
            concatenate_sources => true
            method => "SHA256"
            key => "KEYKEYKEY"
            source => ["ymd","network", "active_time", "ip"]
            target => "log_uuid_fp"
        }

        geoip {
            source => "ip"
            target => "geoip"
            fields => [city_name, region_name,latitude,longitude]
        }

        ruby {
          code=>"
              path = event.get('source').split('/')
              event.set('src_path_prefix', path[-2])
              event.set('src_path_filename', path[-1])
          "
        }

        if !("_grokparsefailure" in [tags] or "_groktimeout" in [tags]){
          if [type] != "foo_bar_type"{
            ruby {
              code=>"
                log_time = Time.local(event.get('year'), event.get('month'), event.get('day'), event.get('hour'), event.get('minute'), '0').to_i
                event.set('log_time', log_time)
              "
            }
          }
          if [type] == "foo_bar_type"{
            ruby {
              code=>"
                require 'date';
                year = Date.parse(event.get('ymd').to_s).strftime('%Y')
                month = Date.parse(event.get('ymd').to_s).strftime('%m')
                day = Date.parse(event.get('ymd').to_s).strftime('%d')
                log_time = Time.local(year, month, day, event.get('hour'), 0, 0).to_i
                log_ym = Time.at(log_time).strftime('%Y%m').to_i
                event.set('log_time', log_time)
                event.set('log_ym', log_ym)
              "
            }
          }
          date {
              match => ["log_time", "UNIX"]
              target => "@timestamp"
          }
        }
    }
  }
}

      output{
  if [log_group] == 'foo_bar_group' {
    if "_grokparsefailure" in [tags] or "_groktimeout" in [tags]{
      file {
        path => "/path_to_error_log/%{src_path_}/%{src_path_filename}"
        codec => line { format => "%{message}"}
      }
    } else {
      if [type] == "foo_bar_type"{
        jdbc {
          connection_string => "jdbc:mysql://db_host:3306/for_bar_db?user=username&password=password&useUnicode=true&characterEncoding=UTF-8&autoReconnect=true"
          # unsafe_statement:允许你在statement中用事件中的变量值,比如for_bar_%{ymd}
          unsafe_statement => true
          # max_flush_exceptions:最大output异常次数,和retry_initial_interval和retry_max_interval三者可以计算出重试时间
          # 比如10次output全部失败重试时间 => 2+4+8+16+32+64+128+128+128+128=638s
          max_flush_exceptions => 10
          retry_initial_interval => 2
          retry_max_interval => 128
          retry_sql_states => ["100","HY00","2015"]
          statement => ["INSERT INTO for_bar_%{ymd} (log_uuid_fp,log_uuid,log_time,ymd,network,active_time,ip,longitude,latitude,region_name,city_name,source_host) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)","log_uuid_fp","log_uuid","log_time","ymd","network","active_time","ip","geoip[longitude]","geoip[latitude]","geoip[region_name]","geoip[city_name]","beat[hostname]"]
        }
        kafka {
          # 输出前格式化为json格式
          codec => json
          bootstrap_servers => "100.200.200.100:9093","100.200.200.101:9093","100.200.200.102:9093"
          topic_id => "for_bar_json"

          # 如果输出到的Kafka开启了安全认证,就需要配置下面这些参数
          # sasl_mechanism => "ONS"
          # jaas_path => "/path_to/jaas.conf"
          # ssl_truststore_password => "KafkaOnsClient"
          # ssl_truststore_location => "/path_to/kafka.client.truststore.jks"
        }
        elasticsearch {
          codec => "json"
          hosts => ["es_host:9200"]
          user => "user_name"
          password => "password"
          index => "foo-bar-%{+YYYY.MM}"
          document_type=>"foobar_doc_type"
          manage_template => true
          template_name => "foo-bar"
          template_overwrite => true
          template => "/path_to/for_bar.json"
        }
      }
    }
  }
}

      {
  "template": "for-bar-*",
  "settings": {
    "index.refresh_interval": "2s",
    "number_of_shards": "3",
    "number_of_replicas": "1",
    "max_result_window": "10000"
  },
  "mappings": {
    "info": {
      "properties": {
        "log_uuid_fp": {
          "type": "keyword"
        },
        "log_uuid": {
          "type": "keyword"
        },
        "log_time": {
          "type": "integer"
        },
        "ymd":{
          "type": "integer"
        },
        "network":{
          "type": "keyword"
        },
        "active_time":{
          "type": "integer"
        },
        "ip":{
          "type": "ip"
        },
        "longitude":{
          "type": "float"
        },
        "latitude":{
          "type": "float"
        },
        "region_name":{
          "type": "keyword"
        },
        "city_name":{
          "type": "keyword"
        },
        "source_host":{
          "type": "keyword"
        }
      }
    }
  }
}

      cd /data/joygames/elasticsearch-head
npm run start &