Realtime Data Analytics at Datadog by Alexis Le-Quoic

So, we'll skip this panel. This is me. I'm @alq on Twitter, so if you have any comments or questions, please let me know. Also, if you have any comments or questions during the talk, fire away, because I want to have it as interactive as possible. So what I figured we'd talk about tonight: there are a few things. First of all, this is not a product pitch, but just to explain what the problem space is, what it is that we're dealing with: what we do and who is the user of Datadog. Then from there, we'll go into a little bit of domain constraints, based on what the problem space is and what constraints it entails.

Problem access patterns, which basically is how do we access the data, or how our users access the data, which then will get to a presentation of our architecture, which is by and large a really iterative architecture. It took us a few years to get there, not designed to be authoritative or anything like that. Also, look at the tools that we use: data stores, languages, libraries, and so on and so forth. And last, a few challenges that we faced, so we'll just discuss a few technical challenges.

So what Datadog is, is infrastructure monitoring as a service. It's a monitoring tool. Who here uses a monitoring tool? Anything from databases, web front ends, load balancers, web tiers, application servers, bare metal, AWS, and so on and so forth. The idea is that you have one place, because your application is typically built on top of a bunch of these components, you have one place where you can see how the app is performing. So far so good.

Now, what is monitoring at the highest level? What does it mean? It's a tool, it's a system, it's whatever you've installed, whatever wakes you up at night when something goes wrong. But really, it's the primary interface you have to understand how your system is doing, where it's failing, the bottlenecks, and so on. That's your monitoring system. It's the primary interface to learn and understand how your system is doing. If you have a data center, you can't just walk in and look at the racks to understand what's happening. You need something to mediate that, and that is your monitoring system. In the worst case, your monitoring system is your customers, because they tell you when something goes wrong and you fix it. That's one way to do it, but not the way I'd recommend. In the better case, you have something that sucks performance metrics all the time, around the clock, from your various components.

First and foremost is metrics. An example of a metric at the lowest level could be how much CPU a particular machine is using, or how busy a CPU is in a given machine. At the highest level, it could be how many users are on the site at a given time. The taxonomy is relatively simple: metrics are either about resource or workload. Resource metrics are stuff you consume: CPU, network, memory, storage, and so on. Workload is what you produce, what your application is producing. To understand how things are going, you need both. In our world, the software world, resources are typically the machines, and workload is whatever you are producing. This taxonomy applies regardless of the application. The monitoring system will first and foremost consume these metrics and make some judgment about whether they are good or bad. So far so good.

Concretely speaking, I took some screenshots. This is the Datadog interface. You like colors and stuff that moves, it's cool. You're graphing because you can quickly see interesting patterns. Different metrics can exhibit the same pattern, whether it's flat, getting bad, or a spike. The brain is amazingly good at figuring out patterns, and that's why a lot of monitoring tools, including Datadog, offer a really rich graphing library. Let's walk through an example of monitoring. Here I have two metrics: one of throughput and one of contention. It could be anything: number of pages served, number of people walking in the door, etc. You have an intuition from experience that there's some kind of relationship between both. What you want to be able to do with your monitoring system is formulate a hypothesis with a few key strokes, transform the data a little bit so you can identify patterns much faster. In this case, if you take the derivative of both metrics, you can see there's some kind of correlation between contention and throughput. For instance, if you have contention on turnstiles in a subway station, the number of people entering the station is going to be affected. The ability to play with the data, slice and dice, and transform it so you can quickly validate hypotheses is key. Does that make sense? I see people nodding, go ahead.

This in particular is the number of locks in a relational database. It could be anything, but in this case, that's what it is. The throughput of that application depends on that relational database. Whenever it starts to lock up, everybody in that application starts to wait for their turn to grab the lock. The second use case is really automatic detection, because you don't have time to watch a thousand metrics. You can define alerts, thresholds, and a number of different things. The third use case is more of a team sport. Whenever you have a complex application, you need to be able to talk about the data. You have a lot of back and forth. This is one discussion we had on a Facebook metric, talking about an improvement. Those are really the three main use cases of a monitoring system: you do a lot of graphing, it's real time, you alert, and you want to learn from the data.

