autocore - a durable workflow engine as a library

uh hello uh we are doing a overview of all the core our generic workflow execution engine and i'm looking at the code of this mr and uh hopefully we'll merge it so we will start with the architecture where we already have multiple databases we yeah so let's look at the first we start with with I guess, overview. So AutoCore lets you run workflows. Workflows are a sequence of activities that are stitched together with some business logic. Activity is a way to run a side effect in the workflow. So to call an API, for example, or to, I don't know, read the file if your thing is running on the computer that has something to read. We will not have that. yeah, let's say just call an API or start a compute job or make a GitLab API or send an event. So workflows, yeah, so workflow and activities are the basic building blocks. They appear, the state for each workflow and information about all the workflows is persisted in the database. We use PostgreSQL, of course. And for scalability reasons, we have Again, let's look at this diagram. We have the central database, which contains workflow identities, let's say. And then we have the workflow databases, which contain the data of each individual workflow. And we can have many databases here for workflows. And so let's look at the schema of the central database. So table database describes... these databases, workflow databases. So there can be one or more. And the DSN is our connection string. We will add the TLS connection configuration here as well, certificate authority certificate and maybe client certificate for mutual TLS and so on. But yeah, each database is identified by an ID. And then we have a separate table charts. This is a table that maps each shard to a particular database. So we have, let's say, 100 shards on each database and they are numbered throughout. So from 1 to 100 on the first database, from 101 to, let's say, 200 on the second database and so on. It doesn't have to be this way. They can be shuffled and whatever, but it just... easier to understand it this way if they are sequentially in each database. We can add shards later to a database and they then won't be sequential but that's absolutely fine that works without the problem. But yeah so this is the table that maps a shard to a database for each shard and then this is the identity table which we use to enforce uniqueness of an identity key in a namespace. Here we will have a namespaces table here as well namespaces at the moment, sorry, at the moment it is in this other database schema in the workflow database. But this just, it is just because I haven't moved it yet, but this will live in the, in the center database. So uh what is the namespace it's well as this says it's our way to represent an organization gitlab organization so we'll have an we have an instance gitlab instance we have one or more cells in an instance and we have one or more organizations in each cell so namespace is just an indirect way to say organization for services for things that don't know the concept of organization It's the same name to decouple from the concepts. It's just a namespace. In GitLab, that namespace is an organization, but it can be something else theoretically. So it's just, yeah, a way to partition the space, I guess. So for each workflow, when you insert it, when you create it, you give an identity key. If your request fails because there is a network problem, for example, you can retry with the same identity key. And if the request succeeded, you will get the previous identifiers. And the second workflow will not be created because you code it two times, right? So if you code the endpoint with this same identity key multiple times, they are deduplicated. These codes are deduplicated. You get the same workflow ID and the same shard. And that's what this table is for, for deduplication, to make this API important. So it records the, for this identity key in this namespace ID, it records the workflow ID, which is a UUID, and the shard ID that was assigned to this workflow. Or the other way around, the workflow is assigned to a shard. So, yeah, you call it again, you get the same workflow ID in the shard. And then once you know the shard, you can look up the database where data for the shard lives in this table, right? Because it maps shards to databases. And once you know the database, you know where to connect to. And, of course, it's open already. You have pools of connections already open to these databases. so yeah you know how to pick the correct database pool and you work with this database for this workflow basically right and that's how all of that is sharded across many databases so the central table uh sorry the central database is quite straightforward right there's not much here only three tables we will also move as i said this namespace table here but yeah that's quite straightforward it's central lookup database um we will be able to do various things here because the program can for example we can add a flag here that this shard should not get any new workflows assigned so read only flag we can do here or we can say i don't know this database is read only at the moment so no no no workflow should be executed on this database at the moment like if it's offline for example and we don't allocate stuff to it and we don't try to read it and so for example you can do maintenance this way you set the flag you wait for all the workflows to to be suspended for this database for all the shards on this database and then that may work for the database is not getting any read or write traffic and then you can do whatever you want and then once you've done like an upgrade. Once you've done it, you flip the flag and everything resumes normally. So because we have that central database, it can work like this orchestration coordination mechanism. None of that is implemented at the moment, but it can be done because we have this central place for it. Okay, let's look at the workflow databases. So I talked about this already. Then this shard table It's to do fencing. So each shard is claimed by at most one instance of relay. And each relay can claim zero or more shards. So typically, well, the way the algorithm works, relays coordinate via redis. They register, they say... uh like i'm here i'm here i'm here and then you can see how many relays there are in this cluster and then they know how many shards there are across all databases they do a look up in the center database you know there are let's say a thousand shards across two databases or 500 uh in each database so they say okay there are i don't know let's say four instances so we will divide four instances sorry, we will divide the thousand shards by four so that each instance gets the same number of shards. And they will try to claim that number. They will claim, they won't try to claim in a single go. They would start with one, two, and four, and eight, or something like this. And I think that goes faster after it. But we are still figuring out, tuning how this works. But basically, instances coordinate like this, they try to claim shards evenly. If there is a new instance, the others would notice and try to release shards slowly so that this instance will get some amount of work automatically allocated to it. This release mechanism exists already and works, but it's not... How to say? It's not... It doesn't wait for workflow execution to finish, to suspend before releasing. We don't do this coordination just yet. It's just we know we need to implement it properly, but it's not there yet. But yeah, anyway, this is the idea that they try to balance the workload for each replica by coordinating through Redis. And so each instance tries to claim shards and shard is unknown when the owner is zero and it is on when owner is not zero and lease hasn't expired yet so lease expires that is in the future or if it's not zero but in the past and the owner is not zero that means the owner is dead so we can reclaim the the shard and because we have a transactional database we can do locks and proper concurrency and so on and atomically claim a record in this table. So each shard gets a single owner and then that owner handles all workflows in that shard. As I said, each CAS, each relay can own zero or more shards and they try to evenly own roughly the same number of shards. So that's the shard table for locking and coordination, like who owns what. And then this is the workflow table, which we probably, I haven't removed anything here since I added this stuff here. So we will probably remove some like identity key. We don't need it here anymore. Creator that we don't need it anymore. Maybe we don't need the whole table. Maybe we do. Anyway, we will figure out what we need and what we can delete later. I tried to reduce the size of the merge request because it's big already. So I haven't touched this. I also didn't want to have any conflicts with other things that are happening and changing. We are changing the schema for other things as well. So this is the execution. This table tracks the state of execution of which workflow. It's the workflow ID and other things like created, updated sequence for concurrency control state of the workflow. The shard it belongs to. Some of that is obviously redundant because we know which shard it belongs to because it's here in this. No, it's not in this table. But yeah, so it's in this table. Yes. where it lives. But again, this now is redundant because it's in the central database. But again, we don't want to, the less load we put on the central database, the better, because it's the central point of, it's the scalability bottleneck. But at the moment, it's just one write and that's it, to create a workflow. And then it's not used for anything else at the moment. Well, it's used to find databases, but that's not in the hot path. that's like asynchronously you discover databases you connect to them you run migrations and then you start using those databases like to to allocate new work and to pick up work from those databases um so this is uh yeah tracks execution which workflow um then history events is a table that uh is the event sourcing table well that's what it says So we record the results of those activities that we execute. So if a workflow calls a function to make an HTTP call, let's say, and we haven't run that before, and we know that we haven't run it because we look at the table for this workflow. in the shard, we look at the sequence ID, which is a monotonically increasing number for a particular, that's what it says here, for a particular workflow. So we saw, okay, this is the first activity in this workflow, hence the first history event. If it exists, it has the number one sequence ID. And okay, is there a number one history event for this workflow? No. That means we haven't called this function before. Or we haven't persist, to be more precise, we haven't persisted the result of that call because maybe we crashed or maybe we never coded. So we need to code. We code, it returns a result. And that's where we persist that result, persist the history event with this payload of that. In this case, for this activity, that would be the result of the call. so maybe the body the status of the call htp call and whatever else you want to persist it's just a blob of data we represent all the payloads as just blobs and they are actually just proto buff messages serialized so we read and then marshal oh so this is our table with all the history of the activities for the workflows and the next one is the These are two tables. This used to be one table. These are two tables. These are tasks for tasks to run for workflows and tasks to run for activities. So there is a periodic process that scans these tables and picks up any tasks that need to be executed. Like if this workflow is ready to be executed, there would be a task here. So when you start the workflow. You do that thing in the center of the database, you insert the record, but then you also start the workflow here. And one of the things that you do is you insert a task to run it. And that's it, you commit the transaction. And well, there are multiple things, but you don't actually do anything. You don't, I mean, you don't actually run any code, any installer code, you don't. You save that, commit the transaction, and that's it. Then there is an asynchronous task. which scans this table and sees, aha, there is a task to run this workflow. It picks it up and it, well, runs that workflow, which has the code for it. And the code for workflows in AutoCore itself is just go code. This is like the polling code, right? Polls the database for a particular... Yeah, so in AutoCore, yes, there is code that poses it, but that's not what I want to show. I want to show the... I want to show the... so Autoflow uses AutoCore. AutoCore doesn't know anything about Starlark. Absolutely nothing. AutoCore, that workflow engine, exposes the primitives to do durably run Go code, Go functions. And where are they? how to find them. Well, I can just look at the register. Register is where you register the name of the workflow to the implementation function. So if you look at register workflow, we can see that, yeah, I'm trying to use that and I'm just learning to use it basically. And I'm a bit lost obviously. Yeah, okay. So this is, I opened it, didn't I? I was looking at the wrong file or something. So this is where the Autoflow engine registers workflows and activities in the AutoCore registry. So it's a registry to look up a map name for the actual implementation. So we have the most interesting thing is the Run Starlark workflow. So just a string name, Run Starlark. This is the function that implements it. As you can see, obviously, this is Go code. It's not Starlark. But this is the workflow that AutoCore invokes. And that workflow actually executes Starlark here because Starlark script itself is an input here. It's viewed in the payload. Yeah, so that's where we execute the Starlark script. So Starlark execution. is part of autoflow and all that the durable execution is a separate and decoupled thing in autocore. Okay, and we also have the activities. Let's look at HTTP do. This was an activity to... Wait. How do I not do this? New editor. Lots of fun. So this is the activity. It gets the context. of an activity and in fact does something and returns the result. So both input and output are protobuf messages and it makes the HTTP call and returns the output and that's yeah that's the activity from the autocorce perspective just go code. Here so workflow code is in Starlark but activities are Well, they are Go code, but it can... I don't know. For this architecture, we'll have an activity that runs a GitLab function on the runner. So like a job, basically, to simplify everything. It would run a compute task on the runner. So we have a Starlark script that imports a module that implements this functionality to run... something. And then it would call the function to actually do it. And then that would be an activity to start something. Okay. I'll just ask a question, Michal. Because it seems like Starlark is like the calling function here. But before in engine.go, it looked like go code was calling Starlark. Yeah. Starlark interpreter is written in Go. So to start the Starlark interpreter, you need to call it. It's a library. You call it. It does things. Okay. That makes sense. Because I thought it was Python, but I'm guessing that it's like the execution environment is Python, but the library itself is Go. Is that? No. Starlark is... This is the... Oh, Starlog Go. Okay. So there's like a language specific. Yeah. So it's an implementation of the interpreter. An interpreter for Starlog written in Go. It's a library. You can embed it and use it. That's what we do. Got it. So we give it a script, Starlog script, and run it. And we also, apart from the script, we also give it some, you can give it variables, functions, and well other things, but basically any symbols and you define what they are. Some of them can be callable, so function-like behavior, and some of them can be called, so they are more like variables or constants. And yeah, basically we represent all of our extensibility bits, all of the modules that extend Autoflow as variables and functions. There's nothing else you can represent the mass. Yeah, so, thanks. Let's, I guess, go back to the database tables. So this is the task that, yeah, tells the system that a workflow task exists and that needs to be executed. And there's activity tasks, same, but... Some activity needs to be executed. Then there is schedule task. This is for timers. So when you say sleep for a week, the schedule task is created, put here, and the fire at is one week in the future. And then your workflow is suspended for a week. And then it is resumed when this is in the past. time has passed and the scanner that scans this table sees that aha there is a task that can run now and yeah it continues that workflow. You can have multiple timers sitting and waiting you can have you can wait concurrently on multiple events like for example you can say you can start an activity it will give you a future And you can say, I wait for the future to resolve or for this timer to expire. Whichever happens first wakes up the workflow. This is the concept of coalescing timers? No, because you may have to, well, I guess at the moment, you can have multiple timers for different purposes. And they can be running concurrently, right? You wait for on that first and then maybe it fires or maybe something else wakes you up and then there is another timer that kind of advances independently. Is that what you're asking about? I think if you have multiple timers, you might wait on the first one, right? Like, yeah. My and I was just looking at the concept because you mentioned like coalescing time is quite a bit in the epic and my understanding of that was that Essentially if you have multiple timers you might not want to Wake something up 50 times You can just kind of make it item potent collect all of those. Yeah, especially if it's happening in like a brief interval. Yeah Yeah, yeah, so there's that just separate These are timers and these are timers. So that thing is we have timers that drive the scanning loop. So when you, for example, say I want to, well, when we scan every, I don't know, 10 seconds or whatever, that's configurable. But if you know that I just inserted a workflow into the workflow task into this table, for example, it's ready to be picked up. Why wait 10 seconds? So we have a system where we notify that relay that's responsible for this workflow, for this shard, that yeah, there is work to do already. You don't need to wait 10 seconds or whatever is remaining to wait. So we either do it in memory if it's the same relay instance or through Redis. if it's a some other instance. And if we get lots of these notifications, we don't want to scan on each notification. That's where we kind of batch them. We enforce that at least, I don't remember how much time, but let's say one second, at least one second has passed since our last scan. And whatever is there, okay, okay, it accumulates for that one second, even if we get to 100 notifications. Okay, one second later, we will scan again, pick up maybe not all of it, but there is also, you don't necessarily pick up everything if you don't have capacity to run a thousand things right now, and maybe you have capacity to run 50, you just pick up 50 tasks. But yeah, that's where coalescing happens. But that's, it's a separate thing. And here, you know in your workflow script you may have multiple timers but they are independent they just exist and how you use them in your workflow script is up to you you can wait on this wait on that wait on both or whatever you want you can have for example separate timeouts you can have a timeout using a timer on the whole section and then you can have a timeout on an individual activities, let's say in a loop, let's say. And then you have two timers, one is tracking the whole execution time and the other one is tracking just this one particular request, let's say. Okay, so this is why what so you implement all the timeouts, all the stuff basically that, sorry, scheduled tasks using this basically using this table. And then this is for workflow signaling, you can send the signal to a workflow that's in the idea is here is that either workflow sends a signal to a different workflow or it could send it to itself I guess but why or you send or maybe it's a third party system not third party but some other system not workflow to workflow but it can be something else to workflow it's a nice like an email comes to you basically from somewhere that's the thing. It does have a payload so you can attach some information there. And that's it. I think there are various indices here. Yeah, that's it for the database part. Then we have this, all these files, they basically implement various logic of working with these tables, signaling the various types and Yeah, workflows identified by the shard ID and the workflow ID. And so on and so on. We also have the worker. No, no, no. This workflow, worker pool. So this is a pool of workers that execute workflow code. So you don't want to run unlimited number because you will crash run out of memory so this is a way to limit set the limit on that and then there is a similar thing here for activities activity worker pool um what else is interesting here we have a client here that allows you to execute a workflow so that's the insert in the main database and then tracing and stuff and then this is the insert into a concrete database that's the whole transaction workflow execution is created history event is created and workflow task is created and that's where that's basically basically it returned the result so this is metrics and uh you see we don't run anything here on workflow creation because we just delegate The scanner would pick it up, basically. But I think we signal to the scanner here somewhere. Yeah, we signal that there is a new task. And this either signals in memory to our own scanner or it signals to via register a different instance. And that's where we do the timer coalescing in one of these. What do I go to? So if there's multiple, like... signals to another CAS or another scanner, then we would coalesce those requests instead of hitting it 100 times. Yeah, so notify local is injected. Yep, notify local. Let's trace this one. Yep. And this one goes to implementation. This is the implementation. It caused reset with zero on a big building blocks here. Well, that's probably it. I can show you the white is it opening everything in two places. This is the so this is the sharp manager. It's still the old name of the file that type is db manager. Again, I will rename it after the Marvel lands. Otherwise the difference will be huge. If you rename it is just okay, this file was gone. This is new file. what changed i don't know so the db manager has the main loop here uh we have a ticker so we tick every every let's say 10 seconds and then on the tick we signal readiness to other relay instances via announcing in redis basically and then we discover and apply and that's where we scan the center database for databases and charts. Then we have this thing called workflow db workers. It's a unique configuration to a worker mapping basically. And it tracks if configuration changes of a particular worker. So here our workers are things that are responsible for running migrations on a database. So our key is the database ID. And that worker is the workflow DB worker, this one. It's an entity that type of whatever that is responsible for everything in this workflow DB. So we have one pair each workflow DB. And this workflow DB workers, it's basically like a map worker manager. Let's call it, we have, we use it for us. multiple things so this is the key db id it's an integer and this is the db config that's the configuration for this worker for this db and it's just the dsn and the list of shards that uh exist or should exist on this database and this is our db worker so this thing basically spawns workers for each db id that's what it does if configuration changes it stops the existing worker and starts a new one instead. Normally it won't change here, but let's say you want to rotate the client certificate and let's say you had client certificates in the central database to talk to the workflow databases. You would update the certificate and you would want all the work using the old certificate to stop and new worker would start and use the new certificate. So that's just stop start this thing handles that and spawning new ones. So let's look at the workflow DB manager. That's the pair workflow DB thing that lots of metrics here and when it runs it basically only does heartbeat. That's the manager. But sorry, I opened the wrong type. We need to look at the worker. the the confusing the name so one is the manager one is the worker worker starts the manager so worker because this is the worker manager so we have the workflow db man worker and that's what runs migrations here so when it starts when you first connect to the database you run migrations so that's what it does and then it opens after migration so it opens a pool connection pool And then it opens another. We have two pools. One is to decouple. We have two pools. One to do the system pool. One to do like a meta work. Like hard bits on the shards, for example. This is important to do in a timely manner. So we have a separate pool of connections. One, two connections. There's not a lot of work that happens through this pool. But it's important to do it quickly. And then there is a separate pool that's for workload. And these are used a lot for actual workflow execution. So if there is a lot of load and you exhaust this pool, this work, this second pool, you can still do hard bits because you have a separate set of connections. But because if we're using the same one and we exhausted the pool, because there's just a lot of work, we would start missing the heartbeat and that means other instances would see that our claim on the shard expired and would try to claim the shard. That would disrupt everything. So we just have a separate pool so that we always have a couple of connections available to do the important in the sense that it's time critical work. So we have two pools, that's why. And then This starts the worker manager. Yeah, there's two worker managers. Because this comes from our library, worker manager. But then this is a different worker manager. This is this worker manager for workflows. So workflow workers. Might need to look at the naming there. Yeah. We had this for quite some time, this worker manager, just generic worker per ID. kind of thing. And this is workers that are... sorry this is the manager that is a different manager. Anyway, so this one does heartbeat and it doesn't actively do anything else that's it's at the moment right. It only does heartbeat in the tick and it does a release also it releases all shards but it doesn't claim any shards by itself. But it does expose a method here to claim shards. And this method is called from the central, from the, sorry, from the DB manager. So from the central thing that discovers the databases and starts these workers. It then has a single loop that drives the shard claiming, shard releasing, like this release one shard. And so this stuff is here, not there. in the in the center thing because we do heartbeat here and heartbeat looks at this map of what this relay instance owns on this database so i'm not sure maybe i should move this to over there but then i would have to i don't know tell tell this manager to do heartbeats on this one somehow for now this is okay okay i think this is basically the main We have maybe one duck to deep for this first session. Nice. Any questions? Probably a lot, but yeah, any general high level? What is not clear? I think overall it makes sense. Maybe you need to go through and look at the architecture a little bit. and come back to you with some questions. I think overall it makes sense. There's a lot of moving paces. Yeah, yeah. It was a bit simpler with one database. So I tried to keep the structure of code, of objects in code closer to this architecture that there is the central database. Then you have multiple workflow databases and then so pair workflow database there is the worker then it does the things migration and stuff it spawns the manager that manager actually manages work by sending work to a worker booth so if it if it runs the workflow it's it sends the work to the workflow worker pool and if it starts activities it sends it there so it's it's kind of a sequence for a change of things i think i do need to understand that a little bit more so we have like the the db worker um why does the worker spawn the worker manager and i suppose not it would seem more natural for the worker manager to be at the highest level how do I I'm starting new one new incognito or something interesting there must be a button like I think I saved subscribing uses and new there is new open it doesn't exist What if you open it in a new, like tab, sorry, like a new incognito window, which is a little bit you start a new one. Okay, maybe there's like arrays, everything or something, but okay, let's just open, as you say, in the new window. So this is our, okay, let's, this will be our database. No, let's put it here. I want to draw a... What is it? What's that? This is our central database. And this is our... Okay. Then we have... This will be DB Manager. This is the thing that books at the central database. And it spawns the DB Worker. One. These things work with their corresponding databases. So they run migrations and on open pools, connection pools. Obviously it's the same thing, just a separate instance of the same type. Okay, once they have done, have run the migration, then they spawn I'm just thinking how to draw it so that it doesn't I will only I will only draw on this one this one let's it obviously is the same thing just let me move it here maybe I'll do it this way do it this way here this one we will use this one as a full representation of everything okay so it runs the migrations and opens the post and then it starts the db manager this is the workflow db manager yeah it's that separate types um then each workflow db manager starts order something like this and this thing uses two worker pools for actually running work and oh oh this is that well the interesting thing is that these two maybe i should these are shared by all the worker managers because yeah if you your instance is limited or instance of relays limited sobs of ram and cpu so we pre-create this and they are shared across all the databases. That way you... it's a way to limit concurrency. It's a way to constrain the resource consumption. Otherwise, yeah, you will run too many coroutines maybe. So this is the structure that's created for each database. But at the leaves, these pools are shared actually. So they are created... either by the DB manager itself or maybe they are even given to the DB manager. I think they're probably given to the DB manager. I refactored it a couple of times, so I don't actually remember. Let's have a look. So this is the DB manager. And it really gets a constructor to construct one of, sorry, one of these workflow managers. And that constructor brings in everything it needs to for the work manager. So this pose abstract, like, DB manager doesn't know about this part. It only gets this constructor function to create the worker manager. So addshard, removeshard and it run. This is just a run there. And then this is code from... Wait, where is it code from? It's code from... Ah, here. So this is the auto core engine and that's where we get the... new workflow manager and new workflow manager, this is the real constructor of it. And if we don't run workers on this instance, let's say we only want to use the client. We only want to inject work into the database, but not pick it up. That's where we use a kind of knob worker manager. So we do everything. We run migrations. We start all these things, but we don't actually pick up any work and we don't actually have any workers. So this doesn't exist. It becomes a noob in this case. Currently we use it in a lot of tests only, but I think it makes sense to have this mode. And this is the real one. And this is where it uses the actual pools. Maybe we still pass them here because we need them for... Wait, we need them here. We need to stop them here. That's why we passed them. Okay. This is the structure of the objects. Anyway, I think that's probably enough for now, and we can continue some other time. Yeah, that's great. One final question just quickly. These workflow worker pools get allocated jobs from the worker manager, right? And the worker manager gets that from the... central DB. Am I mistaken? So worker manager is responsible for picking up work from this database. So this thing connects to it, runs migrations, opens connection pools. Then it starts this worker manager and gives it the connection pool. It starts the worker, sorry, workflow DB manager. And this thing starts this one. Here. Again, it is a new editor. And I'm like, how do I get where I want to go? Anyway, so this thing scans the database. And picks up work. And all the work happens in this worker manager. Oh, okay. It happened in... Wouldn't it happen in one of the instances in the worker pool? Sorry, I mean work in terms of scanning for jobs, tasks, the right terminology tasks and everything. But once it has something to run, either a workflow or an activity, yes, it sends that to a go routine in one of these pools. Yep. Got it. Okay. That's great overview. Good to have it all sketched out like that. Thanks, Mikhail. Okay. Thank you. Let's stop.

https://gitlab.com/gitlab-org/cluster-integration/gitlab-agent