airbnb / plog

Fire-and-forget UDP logging service with custom Netty pipelines & extensive monitoring

plog

Fire and forget unboxed or fragmented messages over UDP or TCP, have them forwarded to Kafka 0.8 or the standard output. That should cover syslog.

Disclaimer

As-is: This project is not actively maintained or supported. While updates may still be made and we welcome feedback, keep in mind we may not respond to pull requests or issues quickly.

Let us know! If you fork this, or if you use it, or if it helps in anyway, we'd love to hear from you! [email protected]

Getting started

$ printf 'plog.server.udp.listeners=[{handlers=[{provider="com.airbnb.plog.console.ConsoleOutputProvider"}]}]' > plog-distro/src/main/resources/application.conf
$ ./gradlew run
$ printf 'yipee!' | socat -t0 - UDP-DATAGRAM:127.0.0.1:23456
$ printf '\0\0statsplz' | socat - UDP-DATAGRAM:127.0.0.1:23456

Build / Upload

• To build a shadow JAR:

    $ ./gradlew shadowJar
  

• To build source JARs and upload to bintray manually:

  • Create an account on https://bintray.com (you need to create and register your own org first)

  • Ask an admin (e.g. Alexis Midon @alexism) to invite your newly created user to Airbnb's account

  • find your bintray API key (Edit Profile > API Key) and set BINTRAY_USER and BINTRAY_KEY environment variables locally. e.g.

    $ export BINTRAY_USER=<your_user_id>
    $ export BINTRAY_KEY=<your_api_key>
    

  • Tag master with a newer version. Use git describe --tags to see the previous version. e.g.

    $ git describe --tags --dirty
    v4.0.0-BETA-36-g21add12
    

    This means the previous tag was v4.0.0-BETA, there's been 36 commits since the tagged commit, and HEAD is at 21add12. We use com.github.ben-manes.versions to apply set the build version to the output of 'git describe --tags --dirty'

    To tag a new version, do something like the following

    $ git tag -a v4.0.1
    $ git describe --tags --dirty
    v4.0.1
    $ git push origin v4.0.1
    

  • Now you are ready to upload to bintray, by running the following command in the root plog directory

    $ ./gradlew build sourcesJar bintrayUpload
    

Configuration

All configuration can be done through system properties. The app settings are read using Typesafe Config.

Kafka settings

Please refer to Kafka's documentation.

plog settings

Please refer to reference.conf for all the options and their default values.

Note that multiple TCP and UDP ports can be configure and have separate settings, and each has their own sink (whether Kafka or standard output).

Building a fat JAR