Now that we know the problem space and the use cases, what kind of constraints does that entail? We start from the highest level use cases, look at the constraints, and that tells us what kind of trade-offs we can make, which helps us get to an architecture. Domain constraints: the first one is latency. When I say real time, there's a lot of understanding of real time. In this context, it's a 10 to 30 second latency between the time you collect a given metric, like how much CPU a machine is using, to the time you can see it on a graph. It's really time to signal: the time from the moment you collect the metric to the time you receive it and your brain can do something with it. It's not page load; the page is a little bit faster than that hopefully. The idea is that it's not 200 milliseconds, it's not one microsecond, because for this domain, you're not going to be able to react to a CPU spike of a microsecond. There's nothing actionable there. It's also not 10 minutes, because you don't want your customers to call you and say the website is slow.

The second constraint is availability. That's standard, but in this case, we get data 24/7, 365. There's no seasonality because it's just machines talking to machines. Thanksgiving will be as busy as tomorrow. It just increases ever so slowly as we have more users and more machines reporting to the system. The third key constraint is access control, which is very generic to any multi-tenanted system. If you run something as a service, you have all your data in one place but multiple users that don't necessarily want to see each other's data. In our case, that means obviously that we need to isolate data per customer.

I talked a little bit about the 100 millisecond round trip. I don't care if it takes 50 milliseconds between the time I gather a metric on a machine and the time I can see it, because if it takes two seconds, it'll be the same result. There's no point in over-engineering to get super fast latency. This is not a financial trading application; this is monitoring infrastructure. So we can get away with that. The second thing is we don't have to engineer for crazy elasticity, like Black Friday or Cyber Monday bursts. In our case, what really drives the volume of data is how many customers or how many machines are reporting. Precision can be antagonistic with latency. Here you have two essentially the same metric in two different configurations: one where it's very precise, with a lot of data points, but it's late. The other is less precise but has lower latency. I'd much rather have this than that. If your pipeline starts to slow down, you can skip a bunch of points to catch up, because I don't care about having every single point. I'd rather make sure I can catch up and be up to date. That's an interesting trade-off that would not work at all in a financial application where you want to count every single money transfer. In our case, we can skip data points. I'm grateful for that, but it doesn't necessarily apply to what you guys are doing.

Constraints: latency 10 to 30 seconds roughly, always available, highly access controlled data. The freedoms we have: I can trade precision for latency. Am I making sense? Okay cool. Do tell me if I'm not making sense.

Having said that, I want to take a quick look at what the monitoring data looks like more concretely. There are two types of data in monitoring: metrics and events. Metrics are basically time series. It could be the temperature of this room. Events are just annotations. I'll show you a little more detail. Metrics, time series: at the most basic level, it's timestamps and values. That's the basic representation. It has a bunch of metadata: the name of the metric, for instance 'number of people in the audience', some context like 'Meetup Data Engineering Meetup', the date, units, an owner, etc. That's pretty much it. In Datadog, that's the core data model. It's very simple, very uniform, not a ton of variation. The resolution we chose is that of a second. We don't see a lot of value in measuring things at a resolution lower than one second, and it's adequate to capture signals of less than 0.5 hertz. When you set up Datadog, you deploy an agent on your laptop that collects metrics about CPU, memory, which programs are running, etc. It typically samples data every 10 to 20 seconds. It could sample more often, but you don't want to waste cycles measuring; you want your laptop or server to do useful things. Even though we can measure things down to a second, typically you look at data at a lower resolution. You can also use something called StatsD, which will pre-aggregate whatever metrics you want to send. Overall, metrics have a pretty high signal-to-noise ratio. If you have a very noisy metric that goes all over the place, it's probably not very useful. You need to derive it to get something useful. This also means you can lose some precision and still understand what the metric is telling you. The second great thing is that metrics are immutable. Because there's a timestamp and nobody's figured out time travel yet, you can persist metrics and they never change.

Here is an example of a query. If you send this query into Datadog, because use case number one is graphing, we're mostly interested in displaying data. There's no point in giving you a million data points back if we're going to compress that to a graph that fits on your screen. Typically, it's 200 to 600 points for display. These points are essentially timestamp and value. When you graph them, this is what you get. This is a measure on a machine. If you look at one month of data, you also get 600 points. There's clearly a time aggregation that takes place so that the system doesn't query two million points and send them to the browser. We time aggregate as we query and return just 200 to 600 points so you have a nice graph. Does that make sense?

