A Modular Cluster Controller for Burst Autoscaling and Graceful Reconfiguration

We recently added a bundle of new cluster features: burst autoscaling and background graceful reconfiguration. I will explain these below by way of example, but to me the more exciting thing is that we managed to add these and improve how our control plane is architected. Which latter I will explain after we first explore the examples.

Say we have an existing big collection demo_events and we want to create a new index on it. This requires creating a new dataflow which can take quite a while to hydrate.

To reduce the time we have to wait, we can now specify an auto scaling strategy on the cluster that will spin up a large burst replica while the regular steady-state replica is hydrating. The idea being that the burst replica is big and expensive to run, but can hydrate quickly. And when the (smaller and cheaper) steady-state replica is finished hydrating the burst replica is retired.

As SQL:

CREATE CLUSTER burst_demo (SIZE = '100cc');

ALTER CLUSTER burst_demo SET (

    AUTO SCALING STRATEGY = (

        ON HYDRATION (

            HYDRATION SIZE = '800cc',

            LINGER DURATION = '60s'

        )

    )

);

CREATE INDEX demo_events_idx

    IN CLUSTER burst_demo ON demo_events (id);

We can check in on the cluster to observe the burst while it happens:

SELECT name, replicas, activity

FROM (SHOW CLUSTERS)

WHERE name = 'burst_demo';

\watch 1

Example output, during the hydration burst:

name    |         replicas         |         activity

------------+--------------------------+---------------------------

 burst_demo | r1 (100cc), r2 (800cc)    | hydration burst at 800cc

(1 row)

And, once the steady state replica is done hydrating and the burst is retired, we get:

name    |  replicas  | activity

------------+------------+----------

 burst_demo | r1 (100cc) |

(1 row)

You can read up on the details of burst autoscaling strategies in our docs: Autoscaling for hydration.

For a while, we've had a feature that allows reconfiguring a cluster without causing downtime: graceful cluster reconfiguration. When issuing an ALTER CLUSTER [...] statement that, say, changes the size, we would spin up a replica of the new size, keep the old replica, and only retire the old-size replica when the new-size replica is hydrated. This worked, but had the limitation that you had to keep the SQL client session open while reconfiguration was happening.

As part of the burst autoscaling work, we took the opportunity to improve our code and factor out all cluster controlling machinery into a shared cluster controller. And as a "side effect", we also made it so that graceful reconfiguration now works in the background, without requiring an active SQL client connection.

As a user you can now use it like the following. Keep querying in one psql session:

SET cluster = burst_demo;

SELECT payload FROM demo_events WHERE id = 42;

\watch 1

In another session, resize and observe:

ALTER CLUSTER burst_demo SET (SIZE = '200cc');

SELECT name, replicas, activity

FROM (SHOW CLUSTERS)

WHERE name = 'burst_demo';

\watch 1

The ALTER statement will return immediately and the reconfiguration will continue in the background.

During reconfiguration, the above watch query will yield something like:

name    |         replicas         |          activity

------------+--------------------------+-----------------------------

 burst_demo | r1 (100cc), r2 (200cc)    | reconfiguring size to 200cc

(1 row)

You can read up on cluster resizing in our docs as well: Cluster resizing

Simplifying quite a bit, the materialize control plane is an environmentd process that drives around the clusterd processes, which constitute the data plane. Within environmentd, there is one central control loop of what we call the Coordinator that receives internal and external commands, processes them, and dispatches new commands to itself or the clusters.

As pseudo Rust code, you can imagine it mostly like this:

loop {

    select! {

        cmd = internal_cmd_rx.next() {

            process_internal_cmd(cmd).await

        }

        cmd = user_cmd_rx.next() {

            process_cmd(cmd).await

        }

        [..]

    }

}

We inherited this design from the very first days of materialize, where using a central sequentializing loop was the easiest approach for providing the correctness experience (read consistency) of Materialize. It comes with the downside that one central loop that has to process everything limits the amount of work you can do, as there are only about 1000ms in a second. Over time, we have grown many optimizations that offload work from this central loop to background tasks and execution stages to squeeze out more performance from this design. But it has its limits, and one of the long-running goals of the team working on the control plane is to make it "infinitely" scalable, both vertically and horizontally.

