Cloud Native Computing Foundation KubeCon + CloudNativeCon Europe 2020 - Virtual, 4 Sep 2020

Previous Meeting Next Meeting

⏯

youtube image

►

From YouTube: Stateless Fluentd with Kafka - Steven McDonald, Usabilla

Description

Don’t miss out! Join us at our upcoming events: EnvoyCon Virtual on October 15 and KubeCon + CloudNativeCon North America 2020 Virtual from November 17-20. Learn more at https://kubecon.io. The conferences feature presentations from developers and end users of Kubernetes, Prometheus, Envoy, and all of the other CNCF-hosted projects.

Stateless Fluentd with Kafka - Steven McDonald, Usabilla

Fluentd is typically deployed as a central aggregator to which everything sends its logs for processing and routing. This superficially simple approach was found to be inadequate at Usabilla. Errors in one part of the processing chain often had knock-on effects elsewhere, leading Usabilla's SREs to search for a more failure tolerant design. Steven will introduce the new stateless fluentd deployment at Usabilla, built around Kafka as a centralised, highly available log buffer. He will also introduce the new components that have been developed to adapt fluentd to be completely stateless, as well as how logs are reliably fed into Kafka from hosts all over the world. Finally, there will be a brief overview of the challenges still remaining.

https://sched.co/Zeik

A

Hello- um and my name is steven mcdonald- I am a psych reliability engineer at uh usability by surveillancy, as you can see a bit about me here, um so I initially joined usabilla in march of 2018 in order to help set up and migrate to new cloud native infrastructure and around the start of 2019. I took on the logging component of that architecture, which is what led to the material being discussed here.

A

So our infrastructure is completely running in ec2. We have a wide variety of different instances, both legacy infrastructure and cloud native infrastructure, and we want to aggregate all of those logs together into the one place. So that's that's the goal that we're working towards here um and so the first iteration that we built was a fairly standard, fluency setup. So we had two uh redundant fluency aggregators with the fluentd forwarding protocol being used to forward logs from ec2 instances, and we used fluent bit, which is a lightweight c implementation of the fluently frauding protocol.

A

To do that from those instances and then those fluency aggregators would of course do any filtering. Do any rewriting and end up writing the logs to cloudwatch and elasticsearch, which is so we were using cloudwatch for archival and elasticsearch for easy searching of recent logs.

A

So that's all fairly standard and it worked well to a point, but we ran into a problem that highlighted quite a few problems throughout the entire blog processing pipeline.

A

So what we ran into was the first thing that went wrong was that our elasticsearch cluster, that we were using turned out to be too small to handle the logs that we were putting into it. So that's a fairly innocuous misconfiguration. It means blocks, can't be written to elasticsearch, but you would expect that the problem would remain isolated to writing the logs into elasticsearch.

A

Unfortunately, that led us into a cascading failure where we had a misconfiguration of the fluently elastic search plug-in, um which caused log duplication. So this requires a bit more explanation.

A

um It is documented in the fluently elasticsearch plug-in that, uh in case of a partial failure from the elastic search api, the failed messages will be re-emitted into the login pipeline and the reason for this is there's no way in the fluency plugin api for the plugin to tell fluentd some of these log events were successfully submitted, but others weren't. So what so? It just re-emits them from the beginning of the blog processing pipeline.

A

So that was how bad we hadn't read that bit of the documentation. So we hadn't configured it in that way, um but that caused log duplication which both meant logs were duplicated in cloudwatch.

A

But it also meant that um it also placed a whole lot more low on the fluidity aggregators because they were processing the same messages, hundreds and hundreds of times over again, uh eventually to the point that they were unable to accept new logs from the fluent bid, clients, which is now a problem, because those start, those logs um start backing up in log buffers on the individual instances and if one of those instances dies, we lose any logs that were not accepted by fluently, but that caused even more problems, because we um we ran into a bug, influence bit where, as the log buffer grows, it creates more files to hold the additional logs and fluid bit has a bug where it will hold all of those files open at the same time, and when it hits its open file limit it just crashes, it doesn't gracefully recover from that situation.

A