If you want to get a little more complex, let's say you want the same metric but averaged across a number of machines. At the end, you get the average across a number of machines. We do what we call spatial aggregation. You take the metric for all the machines and compute the average. Because not all data is sampled at the same time, you have to do some form of interpolation so you can compute the average on the same timestamps. Does that make sense? If it doesn't, please raise your hand. This time aggregation and space aggregation is a really time-based textual annotation. It could be 'I released a new version of the software' or 'somebody got a page because a web machine went down' or 'we started a new marketing campaign'. The data model is very simple: it has title, body, and some metadata like author and tags. As an example, these are events from Nagios, a legacy monitoring system. There's a title, a body, and a timestamp. It's interesting to mix events and metrics because it lets you annotate the data. For example, if traffic spikes, you might think the servers are going to die, but it could be because of a successful marketing campaign. Being able to mix and match is really useful. There are markers on the graph that show events. It's really useful because now you have a sense of what's happening.

That's the core data model in our architecture. The rest is not that important. Now we know what kind of data we produce or gets materialized in the system. Let's look at the way it's accessed, how data flows. In most cases, this is the simplified workflow. You have agents, small programs on a machine collecting metrics, sending them through Datadog, or programs hitting our API to send data. It gets into something that we'll call the intake. Once in a while, you'll go back and look at what happened a week ago or last month. But the vast majority of the traffic outbound is a little bit of recent data and then very frequently more random queries. On the intake, we want to be able to ingest data at an ever-increasing rate. If the business is successful, we'll have more customers, more machines sending more metrics. The intake has to be able to handle that. It's not batch; it's a constant flow of stuff coming in. The latency constraint I mentioned is 30 to 60 seconds. Between the time the data is collected and the time it flows into the intake, it's in the order of tens of seconds, not minutes or milliseconds. When you have this kind of stuff, particularly when you're ingesting data at a very constant rate and you have a fairly constant upper bound for latency, you end up with a bunch of queues. That's what I'll show in a little bit.

Outbound queries are different. You query the data for a time span. Typically, our customers look at the last one to four hours because they want to know what their app is doing now. What it was doing six hours ago is sometimes interesting, but most important is what's happening now. An interesting thing is that when you look at the parameters of the queries, the time span and the metric you're interested in, when you render dashboards, people tend to look at the same time span across a bunch of different metrics. There's locality of the parameters, not completely random. Another outbound use case is alerts, telling you when something goes wrong. It's detecting anomalies. Latency is typically 60 seconds because there's no point in alerting as soon as a metric shoots up above a certain level. You'll get a ton of false positives. You want to know when it's been above a threshold for the past five minutes or the past minute. You don't want to know every second that something's happening. For something like storage or memory that's constrained, you want to know once you're getting close to the limit because there's no extending it. Alerts and spatial aggregation make things a bit more complex when you have a bunch of different time series that come at different points in time. The last thing is events. When you query for events, you say 'show me all the events of this type in this time range'.

I'll go over the stages in a little bit. Basically, things will look at the incoming metrics, look for anomalies, persist them, do roll-ups and aggregations so you can keep your latency fairly low, and do some kind of reporting across the board. The latencies vary. If you want to resolve something right in the flow, like resolving a metric name to an ID, it's 10 milliseconds. If you want to query and assemble time series at a different resolution, for instance, if you don't want to do hourly aggregates, any stage in the processing pipeline that rolls up hourly obviously wakes up every 60 minutes to compute the aggregates for the past hour. Same for dailies, and if you do batch reports every week, it takes seven days. The reason I wanted to show this is to illustrate the different time scales that exist throughout the system. Something that's disk-based at 10 milliseconds is probably not a good idea; that's probably going to be in memory. One second is probably something in memory as well.

The volume of data inbound is way bigger than the volume of data outbound. That's because you have 100,000 machines telling Datadog how they're doing, and then you have 1,000 people looking at 1,000 screens, each looking at 10 graphs, so that's only 20,000 queries at most, whereas one machine can generate 100 metrics. There's a clear asymmetry between how much data we're ingesting and how much we're serving back, and that leads to some interesting architectural tweaks.

The intake has a predictable time. You want to write data and persist it so it can be processed later. The time it takes to get the data should not vary, because otherwise you're backing up at the entrance of the pipeline, and your customers are backing up as well. They won't get the data on time. It needs to be scalable because the intake volume grows. We want to spend less money if we double the number of customers; we don't want to double our bill to ingest the data. We want something that's lower than that.

