A guide to working with Riemann

Changing the config

If you use the Debian or Centos packages, Riemann adds a riemann user, stores its configuration in /etc/riemann/riemann.config, and logs to /var/log/riemann.log. You should always tail the log file when working with Riemann; it'll alert you to configuration errors and help you debug your streams.

In another terminal, use your favorite editor to open /etc/riemann/riemann.config. Let's add a logging stream, so we can see all events that pass through the streams. #(info %) expands into (fn [x] (info x)), which is a function that takes an event and logs it at the INFO log level.

Now we need to reload the config file. Riemann will respond to SIGHUP, or you can use the init scripts. You'll see a message about reloading the config in the Riemann log. If you make a mistake, like a syntax error, Riemann will continue running with the old config and won't apply the new one. Instead, it'll log an explanation of the problem, so you can investigate and fix it.

Now that the config has been reloaded, you should see events flowing by in the log. Some come from external sources, like riemann-health or your own programs. Others are generated by Riemann internally, as a part of its internal performance instrumentation. You can insert logging statements anywhere in your streams to verify what kind of events are flowing through that point.

Reloading is experimental, and is subject to the normal Clojure rules about (def) and (defn). To avoid confusion, use let bindings instead of def; it'll guarantee reloads work correctly. If reloads seem broken, you might need to do a full restart for your changes to take effect.

We present here a minimal configuration which only prints incoming events to standard output. This configuration can prove useful when debugging clients to make sure expected events are reaching Riemann.

This example includes a full configuration, later examples will focus on streams. The following actions are performed:

• logging/init: Setting-up logging to log to standard output (STDOUT).
• tcp-server: Starting a Riemann TCP server on port 5555 (the default).
• instrumentation: Disabling internal event production.

Since streams expects a list of functions to call with individual events, we can output any input by calling prn on them.

Often your Riemann configuration file will get quite large as you add streams and handle more events. Additionally, if you're managing Riemann with a configuration management tool like Puppet or Chef then it can be complex to template a large configuration file. To help with this Riemann also supports including functions via Clojure's namespacing model. These can include your own custom stream functions.

To enable this, the contents of the directory containing the riemann.config configuration file are included in the CLASSPATH.

To use other functions then create a source directory for them. For example, to create functions specific to your organization you might create a examplecom.etc namespace.

Then create a namespace containing the functions we wish to include in a file in this directory, for example email.clj.

We would then require this namespace in our riemann.config. With this we could use the email variable inside our Riemann configuration.

You can read more about Clojure's namespacing and how it works.

This approach is supported in Riemann 0.2.11 and later.

Riemann has three main parts: clients, the server, and dashboard. By default, all listen on the loopback interface. You'll need to configure each to listen to an appropriate interface for your environment. On the host that runs the Riemann server, open /etc/riemann/riemann.config, and change the host that Riemann binds to. To listen on all interfaces, use 0.0.0.0.

Now we need to reload Riemann to tell it about our changes.

To change the address that the dashboard binds to, edit riemann-dash's config.rb.

Riemann-dash only serves some JS code and a small piece of configuration; when you open riemann-dash in the browser, it'll open connections from your browser to the Riemann server's websocket interface to receive events. To connect the dashboard to a remote Riemann server, double-click the text box in the top right of the dashboard, and change the address from 127.0.0.1:5556 to your Riemann server's host and port. When you hit enter, the browser will initiate a websocket connection to that address and begin displaying events. Hit s to save the new dash config; the dashboard will connect to that address every time you load the page.

Riemann-dash is not the only dashboard for Riemann; Riemann decouples data processing and visualization. Opening the Riemann server from your browser *won't* work.

If the dashboard is unable to connect to the Riemann websocket server, you'll see an alert pop up every few seconds. Check that the server is running, that the Riemann websocket server is reachable from your browser, that you used the correct host and port, and so on.

If you don't see *anything* on the dash, but it connected successfully, you may not have any events to show, or you may not have set up any views.

Instead of installing Riemann directly on your system, you can also use Docker to run it. There is a prepackaged Docker image available as riemannio/riemann.