So that means that logs are not even being collected by fluent bit once it's crashed, but additionally, fluid bit doesn't always recover from that situation. At all. Without manual intervention, there seems to be another bug where, if it crashes, while the log buffer is in the wrong state, it will spam a whole lot of empty invalid log messages at fluently.

A

Instead of sending it the um the actual log messages that are in the buffer, which means there's now no way to recover those log messages without diving into the internal fluid bit um buffer format and manually recovering that which the easiest thing to do is just consider those logs lost and delete the buffer so that it just picks up new blocks.

A

So this turned out this one, tiny misconfiguration in elasticsearch, turned out to cause problems through the entire pipeline, even in cloud watch where the vlogs were being duplicated. So we learned a few lessons from this about the way we were doing things the initial root cause and which it was an overload of elasticsearch.

A

We can easily fix that, so that's just a case of giving elasticsearch more resources, but we didn't want a problem in one part of the architecture to be able to cause a whole lot of problems elsewhere.

A

Secondly, the the misconfiguration of the search fluency plugin- um it is documented behavior, but it highlights that it it's difficult to configure fluency plugins for reliability, because um you can't assume that, because this is an output plug-in, it's only going to be able to do the function of outputting locks. It has full access to these internal apis and in this case it was using them to behave sort of like an input plug-in by re-emitting.

A

uh Those log events, which is the an input plug-ins drop, um fluid bits crashing highlights that if fluent d does have availability problems, this can cause problems on every instance on the network. That's 17 logs with fluid bit, which is also undesirable.

A

And so the upshot of all these things, the fact that one component can have knock-on effects earlier on is we need to decouple those somehow. So if it's difficult to configure fluency for reliability and if fluently not being reliable, causes problems elsewhere, then we must have something in between. We need to have a buffer that doesn't break when fluentd goes unavailable.

A

So we came up with a new plan. First of all, we wrote a new talk. uh Sorry not yet. uh First of all, we decided to use kafka as a central logging buffer, so all of whenever any logs went from one system to another, they would always go via kafka. They would never get written directly from one um from one tool to another kafka. If you're not familiar with it is a distributed.

A

Message broker it's it's used in production by a whole lot of companies, it has demonstrated fault, tolerance, we're also using it in our application. So we have confidence in that um there are a couple of added benefits to it as well.

A

Consumers may be stopped entirely, um which means that if fluent d breaks entirely for some reason, such as what I just described, it doesn't that any any log producers don't even see that they just keep writing their logs into kafka, and when the consumer comes back online, it can reprocess word mist.

A

Another added benefit is that you can process the same logs twice by two different applications, which makes testing a new configuration very easy. You don't need to roll out a new configuration and find out if it breaks, you can have a testing fluency instance that reprocesses the same logs on real data with a new configuration and just see what happens.

A

So, given all of these advantages, it seemed like a good fit for logging buffer, at least for our use case.

A

um We also noticed that we only really need half of what fluent bit was doing for us. So what fluent bit does is it takes logs from a variety of sources, puts them into a buffer that it manages internally and then writes those logs out from that buffer to some destination.

A

In our case, it was a fluency instance, but it can also be kafka or whatever else, but we don't actually need that first part reading logs from various sources, because all of our machines um are running systemd and systemd has a component called journal d that maintains a unified login buffer to aggregate all of the logs on a particular host.

A

So we already have a blog blog buffer there, any log exporter that just reads that and writes it into another buffer is providing duplicate functionality. So we only need that second half. We only need something to read from the systemd journal and write those logs into kafka and then, as a matter of architectural design, we can say anything that logs must log to the journal and from there it will be picked up and taken where it needs to go.

A

So that's all well and good. That's the general design that we're going for what code do. We then need to write to to make this this architecture happen. So the first thing we developed is a tool called peripera, which is just a silly pun. So uh fluent d is a japanese project and peripera is japanese for fluent. So it's just reversed. um We looked into using the um existing fluid bit kafka plug-in for this, but what that does so?

A

It uses a library called liberty, kafka, which is a c implementation of the kafka protocol and the way liberty, kafka works for performance reasons is when you write a message to it. It puts it into an in-memory buffer and then once it's aggregated, some messages, it'll write a batch to kafka.

A

