Logging and instrumentation overhaul

[utaal]

For various use-cases we're working on overhauling the logging in timely dataflow, with a couple of requirements in mind:

• instrumentation should be low/zero overhead when it's not in use;
• we should be able to stream the logs out to a separate computation/system for real-time analysis (this is specifically for some research use-cases and should be discussed);
• we'd like to be able to replay the instrumentation in the same computation that produced them, if at all possible.

[utaal]

A work-in-progress branch is here: https://github.com/frankmcsherry/timely-dataflow/pull/92

[frankmcsherry]

Here are some thoughts,

1. I would suspect that it would be pretty low overhead if we have value-typed log events everywhere, each logging event is the equivalent of Vec::push into an allocation with sufficient capacity, and unused logs are just a Vec::clear when they reach capacity. My sense is that logging events are generally much lighter weight than the events they correspond to, and as long as that holds we could build and discard the logs pretty efficiently. This becomes less true if the logged events require allocation, formatting, stuff like that.

   The plus side to this is that we might not have to have a compile-time flag to enable the logging, which would be appealing. Folks could enable and start to use the logs without rebuilding the executable, which sounds like the only reasonable way to run things (other than benchmarking).

2. I think this is super important. I think we should check whether something like capture_into can handle this, as it shouldn't be doing much more work than serializing the log data. It is possible that custom serialization could be better for some reason, or that avoiding timely dataflow mechanisms for logging simplifies the "logging the logging" issue. Though, that might be cool to have too. For more on avoiding logging logging, check out comment later on.

3. This also seems important. Ideally we should be able to attach to the logs as timely streams, even if we don't capture them as such (if it is a pain to do so). If that means attach to logs within a timely process, great! If it means a different remote process with fewer/more workers, should also be great (any thing else sounds like a defect).

Here is some thinking on how we might do this:

Right now we have per-worker logging about everything that goes on. There are various flavors of everything, but each flavor captures all the work done across all the dataflow graphs. This has some issues, like logging timestamps which have heterogenous types (which we format as strings and log).

By comparison, Naiad had per-scope logging. At least, it did for a while; we may have ripped it out (sorry, Michael). But, it makes some sense, especially with the timely scope-as-operator design. One can imagine being interested in the logging events for a scope S without any interest in progress traffic beneath the scope. This would also simplify the timestamp question, as all timestamp types are available to scope-specific loggers.

This could complicate logger acquisition, or might make us want to host them in the Subgraph types themselves (or anywhere else) to avoid constant thread_local! look-ups followed by "do you have a log for my scope identifier?" interactions. I'm slightly in favor of finding out where to host the loggers so that there is less random access to thread local mystery, and more "each channel adapter is constructed with a logger"; it needn't be all logging, but to the extent that we can localize the logging and worry less about global namespaces and unrelated types needing to hit the same objects, great imo.

---

One point of "contention" let's say is whether the logged events should have a homogenous type (perhaps an enum) and be logged in one stream, or should have different types and be logged in different streams. There are some pros and cons:

1. If events are logged in one stream, it is very easy to recover the total order of events from a worker, as they are recorded in exactly this order. If you need to do this and have several independent streams, you'll need to do a merge between the relevant streams, even though we knew at time of recording which ones went where.

2. If events are logged in separate streams, it is very easy to load or track the subset of the events that interest you. For example, if we only want to watch communication between workers, and not scheduling events, it could be very helpful to selectively enable the streams of interest and only record / transport those. Working from a single stream would mean filtering through a larger volume of logged data.

I have a slight preference, but I'm also not the one who has being doing a bunch of the log analysis. :)

[utaal]

In my attempt to re-structure the logging so that it can be easily redirected and included logging from the communication layer, having to deal with the thread_local was definitely a bit of a pain (and sometimes tricky to remember whether I was "on the right thread" for some code/initialisation to make sense): it would probably help to make it a first-class citizen of the computation/scope structure.

---