With its default configuration, Riemann will index all incoming events and print expired events. Make sure to bind the neccessary ports to your host system, or else you won't be able to send events to Riemann.

If you expect lots of traffic and don't need segregating of containers, you can also skip the Docker network and use the host network by using the --net host flag. Note that the default Riemann configuration in Docker will listen on all network interfaces, so make sure to either mount a different configuration (see below) or configure your firewall appropriately.

Riemann will, by default, use a configuration file located at /etc/riemann.config, which you will most likely override by mounting your own configuration into the container.

For this you can either mount your configuration file at the default location or override the default container command (/bin/riemann /etc/riemann.config) if for some reason you want to change more command line options.

To keep things manageable, you can use Docker Compose to define how the Riemann container should be run.

Oracle JDK8 is likely to be the fastest option (and supports the full range of aggressive tuning options with riemann -a), but we test and deploy with JDK 8, from both Oracle and OpenJDK, and JDK 9 from Oracle. You're welcome to give other JVMs a shot, though!

Use the Riemann TCP protocol. If you have to, and know what you're doing, back off to UDP, HTTP, or Graphite.

Use UDP if you intend to have the OS and network discard data automatically, instead of application-level flow control. Don't use UDP "for speed"; it's designed to drop data, it will drop data, and you'll spend a long time trying to figure out why your stats are wrong. UDP may be appropriate for sampling events where discarding large parts of the datastream is OK, but if different nodes drop packets at different rates you'll introduce sampling bias. If you don't understand what this means, don't use UDP.

Use HTTP only if a TCP client is unavailable; it introduces significant encoding and state machine overhead, takes more bytes on the wire, and can't represent the full range of Riemann events.

Use the Graphite server and other compatibility shims only for interop with legacy systems. The graphite protocol can only represent a small subset of the Riemann data model; you'll have to do additional work to reconstruct meaningful events.

Riemann runs servers by default on several ports.

• TCP on port 5555.
• TLS on port 5554.
• UDP on port 5555.
• Websockets on port 5556.
• REPL on port 5557.
• OpenTSBD on port 4242
• Graphite on port 2003

To log to a file, just say

(logging/init {:file "/path/to/some/riemann.log"})

If you'd prefer to only log to stdout, just leave out :file

(logging/init)

Under the hood, Riemann is using Logback.

To configure Logback, you can provide a configuration file via the system property logback.configurationFile.

For more information, please read the Logback documentation.

You can use Riemann across any TCP or UDP VPN, or an SSH tunnel. Riemann also supports bidirectional TLS auth as a part of its TCP protocol.

Before you can use TLS, you need a CA, server key and cert, and a client key and cert. Refer to less-awful-ssl and easy-rsa for more details.

Then, in riemann.config, add a new (or repurpose an existing) TCP server for TLS.

Refer to your client's documentation or source for the client TLS options. In riemann-clojure-client, try something like:

Sometimes, something in Riemann doesn't work the way you'd like. When this happens, you can redefine any function in Riemann in the config file. Just switch to the appropriate namespace, redefine the function, and switch back to riemann.config.

For instance, let's say you wanted to change how emails are formatted. The namespace riemann.common has a function called body, which accepts a sequence of events and returns a string. We'll override it in our config file:

Because Riemann performs arbitrary computation over the event stream, with side effects, it cannot be distributed safely. You can, however, distribute it somewhat less than safely--a technique aligned with Riemann's philosophy that "mostly correct information right now is more useful, than totally correct information only available once the failure is over."

One obvious distribution strategy is to place two Riemann nodes on the same switch and have them share a virtual IP address using, say, Heartbeat or Corosync. On resource acquisition, the node allocating the IP should issue a gratuitous arp broadcast over that interface advertising the new route to the relevant switch(es). Failover in this case looks essentially like restarting a Riemann node, which should be relatively painless--most Riemann deployments fill in missing state within a few seconds to minutes.

What happens if the nodes are isolated by a network partition and both claim the interface? Depends on your network, but the answer is "probably not good things". OTOH, there's no good way for *any* consensus algorithm in an asynchronous network to make latency-bounded decisions in a safe way. FLP rears its ugly head. You're going to lose data in the short term; the question is how to emit as much useful, actionable information as possible, so you can fix the problem.