The fluent bit kafka output plug-in considers the right successful as soon as liberty, kafka has it in its in-memory buffer. So if fluid bit crashes, while there are messages in that buffer, you just lose them and we didn't want to. um We didn't want that behavior, but also all of the problems that we'd encountered with fluent bit were related to the way that it does buffering and we, as we already said, we don't need it to do any buffering because we have a buffer. um So we developed pear apparel, which is a very simple tool.

A

um It doesn't support more than one log input. It doesn't support more than one log output. All it does is read the systemd journal and write everything in it to kafka with no processing minimal filtering. The only thing it does is filter out its own messages.

A

If you enable debug logging to avoid right amplification- and we also use messagepack- which is the same serialization format used by fluentd's native forwarding protocol, because what my reasoning was that it would be most likely to give us the best compatibility with fluency fluently can also read json messages from kafka, but we won't message back um and it the only state that peer appear itself keeps is the journal cursor, which is just an address within within the journal.

A

The the cursor that was last acknowledged by the kafka broker so waits for the broker to acknowledge that something has actually been committed to the kafka cluster before it says. Okay, I've actually written that. So, if pero pera crashes next time it starts up. It starts reading from that last successful right, so, barring any bugs we're guaranteed that no long messages will be lost.

A

The other component, then, is making fluency work statelessly and what I mean by that is. If we want to use kafka as a centralized log buffer, we want fluent d not to keep its own state, because it should be using kafka to persistent state.

A

This is a bit complicated because fluentd has um there are buffered and unbuffered output plugins for fluency and the the choice of whether a plug-in supports buffer unbuffered mode is up to the plug-in author and not the plug-in user. If the plug-in author supports both, then the plug-in user can choose, but most output plug-ins only choose to support buffered mode for performance reasons, because if you're reading one log message at a time, you don't want to make one right call for every log message you want to batch them together.

A

That's not a problem when we're using kafka as a log buffer, because we can batch read from kafka, but this isn't the way that these plugins are designed to work.

A

So we would have to re-implement every output, plug-in fluentd as an unbuffered output plug-in in order to do this, but we didn't want to do that, so we didn't so what we did instead was we um wrote one plug-in so there is an existing output, plug-in fluid called bufferize, which is intended to give a buffer to an output plug-in that doesn't support buffering. So it basically wraps another plug-in and adds a buffer to it, and that's all it does.

A

We wrote the opposite of that plug-in, so we wrote a plug-in called unbufferize and what that does is it's.

A

It presents itself to fluenty as an unbuffered output plug-in and it presents itself to a wrapped plug-in as the fluency buffer api, so that that plug-in is tricked into thinking that it's having messages written to it from a buffer when, in fact, they're coming directly from the input plugin and it also um I'm not going to go into too much detail here. It performs chunking of each stream so that you can use chunk keys in strings in the configuration if you've used.

A

Fluentd you'll know why this is useful if you haven't it'll, take too long to explain, but- um and but this only works well, when you can batch reads, as I said, which is perfect if you're using kafka, because we can easily batch reads from conflict, this wouldn't work well with fluenty's forwarding input plug-in, because that just reads. However many messages the the client sent at a time.

A

So if we put that all together, what actually happens within fluentd is the input plug-in. um The the kafka plug-in will not commit a read. Commit means inform kafka. Yes, I've read these messages, don't don't send them to me again.

A

The input plugin won't actually do that commit until the emit call, which is the name of the function in fluency that sends a log event to be processed returns successfully.

A

Normally this will return when they've been committed to the the buffer within fluentd within the output plug-in, um but with the unbufferized plug-in, we made sure that uh it instead returns once the logs have been written out successfully to their final destination, so we're sure that they're, actually unstable storage wherever they're supposed to go before we say to kafka. Yes, I've read these messages.

A

We also made sure that if an exception is raised in any of this code, it does get propagated back to the input plugin and the read is not committed, so any errors will result in a retry of reprocessing, those locks. That means that we have at least once rather than at most once delivery. So we can get duplicate logs, but that's better than losing locks, at least for our use case.

A

So here's a diagram of how it all fits together instead of fluid bit on the ec2 instance, we have peripera doing nothing but reading from the journal and writing to kafka.

A

