logstash config to break out systemd output from `/usr/lib/systemd/systemd-journal-upload`

logstash.conf

input {
  tcp {
    port => 19532
    type => journald
  }
}

filter {
  if [type] == "journald" {
    multiline {
      pattern => "^$"
      negate => true
      what => "next"
    }

    grok {
      match => ["message", "CODE_FILE=%{GREEDYDATA:CODE_FILE}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "CODE_FUNC=%{GREEDYDATA:CODE_FUNC}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "CODE_FUNCTION=%{GREEDYDATA:CODE_FUNCTION}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "CODE_LINE=%{GREEDYDATA:CODE_LINE}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_CGROUP=%{GREEDYDATA:COREDUMP_CGROUP}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_CMDLINE=%{GREEDYDATA:COREDUMP_CMDLINE}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_COMM=%{GREEDYDATA:COREDUMP_COMM}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_CWD=%{GREEDYDATA:COREDUMP_CWD}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_ENVIRON=%{GREEDYDATA:COREDUMP_ENVIRON}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_EXE=%{GREEDYDATA:COREDUMP_EXE}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_FILENAME=%{GREEDYDATA:COREDUMP_FILENAME}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_GID=%{GREEDYDATA:COREDUMP_GID}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_OPEN_FDS=%{GREEDYDATA:COREDUMP_OPEN_FDS}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_PID=%{GREEDYDATA:COREDUMP_PID}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_PROC_CGROUP=%{GREEDYDATA:COREDUMP_PROC_CGROUP}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_PROC_LIMITS=%{GREEDYDATA:COREDUMP_PROC_LIMITS}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_PROC_MAPS=%{GREEDYDATA:COREDUMP_PROC_MAPS}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_PROC_STATUS=%{GREEDYDATA:COREDUMP_PROC_STATUS}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_ROOT=%{GREEDYDATA:COREDUMP_ROOT}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_SIGNAL=%{GREEDYDATA:COREDUMP_SIGNAL}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_SLICE=%{GREEDYDATA:COREDUMP_SLICE}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_TIMESTAMP=%{GREEDYDATA:COREDUMP_TIMESTAMP}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_UID=%{GREEDYDATA:COREDUMP_UID}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "COREDUMP_UNIT=%{GREEDYDATA:COREDUMP_UNIT}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "EXIT_CODE=%{GREEDYDATA:EXIT_CODE}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "EXIT_STATUS=%{GREEDYDATA:EXIT_STATUS}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "MESSAGE_ID=%{GREEDYDATA:MESSAGE_ID}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "PRIORITY=%{GREEDYDATA:PRIORITY}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "RESULT=%{GREEDYDATA:RESULT}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "SYSLOG_FACILITY=%{GREEDYDATA:SYSLOG_FACILITY}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "SYSLOG_IDENTIFIER=%{GREEDYDATA:SYSLOG_IDENTIFIER}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "UNIT=%{GREEDYDATA:UNIT}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_BOOT_ID=%{GREEDYDATA:_BOOT_ID}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_CAP_EFFECTIVE=%{GREEDYDATA:_CAP_EFFECTIVE}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_CMDLINE=%{GREEDYDATA:_CMDLINE}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_COMM=%{GREEDYDATA:_COMM}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_EXE=%{GREEDYDATA:_EXE}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_GID=%{GREEDYDATA:_GID}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_HOSTNAME=%{GREEDYDATA:_HOSTNAME}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_MACHINE_ID=%{GREEDYDATA:_MACHINE_ID}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_PID=%{GREEDYDATA:_PID}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_SOURCE_REALTIME_TIMESTAMP=%{GREEDYDATA:_SOURCE_REALTIME_TIMESTAMP}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_SYSTEMD_CGROUP=%{GREEDYDATA:_SYSTEMD_CGROUP}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_SYSTEMD_SLICE=%{GREEDYDATA:_SYSTEMD_SLICE}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_SYSTEMD_UNIT=%{GREEDYDATA:_SYSTEMD_UNIT}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_TRANSPORT=%{GREEDYDATA:_TRANSPORT}" ]
      tag_on_failure => []
    }
    grok {
      match => ["message", "_UID=%{GREEDYDATA:_UID}" ]
      tag_on_failure => []
    }

    grok {
      match => ["message", "MESSAGE=%{GREEDYDATA:message}" ]
      tag_on_failure => []
      overwrite => ["message"]
      remove_tag => "multiline"
    }

  }
}

output {
  elasticsearch {
    embedded => true
  }

}

systemd-journal-upload.service

[Unit]
Description=Send journald entries to a remote logstash instance
After=journald.service
Requires=journald.service
Documentation=man:systemd-journal-upload(8)


[Service]
Restart=always
RestartSec=0
# By default systemd-journal-upload wants to connect using SSL on port 19532
# The following line will connect to example.com on port 19532 using HTTP
# vs HTTPS.  Refer to 
# http://www.freedesktop.org/software/systemd/man/systemd-journal-upload.html
ExecStart=/usr/lib/systemd/systemd-journal-upload -u http://example.com

Hi, I've been trying to get systemd-journal-upload output into Logstash. What format does it get sent in?
I had created an RFE on bugs.freedesktop.org asking for JSON export as well as the ability to add in some extra key value pairs if possible. i.e. "Environment": "Dev", "Type": "Appserver", "Project": "Feedback Website". Looks like you're doing a fair bit of grok work to get the data into shape.
I'll create another RFE if it looks like it would help. The man page for systemd-journal-upload is pretty sparse.

@duncaninnes I hacked together a plugin for the journal that may or may not help
https://github.com/stuart-warren/logstash-input-journald

@duncaninnes IIRC the format is sent as chunked HTTP, hence being able to hunt for the blank line using the multiline parser at the top. I would love for json to be supported.

Its weird that JSON isnt supported. Almost all their other tools support json.