Operating Rama clusters

The first step to setting up a Rama cluster is creating a Zookeeper cluster. Zookeeper is used by Rama to store cluster metadata such as which nodes should be running which module workers and replication coordination information (e.g. leadership election). Rama makes use of Zookeeper internally through an abstraction called the "Metastore". The deployment instructions for Zookeeper can be found here.

Next, download a Rama release (will be available in the upcoming Red Planet Labs private beta). When unpacking a release you’ll see contents like this:

rama.jar and lib/ contain the Rama implementation and all dependency jars. If there are other libraries you wish to be on the classpath of every module, you can add those to lib/. log4j2.properties contains Log4j configuration for all Rama daemons and module workers. logs/ is where all log files will go. rama implements the Rama CLI, described below. Lastly, rama.yaml is where you’ll configure Rama’s daemons.

A Rama release is used to launch the daemons comprising a Rama cluster as well as on clients issuing the Rama CLI to launch, update, or scale modules. It’s critical that clients use the same Rama release version as used by the cluster.

Rama daemons and clients can be run on UNIX-based machines, including Linux and macOS. Java 8, 11, or 17 should be installed. You will also need Python 3 installed on the machines.

The next step is setting up the Conductor node. The Rama CLI communicates with the Conductor to launch, update, and scale modules. The Conductor also serves a web-based Cluster UI showing status and telemetry information for modules.

First, unpack the Rama release on the machine. Then edit the rama.yaml with the following configurations:

Here’s an example Conductor configuration:

Other available Conductor configs can be found on this page.

Next, launch the Conductor by running ./rama conductor from the unpacked release. For production you should run this command using a process supervision tool like systemd or monit.

The last step to setting up a Rama cluster is configuring Worker nodes. Each Worker node runs a daemon called the Supervisor which watches for changes to cluster state and starts/stops Worker processes as dictated by Conductor.

Launching Supervisors is similar to launching the Conductor. First, unpack the Rama release on the machine. Then edit the rama.yaml with the following configurations:

Here’s an example Supervisor configuration:

Other available Supervisor configs can be found on this page.

Next, launch the Supervisor by running ./rama supervisor from the unpacked release. Like the Conductor, for production you should run this command using a process supervision tool like systemd or monit. For the Supervisor it’s best that you configure the supervision tool not to kill subprocesses if the Supervisor process dies. Although there’s no reason to expect the Supervisor to crash, if it does it’s been designed to have no affect on running workers and there’s no reason to bring down workers if their Supervisor crashes.

It’s also recommended to configure worker nodes with swap space. Module update launches new workers colocated with the previous module instance’s workers as it hands off responsibility between the two module versions. So each worker node either needs enough memory to run workers for both module versions, or it needs swap space to handle the extra temporary memory needs.

The Rama CLI is used via the rama script at the root of an unpacked release. To use it outside a cluster, you simply configure conductor.host in the rama.yaml in the same directory. You must ensure the rama script you’re using is from the same release version as the cluster you configure with conductor.host.

The commands available in the CLI are:

The optional --useInternalHostnames flag on some of the commands causes the command to use internal hostnames when connecting to the Conductor, Zookeeper, supervisors, or workers.

On the depot, PState, and query topology pages, you saw how to use Java-based depot, PState, and query topology clients to programatically interact with modules on a Rama cluster. Let’s now take a look at how to get ahold of those clients.

Depot, PState, and query topology clients are retrieved using RamaClusterManager. A RamaClusterManager instance is told how to connect to a Rama cluster and then has methods to create the desired clients.

To create a RamaClusterManager, use the static methods open or openInternal. Both are exactly the same, except the former uses external hostnames and the latter uses internal hostnames. If you aren’t configuring separate internal and external hostnames in your rama.yaml files, it doesn’t matter which one you use.

The minimal information RamaClusterManager needs to know is the host and port of the Conductor node managing the cluster. The zero-arity versions of open and openInternal retrieve that configuration from the rama.yaml file on the classpath of the client.

There’s also a one-arity version of those methods that allow you to pass in those configs manually. These versions of the methods will still read in the rama.yaml file on the classpath if it exists, but the manual config will take precedence. Here’s an example of usage:

If you’re using any custom serializations, you would also configure those with the "custom.serializations" key. That’s described more in this section.

New modules are launched using the Rama CLI’s deploy command. Here’s an example of usage:

--action launch specifies this is a deploy of a new module as opposed to an update of an existing module. You’ll see the details of updating an existing module in the next section.

The --tasks flag specifies the number of tasks in the module. The number of tasks specified must be a power of two.