um The kafka topic is then an intermediary buffer. When kafka is, um as I said, highly available, it can easily survive the loss of a node, so parapara should theoretically never have a problem. Writing that now that I've said that it's probably probably going to happen out with it, but um you get what I mean, so it doesn't break it. Fluently goes down. Is the point um fluently reads um from that kafka topic and writes to another topic and here's the other thing that we can do with this? Is we can?

A

We can break up the fluency logic into multiple different fluency processes so, rather than having one fluency that does filtering and also writes to cloudwatch and elasticsearch. We have another kafka topic in between the filtering fluently and the output and the benefit there is if we have another outage with elasticsearch.

A

The only thing that that can affect is that one little fluency on the bottom right of the diagram it that cannot be propagated any further back through the chain, because all that fluenty is doing is reading from a kafka topic. The filtering plugin will have no knowledge of this cloud.

A

Log messages will not be duplicated in cloudwatch because it doesn't affect the top right fluency and it definitely doesn't cause any problems at all with parapara. So putting kafka in between these. These sort of roles makes everything a lot more reliable, at least in theory.

A

But how does it work in practice? So we found it worked a lot better than the previous approach. It actually made things faster overall, because without the buffer within fluentd, we were just processing batches from kafka as soon as they came in the main pain point, for it is now the the way that ruby kafka library works.

A

It only sends heartbeat messages to kafka in between processing log batches, which means we can either have nice big log, batches and very infrequent heartbeats. So it takes a long time to recover from failure, or we can have short heartbeats so that it recovers quickly from failure, but we have to process really tiny batches of logs and then run into stuff like cloudwatch api limits, because we're processing those logs too frequently.

A

So that's that's a bit annoying um that was also difficult to debug, because the um the uh sorry, the kafka plug-in for fluency doesn't log messages from the ruby kafka library by default. So it took me took me a while to figure out. There are actually log messages that I was missing there, but we eventually got to the bottom of that um and also fluentd still uses a lot of memory on the order of several gigabytes per fluency process.

A

So if we go back to the diagram that I had here, each of these fluenties is using several gigabytes of memory and of course, if you wanted, if you want them to be highly available, then you're running multiple fluencies. For each of these fluent d boxes in this diagram, so that's also we can afford it, but it's also not not ideal, especially when we're not processing batches that big had to make various fixes. So we've we submitted a total of nine patches to various open source projects in the process of this work.

A

I won't go through all of them, but a couple of the things that we did were we added metrics to the prometheus plug-in for fluid to report the oldest and newest time keys in the buffer. um This was before we decided to go with the new approach. We wanted more, more visibility into fluency's buffering.

A

One thing that fluentd lets you do is chunk um buffers by time key so you say basically cut logs up into 10, second or 30 second or 60 second increments, and write each one to a separate chunk in the buffer, and I just wanted to have metrics. So we could see okay, what's the oldest time, key, that we haven't yet processed and also what's the newest time key. So we know, if there's some delay upstream before d with kafka.

A

We of course have better ways of finding these metrics using the metrics on of the consumer group lag, but um those metrics are now on the open source project. So, if you're doing things the traditional way with fluentd you can make use of that, we also did some fixes for the way timestamps were handled. So we could use uh kafka's timestamps as our canonical source of um timekeeping, rather than having a separate timestamp within the message.

A

Given kafka already provides that, um and we also implemented correct consumer group rebalancing or previously so the way a consumer group in kafka works is, if you have a group of consumers that work collaboratively and when a group is formed. Each group member sends a join group message to the kafka broker, which will wait until it receives well it'll wait for a timeout or until it receives a joint group from all known consumers. In the case of a consumer group, rebalance.

A

There was a bug in ruby kafka where it was using an old version of the join group message that um that didn't that caused the broker not to wait for a timeout. So every time a new consumer tried to join, it would end up in a consumer group of one and all of the other. Consumers would be kicked out of the group, so they were fighting each other over processing logs in serial, instead of processing them in parallel as they're supposed to so.

A

We also fixed that, and so there are a bunch of improvements there that um that other projects can also make use of, and now there is some bad news. So I talked a bit about the fact that the the ruby kafka plug-in makes it difficult to balance nice big log batches with reasonable failure recovery times.

A

For that reason, after this talk was submitted, and it already had a title, we decided to drop fluently for processing log messages from kafka.