$ ./gradlew shadowJar
$ ls -l plog-distro/build/libs/*-all.jar

Operational tricks

• To minimize packet loss due to "lacks", increase the kernel socket buffer size. For Linux, we use sysctl net.core.rmem_max = 4194304 and configure plog.udp.defaults.SO_RCVBUF accordingly.

• Hole detection is a bit difficult to explain, but worth looking into (the tests should help). It is enabled by default, but can be disabled for performance.

Event logging at Airbnb

We use JSON objects with the following fields:

• type: String. Only very few values are acceptable due to our pipeline splitting event streams by type.
• uuid: String.
• host: String.
• timestamp: Number. Milliseconds since Epoch.
• data: Object. Arbitrary.

Statistics

Let's go through all keys in the JSON object exposed by the STAT command:

• version: the current version if available from the JAR manifest, or unknown
• failed_to_send: number of times Kafka threw FailedToSendMessageException back
• exceptions: number of unhandled exceptions.
• udp_simple_messages: number of unboxed UDP messages received.
• udp_invalid_version: number of UDP messages with version between 1 and 31.
• v0_invalid_type: number of UDP messages using version 0 of the protocol and a wrong packet type.
• unknown_command: number of commands received that aren't known (eg KLIL instead of KILL).
• v0_commands: number of valid commands received.
• v0_invalid_multipart_header: number of v0 fragments received with invalid headers (could not be parsed).
• v0_fragments (array): count of fragments received, whether valid or not, clustered by log2 of their index. Ie, the first number indicates how many first packets we've received, the second number how many second, the third number how many 3rd and 4th, the fourth how many 5th, 6th, 7th, 8th, etc.
• v0_invalid_fragments (array of arrays): count of invalid fragments received, clustered first by log2 of (their message's size - 1), then by their fragment index. A fragment is considered invalid if:
  • its header provides a fragment size, fragment count, or checksum that doesn't match the values from the first fragment we processed
  • its length is incorrect: the payload length should always match the fragment size provided in the first fragment we processed, unless they're at the end of the message, in which case they should exactly match the message length provided in the first fragment we processed.
• v0_invalid_checksum (array): count of messages received where the MurmurHash3 did not match the payload, clustered by log2 of (their fragment count - 1).
• dropped_fragments (array of arrays): count of fragments we expected to receive but didn't before evicting them from the defragmenter, clustered first by log2 of (their message's size - 1), then by their fragment index.
• cache (object):
  • evictions: count of yet-to-be-completed messages were evicted from the cache, either because they expired or we needed to make room for new entries (see defrag.max_size and defrag.expire_time in the config).
  • hits: how many times we tried to add fragments to an already known message. Note that this operation will fail for invalid fragments.
  • misses: how many times we received fragments for a message that we didn't know about yet (we don't hit the cache for single-fragment messages).
• kafka (object):
  • Keys: byteRate, messageRate, failedSendRate, resendRate, droppedMessageRate, serializationErrorRate
  • Values: objects with keys count and rate, an array offering 1-min, 5-min and 15-min rates.

TCP protocol

Line-by-line separated, lines starting with \0 are reserved.

UDP protocol

• If the first byte is outside of the 0-31 range, the message is considered to be unboxed and the whole packet is parsed as a string.

• Otherwise, the first byte indicates the protocol version. Currently, only version 00 is defined.

Version 00

• Byte 00: version (00)
• Byte 01: packet type

Packet type 00: commands

Command packet. Commands are always 4 ASCII characters, trailing payload can be used. Command matching is case-insensitive.

• KILL crashes the process without any attention for detail or respect for ongoing operations.

    $ printf '\0\0kill'|socat -t0 - UDP-DATAGRAM:127.0.0.1:23456
  

• PING will cause the process to reply back with PONG. Trailing payload is sent back and can be used for request/reply matching.

    $ printf "\0\0PingFor$$\n\n"|socat - UDP-DATAGRAM:127.0.0.1:23456
    PONGFor17575
    
    $
  

• STAT is used to request statistics in UTF-8-encoded JSON. By convention, the trailing payload should be used for politeness.

    $ printf "\0\0statistics please, gentle service"|socat - UDP-DATAGRAM:127.0.0.1:23456
    {""udpSimpleMessages":0, [...]}
  

• ENVI returns the environment as a UTF-8-encoded string. The format is not defined further.

Packet type 01: fragmented message

Note that 1-fragment fragmented messages are perfectly possible.

• Bytes 02-03: unsigned, big-endian, 16-bit integer. Fragment count for the message (between 1 and 65535).
• Bytes 04-05: unsigned, big-endian, 16-bit integer. Index of this fragment in the message (between 0 for the first fragment and 65534).
• Bytes 06-07: unsigned, big-endian, 16-bit integer. Byte length of the payload for each fragment in the message.
• Bytes 08-11: big-endian, 32-bit integer. Second half of the identifier for the message. Messages are identified by the UDP client port and this second half. Needs to increment with each message for hole detection.
• Bytes 12-15: signed, big-endian, 32-bit integer below 2,147,483,647. Total byte length of the message.
• Bytes 16-19: big-endian, 32-bit MurmurHash3 hash of the total message payload.
• Bytes 20-21: unsigned, big-endian, 16-bit integer. taglength: Size used to represent tags.
• Bytes 22-23: zeroes. Reserved, might be used in later revisions.
• Bytes 24-(24+taglength): Bytes. List of tags (\0-separated UTF-8 strings; can be \0-terminated or not).
• Bytes (24+taglength)-: Bytes. Payload. Will only read the payload length.

plog's Issues

Allow for native transport under Linux

Netty offers a native transport.
It should soon allow for great perf improvements by using recvmmsg/sendmmsg.
See netty/netty#2719 for early support.

Sadly Plog currently hardcodes the transport to Nio. This should be configurable.

Allow tags in UDP v0 protocol

• One extra field (2 bytes) in the header is for the metadata length.
• The metadata is prepended to the payload.
• The metadata consists of one MessagePack object (we could add more afterwards), an array of strings, called "tags".
• Those tags get mapped to byte[][] in Message.
• Tags from the last processed fragment win; well-behaving clients only set tags on the last fragment to avoid reducing the chunk size or increasing packet count.

Message filters

We should keep Plog simple and focused. However app-specific filters could provide value: fire-and-forget eventing in apps is great, but when events need some extra work, there's currently no good way to keep that work local (hence distributed horizontally and away from network spies) yet out of the app workers.

Such tasks could include filtering out sensitive information, converting messages between formats, annotating them with information about the host or its state, etc.

We could have per-port filters, with the following design decisions for a first shot:

• In the port config, add the optional:

  filters = [
    { provider: "your.class",
      ...
    }, ...
  ]
  

• your.class would be found in the classpath, have to implement:

  interface FilterProvider { PlogFilter getHandler(Config cfg); }
  

• And PlogFilter in turn:

  interface PlogFilter extends ChannelHandler { /* extras? */ }
  