Before our work on the cluster controller, the existing (foreground) graceful reconfiguration feature was implemented as messages and execution stages that are sequenced by this central control loop. And the obvious, but in my opinion misguided, approach for adding burst autoscaling would have been to add the functionality right there and in similar fashion to the existing graceful reconfiguration.

Instead, what we did is take a step back, think holistically about what the (long-term) architecture should be, and moved toward that.

The long-term vision for the control plane requires that environmentd (or its replacement processes) be both vertically and horizontally scalable. Vertical scalability here means that we can do more useful work when giving our process more CPU cores and so handle larger query workloads, larger numbers of objects and clusters, and so on. Horizontal scalability here means that we can have multiple processes coordinating and sharing the control plane workload for a given environment.

In our work here, we focused on vertical scalability, while making sure that the architecture will also work in a horizontal-scaling world.

The overall approach to vertical scalability is simple: extract as much work out of the central processing loop into independent "tasks" (in this case, async Rust tasks). Put in place a minimal boundary between these tasks and the Coordinator loop.

Putting that into practice with the cluster controller meant that we had to do roughly these things:

1. Add the SQL surface for creating auto scaling strategies and kicking off cluster reconfigurations.
2. Add commands and responses that the controller sends to the Coordinator.
3. Add the new cluster controller that enacts things from #1.

Previously, the flow of processing SQL commands would be handled by the Coordinator loop, across multiple execution stages that sent things to background tasks and then required processing time on the main loop again.

Now, processing the SQL statement only requires writing down the intent in our durable catalog, and then the cluster controller reads those in the background, figures out what to do, and sends messages to the Coordinator when needed.

The cluster controller itself is a "reconciler". It runs its own loop, and on every tick, it checks the current state: clusters, replicas, configured autoscaling strategies, in-progress graceful reconfigurations and hydration signals. From that, it determines state updates that should be applied, like creating replicas, retiring old-size replicas, or retiring burst replicas. The controller then tries to apply those updates by sending a message to the coordinator that contains a "witness" of the catalog state as of when it made its decisions, and the Coordinator atomically either applies those or rejects if the state of affected clusters or replicas has changed in between.

This design allows the cluster controller to be modular and separate from the Coordinator. And it would also allow running multiple instances of it, as long as everyone follows the same rules for deriving ought from is and the Coordinator prevents stale decisions from being applied. The system is self healing in the sense that it will eventually converge to the state that it should have.

Additionally, the separation via a minimal interface allows testing cluster controller behavior in isolation, and will enable us to swap out the communication channel between Coordinator and cluster controller when we have to, for our work on horizontal scalability.

Here's a figure that summarizes the work, for your enjoyment:

Cluster controller and Coordinator request/reply architecture

Since we finished this work, we've used a similar approach for collecting hydration history. A background task periodically reads hydration information from replicas and writes them down through a client interface to the Coordinator. Both collection and regular cleaning up of old data can stay off the Coordinator main loop, while using the same concurrency-control mechanisms that we use for regular table writes. So far, this has been a fruitful way to expand control-plane functionality while containing complexity. See the list of links right below for this work, if you're curious.

• Design: Cluster Autoscaling and Background Reconfiguration
• Implementation: Hydration burst autoscaling (#37671)
• Implementation: Background graceful reconfiguration (#37452)
• Implementation: Reconfiguration observability and SHOW CLUSTERS (#37628)
• Docs: Autoscaling for hydration
• Docs: Cluster resizing
• Docs: SHOW CLUSTERS
• Background collection of durable object hydration history (#38347)
• Durable replica hydration history (#38512)

For readers interested in the shared controller architecture:

• Implementation: Cluster controller framework (#37214)
• Original design discussion (#36691)

Header photo by adam roye on Unsplash.