--threads specifies the number of task threads for the module, and --workers specifies the number of worker processes. These can be any positive number and do not have to be a power of two. Rama will spread tasks as evenly as possible across task threads, and it will spread task threads as evenly as possible across workers. In this process it also balances the number of tasks per worker as best as possible. The number of tasks must be greater than or equal to the number of threads, and the number of threads must be greater than or equal to the number of workers.

If you specify --threads and --workers as a power of two, then load will be perfectly distributed across all your workers. If not, then some workers will have higher loads than others. For example if you have 64 tasks, nine threads, and nine workers, then eight workers will have seven tasks and one worker will have eight tasks. That one worker will process about 14% more load than the other workers. It’s inevitable to eventually have skewed load like this in a module deployment since you’re likely to only increase the number of resources by a small amount when you scale (e.g. 25%). Note that the more tasks you have in your module, the less the skew will be. However, there’s overhead per task so you don’t want too many tasks. We recommend starting off with no more than 32 tasks per thread.

Unlike --tasks, --threads, and --workers, --replicationFactor is not a required flag. The default replication factor is one, but we recommend using a higher replication factor for fault-tolerance. The number of JVM processes that will exist for a module deployment is the replication factor multiplied by the number of workers specified. Correspondingly, the number of threads that will exist is the replication factor multiplied by the number of threads specified. Each task thread is duplicated by the amount of the replication factor, and two task thread replicas will never exist on the same node. In this example there will be 24 JVM processes deployed across the cluster.

The --jar flag specifies where the module definition exists, and --module specifies the class implementing the module. A module’s class is the same as the module’s name by default, but if the module definition overrides getModuleName they will be different. Most other commands in the Rama CLI take in the module name as input, not the module class.

If using the Clojure API, --module should refer to the namespace-qualified var defining the module, like com.mycompany/FooModule.

The jar you submit must contain your module definition as well as all libraries you’re using that aren’t already provided by Rama. We’ll show one way to build such a jar in the next section.

A module currently running on a Rama cluster can be updated to a new version with the rama deploy command. Here’s an example of usage:

Whereas a module launch uses the launch action, a module update uses the update action. Like module launch, the --jar and --module flags specify the location of the module code and the module class to deploy. Parallelism settings (tasks, threads, workers) are not specified on module update since the new version of the module will use the same settings as the old module. However, any config overrides from a previous module deploy are not inherited. So those must be specified with each module update (e.g. custom serializations, worker JVM settings).

A module update can perform any change to a module: updating topology code, adding new topologies, removing topologies, adding new PStates/depots, or removing PStates/depots. The deploy will validate the changes you’re making are valid: for example, if you try to remove a PState or depot that’s currently depended on by another module, the update will not go through.

The naming of depots and PStates in a module definition determine what carries over to the new module instance. Any depot or PState defined in the old module that’s not defined in the new module is considered destroyed. There’s currently no way to rename a depot or PState in a module update.

Because removing a PState or depot is destructive – Rama will delete all partitions from the filesystems of worker nodes – Rama requires you to explicitly specify objects being deleted as part of an update command. Here’s an example:

The specified --objectsToDelete must match exactly the objects being removed from the module or the update will throw an exception.

A module update is a highly coordinated operation to transition from one module instance to the next. In general, it works by launching the worker processes for the updated module instance on the same machines as the old module instance. It then gracefully transitions data and read/write responsibilities between the two instances in order to minimize downtime.

PState clients have zero downtime during the transition and will automatically retarget queries to the new module instance when it’s ready. This includes regular queries as well as reactive queries. Existing reactive queries will automatically perform a resync to the new instance. Depot appends may have some downtime where appends are buffered clientside while waiting for the new instance to be ready to accept appends. Likewise, stream and microbatch processing may be offline for a brief time during the transition. The downtime depends on the number of PStates and size of the individual PStates that are being transitioned, and can be anywhere from two seconds to thirty seconds (we have it on our roadmap to implement zero-downtime module update for appends and processing as well).

The one case where module update doesn’t deploy the new worker processes on the same machines as the old module instance is when transitioning from the dev scheduler to the isolation scheduler. In this case, it may be forced to relocate workers in order to achieve isolation. There’s no difference in semantics and all data will be transitioned to the new machines.

During a module update, you’ll see a variety of statuses for the module in the Cluster UI. You’ll see states like [:updating-next-assigned], [:updating-prepare-handover], and many others. These are intermediate states as the Conductor shepherds along the update process to perform the graceful transition. When it finishes the old module instance will be destroyed and the module status will go back to [:running].

