Shard Manager

The shard manager assigns shards to worker executors, publishes the routing table, and rebalances assignments when executors join or leave. It also manages the resource leases used by quotas. This page covers operating the service; see Architecture for its place in a Golem deployment.

Choose a persistence mode

All replicas of a distributed shard manager must use the same etcd cluster and shard count. The default number_of_shards is 1024. A shard manager refuses to start if its configured count differs from the stored state.

Keep the persistence store across restarts. Changing from SQL to etcd does not transfer quota state: executors acquire new quota leases, and capacity budgets start from their full limit.

Configure distributed mode

Set the following in the shard manager’s configuration. Replace the endpoint with a private load-balanced address for your etcd cluster, such as a Kubernetes Service:

[persistence]
type = "Etcd"
 
[persistence.config]
endpoints = ["http://etcd.internal:2379"]
connect_timeout = "10s"
request_timeout = "5s"
leader_lease_ttl = "10s"
compaction_retention_revisions = 0

Use one load-balanced endpoint rather than a list of etcd members. An unavailable member in the client’s rotation can delay writes and leader election: reads and writes that never reached etcd, and compactions, are sent again after a short delay within the retry budget, and a leadership campaign can spend tens of seconds landing on the dead member. A write that reached etcd is never sent again, so a refusal stays final. The leader’s ready mark is retried on any retriable failure, because it sets the same value again.

Check these settings before starting replicas:

Compaction removes old revisions across the entire etcd cluster, not just Golem’s keys. Disabling Golem’s compaction does not remove the need to compact: configure etcd’s automatic compaction or another operator-managed policy.

For all settings and their environment variable equivalents, see the configuration  and sample environment file .

Advertise each replica’s address

Each replica must set grpc.advertised_host (GOLEM__GRPC__ADVERTISED_HOST): the host at which other processes reach its gRPC port. On Kubernetes this is the pod IP, taken from the downward API. In Docker Compose it is the replica’s own service name. A standby uses it to name the leader, and clients connect to the leader at its advertised host and port, so executors and worker services must be able to reach every replica directly, not only through a Service.

The shard manager takes a host, which can be a DNS name or an IP address, because in Docker Compose the service name is the natural address. The worker executor setting grpc.advertised_ip is always an IP address.

• grpc.advertised_port (GOLEM__GRPC__ADVERTISED_PORT) is optional and defaults to the bound grpc.port. Set it when a port mapping or NAT puts the replica behind a different port. The value 0 is refused.
• A replica in distributed mode refuses to start without grpc.advertised_host, or with 0.0.0.0 or ::. It warns about a loopback address, which works only when every process runs on one host.
• A host that does not form a valid address with the port is refused at startup: for example one with a space, a #, a zone suffix such as fe80::1%eth0, or a second colon outside an IPv6 literal.
• An advertised IPv6 address needs an IPv6 bind address. Set GOLEM__BIND_ADDRESS: "::"; startup refuses an IPv6 host when bind_address is IPv4.

Each replica needs its own advertised address. Never use the Service name. A replica that finds another election key with its address watches that key’s lease. A live replica keeps renewing its lease, which moves the end of the lease later. When the new replica has seen the end move later twice, without a change of etcd’s raft leader in between, it stops with an error. A key left by an earlier process of the same pod is never renewed: its lease only runs down and the key disappears, so the replica waits it out and then leads. This holds even if the lease TTL changed before the restart, or an etcd leader change extended the lease once.

The replica identity, which is diagnostic only, comes from the POD_NAME environment variable, or else from HOSTNAME. Kubernetes sets HOSTNAME to the pod name, so no extra identity variable is needed there.

Use TLS with the leader address

With gRPC TLS enabled, a client without shard_manager.tls.config.server_domain_name (GOLEM__SHARD_MANAGER__TLS__CONFIG__SERVER_DOMAIN_NAME) checks the leader’s certificate against its configured shard_manager.host, when that host is a DNS name. The certificate of the shard managers must then name that host, for example the Service name. A configured server_domain_name takes precedence.

Secure the etcd connection

