apiVersion: v1
data:
  data.conf: "input {\n\tkafka {\n\t\tbootstrap_servers => \"${KAFKA_BOOTSTRAP_SERVERS}\"\n\t\tclient_id
    => \"context-broker-logstash\"\n\t\ttopics_pattern => \"^data.*$\"\n\t\tcodec
    => json\n\t}\n}\n\nfilter {\n\truby {\n\t\tcode => \"event.set('central_logstash_time',
    Time.now());\"\n\t\tadd_tag => \"central_logstash_time\"\n\t}\n\n\tmutate {\n\t\tconvert
    => { \"central_logstash_time\" => \"string\" }\n\t\tadd_tag => \"central_logstash_time2string\"\n\t}\n\n\tdate
    {\n\t\tmatch => [ \"central_logstash_time\", \"ISO8601\" ]\n\t\ttarget => \"central_logstash_time\"\n\t\ttag_on_failure
    => \"central_logstash_time2data_failure\"\n\t\tadd_tag => \"central_logstash_time2date\"\n\t}\n}\n\noutput
    {\n\telasticsearch {\n\t\thosts => \"${ELASTICSEARCH_HOSTS}\"\n\t\tindex => \"data\"\n\t}\n\n\tif
    \"debug\" in [tags] {\n\t\tstdout {\n\t\t\tcodec  => rubydebug {\n\t\t\t\tmetadata
    => true\n\t\t\t}\n\t\t}\n\t}\n}\n"
  event.conf: "input {\n\tkafka {\n\t\tbootstrap_servers => \"${KAFKA_BOOTSTRAP_SERVERS}\"\n\t\tclient_id
    => \"context-broker-logstash\"\n\t\ttopics_pattern => \"^event.*$\"\n\t\tcodec
    => json\n\t}\n}\n\nfilter {\n\truby {\n\t\tcode => \"event.set('central_logstash_time',
    Time.now());\"\n\t\tadd_tag => \"central_logstash_time\"\n\t}\n\n\tmutate {\n\t\tconvert
    => { \"central_logstash_time\" => \"string\" }\n\t\tadd_tag => \"central_logstash_time2string\"\n\t}\n\n\tdate
    {\n\t\tmatch => [ \"central_logstash_time\", \"ISO8601\" ]\n\t\ttarget => \"central_logstash_time\"\n\t\ttag_on_failure
    => \"central_logstash_time2data_failure\"\n\t\tadd_tag => \"central_logstash_time2date\"\n\t}\n}\n\noutput
    {\n\telasticsearch {\n\t\thosts => \"${ELASTICSEARCH_HOSTS}\"\n\t\tindex => \"event\"\n\t}\n\n\tif
    \"debug\" in [tags] {\n\t\tstdout {\n\t\t\tcodec  => rubydebug {\n\t\t\t\tmetadata
    => true\n\t\t\t}\n\t\t}\n\t}\n}\n"
  proxy-data.conf: |
    input {
      kafka {
        bootstrap_servers => "${KAFKA_BOOTSTRAP_SERVERS}"
        client_id => "context-broker-logstash"
        topics => ["AstridProxyReadData"]
        metadata_max_age_ms => 60000
        codec => json
      }
    }

    filter {
      ruby {
        code => "event.set('central_logstash_time', Time.now());"
        add_tag => "central_logstash_time"
      }

      mutate {
        convert => { "central_logstash_time" => "string" }
        add_tag => "central_logstash_time2string"
      }

      date {
        match => [ "central_logstash_time", "ISO8601" ]
        target => "central_logstash_time"
        tag_on_failure => "central_logstash_time2data_failure"
        add_tag => "central_logstash_time2date"
      }
    }

    output {
      elasticsearch {
        hosts => "${ELASTICSEARCH_HOSTS}"
        index => "proxy-data"
      }

      if "debug" in [tags] {
        stdout {
          codec  => rubydebug {
            metadata => true
          }
        }
      }
    }
  packetbeat-data.conf: |
    input {
      kafka {
        bootstrap_servers => "${KAFKA_BOOTSTRAP_SERVERS}"
        client_id => "context-broker-logstash"
        topics => ["PacketBeatData"]
        metadata_max_age_ms => 60000
        codec => json
      }
    }

    filter {
      ruby {
        code => "event.set('central_logstash_time', Time.now());"
        add_tag => "central_logstash_time"
      }

      mutate {
        convert => { "central_logstash_time" => "string" }
        add_tag => "central_logstash_time2string"
      }

      date {
        match => [ "central_logstash_time", "ISO8601" ]
        target => "central_logstash_time"
        tag_on_failure => "central_logstash_time2data_failure"
        add_tag => "central_logstash_time2date"
      }
    }

    output {
      elasticsearch {
        hosts => "${ELASTICSEARCH_HOSTS}"
        index => "packetbeat-data"
      }

      if "debug" in [tags] {
        stdout {
          codec  => rubydebug {
            metadata => true
          }
        }
      }
    }
kind: ConfigMap
metadata:
  #creationTimestamp: "2020-11-20T14:25:51Z"
  managedFields:
  - apiVersion: v1
    fieldsType: FieldsV1
    fieldsV1:
      f:data:
        .: {}
        f:data.conf: {}
        f:event.conf: {}
    manager: kubectl-create
    operation: Update
    #time: "2020-11-20T14:25:51Z"
  name: logstash-pipeline
  namespace: astrid-kube
  #resourceVersion: "6704517"
  selfLink: /api/v1/namespaces/astrid-kube/configmaps/logstash-pipeline
  #uid: 9de5c51c-613d-49d3-9fc1-3a14ce632b68