I've got a sketch worked out for a highly-available clustered Riemann which tries to negotiate these tradeoffs in a sane way, but it involves minimum latency costs, severe throughput costs (I'm estimating three to five orders of magnitude slower), disk IO, all kinds of esoteric failure modes, and some moderately challenging mathematics. It'd take a long time to build and would require some serious educational work on my part. There's just no way around it: distributing arbitrary latency-bounded computation is hard. You're going to have to reason about consensus. I'm really sorry about this. :(

Because Riemann performs arbitrary computation, it can't shard your workload automatically. You'll need to pick some axes along which to shard, perform local aggregation on those nodes, and reduce the results with some higher-level Riemann node. The forward stream can help you connect nodes to one another, and async-queue! can allow some elasticity around latency jitters and downstream node restarts, but the actual sharding and discovery mechanics you'll have to build yourself.

Event processing in Riemann's model is synchronous (though there's lots of asynchronous stuff going on internally). When you send an event to Riemann, it flows through every top-level stream, and possibly those streams children, and *their* children, and so forth. Once the event has flowed through every stream, Riemann sends an acknowledgement back to the client. This model enables *backpressure*: clients can use the acknowledgements to avoid overwhelming Riemann with more events than it can process.

Riemann keeps track of some internal metrics, exposed as events, for stream latency and TCP server latency. Stream latency is how long it took for an event to flow through every stream. Server latency is how long it took for an event to be picked up by Riemann's TCP server, decoded, handed off to the streaming system, processed, and for the acknowledgement to be sent back to the client.

If you wait for an acknowledgement every time you send a message to Riemann, each process will behave synchronously. This approach is just fine when there are lots of clients sending requests infrequently, where "infrequently" means, via Little's Law, that the typical period between events is greater than the tcp server latency plus the round-trip network time to Riemann itself.

Sometimes, though, you'll want to send events *faster* than that. You've got a few options here.

If you ignore the acknowledgement messages, you can just shove events at Riemann as fast as your runtime will let you. Those events will fill up a queue in front of Riemann's streaming system. If that fills up, they'll spill over into the Netty queue associated with the channel. If that fills up, they'll spill over into the TCP buffer on the Riemann box, and by extension, the queues in the network and the sending node. At some point TCP backpressure will come into play, but understanding that interaction can be complex.

Doing this kind of infinitely-asynchronous send will manifest as high memory consumption on the Riemann node (to store all those queued operations), and correspondingly high wait times for events. If events arrive *too* far delayed, they may no longer be valid by the time they arrive, and you'll start seeing services expire. It's like sending a message via a courier who gets stuck in a waiting room and dies en route.

You can detect this (and alert on it) by watching the service called "riemann netty event-executor queue size"; if it starts getting above a thousand events or so, you're likely overloading Riemann. This queue size is directly related to the latency distribution given by "riemann streams latency 0.5" and friends, and "riemann streams rate". The higher the latency, and the higher the throughput, the larger the queue.

So: infinite asynchrony is a problem. You want *some* degree of asynchrony, especially when Riemann processes faster than the time it takes to send events over the network. But too much and you'll crash into finite resource limits. We need *backpressure*: bounds on the number of outstanding asynchronous requests, or the rate at which we send them.

One option is simply to spin up more threads and have each send synchronously. Depending on the client's design, you may be able to have many threads concurrently sending operations over the same connection. Riemann-java-client and riemann-clojure-client, for instance, will interleave operations from multiple threads on a single network connection. If threads are cheap in your runtime, or you don't need very many (on the JVM, more than 10K IO threads is probably a warning sign), this is perfectly fine. If you're running in Erlang or Haskell, threads are dirt cheap; fire away.