Distributed mode can talk to etcd over TLS, with a client certificate (mTLS), and as an etcd user. With persistence.config.tls set, every entry in persistence.config.endpoints must start with https://; without it, every entry must start with http://, and a replica configured otherwise refuses to start. Each TLS item comes from a PEM file or from an inline value (the PEM in standard Base64 with padding, on one line, as base64 -w0 writes it), not both: tls.ca_cert_file or tls.ca_cert is the CA that signed etcd’s server certificate; tls.client_cert_file or tls.client_cert, and tls.client_key_file or tls.client_key, are the shard manager’s client certificate and its unencrypted key (PKCS#8, PKCS#1 or SEC1; set both or neither); tls.server_name is the name to check etcd’s certificate against when the endpoint host is an IP address or an alias that the certificate does not name. The ca.crt in a Secret that cert-manager makes for the client certificate is the CA of that certificate’s issuer, so it works only when the same issuer signed etcd’s server certificate; otherwise mount a trust bundle that holds etcd’s CA.

At startup a replica reads every TLS item once and logs each certificate’s subject, end of validity and SHA-256 fingerprint. It refuses to start when an item cannot be read or holds no certificate or key, when the key is encrypted or does not belong to the client certificate, when the client certificate has expired or is not valid yet, when every certificate of the CA item has expired, when etcd refuses the TLS handshake (a certificate that etcd rejects, or a server certificate that the CA did not sign or that does not name the endpoint), and when the user’s role may not write /golem/. Each refusal names the setting or the reason. An etcd server certificate that has expired, or is not valid yet, is for etcd to renew: the replica then waits, as for an etcd that is down, and logs the certificate’s end of validity on every attempt.

Renew certificates

A renewed certificate takes effect on a restart. A connection that is open keeps working after a renewal; a new connection presents the certificate read at startup. While a replica runs, a TLS handshake that etcd refuses, or that refuses etcd’s certificate, is retried like a network fault, because the cause can be on etcd’s side. Every persistence.config.retry_max_delay the replica checks, through both of its etcd connections, whether etcd refuses its certificates. When it does and the TLS files have changed since startup, the replica tries the files on disk on a new connection; when etcd accepts them, the replica stops, so that its restart reads them. When etcd refuses the new files too - as when etcd’s own certificate has expired - a restart would not help, and the replica keeps retrying. A controller that rolls the pods when the Secret changes saves the wait for that first refusal. Mount the Secret as a directory, not with subPath: with subPath the files in the pod never change, and a replica keeps retrying with the old certificate. An inline value changes only with a rollout.

# In the shard manager's pod template
containers:
  - name: golem-shard-manager
    env:
      - name: GOLEM__PERSISTENCE__CONFIG__ENDPOINTS
        value: '["https://etcd-client.etcd.svc:2379"]'
      - name: GOLEM__PERSISTENCE__CONFIG__TLS__CA_CERT_FILE
        value: /etc/etcd-tls/ca.crt
      - name: GOLEM__PERSISTENCE__CONFIG__TLS__CLIENT_CERT_FILE
        value: /etc/etcd-tls/tls.crt
      - name: GOLEM__PERSISTENCE__CONFIG__TLS__CLIENT_KEY_FILE
        value: /etc/etcd-tls/tls.key
      - name: GOLEM__PERSISTENCE__CONFIG__TLS__SERVER_NAME
        value: etcd-client.etcd.svc
    volumeMounts:
      - name: etcd-tls
        mountPath: /etc/etcd-tls   # the whole Secret as a directory, never subPath
        readOnly: true
volumes:
  - name: etcd-tls
    secret:
      secretName: golem-shard-manager-etcd-client   # the cert-manager Certificate's Secret

Authenticate as an etcd user