The main use case is seeing all my graphs for what's happening now. You want something that gives you fast access. You also want fast access to persistent data, i.e., past data, for example, to see how many users you had a week ago at the same time. That's going to live somewhere. If you have a lot of memory, it can be in memory, but at some point you have to persist it. You need something that gives you access to it. Internally, the constraints are that it's a bunch of queues, so they've got to be pretty fast.

There's a range in durability and access that are inversely related. You can have a low durability database, meaning the data can go fairly quickly and you may lose it, but it's very fast. That's typically stuff in memory. If you lose the box, the memory is gone. All the way to super high durability, meaning I don't care if it takes a second to get the data, but that data has got to be there no matter what. We store historical time series, and we can't say 'oops, we lost the data, sorry'.

The machine sends data to the API, for example, how much CPU it's consuming. There's a lot of machines doing that. The intake ingests the data, storing it in both memory and on disk. You want it in memory because if somebody is asking for the data in real time, you don't want to read from disk what you just wrote. If it's in memory, the web tier can directly hit the storage and say 'give me the up-to-the-second data', and it's going to be fast. Otherwise, the path is through queues. Stuff gets queued up, processed, and written to some durable store. I'll explain which store we picked. There's also a bunch of cataloging that happens. All of that, which we call the data layer, looks like something you ask 'give me this metric between time A and time B', and it returns the time series. It does the magic. If you're asking for data that spans the last six hours, some of it will be served from memory, some from the durable store, Cassandra in this case, and that's completely transparent.

At the core is a bunch of data stores and a bunch of glue code. There's not much more than that. When we started in 2010, there were about two open source time series databases around: Graphite and OpenTSDB. Graphite is hard to scale to the scale we're at. OpenTSDB is HBase-based, and we didn't want to spend too much time dealing with it. We ended up building something on top of Cassandra. Since then, there have been a few interesting open source projects. If you're interested, I can definitely get into details, but that's really what the data layer is. It's not very complicated, just a bunch of ways.

The tools we used: Redis is for all the stuff that's in memory. It's nice because it gives you very regular access to your data. Time series flow naturally in Redis. That's the low durability, super fast access. Cassandra is somewhere in the middle, much more durable. It's a distributed key-value store. When you're ingesting a ton of data, it actually works pretty well. I realized I forgot one thing: the queues we use are Kafka from LinkedIn. It's a queuing system that is much simpler than most, like MQ Series or QP. It's a thin queuing layer on top of the JVM, which is on top of the kernel. It's really fast, good for sequential writes and reads. We also use ZooKeeper for coordination across a lot of services. Elasticsearch is mostly used for events. When you say 'show me all the marketing launches in the past week and overlay them on a graph', they come from Elasticsearch. Postgres is used for storing the application catalog, who the users are, what the graph definitions are, and so on. The glue is in Python. One of my teammates gave a great talk at PyData about how we use Python. If there are Python users in the audience, yes, if you do anything with Python already, it's unfortunately not very memory friendly, which leads to some interesting optimizations. That's the talk you want to watch if you want to get into the details. I'm happy to go into details on the questions as well.

A few challenges, and I think we're getting towards the end of the talk. To give you a sense of scale, we receive about 30 billion data points every day. It's just machines sending data, 30 billion data points, running 24/7. One of the challenges is that it's hard to monitor. We are a monitoring company, but it's interestingly hard to monitor our own infrastructure. The more complex the monitoring becomes, the more you need it, which is why we have the business, I guess. Long-term operation is non-trivial. One challenge I mentioned a few times is the ever-increasing amount of data coming in. It never stops, never takes a break. That has implications both technically and operationally. Managing memory becomes the most precious resource. Because latency is so important, even though it's bounded at a few seconds to tens of seconds, it can't move much higher than that, so that doesn't give us a lot of leeway. When we're running, the number one game we play at Datadog, on top of building new products, is basically chasing latency bubbles. Anything that slows down, we measure. Everything is instrumented. We find a jump in the time it takes to render something or query one of the data stores, then we profile the code, tune it, test it, hotfix it. That's a very constant operation we play pretty much every day.

The interesting thing is that our first implementation was based on Postgres, and we knew it was naive. It took about six hours to figure out that we had to change it after we started to ingest some data. The architecture I described, even at a high level, is really iterative. It's just basically doing this all the time: replacing this, replacing that. We had queuing based on Cassandra at some point, which was a terrible idea, but we had it. We said 'okay, we gotta change this', so we put Kafka in. When we introduced Redis, it was in response to a bottleneck somewhere. The day-to-day is a lot of that. If you are our customers, that's great because we interact with you guys, and we want to make something that's useful. The last thing I want to say is that we are hiring. If you want to become a Datadog, and it took me a while to understand the joke, if you don't know, I'll explain: 'Data' is like a starter guy, and 'dogs' sleep, dog. Anyway, if you want to become one of us, [email protected]. That concludes my presentation. Thank you. I covered maybe a lot of material, maybe too much, so I'm happy to take questions.

