Selinon - dynamic distributed task flows

Speaker 1 [00:03]

So, thank you. Just first question, who of you worked with Celery? Okay, I see like 50%. Nice. So, welcome to this talk and this talk will be about Selenon and about distributed computing and basically how we enhanced Celery. Before I start, let me introduce myself. So, as stated, my name is Fridolin. I live in Brno, Czech Republic. I used to be a reverse engineer, then I moved to Red Hat. Now, I'm in a team that is called AICOE, that stands for Artificial Intelligence Center of Excellence, and we do machine learning, big data processing, and we do, for example, optimization of software stacks, so we optimize TensorFlow and stuff like that. So I worked during my career on, for example, AFK-TLS, that is an internal module that allows you to transfer TLS or DTLS or use TLS and DTLS protocols directly inside Linux kernel. I worked on this project Selenon and now I'm working on project TOT that is optimizing applications and machine learning applications. Now Selenon is more like free time activity for me so I will show you how it works. Okay so first let's start with distributed computing. So in distributed computing we have something that is called message and this message basically describes what should be computed. It doesn't state how it should be computed so it carries information like send an email to every participant of this talk. This message is then queued on a message broker and the main aim of message broker is basically queue these messages. So these messages arrive and they are queued on message broker. Then we have workers that are connected to message broker and basically these workers wait for messages and if there is a worker available for computation it simply picks a message and starts to computing it. Meanwhile there are queued more and more messages so if there is a new or available worker it simply picks it and starts to processing it. If you would like to store results of what's been computing you need to have something that is called result backend and simply when a worker computes results these results are stored into result backend and so applies also for worker one when it finishes a computation. Basically this whole architecture is like very high level overview of Celery and this terminology was brought by Celery so there is result backend, message broker, workers. Celery is quite popular project and I think it's one of the biggest Python projects out there. It's quite large community around it and it's basically implementation of a distributed task queue. You can find it used on many places for example Django uses Celery for background tasks. So now let's try to do something like task flows. In many cases you don't have only one task that should be accomplished but you have rather many tasks that you would like to run and these tasks have data or time dependencies between them. That means that you would like to execute some tasks first and then run subsequent tasks. In this example we have task 1 to task 6 and there are dependencies between them. dependencies could be error data dependencies so in order to execute task five you need for example results of task one being available or you can have some time dependencies meaning you are relying on some remote REST API and you need to first query the REST API in order to proceed because there is some some background work going on. What we could do we could create one giant like master task and put every single task into this task and execute it as one unit. The main downside thought is that if you execute all of them what which one should be like first executed then if some fails how do you want to recover from these failures and if you execute it in one worker you don't have basically this parallelism and you are relying on CPU parallelism or threat parallelism that we would like to avoid when using cluster parallelism. So we can utilize something that is called Celery Primitives and these Celery Primitives basically allow you to run tasks in parallel. So you can state I would like to run three tasks in parallel or they allow you to chain tasks. So you execute one task after another. So, let's use these salary primitives. This was our flow design and when we will use these primitives, we introduce something that could be called phases. So, first we introduce phase one and we run task one to task four all in parallel and once this phase one finishes, we execute the subsequent phase, phase two and then there can be run task five and task six in parallel. Initially at Red Hat when we used Celery we used this approach and there are a couple of downsides. If you would like to introduce a new task that has new dependencies, so imagine we would like to introduce task 7 that has dependencies on task 2 and task 6, then you'll need to somehow restructure your flow. It's important to state that Celery has everything written down in code. So basically, these phases are directly Python code that is written. And if you would like to introduce Task 7, that means that you need to somehow restructure it, possibly create Phase 3 and put Task 7 into Phase 3, or you would like to create something like Phase 1.5 and execute Task 7 before Task 5, and stuff like that. There are also other issues. And another issue is timing. So in the real world, these tasks have different execution time. And for example, in this case, task 3 takes 30 minutes to accomplish in phase 1, whereas task 1, task 2, and task 4 all finish in one minute in this scenario. So task 3 basically blocks all phase 1, and we cannot proceed to phase 2. And we get results in 40 minutes, whereas we could get results of task six in two minutes in good conditions. Okay, so these are all the pitfalls that I stated. So we were not able to add new tasks into workflow management. It was not straightforward. When a new colleague came, it was basically a nightmare to introduce him into the workflow management we had. And we then came with issues like, okay, what if one task fails? How should we recover? So Larry has some features for that, but it was not that straightforward to use. And we would like to also reuse some tasks. So when we define some task flow, we would like to say, okay, I would like to run only some particular task, and everything else should be somehow find out what should be run. And over time, we find out that our data have different structures. So for some scenarios, we would like to use Postgres when we have relational data. For some scenarios where we have graph data or data that construct graphs, and we would like to explore them, we would like to use, for example, Janus graph. And it was not that straightforward into the existing Celery solution. So everything was kind of a mess. That's why we introduced Selenon. So Selenon means celery in Greek. And basically, the main idea of Selenon is to separate task logic or configuration into YAML files. So you have simply YAML files that state how your task flow or workflow should look like. And then you have Python code with very small units describing how you would like to compute these results. And then you have other primitives like graph adapters and stuff like that. So let me show you. So in Selenon, there is a primitive task, similar to the Celery one, and if you would like to create a task, you simply derive from Selenon task that implements a simple interface, and this task has one entry point that is called run method, and it accepts arguments that are node args. These arguments can be dictionary, or it can be an array, or anything that can be serialized on your message broker. In this example, we have a very simple task that basically multiplies, I guess, integers A and B and return them as a result in JSON. As you can see, there is nothing with storage or how you store data. You simply define how you compute data and you return results. Then, once you have this implementation ready, you create a YAML configuration file. And in this YAML configuration file, in the task section, you register your task. So you state its name and from where it should be imported. In the next section, that is called flow definitions, you state or you describe how your flow should look like. So in this example, we have flow that is named flow1. And then you describe edges that basically state transitions between tasks. So we start with task 1 because task 1 has no source dependency and from task 1 we continue to task 2 and task 3 and task 2 and task 3 are both executed in parallel depending on available resources in the cluster. You might be wondering, okay, what are these octagons that are shown on the picture? By the way, this picture is directly generated by selenon, so when you describe declaratively how your flow should look like, you can use selenon then to visualize this flow. And here you can see these octagons that are automatically injected by selenon in this case. So what are these octagons? These octagons are conditions. allows you to conditionally execute or conditionally fire some edge in your task flow so if we have edge that starts with task one and continues to task two and task three we can say that we would like to run task two and task three under certain conditions. These conditions can be for example results of data so you can you can examine what was the result of task one or can be some external event or can be based on environment variable as shown in the in the example here. Basically these conditions are Python functions you can register your own and you can directly use them in the YAML configuration. So in this case we proceed from task 1 to task 2 and task 3 in our flow when there is key proceed with value yes and or if there is environment variable that is set named testing. These conditions can become compound using boolean operators as desired so you can do and or or not. If we visualize our flow we can see something like this so task 1 is executed and then task 2 and task 3 both run in parallel if the given condition is met. Okay, you might be wondering, okay, we executed task one, but where were these results stored that were computed by task one? And the answer is data storage or databases. So in order to use databases in Selenon, there is an abstraction that is called data storage. It's a class, abstract class, that defines what methods you need to implement in order to use some storage so in this case we are implementing a Redis adapter and this Redis adapter in order to use this Redis adapter we need to define how to connect how to retrieve and how to store results. By the way Selenon comes with a pack of storage adapters so you can directly use for example Redis, MongoDB Amazon S3 or don't know Postgres there is also supported. Once we implement this adapter then we state it into the YAML configuration file under the storage section so we state name of the adapter from where it should be imported from which package or Python module and then you state configuration. This configuration is basically how to connect to database so in case of PostgreSQL, you define connection string. Then when you have tasks you basically assign these storages to tasks. So our task 1 uses PostgreSQL and Redis is used by task 2. Okay, nothing is perfect and from time to time a task can fail and we want to recover from these failures. For this, Selenon offers fallback tasks or fallback flows. So what does it mean? Basically in your YAML configuration section you can state something that is called failures and these failures state what task or what flow should be run in case of failures. So in this example we run fallback one if task two and and task three fail at the same time, or we run fallback two if only task two fails. You might be wondering, okay, what about time dependencies, so what if task two fails and task three is still running? So in this case, Selenon waits for task three, and if it finishes successfully, then there is run fallback two, and if it fails, then there is run fallback one. If we visualize it, we can see something like that. You can see these red arrows that basically state these fallback scenarios or error scenarios in case of task failures. It's important to state that these fallback tasks and fallback flows are tasks or flows as any others. So it's basically a task implementation. With Neo-Selenon, you can also apply conditions. So these edges have also conditions, meaning a fallback task or fallback flow is executed conditionally as in the main flow. I stated something like subflows or fallback flows. So what does it mean? Basically, Selenon does not strongly distinguish between tasks and flows, so they share namespace. And if you define a flow, it can become a node in graph. So in this example, there is run init task, and after init task finishes, then there is executed flow 1. So there are executed all the tasks and all the flows that are defined in flow 1. After flow 1 finishes then there is continue, so we continue to task 2 and you can basically nest flows as desired. You can even recursively call flows and stuff like that. Okay, so this was like very brief overview what Selenon offers and you might be wondering how does Selenon works. So the key idea is in dispatcher task or you can call it also scheduler and this dispatcher is periodically scheduled on nodes in the cluster and it basically checks which tasks have finished, which tasks should be scheduled based on state in the flow. So in this example we have this simple flow that starts with task one and task two and And once task 1 and task 2 finish at the same time, or finish, we run task 3. Imagine we decided to run this flow, so at time 0 there is scheduled dispatcher task, and this dispatcher task knows, okay, there was no task run in this flow, so it basically schedules task 1 and task 2, and it reschedules itself in 5 seconds, for example. After 5 seconds it finds out, OK, task 1 and task 2 haven't finished yet, so it reschedules again after 5 seconds. Meanwhile task 1 finishes, so after 10 seconds dispatcher knows, OK, task 1 has finished, but I cannot schedule task 3 because task 2 hasn't finished yet. So it reschedules itself after another 5 seconds. Now task 2 finishes and after 15 seconds when dispatcher task is scheduled it knows that it can schedule task 3 because both task 2 and task 1 succeeded and reschedules itself in another 5 seconds. It reschedules itself because there can be failure with task 3 and in that case it should know that, okay, there is a failure and there is no failure path or fallback defined in this flow. So what Dispatcher does, it basically propagates this failure to parent subflows, if there are any. There are also other features, so for example, you can use distributed caches. These distributed caches are used to optimize network traffic inside your cluster. You can also apply something like task and flow throttling. So when you decide I would like to execute this particular task no more than twice a minute you can do so with Selenon. This also applies for flows so you can throttle wall flows and this is especially helpful when you are relying on some remote API and this remote API is not capable of serving your requests. Then you can do something in reverse so you can do also task prioritization. There is also optimization of dispatcher scheduling, so you don't need to schedule a dispatcher each five seconds, but you can decide, okay, I would like to schedule dispatcher after ten seconds. These are more optimization techniques. Then Selenon comes with trace points, and these trace points are some events in the cluster that you might be interested in. For example, a task has started, task has failed, task has finished, a new flow has been started, also ID of that flow and stuff like that. If you log your cluster logs into, for example, elastic logs-kibana stack, then this is especially helpful when you would like to debug what went wrong with some flow and stuff like that. Then there are migrations. I will talk about them briefly and selective task flows. I will talk about them as well. Okay, so what are migrations? Imagine you have a system running in production and then you decide, okay, I don't want to run task 3 or flow 1 after init task but I would rather switch it to task 3. but I have already running system. So in this case, Selenon comes with migrations and these migrations help you to basically change behavior of production system on round time. So Selenon automatically computes what should be executed next on configuration change. There is something called tainted flows and the term means if you're running in production and imagine this flow 2 was already executed or it's or init task in flow 2 was already executed and you change behavior of this flow. In this case you can state how the flow should behave and you can choose from three options here you want to retry flow, this tainted flow that would behave differently with the old configuration in comparison to the new one or you would like to ignore it completely or you want to mark that flow as failed so you probably aggregate these failed flows. Then there is another feature that is called selective task runs and we found it especially helpful when we were executing flows periodically. So in one task we aggregated some data from GitHub and we wanted to refresh this data every, for example, five days. So in that case we just wanted to state run only task seven, for example, and we don't care about other tasks. What Selenon does for you, it computes all the dependencies in the flow so it knows, okay, in this flow, in order to run task 7, I need to run task 2 and task 6. And in order to run task 6, I need to run task 3 and task 4 as well. So it basically captures these dependencies and executes these tasks for you. The only thing you state is run task 7. so you don't mind about the internal structure of flow. Besides that, Selenon can be configured in a way that it reuses all the results of some tasks. So if task 3 and tasks 4 do not change over time, the results of these tasks do not change over time, you can say, okay, I would like to use all the results of these tasks and Selenon automatically picks them from database or storage you assigned to these tasks. So Selenon is built on top of Celery. The main idea is creating a YAML configuration file that states how your cluster should behave and then you write your Python code in which you state how you would like to accomplish or compute certain results and And if you think about it like more high-level, you can have YAML configuration of your cluster, like Kubernetes cluster or OpenShift cluster, where you state what should be deployed, how it should be deployed. Then you have YAML configuration files where you state how your workflow should be done. And then you have Python logic that is basically code that you compute. Selenon offers you these conditional task executions, and this makes the flow dynamic, so it means it can behave differently over time. You can group tasks into flows, and you can create nested flows. You can even create cyclic graphs. As far as I know, Selenon is the open-source tool that allows you to create cyclic graphs and cyclic flows. So you can nest them, you can create cycles and things might get pretty crazy. Then you have this advanced task flow, failure handling and fallbacks and you can also do system diagnostics and if you do aggregated logging as stated before, Selenon can help you. Besides that Selenon offers integration with Sentry, so if you use Sentry for monitoring, Selenon can directly send error reports into Sentry. We successfully used Selenon in a project that is called OpenShift IO. We use Selenon on other parts in Red Hat internally for workflow management and I think, or it's used by others, it's open source. The funny part is that I didn't know that it's used elsewhere unless I released a buggy version. Then people started to complain, hey, this doesn't work. But until then I didn't know that it was used elsewhere. These are examples of flows that we run inside Red Hat and these are basically pretty crazy flows that nest into each other and it's very easy for newcomers to introduce them into the system, show them YAML configuration file and they can directly navigate and contribute to your project. Then I have prepared demo. I had one issue with connection, so let me connect to my mobile phone. it works Thank you. Meanwhile, I would like to thank the European Union because I'm doing data, so it's not expensive demo. So you can find Selenon on Github, it's open-sourced, it's free. Here you can find Selenon itself and there is a demo deployment that is deployment into OpenShift cluster and there is a Verkr implementation. So Verkr is a Python code that states some tasks, then there are implemented some storages, then there is configuration files. So here are nodes, you can structure your configuration files into multiple YAML files. So in this file you can see all the tasks that are defined in the system, so we have for example hello task, Travis active repos tasks, and there are additional configuration options that can be found in the documentation. Then there is a list of flows and there are defined some storages or storage adapters so there is in-memory, Redis and stuff like that. Then there is flows section under Travis logs you can find a flow that is called Travis org logs and other flows as well. So let me show you what we will demo. So Selenon uses Travis CI for testing if everything is working correctly and here you can see that it's tested against Python 3.4, 3.5, 3.6, 3.7 and in each case or job, you can find some output like what went, if everything is correct. I deployed this demo into an OpenShift cluster, hopefully we will be able to load them, so I will show you the infrastructure. We have some API server that will allow us to communicate or run flows, then we have worker, then we have worker dispatcher. It's kind of optimization. We found out that if we schedule tasks on a separate worker or we schedule this dispatcher task on a separate worker we get much better performance and we leave computational tasks on a separate worker. Then we have a message broker that in this case is RabidMQ and then we have some storages that in this case is Minio and Redis. Minio is open source alternative for AWS. If we take a look at flows that are defined, so in these flows we will aggregate logs from these Travis runs. So first we will have one flow that is called Travis organization logs and it basically checks API of Trevis for active repositories, for repositories that are used in Trevis CI in Selenium organization and then for each of these repositories we will run another subflow that is called Trevis repo logs. You might be wondering okay what is this for each? That's basically a way how you would like how you describe how how many flows you would like to run in parallel, but you don't know number of flows that should be executed. Then for each repository you check these builds that are available and for each build you, so if it is Python 3.4, Python 3.5, for each of these builds you basically aggregate logs and store them in TravisLogsStorage. You can see that these tasks have different storages assigned, so here we have Redis and here we have TravisLogsStorage that is in fact our Minio or S3 alternative. So when we access API, we can access it using this URL and here we can see exposed endpoints. These are more like low-level endpoints that will show you all the flows that are registered inside the cluster or you can use run flow that is some kind of low level that just puts the given message with the given flow on queue but more important ones are these Travis CI endpoints so here we can say aggregate all logs in the Selenon organization on GitHub or we can say aggregate all logs for Selenon organization on GitHub but for repo Selenon. In fact this flow is run by the upper one so we can run this one. So this is how our message broker looks like. You can see that there are no messages queued. Now we run this flow so it responds the API server responds with some some response here you can find ID that is a unique ID generated for the given flow and once our RabbitMQ refreshes so we should be able to see some messages queued yeah I'm running data. So, nothing. This is the issue with online dimmables. What's going on? Ah, here you can see like explosion of messages that are queued and we can see I see what was wrong so we found out we're looking at some particular queue on RabbitMQ so if we check all the available queues you see there is 3,000 messages queued and if we check queues we can see separate queues for each task so we can see for example that the last task that is aggregating textual representation of these logs has some number of tasks already queued, so that was the queue I was checking previously. If we refresh our Minio, I'm not sure if we will see some data already computed, so let's see how the cluster behaves. So we have 6,000 messages already queued. It's important to state that the infrastructure is running on restricted or there are not enough resources allocated so from time to time Redis went down because of memory issues so here we can see some JSON files that are aggregated logs so I can download one and show it to you. So we can go to downloads and we will wait until it's downloaded. So here you can see a JSON file that is output of our last task that aggregated for Selenon organization, Selenon repo. This is some build number from Travis CI job and the actual log that is the text tool representation. Okay, now when we go to repository and we go to worker, I prepared a patch and we found out after some time after running in production that some logs have some special characters that are not that well formatted inside text files. So what we did we basically implemented a new task that is called TravisLogCleanup. We registered it into our system and we assigned TravisLog's storage adapter for us and here we created migration or Selenon did it for us where we stated using the following command that we would like to migrate the flow that is run in the production and that all the flows that are tainted, meaning they have some tasks in progress, should be retried. So when I merge this pull request and now I populate these changes to worker. I go to worker and to build and I built a new worker with new configuration that is picked automatically from GitHub. On cluster overview you can see that there's going on a build of worker so So, hopefully it finishes soon. Meanwhile, are there any questions? Yes? I'm new to... Great talk so far. I'm rather new to Celery, and Celery, I've never heard of it before. It looks very interesting, and I'm wondering how the task scheduler allows it to fail over. Because if the task scheduler dies, as far as I can see it right now, no more tasks will be executed. Is there any way that you can make it a self-healing system, so to say? So this is not an issue of Celery or Selenon, but what you would like to do is basically configure RabbitMQ to have persistence. So these tasks are nothing else or these dispatcher tasks are described by messages. So when you publish some message you would like to make sure that it's persisted on RabbitMQ or message broker and once it is persisted, if message broker goes down, then you bring it up again, it basically picks the persisted task from hard disk and if a worker goes down, basically nothing bad happens because these workers, how they are configured, when they pick a task, start processing them, basically tell message broker, hey, I picked this message and I'm working on it. But if it dies, meanwhile, nothing bad happens because this message is not acknowledged. So when worker finishes task, it tells message broker, hey, I finished this task so you can throw it away. And meanwhile, no more workers pick the same message because it was marked as one worker is already working on it and I see our build finished so I scale up over our dispatcher worker again meanwhile we can check how our queues are going on so we have still 15,000 messages queued there is quite a lot of a lot of logs I think it's like four gigabytes and they are processed so now our workers should start consuming these these messages again so if we check queues you can see there is some incoming and delivered messages and when we go to to our OpenShift console and we decide okay we would like to run more workers it's as simple as this so now we have eight workers processing tasks we can also scale up our our workers for dispatcher so hopefully that infrastructure can hold it and in the in the change there was basically the merge change I change also the number so we after a while we should see under directory to messages that are consumed with a new configuration with filter of special characters so hopefully we will get there soon any other questions if we Do we have time? We have three more minutes. Three? Three, yes. Oh, nice. Yes? Thank you very much. You have the failure links that trigger another node if a node fails, but you specify those to be when that specific node fails, do this. also possible to have like a global failover where if anything in the in the entire flow fails trigger a note that sends me a slack message that something went wrong something like that good question actually we read it something like this and the way how we did it it's basically you put everything into a flow and you state if that flow fails then this is our failover task so anything in that flow or any sub flow from that flow fails handle it and you will receive information what went wrong like what task what particular task failed with trace back so you can basically inspect like what went wrong in the fellow retask awesome thank you very much thank you how would you go about enabling or disabling a flow do you have to run a migration each time or is there another way run migration or yes run basically direction migration where you state I don't want to run that flow and that's it any other questions You briefly mentioned caching of intermediate results, how does that happen? Caching of results? You mentioned in one of the slides, you mentioned distributed caching of intermediate results. So this is helpful if you have, for example, Postgres and Postgres keeps context. So for each worker that is connected, it keeps, I think, 30 megabytes of context, so it can get pretty unscalable because if you have, for example, 100 workers and each is taking 30 megabytes just to keep context, then you would like to use something like caches or optimization of network inside cluster. how it works you can in the configuration you can state this is my cache and cache has has special methods cache and before there is access the actual database then you can specify or selling on for you accesses this cache and if it does not find it in side cache then it goes to database and it's also good that all adapters in Selenon are lazy so they connect only when you need to have results so it's completely fine that some worker does not connect to database because it doesn't need to be connected okay thank you our time is over. Thank you very much. It was a really nice talk.