With etcd authentication enabled, set persistence.config.username and the password, either as persistence.config.password_file - a file that holds it, such as a key of a Secret mounted as files; only one trailing line ending is removed - or as persistence.config.password, usually from a Secret through GOLEM__PERSISTENCE__CONFIG__PASSWORD. Neither the logs, the configuration display nor a refused configuration shows the password. The configuration loader reads an environment value made only of digits, or true or false, as a number or a boolean, and a replica refuses it without printing it; the loader also removes spaces around a value and double quotes around a whole value, without a warning. Use password_file for any password with such characters (or, in an environment variable that you write yourself, put double quotes around the value). A user needs TLS as well: a replica refuses a user name and password over http://, because the password, and then the auth token on every request, would cross the network in plain text, so switch TLS and the user on together. The user’s role needs readwrite permission on the key prefix /golem/. With client certificates, etcd can also take the user from the certificate’s common name (CN) instead of a password: leave username and the password unset. etcd drops an auth token that goes unused for its --auth-token-ttl (300s by default), a JWT token at the end of its lifetime, and its simple tokens when a member restarts. When etcd answers that a token is not valid, the shard manager logs in again and sends the request once more, so no token setting is needed. With JWT tokens (--auth-token=jwt), etcd also refuses every token that was issued before a change to its users or roles, such as a new user or a changed password: the campaign, each replica’s watch of the election, the leader’s ready mark and the step-down then log in again and send the request once more, a new leader logs in again before its first read, and a leader whose state write is refused stops, so that its restart logs in.

Executor leases

Each executor registers with the shard manager and holds a renewable lease on its shard assignment. Registration is persisted before it is acknowledged. Assignment updates are persisted before they are sent to executors.

The shard manager checks for expired leases periodically, normally every third of shard_lease_duration. Persistence delays, health checks, and rebalance work can extend recovery time; this cadence is not a failover deadline.

Each renewal carries the epoch of every shard the executor holds. A renewal whose claim does not match the shard manager’s view is still renewed, and the response carries the shard manager’s set, which the executor adopts as it would an assignment push. Pushes deliver a change at once, and renewals guarantee it arrives within a third of the lease even if a push was lost. Every delivery carries the revision of the state it was read from, and an executor ignores a delivery older than one it has already applied. Revisions are comparable only within one shard manager process, so every delivery also names the process that sent it. An executor follows the process that answered its latest registration or renewal, ignores a push from any other process, and renews at once to hear from the one in charge. This is how a shard manager that failed over, or came back on a wiped or restored store, gets its sets applied although its revisions start lower.

A renewal never moves an executor’s deadline earlier, so a shard_lease_duration reduced across a restart takes effect only once the new length outruns the deadline each executor was last told.

Keep executor clocks synchronized before starting them. Lease exchange does not compare absolute clocks, but clock-rate differences still affect expiry timing. An assignment push does not renew a lease.

Restarts and leader failover

In distributed mode every replica serves its gRPC port (grpc.port) from startup, but only the leader runs calls. A standby answers each call with UNAVAILABLE and a golem.shardmanager.v1.NotLeader detail that holds the leader’s host and port (leader_host, leader_port). It takes them from an in-memory view of the etcd election, so answering does not read etcd. The worker executors and worker services follow the detail: they connect to the leader directly and keep using it until a call to it fails, and then ask their configured address again. The leader publishes its address only after its own startup has completed, so a client is never sent to a leader that cannot serve yet. A replica that has not reached its own advertised address is a standby and names the ready leader in the same way.

A new leader restores shard and quota state before it serves. Existing quota leases survive a leader change because their state is in etcd. A restart or failover restarts the lease of every executor the shard manager keeps, so it never evicts a healthy cluster.

If the initial health check runs out of time, the shard manager still removes the executors that had already answered that they are unhealthy. It keeps the executors it could not finish checking and leaves them to the periodic check. An executor whose own lease lapsed during the outage resumes admitting work after its next successful renewal.

If the leader loses its lease part-way through a registration write, the executor receives an error instead and repeats the registration within its shard_manager.retries budget.

Neither the leadership lease nor any other single setting is an upper bound on the interruption: the lease, etcd availability, the startup checks and the retry policies all add to the recovery time.

How clients wait out a failover

During a failover no replica serves. After a leader crash, the replicas name the old leader until its lease expires. Then they answer UNAVAILABLE without naming a leader until the new leader’s startup has completed. Clients keep asking and reach the new leader once its startup has completed.

