logstash-simple.conf

input {
 redis {
    data_type => channel
    host => "${LOGSTASH_REDIS_HOST}"
    id => joblogs_in
    key => "logstash:in"
  }
}

filter {
  if "wflow-nonint" in [kubernetes][event][involved_object][name] {
    grok {
      match => { "[kubernetes][event][involved_object][name]" => "wflow-nonint-%{UUID:wflowid}%{GREEDYDATA}" }
    }
    grok {
      match => { "[wflowid]" => "(?<idpart1>[A-Fa-f0-9]{2})(?<idpart2>[A-Fa-f0-9]{2})(?<idpart3>[A-Fa-f0-9]{2})%{GREEDYDATA:idrest}" }
    }
    mutate {
      add_field => { "logdir" => "%{idpart1}/%{idpart2}/%{idpart3}/%{idrest}"}
      add_field => { "msg_type" => "wflow_event"}
      add_field => { "msg" => "%{[kubernetes][event][message]}"}
      remove_field => ["idpart1", "idpart2", "idpart3", "idrest"]
    }
  }
  if "wflow-job" in [kubernetes][event][involved_object][name] {
    grok {
      match => { "[kubernetes][event][involved_object][name]" => "wflow-job-%{UUID:jobid}%{GREEDYDATA}" }
    }
    grok {
      match => { "[jobid]" => "(?<idpart1>[A-Fa-f0-9]{2})(?<idpart2>[A-Fa-f0-9]{2})(?<idpart3>[A-Fa-f0-9]{2})%{GREEDYDATA:idrest}" }
    }
    mutate {
      add_field => { "logdir" => "%{idpart1}/%{idpart2}/%{idpart3}/%{idrest}"}
      add_field => { "msg_type" => "subjob_event"}
      add_field => { "msg" => "%{[kubernetes][event][message]}"}
      remove_field => ["idpart1", "idpart2", "idpart3", "idrest"]
    }
  }
  if [kubernetes][labels][component] == "wflow-engine" and [kubernetes][labels][interactive] == "false"  {
    grok {
      match => { "[kubernetes][labels][wflowid]" => "(?<idpart1>[A-Fa-f0-9]{2})(?<idpart2>[A-Fa-f0-9]{2})(?<idpart3>[A-Fa-f0-9]{2})%{GREEDYDATA:idrest}" }
    }
    mutate {
      add_field => { "logdir" => "%{idpart1}/%{idpart2}/%{idpart3}/%{idrest}"}
      remove_field => ["idpart1", "idpart2", "idpart3", "idrest"]

      add_field => { "msg_type" => "wflow_log"}
      add_field => { "wflowguid" => "%{[kubernetes][labels][wflowid]}"}
      add_field => { "msg" => "%{[message]}"}
      add_field => { "date" => "%{@timestamp}"}
    }
  }
  if [kubernetes][labels][component] == "wflow-job"  {
    grok {
      match => { "[kubernetes][labels][jobid]" => "(?<idpart1>[A-Fa-f0-9]{2})(?<idpart2>[A-Fa-f0-9]{2})(?<idpart3>[A-Fa-f0-9]{2})%{GREEDYDATA:idrest}" }
    }
    mutate {
      add_field => { "logdir" => "%{idpart1}/%{idpart2}/%{idpart3}/%{idrest}"}
      add_field => { "msg_type" => "subjob_log"}
      add_field => { "msg" => "%{[message]}"}
      add_field => { "date" => "%{@timestamp}"}
      remove_field => ["idpart1", "idpart2", "idpart3", "idrest"]
    }
  }

  if [msg_type] == "wflow_state" {
    grok {
      match => { "[wflowguid]" => "(?<idpart1>[A-Fa-f0-9]{2})(?<idpart2>[A-Fa-f0-9]{2})(?<idpart3>[A-Fa-f0-9]{2})%{GREEDYDATA:idrest}" }
    }
    mutate {
      add_field => { "logdir" => "%{idpart1}/%{idpart2}/%{idpart3}/%{idrest}"}
      remove_field => ["idpart1", "idpart2", "idpart3", "idrest"]
    }
  }

}

output {
  if [msg_type] == "wflow_event" {
    file {
      path => "${LOGSTASH_WFLOW_LOGBASE}/%{logdir}/events.log"
    }
    redis {
      data_type => channel
      host => "${LOGSTASH_REDIS_HOST}"
      id => wflowevent_out
      key => "logstash:out"
    }
  }

  if [msg_type] == "subjob_event" {
    file {
      path => "${LOGSTASH_JOB_LOGBASE}/%{logdir}/events.log"
    }
    redis {
      data_type => channel
      host => "${LOGSTASH_REDIS_HOST}"
      id => subjobevent_out
      key => "logstash:out"
    }
  }
  if [msg_type] == "wflow_log" and [kubernetes] {
    file {
      path => "${LOGSTASH_WFLOW_LOGBASE}/%{logdir}/log.log"
    }
    redis {
      data_type => channel
      host => "${LOGSTASH_REDIS_HOST}"
      id => wflowlog_out
      key => "logstash:out"
    }
  }

  if [msg_type] == "subjob_log" {
    file {
      path => "${LOGSTASH_JOB_LOGBASE}/%{logdir}/run.log"
    }
    redis {
      data_type => channel
      host => "${LOGSTASH_REDIS_HOST}"
      id => subjoblog_out
      key => "logstash:out"
    }
  }

  if [msg_type] == "wflow_state" {
    file {
      path => "${LOGSTASH_WFLOW_LOGBASE}/%{logdir}/state.log"
    }
    redis {
      data_type => channel
      host => "${LOGSTASH_REDIS_HOST}"
      id => wflowstate_out
      key => "logstash:out"
    }
  }
}