A question: 'You or they call you a day on our C and say hey I can't see my metrics and you look and say yeah there's nothing.' It happens that either their graphs are basically in the past because the machine is sending data thinking it's in the past or in the future. We've had cases where we look for the data and can't see it, and it turns out the machine is not time synchronized, maybe two hours behind or a week ahead. We've had that problem fairly often unfortunately.

We roll up past data. We summarize it so that whenever you're asking for data to graph for the past month, we give you 200 to 600 data points back. We do that because we have pre-computed the time summaries. The question is: we save everything. We save the highest resolution data, which for us is down to the second. From a product perspective, it's like almost science fiction. You look at a month of data, you zoom in and see more detail, zoom in and see more detail until you get to the finest resolution. So yes, that's available.

Question: 'More than one data store for a single query?' Yes. That's simply based on the time boundaries of the query. For us, the past four and a half hours are in memory. If you ask for six hours starting six hours ago ending now, we'll still serve the last four and a half hours out of memory, partly because the processes down the pipeline that are supposed to persist the data may not have flushed it yet. So you go to Cassandra, the durable store, for the intermediary data. We stitch the data on the fly when we query. What's hard about that? Mostly, the main thing is a gap between stuff served from memory versus from disk. That's a fairly well understood problem for us.

Question: 'How do you think about the cost of operations while you double the number of customers?' The raw intake is not necessarily the vast bulk of the expense. The bulk of the data is going to live in the durable store. What we do is replace pieces as they get too slow. That's the idea to keep a fairly high gross margin for the business.

I also forgot to mention S3. Elasticsearch: we store events, the textual annotations of the data, in Elasticsearch. Until you materialize that there's a problem and you missed something, but that's because you're so busy with the rest. For us, Elasticsearch, Redis, Postgres are run almost on their own. S3 definitely runs on its own. The thing I wanted to say about S3 is that because our data is immutable, past data makes a perfect case for S3. We never need to randomly seek and change a little bit here and there.

Question: 'Is it running on AWS?' Yes. EBS is kind of shitty. It's very bursty. In general, AWS is a great way to start, but that's not where we'll end. EBS, I blogged about it a while back. If you read the fine print, even if you get provisioned IOPS, they guarantee performance only in X percent of the time and only if you read and write within certain boundaries. It's a hard problem, and they're doing a great job, but it's not as good as dedicated hardware.

Question: 'How much data are you serving?' The truth is I don't even know. The thing I obsess about is how much data we're receiving. That hasn't really changed. There was a dip, but we couldn't see much. How much data we are serving, I don't really care. The marketing team and I care about the product as a service, but otherwise I wouldn't know.

Question: 'How do you monitor the queue system?' The queue system we use is Kafka. It has topics, places you write to. We monitor the writing to the queue and the reading from the queue. We use workload metrics, the amount of work being done. Our processes read from the queue to do something, like roll up or anomaly detection. The metrics I watch are how many points have been consumed. We read messages from Kafka, open them up, see how many points are in each message. We don't have a metric for the number of messages entering the system. Assuming the messages entering are constant, if there's a drop, we look at how many messages are being consumed and what the machine is doing in terms of I/O. That tells us the story. If there's a lot more I/O than before, Kafka is having some issues. The processes themselves feed metrics into Datadog, and we have alerts. The other thing is the learning engine runs on a schedule. Every minute, it wakes up and looks at everything to see if everything is fine. If it's not producing any alerts, we can check if the producers are not doing anything because they crashed. We monitor all of them at once, so we can know if one crashed or restarted. That's usually the starting point of the investigation: we expect to receive X amount of data points per second, but we only got 60. How many processors are we running? Does that make sense? Okay cool.

Question: 'What about the failure modes of Kafka in production?' The queuing system has some replication built in in the newer versions. We don't use that yet. I'm a little bit cautious about what the failure modes are going to be. We were relatively early users of Cassandra, and I've kind of been there, done that, so I can wait a little bit. If you lose a data center, you have that four and a half hours of data in memory that could be lost. We have to think about that.