Luigi: Distributed central scheduler support

[ntdef]

I recently asked on the mailing list if the central scheduler would support being run in a distributed environment. From the answer, and from looking at the code, it's pretty clear that it would not, for the following reasons:

1. Central scheduler state is pickled and saved to disk
2. Tasks are kept from running multiple times via pid locks
3. @erikbern mentioned something about transactions in his reply on the mailing list

I'm trying to put together a PR that would add support and I just want to make sure I'm not missing anything. From the looks of it, (1) and (2) would be pretty simple to remedy using Redis to handle task locks and as a pub/sub for the scheduler state. As for (3), I'm not sure what @erikbern was referring to. Are sql transactions (from the contrib modules) distributed across tasks?

If there are any other points that I'm missing please let me know.

[erikbern]

I don't think number 1 or 2 are issues. 1 can be solved by storing the data somewhere else (it's only for restarts), number 2 is on the worker side, so doesn't apply to the scheduler.

Number 3 is the hard one. What if two workers each ask a separate scheduler for new work and both of them assign task X? Now you have task X running in two places.

You would have to have some sort of lock mechanism here: https://github.com/spotify/luigi/blob/master/luigi/scheduler.py#L816

Another issue is that get_work is currently O(n) and will traverse over all the tasks to find which one has the highest priority that is possible to run. This is not terrible if all the tasks are kept in memory (although not ideal) but I think performance would drop significantly if you had to fetch the entire task list from an external service every time. Maybe this could be solved using pub/sub? That could probably lead to occasional resource constraints being broken (not a super big deal)

Aphyr completely crushed Chronos a while ago and I'm pretty sure it would take a lot of effort to make Luigi resilient to all kind of network issues. But for most use cases I think what you outlined might work. https://aphyr.com/posts/326-jepsen-chronos

Besides from that, I would _love_ moving the scheduler state to a real database

[ntdef]

I need to sit and ruminate on number 3. Suppose that you impose some kind of lock with some kind of window for a race condition. It's possible that once in a blue moon your tasks might get executed twice, but if your tasks are idempotent, which Luigi seems to encourage anyways, this wouldn't be too much of an issue. The idea of implementing a lock in this case would just be to minimize unnecessary usage of resources. But again, there might be a solution I haven't considered yet.

The complexity of get_work could be a problem if you have a significant number of tasks. But if you're at the point where that's an issue maybe you're already looking for an alternative to Luigi anyways, right? And beside, Redis tends to be pretty darn quick for reads like that and you could always migrate to its pubsub features if it proves to be an issue.

The Aphyr link is telling, and certainly a high bar to reach. However, having Luigi running on a sole instance exposes one to a single point of failure anyways.

Moving the state to a real database should be an option for users, and should be pretty straightforward to implement. My current tests just saving the pickle file to a cache have been working just fine. Another easy option would be to save the pickled state to an s3 bucket. As Docker becomes more and more popular for deployments saving stuff directly to disk becomes less of an option for users so I think adding external state support would be a really desirable feature in the long-run.

[nugend]

There is one thing here though: Chronos has a vastly more complicated model of work. It actually tries to execute jobs based on timers. Luigi is _specifically_ structured around submitting tasks and being okay with them not being able to execute and then re-running them blindly.

What if Zookeeper was used to validate that a single Task is assigned to a single worker? If the tasks are correctly written idempotently, then it doesn't seem terribly hard. You don't have to share the Task lists among schedulers in that case either. Just have some way of uniquely identifying tasks so that ZK can validate the work assignment.

Any other state that the schedulers and/or workers need to persist could be written to a database acting as a write only log.

I realize that this ends up being a fairly complicated coordination of processes and daemon managers and all, but each piece would at least be fairly simple.

Or maybe I'm not thinking of some subtle issue?

[erikbern]

I think relying on ZK would be cool but it's more complicated than that – how do you pick a task in such a way that that all its dependencies are met, all resource constraints are met, etc? You would have to store the dependency graph somewhere and maybe ZK would work but it would be a bunch of work

[nugend]

I think I don't quite understand Luigi's scheduling model then. My understanding was that workers submitted tasks to the scheduler and then the scheduler told the worker which one it should do.

In that case, you can punt on the task picking issue by only assigning tasks to workers that generated them.

Fair warning: My understanding of Luigi is only a few weeks old at this point and I _don't_ need it for a distributed task, so I'm only talking in the abstract.

[erikbern]

Yes that's correct. Maybe I'm misunderstanding your point. I also don't know much about ZK. How would you keep track of the dependency graph in a distributed system?

[nugend]

You would basically flatten the dependency graph into a namespace and use ZK to manage locks around each task. Where the name of the task corresponds to a lock.

There are probably some failure modes I'm not totally considering regarding the scheduler and worker going offfline though as I think more about it. I dunno, it might be possible to get working, but I'm 100% not an expert.

[davemt]

Just to clarify (and to check my understanding): in this case ZK would be an additional representation of the tasks, _only_ used for locking the assignment to run state (a distributed lock keyed on task id). It would not take place of the dependency graph itself which we would also need to store separately. Redis distributed locks would also be usable for this, as an example.

[nugend]

That's what I was thinking, basically.

Things that would also be needed would include heartbeat tracking for the scheduler and worker, an explicit policy that the system can only work if your tasks are idempotent (because we may lose a worker), a notion to the model that there is a time after which a worker goes away that it is lost (so the worker knows to halt any running tasks if its heartbeats are unable to reach the cluster and the schedulers can safely begin a reschedule), and finally, you'd also want to use the distributed locks to ensure that each worker is assigned to only one scheduler (this could probably be extended to let workers reassign themselves to a different scheduler if their scheduler dies).

Anyway I think that makes sense and wouldn't lose work in an unexpected way?

It punts on the idea of ensuring you have enough workers which have been assigned the right resources and puts that burden on the end user to schedule a daemon to restart workers (and restart nodes). It also punts on _ensuring_ that work is never lost and, as is mentioned in the Luigi docs, treats each task like a checkpoint.

[afarbos]

• Can that not be "fakely" done today by using the ECS Task for example?
• I really like this idea, Could that be a feature request?

[colemanja91]

Bringing this back up from the depths of the issue tracker! I'm running the central scheduler + jobs on Kubernetes/Openshift, it's great for rapid deployment and running health checks, but if, for some reason, my scheduler pod crashes while there are workers running, it can take up to 30 seconds for a new pod to fire up.
If the scheduler could run distributed, then I could set a minimum of 2 pods and never have to worry about downtime.

The two options I'm seeing discussed here are 1) move scheduler state into a database, or 2) use Zookeeper to manage state.

I can start some dev work for a PR if moving state into a DB is preferable, Zookeeper is a bit out of my current skills. Maybe state/state-manager could be specified in the config file?

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. If closed, you may revisit when your time allows and reopen! Thank you for your contributions.

[dev-head]

Hi ya'll,

I'm following up on this; wondering if there will be any movement towards supporting a distributed state file? Not being able to scale or be fault tolerant makes this less useful for production. I went through some of the PR's and saw a redis backend that was denied; so i'm guessing it really isn't happening?

thanks

[lallea]

I have been running Luigi in large scale production for years in various settings, and luigid downtime has never been an issue. As described above, if you run it on Kubernetes, configured to dump state to a persistent volume, Kubernetes will fire up a new pod in case of failure, and no one will notice the downtime. Hence, this is not a pain point in most settings. If you have a context where it actually causes some pain in practice, feel free to either submit a PR or describe your setup, and someone might have good ideas for how to alleviate the pain.