As a possible "compromise" between 1 and 2, what about we have a single stream (so you don't have to re-sort and combine things at the other end), but it's still possible to enable/disable specific streams? On the other hand, reconstructing things on the receiving end is probably relatively cheap anyways, especially if we are smart about flushing things when they get big enough (we should probably flush when we filled a reasonably-sized chunk, rather then after each .step()).

The other thing I'm not sure about is, if you're shipping the logs to a separate computation and they are separate streams, you may end up having to open a whole lot of sockets. Not a big deal of course, we can probably just multiplex all the log streams on a single socket / channel per process.

[frankmcsherry]

For more on avoiding logging logging, check out comment later on.

I forgot the comment. If we have separate logging streams for different scopes, we can just disable the logging of the logging scope.

As a possible "compromise" between 1 and 2, what about we have a single stream (so you don't have to re-sort and combine things at the other end), but it's still possible to enable/disable specific streams?

This seems like a reasonable thing to start with. If we've not screwed things up, we should (ideally?) be able to turn on and off these streams, so that if (hypothetically) I want one single log and you want lots of little logs, I could turn yours off and you could turn mine off. It doubles the amount of possibly wasted work with each logging event (the numbers of Vec::push and Vec::clear), but it seems like a fine place to start. It also has the advantage that if either of the approaches are determined to be silly, we can just comment them out.

The other thing I'm not sure about is, if you're shipping the logs to a separate computation and they are separate streams, you may end up having to open a whole lot of sockets.

Yeah, this is a fine question. Probably same questions, issues, but could be resolved independently. My guess is

1. Logging should be enabled / disabled in process.
2. What to do with the logged data should also be determined in process.
3. If you want to use one TCP connection, you should be able to copy/paste whatever we did in networking.rs, and if you want to use multiple connections that should be fine too. In any case, someone would need to write that code.

This seems a bit like "columns" in streams. We could imagine one stream with lots of data, several possible consumers, and some angst about which way is best "one true way" to store the data. I don't know that we've solved that issue either though.

[utaal]

This seems like a reasonable thing to start with. If we've not screwed things up, we should (ideally?) be able to turn on and off these streams, so that if (hypothetically) I want one single log and you want lots of little logs, I could turn yours off and you could turn mine off. It doubles the amount of possibly wasted work with each logging event (the numbers of Vec::push and Vec::clear), but it seems like a fine place to start. It also has the advantage that if either of the approaches are determined to be silly, we can just comment them out.

To be more specific, I was suggesting to have a single log stream that carries an enum LogEvent, and logging configuration can enable/disable specific sub-streams. This would possibly turn into a single Vec::push / Vec::clear per event, it gets us multiplexing for free, and you can still separate the streams at the receiving end. Of course, it's a bit less clean/type-safe.

1. Logging should be enabled / disabled in process.
2. What to do with the logged data should also be determined in process.

In the current experiment, you get a handle to the "raw stream" of logs, and can decide to
send_to_tcp() or to replay_into() another scope within the computation. If we do it right we may be able to make the replaying/sending extensible, by exposing the right hooks from the logging handle.

[frankmcsherry]

I was suggesting to have a single log stream that carries an enum LogEvent, and logging configuration can enable/disable specific sub-streams.

I propose we also be able to turn off the single stream. :) My sense is that it will be the expensive one, as putting e.g. all the progress timestamps into a homogenous format will most likely involve allocation.

In the current experiment, you get a handle to the "raw stream" of logs, and can decide to
send_to_tcp() or to replay_into() another scope within the computation. If we do it right we may be able to make the replaying/sending extensible, by exposing the right hooks from the logging handle.

My thinking was that if we have a Rc<RefCell<Logger>> everywhere that a logger needs to be used, then we could in-prinicple have enable() and disable() or more RAII-ly get_listener() and Drop logic that lets the logger know whether it should flush() or clear() when its buffer fills.

[frankmcsherry]

More concretely, one design might look like how differential's Arrange streams batches out to other computations: there is a listeners: Vec<Weak<RefCell<Box<Pusher>>>> which has a bunch of folks ready to hear about logging events; each time a batch of events fills, the logger swings through this list and drop a copy of the batch of logged events (.clone(), I guess?) in each reference it can upgrade to Rc, and then clear the local stash.

You sign up for this by minting a Rc<RefCell<Box<Pusher>>>> and handing a copy to the logger, which downgrades it to Weak. Now for as long as you/your code holds the handle, you'll get to see events. If you want to stop, you just drop your strong Rc.

[utaal]

More concretely, one design might look like how differential's Arrange streams batches out to other computations: there is a listeners: Vec<Weak<RefCell<Box>>> which has a bunch of folks ready to hear about logging events; each time a batch of events fills, the logger swings through this list and drop a copy of the batch of logged events (.clone(), I guess?) in each reference it can upgrade to Rc, and then clear the local stash.

Makes sense. Should this be one Vec per worker - for the timely logging, and one Vec per process - for the communication logging?
One thing we may want to consider is letting timely thin out the logs if "nothing of interest has happened" (e.g. we were waiting from input from other workers and we don't want to send out a bunch of .step()s where nothing happened).