If threads in your runtime are more expensive, or you don't have them at all--say, node.js--you can emulate them using other asynchronous constructs. Riemann-java-client and riemann-clojure-client, for instance, have async-send functions that return promises; you can dereference those promises to find out whether Riemann acknowledged the message. Deref'ing blocks until acknowledgement occurs, or throws an exception if something in the network or Riemann went wrong. You can specify a timeout for deref; if the timeout is zero, you get a nonblocking check ("did it go through yet?"). Fixed timeouts are useful for retries of idempotent operations. "Allow this much time, then send again". Same semantics as any other asynchronous or stateful construct in Clojure.

The important thing for *backpressure*, if you're emulating threads using asynchronous objects, coroutines, continuations, fibers, callbacks, or what-have-you, is that you a.) eventually dereference the results of the send, and b.) only allow a fixed number of outstanding asynchronous requests. Riemann-java-client automatically throws (or returns promises which throw) when you try to send more than a few thousand concurrent messages, forcing you to implement backpressure and flow control logic. You can tune this option if you like, but it's there to protect you against unbounded resource consumption on either Riemann's side or the client.

Different clients and different languages will expose these constructs in different ways, but you'll find the same core ideas of limited asynchrony, queuing, and backpressure in every networked system.

Streams are just functions, so you can use any function that prints an event to stdout to see which events pass through a particular stream. prn prints an object to stdout, so adding prn to any stream prints the events at that stream to the console.

If you're running Riemann as a daemon, it probably won't have a stdout to print to. To write to the logfile (which is *also*, by default, printed to stdout), use #(info %), which is a shorthand for the anonymous function (fn [x] (info x)). Note that info logs events a little differently than prn, but you can use pr-str to recover the clojure representation if desired.

First, determine whether the events you're looking for are actually making it into Riemann. This will help localize the problem to the network or the streams. Add a new top-level stream which simply logs all inbound events. Use a (where) filter around that logging stream to cut down on noise.

If you don't see the events you're looking for in the logs, they must be dropped before arriving at the streaming system. Check:

• Riemann clients are sending to the correct Riemann host and port.
• Riemann is listening to that host and port. Double-check the config (e.g. (tcp-server) options), that the appropriate server startup message appears in the Riemann logs, and that and netstat -lntp | grep riemann shows the port bound.
• The network can relay packets from the client machine to the Riemann server. Use telnet some.host 5555 or nmap -sT some.host -P 5555.
• Your packets are making it from the client to the server. UDP messages can be dropped, delayed, duplicated, or re-ordered at any time by the network, or discarded if the receiving node's receive buffer is too full. Try using TCP.

If your messages are arriving in the streaming system, but aren't making it to the index or other output streams, gradually slide the #(info %) stream downstream towards the exit point, to verify at which point in the chain the events are not what you expect.

