Cluster state refactor Part 2

[thinkharderdev]

Which issue does this PR close?

Closes #554

Rationale for this change

See original ticket for full description but the tl;dr; is:

1. The existing interface for managing shared state among schedulers is too low-level and both pushes too much complexity into other layers and prevents us from taking advanatage of data store specific freatures (atomics, transactions, etc) to avoid locking in the application layer.
2. By forcing everything through the KV interface we force all the serialization overhead even if we are only using in-memory state.

What changes are included in this PR?

There are a lot of changes because the StateBackendClient was baked in everywhere but at a high level:

1. Create JobState trait to complement ClusterState created in last PR. This in the interface for storing/managing global state w/r/t jobs and sessions.
2. Refactor in-memory implementation to not serialize things to protobuf
3. Provide KeyValueState implementation based on existing KV interface so we can continue to use etcd and sled for state without any changes.
4. Take advantage of curated job architecture to minimize distributed locking. Since jobs are owned by a single scheduler we don't need to lock beyond the node level.
5. Cleanup configs.
6. Move everything related to state the ballista_scheduler::cluster module

I removed the separate configs for cluster-backend and config-backend because it seems confusing. So currently if you want to use different implementations for the ClusterState and JobState you would need to write your own entrypoint. We might introduce standard "profiles" (standalone/ha/etc) for setting up the two state backends on different implementations but for now it seems too noisy config wise.

NOTE: I was not able to run the integration tests using the standard configuration due to weird timeouts. When using push scheduling everything worked fine so not sure what's going on. I also checked on main and see the same issue so maybe something related to the M1 macbook?

Are there any user-facing changes?

The storage layout for etcd and sled are simplified so existing state would not be valid after upgrade.

[thinkharderdev]

@mingmwang @yahoNanJing @Dandandan

[thinkharderdev]

Not sure what else to do here. Clippy complains when I inline variables in the format string in the assert and also complains when they are not inlined so just disabling this lint for the module.

[thinkharderdev]

Not sure why this is failing (https://github.com/apache/arrow-ballista/actions/runs/4123347318/jobs/7121701363) but it doesn't seem to be an issue with the actual formatting

[andygrove]

It looks like this PR drops bind_host? That can be different from external_host, so I'm not sure we can do that.

Or maybe that is handled somewhere else in this PR and I haven't seen it yet?

[andygrove]

It looks like bind_host is not currently used in the scheduler, but it is in the executor, so this seems like an oversight.

cc @avantgardnerio who may have opinions here

[thinkharderdev]

Sorry, this is slightly confusing :). It is still used to build the bind address for the server

    let addr = format!("{}:{}", opt.bind_host, opt.bind_port);
    let addr = addr.parse()?;

The reason external_host was added to SchedulerConfig is so that we can pass it in to start_server and build the BallistaCluster there only from the SchedulerConfig. In order to do that I had to pass a few more things from the Config struct generated by configure_me which isn't available in the lib code.

So really the only point is to further minimize what we do in main.rs to just setting up logging and validating CLI arguments

[avantgardnerio]

FWIW, I believe external_host is required for FlightSQL.

[avantgardnerio]

I stand corrected - it appears to have morphed into its own advertise_flight_sql_endpoint.

[andygrove]

LGTM. I think we can have a follow-on PR for the scheduler bind_host issue.

[thinkharderdev]

I'll plan on merging this weekend unless someone else would like more time to review.

[Ted-Jiang]

@thinkharderdev Sorry to interrupte, plz take look, could you plz tell me why remove the async , same in other functions 🤔

[thinkharderdev]

This function is just removing the ExecutionGraph from the local active cache so it doesn't need to be async. We only need an async function when we modify the global state.

[Ted-Jiang]

I think active_job_cache here is a global states here in TaskManager -> SchedulerState, Is there something wrong ? plz tell me 🤔

[thinkharderdev]

Sorry, by "global state" I mean state shared across multiple schedulers. The active_job_cache is local to a single scheduler and and a purely in-memory data structure