This model would allow handlers to hold state, turn each message into zero or more messages, let exceptions get logged and counted in our stats (TBV).

Questions left out of this initial implementation, hopefully forever as they would add complexity:

• Filter metrics reporting (poll by adding to the interface, push via the pipeline?)
• ?

Revamp stats

Stats are a bit limited right now; per-port stats would be really useful, but keeping global stats would be nice too.

Look into Guava service for plog-server

Collect all failure counts into a single object in stats

... so we can sum them into a single metric.

Support for kt: tags in plog-kafka

• If a message has kt:foo and kt:bar tags, forward the message to the foo and bar topics.
• Configuration now specifies a default topic, used for messages that do not have such tags.
• If the default topic is not set, a warning is printed on startup and messages are ignored (allows for more clever pipelines)
• The magical value null can be used for the default or tag to avoid sending to Kafka, for example for warmup or to use the same port for multiple sinks and client-side routing.

Expose GC stats

It would be useful to expose minimal histograms of GC stats.

See ManagementFactory.getGarbageCollectorMXBeans()

Expose stats over JMX

Could be useful to companies relying on JMX for monitoring, eg via Jolokia.

Write documentation for filters

As soon as that feature is ironed out and used in production, we should document them filters properly.

Publish artifacts to jcenter

Expose memory usage stats

See ManagementFactory.getMemoryMXBean().

Replace sinks with out-of-tree filters

• Removes the dependency to Kafka
• Removes a bunch of code from plog-server into separate modules
• Unifies the model around Netty 4 channels

Would break the config format.

Revamp TCP protocol

Proposal to be reviewed:

• Use \0 as a line prefix for advanced protocol.
• Advanced protocol supports commands, including stats.
• Advanced protocol allows to specify tags, ask for acknowledgement(?).

Provide defaults for listeners

Currently, each UDP and TCP port has to be fully configured.
Offer UDP & TCP default configuration sections that get merged with each port at startup.

Report end-of-pipeline messages

If messages reach the end of the pipeline, we should do something about it.

Separate but similar to end of pipeline exceptions counting mechanism?

build.gradle: establish version using git

~~README.mdon https://github.com/ajoberstar/gradle-git should cover it.~~ (doesn't supportgit describe`-like naming)

Tags like v4.0.0 should turn version into "4.0.0", and non-tagged releases should contain an unique suffix.

Bring the Datadog monitoring up to speed for v4

The stats format changed with the move to custom handler chains.

Adapt the datadog plugin, make use of the YAML config file to specify what should be reported.

Groovy REPL as a Gradle task

The classpath should match testCompile in plog-distro.

This could be very useful to quickly hack around stuff.

Double-log stats give unnecessary numbers, and seem b0rked

plog.Stress: cannot send only 1 byte payloads

With min_size=1, max_size=1, size_increments=1,

Exception in thread "main" java.lang.RuntimeException: No sizes! Decrease plog.stress.size_increments

Publish artifacts to bintray

This will make it easy to spin separate projects off, like plog-upshot.

Listeners: log which port is listened on

Support for binding locally, enabled by default

So no need to firewall off.

Statistics for filters

Allow filters to report stats.

Support stdout as a target instead of Kafka

Based on #10.

Provide full isolation for UDP

Verify that each UDP listener has its own receiving threads.

Clean up integration tests

• Integration tests should pass
• We shouldn't play around around with the console as we can trivially have our own in-memory "sink"

Allow multiple `bind()`s

Netty allows us to bind multiple times, which is a very useful performance improvement for UDP sockets under Linux as it allows for multiple listeners.

In that case, it would be useful to allow threads = 0 and skip the thread pool altogether in UDPListener, to avoid the dispatch overhead (including extra context switches).

Kafka settings coming from Typesafe Config

System properties aren't that great.

Support multiple TCP/UDP ports

• Allows multiple buffers, one per app would be a good practice
• Per-port Kafka topics