The Rama CLI provides the command rama scaleExecutors to change the number of workers, number of threads, or replication factor of a module. It can be used to increase or decrease any of these values.

Here’s an example of using this command:

Unlike the rama deploy command, --module refers to the module name and not the module class. The rest of the flags specify the new values for the number of threads, number of workers, and replication factor.

--threads, --workers, and --replicationFactor are all optional. If unspecified, the value from the current module instance will be used. You can even run rama scaleExecutors without specifying any of these flags (called "no-op scaling") which is useful when transitioning a cluster from the dev scheduler to the isolation scheduler.

All other aspects about a module are preserved when scaling, particularly the module instance config. So whereas with a module update you must specify the module instance config each time, with scaling it carries over to the new module instance. This is because a module update could be changing or deleting configs, whereas a scaling operation doesn’t change the module itself.

Scaling takes longer to complete than module update because Rama needs to copy data around to move tasks to new nodes. The scaling operation will still colocate the new module instance as much as possible with the old module instance to minimize this copying, but some copying will be necessary.

Rama provides a special "monitoring module" that will automatically collect, analyze, index, and visualize detailed telemetry on all modules running on a cluster. This telemetry is displayed throughout dozens of charts on the Cluster UI and is invaluable for diagnosing and debugging issues.

To active self-monitoring, all you have to do is deploy the monitoring module using a command like the following:

The only difference in the rama deploy command here as compared to deploying your own modules is the lack of a --jar flag and using --systemModule instead of --module. Otherwise, this module is deployed and operated just like any other module, and you should make sure to scale the module as appropriate as the number of modules on the cluster grows. As a rule of thumb, the monitoring module should have one worker per 70 workers from other modules.

If you’re using the isolation scheduler, you can reference the isolation module in the config as monitoring. For example, this could be the isolation config in the rama.yaml for the Conductor:

We highly recommend running the monitoring module on every cluster since it provides a huge amount of fine-grained, useful information and uses a relatively small amount of resources.

The Cluster UI is a web-based interface providing a wealth of information on a Rama cluster and modules running on that cluster. By default it runs on port 8888 on the Conductor, but the port can be changed in the Conductor’s rama.yaml via the cluster.ui.port config.

Let’s take a brief tour of the information available in the Cluster UI. We won’t look at every page, but we’ll look at enough to give a general idea of what’s available. Here’s the main page of the Cluster UI:

All deployed modules are listed along with their current state. The state refers to how the module is progressing through module operations such as launch, update, or scaling. When these operations have completed the module will be in the [:running] state. Sometimes you may see modules in the [:leadership-balancing] state. In this state Rama has found an imbalance of leaders across module workers and forces the minimal number of leader switches necessary in order to balance leadership. This is important since all work goes through leaders, so an imbalance would cause unnecessary skew of load for a module. Leaders can become imbalanced when a cluster has node outages.

The Supervisors tab lists all Supervisors for the cluster, and clicking on a Supervisor displays general information and telemetry for the node. The telemetry includes memory usage and garbage collection stats for the Supervisor and a graph of the system load average over time. The Conductor tab shows similar telemetry for the Conductor process and node.

On the top-right of the UI you can change the time window for charts, such as looking at the past hour of data or looking at the past year of data. The monitoring module computes and indexes every stat at multiple granularities to be able to support these large time windows efficiently. When the autorefresh button is on, all information on a page will refresh every ten seconds.

Clicking on a module brings up the module page. For example:

This page lists each module instance for the module. You’ll only see one module instance here unless the module is undergoing a transition like module update or module scaling. The "read target" and "append target" refer to internal flags used to coordinate a module transition.

On this page you can also set module level dynamic options. The ⓘ symbols, when hovered over, display a short string describing the dynamic option. You can set topology, depot, and PState level dynamic options on the pages for those entities.

Clicking on the module instance brings up this page:

On this page you can see all the depots, PStates, and topologies defined for this module instance. You’ll also see internal PStates used by Rama in its implementation in the PStates section. Some of these PStates are in-memory and unreplicated (e.g. buffers for two-phase aggregation), and some of them are durable and replicated (e.g. stream topology progress state).

Below those are many charts aggregating telemetry for the module instance across all workers. For example:

"Task groups load" is one of the most important metrics. It represents the percentage of time a task thread spends executing events vs. sitting idle waiting for events to arrive. It can be a strong indicator for when a module needs to be scaled. On the module instance page, you can see the average task thread load across all task threads in the module as well as the maximum task thread load across all task threads.