A

um We, this approach in general, is still possibly useful for anyone else who might want to try a deployment like this with fluency, um but it's not all bad because we are still using fluency in one case so before the logs even get to peripera on our kubernetes cluster, we're using fluentd to write the kubernetes logs into the journal.

A

So that's what I said before we've made an architectural decision that all logs on a node go to the journal, and that is our canonical buffer, so we use fluentd as a way of getting logs into that buffer for kubernetes.

A

um This is what the third iteration looks like, and this also illustrates a nice property of the design that we went with, which is that components can be replaced. So we could easily just take fluency out of the the slots here in this diagram where it was and put in kafka streams and kafka connect.

A

um We also switched to s3 to um for not archival for other reasons, but um the thing here is we could still use fluency if we wanted to so we could still use fluency for filtering and use kafka connect for output or we could use. We could go the other way and use calculus streams for filtering and use fluency for output, so breaking things up with kafka buffers in between makes it possible to independently replace each component of the architecture without needing to consider dependencies.

A

As long as as long as each component agrees on the the protocol that gets written to the kafka topic, um we peripera didn't need any modification to deal with the fact that we were placed fluently with kafka streams. So that's that's the real power of this um this solution and we may bring fluency back into it at some point. It's easy to add it back in because we can just slot it in wherever we want it so and the future of this we um I mentioned some of the open source contributions we've made.

A

We are looking at the possibility of open sourcing, some of the new software we've written. If we find that it's not too specific to our setup, currently I can't make any promises at all there, but we are looking at the possibility and for updates you can look at a github blog or you can feel free to reach out to me personally, if you, if you want and yeah so that's about it, are there any questions hello, um so I'm going to go through these questions in the order that they came in.

A

So the first one which should be appearing now, um how did you detect um the influence d that the logs were being sent duplicated? We actually detected that in cloudwatch, because we were noticing that cloudwatch had hundreds and hundreds of the same log over and over again, um so that was um actually tracking down where that came from did involve looking at the fluency logs, the we in the logs.

A

It was just saying that there were partial failures running to elasticsearch and I actually went and looked at the elasticsearch plug-in code and eventually found that it was documented in the readme and we'd missed it. um But the initial way we found it was just seeing that there were duplicate clocks in cloudwatch.

A

um So how did you handle this configuration that caused uh duplicates within fluentd due to partial success? Requests um in the end we handled it by completely changing and changing the architecture as I've described in the talk, um but the short-term fix is um the elasticsearch plug-in, will re-emit those logs with a particular um label and you can match on that label and just throw them away. If you, the short term fix, would just just throw them away.

A

So that they stopped being duplicated, uh but ideally you're supposed to handle those three emissions properly, um this one I'm not entirely sure what it's.

A

A difference between fluent d fluid, but with kafka integration. um I assume that's asking about the difference between the two architectures um with the fluency forwarding protocol fluent bit sends log messages directly to fluency over a tcp connection, whereas with kafka.

A

Well, it's still going on for tcp connection, but kafka isn't doing anything smart with it. um So it's much less likely to break.

A

Is is there a way to know maybe looking into the code if an output plug-in does or does not support buffering? Yes, there is um at the very least you can look at the code. There are probably other ways, um but if you look at the code they will. There are two methods that can implement.

A

There's one called processor, one called right: one of those is buffered and one of them is unbuffered and I don't remember which ones which, but yes, you can easily see from looking at the code um which ones it implements um this next. One is a good question: um how many logs can you ingest and consume um with this solution?

A

So far, we um so far it's handled all of the log volume that we've thrown at it. So I don't know what the upper limit is, but um it's higher than the most logs that we've had to try to use it. I can get you better numbers if you want to ask that one again in the slack channel after the talk, I can take a look and um give you a more concrete answer to that um this next one.

A

Why don't we use fluentd memory buffer instead of unbuffered mode, because we want to make sure that, uh if fluentd crashes, while something is in its buffer, it gets properly flushed. uh Sorry it gets it it doesn't. We don't lose those log messages, so um what we were doing previously was using disk buffering and we um we actually um had that on an ebs volume attached to the instance, so that if the instance died completely, we still had that ebs volume with the buffer. So we could recover those locks with a memory buffer.

