Scaling service protocol

A bare-metal cluster can scale its worker pool while it runs, instead of holding a fixed number of workers. Polars Nanook delegates creating and destroying machines to a scaling service that you implement and run, over REST or gRPC. The service uses the infrastructure already in place to provision machines, such as a cloud autoscaling group, a VM API or a batch scheduler.

Autoscaling describes when the scheduler, which runs on the leader node, changes the pool size and how to point it at the service, and Workers per query how queries ask for workers. The [scaling] section of the config file reference lists every key.

Operations

The REST and gRPC transports carry the same two operations: move the pool to a size, and report configuration and state. REST is the simpler choice when an HTTP service is already in place.

Scale-to request

The scheduler asks the scaling service to move the pool to a target size. To add a worker, the service starts polars-nanook-compute service with a worker node configuration and a unique instance_id.

The workers listed in workers_to_keep are running a query, and removing one of them makes that query redo the work stored on it. A service that cannot choose which specific workers go away, for instance because it only sets a replica count, should refuse to scale down while that list is non-empty instead of removing an arbitrary worker. Alternatively, set scaling.rest.scale_down_on_idle_only = true (or scaling.grpc.scale_down_on_idle_only). The scheduler then only asks to scale down while no query is running, so workers_to_keep is empty on every scale-down request.

Worker names in workers_to_keep are the identifiers the workers registered with the scheduler (i.e. instance_id). The list is a plain array of names over REST, and a ValueList message wrapping a values field over gRPC.

Answer as soon as the target is accepted, without waiting for machines to boot. Report the new target as desired right away, and let available catch up as the workers register with the scheduler (see below).

Failing the request is safe. The scheduler logs a warning, keeps the size it had, and asks again at its next scaling decision, so an error is a reasonable way to refuse a change that cannot be applied yet. Queries waiting for those workers still fail once the scale-up timeout passes.

Scaling configuration and state

The scheduler asks the scaling service for its current state and limits. The leader fails to start if the service does not answer at startup, so start the scaling service first.

min must not exceed max when max is present. These values can change whenever the service's own limits change, for example when an administrator raises a quota, and the scheduler uses the new range without a restart.

REST scaling service

A REST scaling service exposes two endpoints, based at the uri configured in scaling.rest.

GET /scale_config returns the current configuration and state.

{
  "desired": 4,
  "available": 4,
  "min": 0,
  "max": 16
}

POST /scale_to receives the target size, and should answer 204 No Content.

{
  "amount": 6,
  "workers_to_keep": ["worker-a", "worker-b"]
}

The schema also has a workers_to_delete field, which the scheduler sends as null.

The scheduler allows 5 seconds per request. It retries a request that times out, cannot connect, or is answered with 429 Too Many Requests, up to four times, waiting a second longer before each attempt. Any other error response is not retried. It re-reads /scale_config every 5 minutes, so a change to the limits takes up to that long to be noticed.

{
  "openapi": "3.1.0",
  "info": {
    "title": "Polars Nanook Scaling",
    "description": "REST API specification for the Polars Nanook auto-scaler.",
    "version": "1.0.0"
  },
  "paths": {
    "/scale_config": {
      "get": {
        "tags": [
          "Scaling"
        ],
        "description": "Get the current auto-scaling config.",
        "operationId": "getScaleConfig",
        "responses": {
          "200": {
            "description": "",
            "content": {
              "application/json": {
                "schema": {
                  "$ref": "#/components/schemas/ScaleConfigResponse"
                }
              }
            }
          }
        }
      }
    },
    "/scale_to": {
      "post": {
        "tags": [
          "Scaling"
        ],
        "description": "Scale to a target amount of workers.",
        "operationId": "postScaleTo",
        "requestBody": {
          "content": {
            "application/json": {
              "schema": {
                "$ref": "#/components/schemas/ScaleToBody"
              }
            }
          },
          "required": true
        },
        "responses": {
          "204": {
            "description": "no content"
          }
        }
      }
    }
  },
  "components": {
    "schemas": {
      "ScaleConfigResponse": {
        "type": "object",
        "properties": {
          "available": {
            "description": "The currently available amount of workers.",
            "type": "integer",
            "format": "uint32",
            "minimum": 0
          },
          "desired": {
            "description": "The desired amount of workers to auto-scale to.",
            "type": "integer",
            "format": "uint32",
            "minimum": 0
          },
          "max": {
            "description": "The upper-bound of workers to auto-scale (up) to. Absent means no upper bound.",
            "type": [
              "integer",
              "null"
            ],
            "format": "uint32",
            "minimum": 0
          },
          "min": {
            "description": "The lower-bound of workers to auto-scale (down) to.",
            "type": "integer",
            "format": "uint32",
            "minimum": 0
          }
        },
        "required": [
          "desired",
          "available",
          "min"
        ]
      },
      "ScaleToBody": {
        "type": "object",
        "properties": {
          "amount": {
            "description": "Target amount of workers.",
            "type": "integer",
            "format": "uint32",
            "minimum": 0
          },
          "workers_to_delete": {
            "description": "Workers that must be deleted when scaling down.",
            "type": [
              "array",
              "null"
            ],
            "items": {
              "type": "string"
            }
          },
          "workers_to_keep": {
            "description": "Workers that must be kept when scaling down.",
            "type": [
              "array",
              "null"
            ],
            "items": {
              "type": "string"
            }
          }
        },
        "required": [
          "amount"
        ]
      }
    }
  }
}

gRPC scaling service

A gRPC scaling service implements the ScalingService service. ScaleConfig is a server-streaming RPC: answer immediately with the current configuration, then send a new message every time it changes. A ScaleTo call answered with UNAVAILABLE is retried, but other statuses are not.

syntax = "proto3";

import "compute_scaling/client/v1/scale_to.proto";
import "compute_scaling/client/v1/scale_config.proto";

package compute_scaling.client.v1;

/**
 * ScalingService is a service which provides scaling up/down of the cluster.
 */
service ScalingService {
  // Scale the cluster to the requested size. This is called when the cluster wants to scale up or down.
  rpc ScaleTo(ScaleToRequest) returns (ScaleToResponse);
  // Sync the autoscaler's configuration and state. The autoscaler should respond with its current configuration and state as quickly as possible and then send additional responses whenever its configuration changes
  rpc ScaleConfig(ScaleConfigRequest) returns (stream ScaleConfigResponse);
}

syntax = "proto3";

package compute_scaling.client.v1;

// This request is sent when we want to get the latest config information.
message ScaleConfigRequest {}

message ScaleConfigResponse {
  // Amount which currently is requested from the AutoScaler.
  uint32 desired = 1;
  // Amount of workers currently scaled up.
  uint32 available = 2;
  // Lower bound of allowed scaling.
  uint32 min = 3;
  // Upper bound of allowed scaling. Absent means no upper bound.
  optional uint32 max = 4;
}

syntax = "proto3";

package compute_scaling.client.v1;

message ScaleToRequest {
  message ValueList { repeated string values = 1; }
  // Number of workers to scale to.
  uint32 amount = 1;
  // Workers which should NOT be removed (when scaling down).
  optional ValueList workers_to_keep = 2;
  // Workers which should be removed first (and replaced if amount requires it).
  optional ValueList workers_to_delete = 3;
}

message ScaleToResponse {}