If you're indexing events, but they don't appear in queries, they may have expired from the index. Try (expired #(warn "expired" %)) to warn about expired events. They may be expiring because their TTL is shorter than the interval between events. They may also expire because the Riemann clock is in the future relative to the originating event, due to clock skew or network/processing latency.

Check the internal instrumentation for the Riemann queue depth and core latencies; if the queue is more than a handful of events deep, or stream latencies are on the order of the relevant event TTLs, you might need to reduce the load on the Riemann server or investigate downstream services that might be slow.

Check clock skew between nodes using watch date or similar. Make sure your clocks are in UTC, not local time. Never local time. Especially never Daylight Savings Time. You can use Riemann's clock-skew stream to measure clock skew as seen by the Riemann node as well.

Riemann has integrated support for building fast, repeatable tests into your config itself, including virtualized time and scheduler side effects and suppressing unwanted IO during test runs.

In your streams, use riemann.test/tap to declare named points where you'd like to observe events. For instance, we might want to track any events flowing into the index, so we'd surround it with a tap and call that tap :index.

Then, we can inject events into the config's streams to see what happens. Define tests in any config file using tests and deftest.

(inject! events) returns a map of tap names to a vector of the events that arrived at that tap. Time is always reset to zero at the beginning of each test, and is automatically advanced to each event time just prior to event arrival. You may manually advance the clock using riemann.time.controlled/advance!. Scheduled operations like rate, rollup, etc will take effect atomically as the virtual clock advances.

You can also inject events into specific streams using (inject! [stream] events).

Under the hood, tests expands to a new ns declaration based on the current namespace--if the current namespace is foo, tests take place in foo-test. tests automatically refers clojure.test's is and are, plus riemann.test's version of deftest and inject! for inserting events into the config's streams.

You can define regular clojure tests too; no need to call inject. Every namespace ending in -test is eligible for testing.

Some streams may have *side effects*, like sending events over the network or emitting emails. You can suppress those side effects by wrapping them in a test/io stream, which forwards events to its children normally in production, but ignores them in test mode.

io suppresses events, which means that (io (tap :foo)) will never see anything. Always place taps *outside* io expressions.

Note that tap and io are eliminated at compile-time during production mode, so there's zero performance penalty; won't count against your Hotspot inlining budget, etc.

From Riemann 0.3.0, the test framework runs a core. Functions requiring a core like riemann.config/reinject, indexing events, index querying and expiry will work. The index is cleaned at the beginning of each deftest.

Before Riemann 0.3.0, the test framework didn't actually run a core, so functions that require one, (like riemann.config/reinject), didn't work from within the test framework.

Riemann has an nREPL server built in. You can enable this in your config with:

You can then use Leiningen 2 to connect to it.

you can reload the config by sending riemann a sighup, but you can also do it from the repl.

Riemann comes with instrumentation built-in, which is sampled periodically and injected into the event stream. You can query the dashboard for service =~ "riemann %" to see Riemann's internal metrics. All latencies are in milliseconds.

riemann streams refers to the stream processor: its throughput measures the number of events being passed to (streams ...) per second, and its latencies indicate the time it takes to process a single event through the streaming system.

The TCP, UDP, and Websockets servers instrument their throughput and latency where possible. TCP latency measures the time from initial message deframing to Netty write of the corresponding response. For instance, riemann server tcp 1.2.3.4:5555 in latency 0.95 measures the 95th percentile time, in milliseconds, for the TCP server on port 5555 to parse, queue, process, and dispatch a response for a message. The rate for servers measures *message* rate, not *event* rate.

The main queue connecting Netty IO threads to the parsing/execution threadpool is measured by riemann netty event-executor queue size, and is a critical measure of whether the streaming system is overloaded relative to inbound request load.

To disable instrumentation, or control how often events are sampled, see riemann.config/instrumentation.

Riemann-tools includes a program called riemann-health, which measures the local host's cpu, memory, and disk use. You can install it from rubygems.

... and run riemann-health with the address of your riemann server like so:

Riemann-health reports utilization fractions by default, ranging from zero to one. Load average is divided by the number of cores, and disk use by capacity. You can adjust the polling interval, what percentage of CPU usage is considered critical, and more. riemann-health --help will tell you more.

Streams can produce exceptions. A special event containing the exception details will be sent to the first child of exception-stream if an exception occurs on other child streams.

Riemann-tools includes riemann-riak, which uses Riak's HTTP stats interface, some filesystem checks, and optionally, some erlang RPC requests to measure Riak's get and put latencies, request throughput, ring status, and disk use. Run riemann-riak on each Riak node.

Riemann-riak comes with defaults for the Riak .deb packages shipped by Basho, but it's tunable for other configurations. See riemann-riak --help for more options.

You can include arbitrary additional fields in Riemann events. These are not a replacement for service, host, and time: the composite primary key of a logical "thing" in the Riemann universe, but they *do* let you more easily aggregate and filter events, or carry richer contextual information associated with a single message.

In most clients you can simply pass an extra key and value in the event map. In strongly-typed clients like riemann-java-client, special methods like EventDSL.attribute(key, value) are present.

Custom attributes are always encoded as strings in Protocol Buffers (pull request welcome!), but you can replace them with parsed variants in your streams using smap, adjust, and friends.

In the Riemann server, custom attributes work just like normal attributes on events--they're just a bit slower, owing to the hashmap lookup. Just like normal fields, you can use get, assoc, dissoc, update, and all the other Clojure functions over maps. Custom attributes do *not* have shorthand syntax in (where) (to prevent arbitrary symbol capture), but are accessible through (:my-field event).