A

You don't even have that option if, if the fluency process dies, while there are logs in the buffer, you've just lost those um lost those logs forever. So the advantage of using unbuffered mode is um that the input plug-in the kafka input plug-in won't acknowledge that it's written sorry, that's read those messages from kafka until the output plug-in has actually written them. um So it guarantees that we don't lose lots.

A

Did we consider apache pulsar instead of kafka? uh No, to be perfectly honest, I've never heard of apache pulsar before, but one of the reasons we used kafka is that we're also using it in our application, so it made sense to use technology. We were already familiar with.

A

Is peripera open source? uh No, um no slash not yet. As I said, I can't make any promises at the moment, but it's one of the components that we're looking at the possibility of open sourcing did.

A

Did you ever encounter missing logs in journal d? um I'm not sure what so we journal d is kind of our canonical. That's what we treat as a canonical source of logs to export. So that's in a way the definition of a log message existing is going into journal d. um I can't say that we've ever been encountered a situation where we've um had a.

A

We found a lot message that we thought should be in journal d, but isn't uh the only thing is if journal d has limited retention, so if pero pera falls behind it can miss log messages if the journal gets rotated, um but that's a matter of um configuration. First of all, power repair doesn't tend to fall behind, often and second of all, you can adjust the the general rotation. If you need to.

A

So how many people do we have on this team?

A

um It's a good question. The team's been we've shuffled a bit in terms of people actually working on this I'd say um two people who have worked on it in any significant capacity.

A

um Did we evaluate using file b as log shipper? No, to be honest, I haven't that's something I haven't heard of either.

A

So this next one is an interesting question: what were the red flags before deciding to improve the solution during the various iterations.

A

That's kind of a really broad question.

A

I would I'd say that the most important ones I described during the talks so um clearly the the initial problem that led to this restructuring is um a problem with the problem that was caused by um elasticsearch having problems and backing up throughout the the stack. um All of those issues were kind of red flags. None of them by themselves was very serious but all put together. They kind of convinced us that we had to um had to change something.

A

Next question is: we show the k stream? Why not use a fluency source and domestic, such sync connector? um We could do that and we we were doing that. We, the reason why we stopped doing that was. um As I said, there are problems with the ruby kafka library that um that make it difficult to consume batches of messages effectively. You can either have really big batches and very infrequent heartbeats, so that it takes a long time to recover when something goes wrong or really small, batches and um really short heartbeats.

A

So you recover quickly if things go wrong, but you end up doing a lot of api requests such which, to be honest in the case of elasticsearch, was not that big of an issue. It was more cloud watch because of its api limits, but we decided to just go with tougher streams for everything.

A

So this question is asking: did we have a particular config for the topics in kafka? um We currently we have 24-hour retention, which is more than enough. We always process those logs in a timely fashion. Replication factor, I think, is three or so it's a fairly standard configuration.

A

Someone's asking uh can, can I explain what kafka does um it's? Well, as I said, it's a distributed um messaging broker, so you it. You essentially have one thing that writes messages to it and another thing that reads: messages from it and the whole point of kafka is to be very reliable and distributed and fault-tolerant, so there's obviously more to it than just reading and writing messages. The idea is that it should basically never go away because it can survive all kinds of failure modes.

A

That's it in a nutshell, any more than that. I think it's out of scope, but there is excellent documentation on the kafka website. If you want to read.

A

A

um Next question is: did we think of using flink over fluently when it comes to stream processing? uh No, I also I haven't heard of flink. To be honest, um we did. I did look briefly at um I'm completely drawing a blank. Now.

A

The the um there is another tool that I looked at, that I can't remember the name of, but um in the in the end it was um fluently, seemed to be the thing that um had all the plugins that we wanted for filtering and and so forth, um and I mean it would have worked well if it hadn't been for the the problems that we had reading messages from kafka in a reasonable fashion, but um in the end the kafka connects and kafka streams turned out to be, I think the best fit, partly because we're already using kafka, we might as well use the um the methods kafka provides for this stuff.

A

um I mean that's all the questions for now. I think well thanks very much for coming thanks very much for all the questions and yep. Please join in this lac channel. That's on your screen now, bye.