• A worker service does not ride the window out on a warm routing table. The first executor-level error it hits discards the cached table, and the worker service then waits until a leader serves again.
• A worker executor that starts inside the window waits it out too. Registration is a call that may be repeated, and the executor repeats it within its shard_manager.retries. With no ready leader at all, a registration fails only after each repeat has waited out the budget described below.

The worker executors and worker services wait out a failover themselves, so no client setting needs tuning for it. A call that may be repeated keeps asking for about 15s once a replica has named a leader that cannot be reached, or a replica itself answers UNAVAILABLE. The 15 seconds count from the start of the call. If the shard_manager.retries_on_unavailable schedule takes longer than that in all, the call keeps asking for that long instead. With the default settings this covers one leadership lease plus the startup.

• A shard manager that cannot be reached at the configured address at all is not a failover. shard_manager.retries_on_unavailable still decides how long the client keeps trying there.
• The 15 seconds are fixed. With a longer leader_lease_ttl or initial_health_check_timeout, raise shard_manager.retries_on_unavailable on the clients, because a call waits for whichever is longer.
• A call that must not run twice is not repeated.
• To notice a replica whose node lost power in time, the clients probe their connections to a named leader and to the configured address at least every 2s and wait at most 5s for the answer, also while idle. They also use one connection to the configured address that is not replaced for its age, whatever connection_pool_size and max_connection_age say. A sidecar proxy answers such probes itself, so behind one the clients find a dead proxy but not a dead replica.

Use other tools against a standby

Tools such as grpcurl and your own scripts must follow the NotLeader detail themselves. A replica that refuses a call before it ran adds the detail type.googleapis.com/golem.shardmanager.v1.RefusedBeforeRunning to the status. It does so when it is starting, stopping, no longer sure that it holds its lease, or knows no ready leader. The detail has no message type in the proto and its value is empty, so a tool recognises it by that type URL alone. A call refused with that mark never ran, so a tool may send it to another replica whatever the call is. An UNAVAILABLE with neither detail may have run.

A kubectl port-forward to the Service can land on a standby. Its answer then names a pod IP that a laptop cannot reach. Port-forward to the serving pod instead, which is the one whose /metrics shows shard_manager_is_serving 1.

Health probes and monitoring

Use two different probes on the HTTP listener (http_port, 8081 by default):

/readyz answers 200 or 503:

• A leader that serves is ready while it believes it holds its lease.
• A standby is ready while etcd has answered it within one lease TTL. A replica cut off from etcd is therefore taken out of a rollout’s count. Cut off from etcd, it keeps naming the leader it last saw.
• A just-elected leader that is still running its startup is ready on the same terms as a standby.
• A replica is not ready before it has connected to etcd, and in local mode not until the startup has completed.
• In distributed mode a replica is also not ready until it has reached its own advertised address once. It makes a TCP connection to the advertised host and port and, when gRPC TLS is off, a gRPC health check. With gRPC TLS on, only the TCP connection is made. It waits one second between attempts. While the check fails, it logs a warning that names the address, at most every 30 seconds.

The own-address check finds a wrong grpc.advertised_host or grpc.advertised_port: the replica never becomes ready. A setup in which a pod cannot reach its own externally advertised address, such as some NAT or port-mapping layouts, keeps the replica unready. Without gRPC TLS the health check asks for an entry that only this process has, so an address that another process answers at is refused too, for example a Service name that reaches only other replicas.

The check does not catch every mistake. It passes as soon as one attempt reached this process, so a name that resolves to several replicas including this one passes (a headless Service, the Compose alias), and a ClusterIP name passes after a few attempts. A name that several replicas advertise is caught by the duplicate-address rule above, not by this check.

A replica that has not reached its own address does not campaign. It neither queues for the leadership nor leads, and it sends callers on to the ready leader as any standby does. If every replica fails the check, no replica leads, and the warning in the log is the first sign.

The gRPC health service reports the overall status "" as SERVING on every replica. The entry golem.shardmanager.v1.ShardManagerService is SERVING only on the replica that serves. Routing by a leader label is not needed.

Monitor these metrics across all replicas:

