apiVersion: v1
kind: ConfigMap  # 定义logstash入数据和出数据的规则
metadata:
  name: logstash-config
  namespace: elastic

data:
  logstash.conf: |-
    input {
      kafka {
        bootstrap_servers => ["ip:9092"]
        group_id => "logstash-kafka"
        auto_offset_reset => "latest"
        consumer_threads => "12"
        decorate_events => "false"
        topics => ["test"]
        codec => json
      }
    }

    filter {
      date {
        match => [ "startDate", "UNIX_MS" ]
      }
      ruby { 
        code => "event.set('timestamp', event.get('@timestamp').time.localtime + 8*60*60)"
      }
      ruby {
        code => "event.set('index_date', event.get('timestamp').time.localtime.strftime('%Y.%m.%d.%H'))"
      }
      mutate {
        remove_field => ["timestamp"]
      }
    }

    output {
      elasticsearch {
        hosts => ["es 地址"]
        user => "username"
        password => "password"
        index => "test-%{index_date}"
      }
    }