The "System event throughput (sampled)" and "System event execution duration (sampled)" metrics provide extremely fine-grained information on what’s going on inside a module. A random sample of events in Rama is categorized and timed, and these charts allow you to explore the throughput and durations for all the categories. The sample rate can be adjusted with the dynamic option worker.event.telemetry.sampling.rate, but keep in mind that adjusting it too high will have a noticeable impact on module performance. The events displayed here include both the events executing module code as well as internal events supporting the execution of modules (e.g. replication, telemetry collection, cleanup events).

The event categories are organized into a hierarchy, and you can see the aggregated metrics for any node in that hierarchy by selecting that node in the drop-down on the top-right of the chart. For example, selecting in that dropdown [:all :replication] shows the combined stats for all replication related events. Selecting [:all :replication :leader :forwarding] shows stats only for replication events related to leaders forwarding replog entries to followers. Categories containing :topology-event are events that are executing module code (e.g. [:all :microbatch :whoToFollow :topology-event] is the category for events executing the topology named "whoToFollow").

These are only a few of the metrics available on this page. All the charts available via the Cluster UI are invaluable for diagnosing any issues that occur.

The Workers link lets you navigate to a page specific to each worker, where you’ll see the same telemetry as the module instance page except specific to that worker. If you had skew in the amount of data being processed by each worker, you’d be able to see that by comparing the worker pages.

Clicking on depots, PStates, or topologies navigates to pages specific to those entities. Those pages have their own telemetry specific to them. For example:

On microbatch topology pages, you’ll also see a button you can use to "Pause" or "Resume" the topology. This is the only action available in the Cluster UI, as currently all other actions are performed through the Rama CLI.

Machines can fail for all sorts of reasons. The Rama CLI command for module update provides the option --reassignSupervisors to get workers currently assigned to a failed machine relocated to another node. Rama never reassigns workers automatically and will only assign workers in response to a module operation (module launch, module update, or module scaling).

The additional argument given to module update is used like so:

Every Supervisor has a unique ID which you can look up on the Cluster UI. Running this command will relocate workers for the module running on that node to other nodes. The command respects the isolation scheduler, and it requires sufficient spare capacity be available for moving the workers. You can reassign multiple supervisors at once by passing a comma-separated list to --reassignSupervisors.

If you have multiple modules with workers assigned to that node, you’ll need to run a separate module update for each module.

The reassignSupervisors option has no reliance on communicating with workers on the target nodes or executing any computation on those nodes. It assumes the worst case that the processes are down, unresponsive, or faulty in some way. It’s recommended you kill the Supervisors and all worker processes on those nodes before running the update command to ensure other modules don’t get assigned to those nodes later.

The reassign command as shown also requires that workers on the target node not be the only members of the ISR for any of their task groups. This can happen if the replication factor is one or if all other replicas happen to be in the OSR.

You can force the reassign to relocate all workers on a node regardless of ISR status with the flag --forceReassign. The reason this isn’t automatic is because reassigning a task group that’s the only member of its ISR means all data for PState and depot partitions on that task group is lost. Because Rama presumes the worst case of the node being inaccessible or broken in some way, it doesn’t try to transfer any data off of that node in this case. Keep in mind that with replication and a sensible Min-ISR setting, this scenario is extremely unlikely. We recommend a replication factor of three and a min-ISR of two.

Rama has built-in mechanisms for upgrading a cluster to a new Rama version and transitioning all data between the old and new versions. There are two ways to upgrade a Rama cluster: "patch version upgrades" and "atomic version upgrades". During a patch version upgrade the Rama cluster and modules are fully operational, but it can only be done when only the patch version changes (e.g. 0.13.0 to 0.13.1). An atomic version upgrade can be done between any Rama versions and requires taking some downtime on the cluster. Though if you prepare, the whole process for an atomic version upgrade can be done in 5-10 minutes.

Rama provides the tooling you need to operate clusters over long periods of time. Self-monitoring gives you insights to understand performance and diagnose issues. Module update and module scaling let you evolve running applications to incorporate new features, fix bugs, and handle increased load. Lastly, the rama reassign command lets you handle the need to decommission nodes when they go bad.

There’s a lot to learn about operating Rama clusters. The breadth of scope of what Rama does is far greater than most other software tools. But the alternative to using Rama is to use many different tools, sometimes dozens, each of which handles only a narrow function of building a backend. So although there may feel like a lot to learn to use Rama, it’s vastly simpler in the long run to operate one tool than many tools with inconsistent designs around how they work and how they’re operated.