The two gauges differ while a new leader starts. shard_manager_is_leader is 1 from the moment the replica wins the election, because it holds the lease from then. shard_manager_is_serving becomes 1 only after the replica has restored its state and checked the executors: a short time after a crash, and up to initial_health_check_timeout after a longer outage.

Both gauges read 0 on a healthy standby and on a stuck one alike. /readyz and shard_manager_campaign_attempt_failures_total tell them apart. A replica that etcd stopped answering, or that has not reached its own address, is not ready. A rising counter means the replica cannot reach etcd and is retrying. A ready standby with a flat counter is waiting for the leadership. See Metrics for metrics collection.

To read the election itself, run etcdctl elect -l /golem/shard-manager/leader. It prints the leader’s election value, a JSON document with its identity, its gRPC address, a ready flag and its build version.

Roll out replicas

A Deployment or StatefulSet RollingUpdate, health gates such as Helm --wait, Flux wait: true or Argo, and a PodDisruptionBudget all work with these replicas.

A rolling update to a newer version hands the leadership over once, to a new replica, without any setting. Each replica writes its build version into its election entry, and an old standby moves to the end of the election queue when a replica of a newer version queues behind it. Versions compare by semantic-version precedence, with the numbers in a release candidate and in a development build compared as numbers: 1.6.0-rc10 is newer than 1.6.0-rc9. A development build compares as a pre-release of the release its last tag belongs to. 1.6.0-rc2-156-g… is newer than 1.6.0-rc2-9-g… and 1.6.0-rc2, but older than a later tag 1.6.0-rc2.2 and than 1.6.0-rc3. 1.6.0-12-g… is newer than every candidate and older than 1.6.0.

A replica of an older release that writes no version, or a version text that is no semantic version, never causes a move and never moves, and the queue keeps its order. A rollback, which queues an older version behind newer ones, moves nothing, and neither does a rollout of the same version.

Kubernetes example

Three replicas behind one ordinary Service:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: golem-shard-manager
spec:
  replicas: 3
  minReadySeconds: 20
  strategy:
    type: RollingUpdate
    rollingUpdate: { maxSurge: 1, maxUnavailable: 0 }
  selector:
    matchLabels: { app: golem-shard-manager }
  template:
    metadata:
      labels: { app: golem-shard-manager }
    spec:
      containers:
        - name: shard-manager
          image: golemservices/golem-shard-manager:<version>
          env:
            - name: GOLEM__GRPC__ADVERTISED_HOST
              valueFrom: { fieldRef: { fieldPath: status.podIP } }
            - name: GOLEM__GRPC__PORT
              value: "9092"
            - name: GOLEM__HTTP_PORT
              value: "8081"
            - name: GOLEM__PERSISTENCE__TYPE
              value: Etcd
            - name: GOLEM__PERSISTENCE__CONFIG__ENDPOINTS
              value: '["http://etcd-client.etcd.svc:2379"]'
          ports:
            - { name: grpc, containerPort: 9092 }
            - { name: http, containerPort: 8081 }
          livenessProbe:
            httpGet: { path: /healthcheck, port: http }
          readinessProbe:
            httpGet: { path: /readyz, port: http }
---
apiVersion: v1
kind: Service
metadata:
  name: golem-shard-manager
spec:
  selector: { app: golem-shard-manager }
  ports:
    - { name: grpc, port: 9092, targetPort: grpc }
---
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: golem-shard-manager
spec:
  maxUnavailable: 1
  unhealthyPodEvictionPolicy: AlwaysAllow
  selector:
    matchLabels: { app: golem-shard-manager }

This example connects to etcd over http://. For an etcd with TLS, give the container the GOLEM__PERSISTENCE__CONFIG__TLS__* variables, an https:// endpoint and the Secret mounted as a directory, as in the pod template under Renew certificates, and add the Secret volume to the pod. With an etcd user, add GOLEM__PERSISTENCE__CONFIG__USERNAME and the password as described in Authenticate as an etcd user.

Point the executors’ and worker services’ shard_manager.host (GOLEM__SHARD_MANAGER__HOST) at the Service name, and set their shard_manager.port (GOLEM__SHARD_MANAGER__PORT) to the Service’s gRPC port, 9092 in this example. The host alone is not enough: the client’s default port is 9002, which the Service does not expose. A rolling update that replaces the leader’s pod sends it SIGTERM, so a standby takes over within milliseconds and serves once its startup completes.

minReadySeconds: 20 matters for rollouts. A new pod is Ready at once, because a standby is ready while etcd answers it. Without a pause, Kubernetes then stops the next old pod while a just-elected leader is still running its startup, and several startups can follow in a row. A rollout to a newer version hands the leadership over once, whatever order Kubernetes stops the old pods in. A rollout that keeps the version, such as a configuration change, and a rollback keep the queue’s order, so minReadySeconds is what keeps a just-elected leader from being stopped in the middle of its startup.

unhealthyPodEvictionPolicy: AlwaysAllow in the PodDisruptionBudget (Kubernetes 1.27 or later) matters during an etcd outage. Every replica is then unready, and a budget that counts unready pods would block every node drain.

Docker Compose example

The same layout for three replicas. The replicas share one network alias, and each one advertises its own service name:

services:
  golem-shard-manager-1: &shard-manager
    image: golemservices/golem-shard-manager:<version>
    environment: &shard-manager-env
      GOLEM__GRPC__PORT: 9002
      GOLEM__HTTP_PORT: 8081
      GOLEM__PERSISTENCE__TYPE: Etcd
      GOLEM__PERSISTENCE__CONFIG__ENDPOINTS: '["http://etcd:2379"]'
      GOLEM__GRPC__ADVERTISED_HOST: golem-shard-manager-1
    networks:
      default:
        aliases: [golem-shard-manager]
  golem-shard-manager-2:
    <<: *shard-manager
    environment:
      <<: *shard-manager-env
      GOLEM__GRPC__ADVERTISED_HOST: golem-shard-manager-2
  golem-shard-manager-3:
    <<: *shard-manager
    environment:
      <<: *shard-manager-env
      GOLEM__GRPC__ADVERTISED_HOST: golem-shard-manager-3

The worker executors and worker services set GOLEM__SHARD_MANAGER__HOST: golem-shard-manager and GOLEM__SHARD_MANAGER__PORT: 9002. The complete example, with etcd and the other services, is in docker-examples/distributed-etcd .

Upgrade from earlier replicas

Replicas of an earlier version, whose standbys keep their gRPC port closed, write an election value that this version cannot read. A standby of this version then names no leader, and clients retry until the old replicas are gone. Roll all shard manager replicas before you rely on redirects.

Oplog fencing and storage

Every shard has an epoch, an ownership generation that advances when the shard moves. Indexed storage records the epoch allowed to write an agent’s oplog. Once the new owner raises that record, writes from an older owner are rejected atomically. The same protection applies to oplog deletion and archive metadata.

This protection has limits:

• It needs an epoch record. A new or recreated oplog is claimed by the first opener. Leases remain necessary to prevent stale executors from admitting work.
• It protects the oplog, not every agent write. Do not infer the same guarantee for key-value data, status blobs, or arbitrary blob payloads.
• It cannot undo external effects. A call that reached an external service before its oplog entry was recorded can run again after recovery, as it can after a crash.

Executors that share shards need shared indexed storage, such as PostgreSQL or Redis. A local SQLite file serves one executor. Redis failover that loses recent writes can also lose fencing records; Redis Cluster is not supported. See Persistence for storage configuration.

Restore shard-manager state carefully

Shard epochs and indexed-storage fencing records must agree. Replacing, wiping, or restoring the shard-manager store can produce epochs lower than those already recorded in oplogs. The affected agents then cannot write.

Treat this as coordinated storage recovery, not a routine restart:

1. Stop every worker executor before changing fencing records, and prevent the orchestrator from restarting them.
2. Back up the shard-manager store and indexed storage. Determine which state must be restored together before changing either store.
3. If epoch records must be reset, account for every indexed-storage layer. SQL uses indexed_key_epoch; Redis uses the oplog-epoch keys. MultiSqlite also retains epoch snapshots in epochs.db, separate from the live oplog databases. Clearing only the live databases is incomplete: the saved epochs can be restored when a database reopens.