This is an automated email from the ASF dual-hosted git repository.
zaynt4606 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/celeborn.git
The following commit(s) were added to refs/heads/main by this push:
new e107e2d0c7 [CELEBORN-2466] Support per-zone worker StatefulSets and
autoscaling in the Helm chart
e107e2d0c7 is described below
commit e107e2d0c70e59890b0e428f197947a35cc0b359
Author: strelok89 <[email protected]>
AuthorDate: Tue Oct 13 11:41:24 2026 +0800
[CELEBORN-2466] Support per-zone worker StatefulSets and autoscaling in the
Helm chart
### What changes were proposed in this pull request?
Three values for `charts/celeborn`, all disabled by default.
**`worker.zoneAwareReplication`** renders one worker StatefulSet per zone
instead of a single
one for the whole cluster. Each carries a `celeborn.apache.org/zone` label
as part of its
selector, so a zone's worker count is declared rather than inferred from
scheduling.
`worker.replicas` becomes the total across zones, split `ceil(replicas /
zones)`, with a
per-zone override; `zones[].nodeSelector` merges over `worker.nodeSelector`
and
`zones[].affinity` replaces `worker.affinity`. All zones share the existing
headless service,
so the service and pod monitor keep selecting every zone.
**`worker.zoneAwareReplication.metricsLabel`** passes the zone into
`celeborn.metrics.extraLabels` so the worker stamps a `zone` label on
everything it emits. It
is injected as a system property, since master and worker share one config
map:
`Utils.loadDefaultCelebornProperties` applies the file with `setIfMissing`,
so a system
property takes precedence, and `start-worker.sh` builds
`CELEBORN_JAVA_OPTS` from
`CELEBORN_WORKER_JAVA_OPTS`, so it composes with whatever `worker.env` sets.
**`worker.drain`** adds the `preStop` hook described below, independent of
autoscaling, since
any rollout removes workers too.
**`worker.autoscaling`** creates a KEDA `ScaledObject` per worker
StatefulSet — one per zone
with the above enabled. Two triggers ship, on the two resources whose
exhaustion takes a
worker out of service, both scoped by the `zone` label and both excluding a
decommissioning
worker:
- `diskUsage`, on `DeviceCelebornFreeBytes / DeviceCelebornTotalBytes`,
default 0.70.
- `memoryUsage`, on `DirectMemoryUsageRatio`, default 0.70 — below
`celeborn.worker.directMemoryRatioToPauseReceive` (0.85) so capacity
arrives before pushes
pause.
Both use `metricType: Value`, since a ratio must not be divided across
replicas.
`worker.autoscaling.triggers` appends user triggers, rendered through `tpl`
against the zone's
context.
`worker.drain` adds a `preStop` hook calling `POST /api/v1/workers/exit`
with
`DECOMMISSION`, because Kubernetes scale-in only sends `SIGTERM`, which
never enters the
decommission path — the worker would leave with its shuffle data. The hook
must not fire on a
rolling update or node drain, where waiting for shuffle data to expire
would block every pod
replacement for hours, so it distinguishes them by ordinal: the StatefulSet
controller lowers
`spec.replicas` before deleting pods on a scale-in, so an ordinal at or
above the desired count
is being removed for good, while anything else shuts down gracefully and
returns. A desired
count that cannot be read falls back to a graceful shutdown, so a failed
lookup cannot stall a
rollout. The hook reads the count with the pod's own service account, so
the role gains `get`
on `statefulsets/scale` when the drain is enabled.
### Why are the changes needed?
Workers all live in one StatefulSet at a fixed replica count, which leaves
two gaps.
Their spread across availability zones is uneven. The scheduler decides
where a worker lands,
so pod-level spreading (`worker.affinity`, or
`worker.topologySpreadConstraints` from
CELEBORN-2446) balances it only on a best-effort basis: the balance is
re-derived on every
scheduling decision, holds while the constraint is satisfiable, and
degrades quietly when it is
not — a zone out of capacity for the workers' instance type, or several
nodes replaced at once.
And a zone cannot be scaled at all. Load per zone is genuinely uneven when
applications select
workers by tag (`celeborn.tags.tagsExpr`), so the fleet has to be sized for
the busiest zone
and left that size. A single StatefulSet also cannot be rolled or paused
one zone at a time.
Note that this controls where worker pods run and how many there are. It is
not zone-aware data
placement: which workers a client may use is decided in the master by
worker tags, not by pod
topology.
### Does this PR resolve a correctness bug?
- [ ] Yes
### Does this PR introduce _any_ user-facing change?
- [x] Yes
New chart values: `worker.zoneAwareReplication` and `worker.autoscaling`.
Both default to
disabled and the rendered output with default values is unchanged, so
existing releases are
unaffected until they opt in. `charts/celeborn/README.md` documents both,
including that
enabling zone-aware replication renames the worker StatefulSet and so
recreates every worker —
a maintenance-window change rather than a rolling update.
### How was this patch tested?
- `helm unittest charts/celeborn --file "tests/**/*_test.yaml" --strict` —
121 tests pass, 36
of them new, covering per-zone StatefulSets and selectors, the replica
split and per-zone
override, the node selector merge, the zone metric label and its
composition with existing
extra labels, the per-zone ScaledObjects and their default triggers, the
`preStop` hook with
its environment and scoped RBAC rule, and each `fail` guard.
- `helm lint`, with default values and with
`charts/celeborn/ci/values.yaml`.
- `helm template` with default values, diffed against `main`:
byte-identical.
- `helm template` with three zones and autoscaling enabled: three
StatefulSets with distinct
names and selectors, integer replica counts, one shared `serviceName`,
unchanged service and
pod monitor selectors, and one ScaledObject per zone targeting its own
StatefulSet.
- The trigger queries were run against a live Celeborn deployment's
Prometheus to confirm the
metric names exist and each query returns the single element KEDA
requires.
- `files/worker-drain.sh` was run directly over scale-in, rolling update,
always-decommission,
unreadable replica count, missing pod name, IPv6 bracketing, and an
unreachable exit
endpoint.
Closes #3855 from strelok89/helm-worker-zone-aware-statefulsets.
Authored-by: strelok89 <[email protected]>
Signed-off-by: zhengtao <[email protected]>
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Co-Authored-By: Codex <[email protected]>
AI-Model: gpt-6
AI-Contributed/Feature: 0/860
AI-Contributed/UT: 0/852
---
charts/celeborn/README.md | 332 ++++++++++++++++
charts/celeborn/files/worker-drain.sh | 130 +++++++
charts/celeborn/templates/configmap.yaml | 5 +
charts/celeborn/templates/role.yaml | 19 +
charts/celeborn/templates/worker/_helpers.tpl | 71 +++-
charts/celeborn/templates/worker/scaledobject.yaml | 77 ++++
charts/celeborn/templates/worker/statefulset.yaml | 121 +++++-
charts/celeborn/tests/configmap_test.yaml | 18 +
charts/celeborn/tests/role_test.yaml | 52 +++
.../celeborn/tests/worker/scaledobject_test.yaml | 361 ++++++++++++++++++
charts/celeborn/tests/worker/statefulset_test.yaml | 421 +++++++++++++++++++++
charts/celeborn/values.yaml | 105 ++++-
12 files changed, 1705 insertions(+), 7 deletions(-)
diff --git a/charts/celeborn/README.md b/charts/celeborn/README.md
index c6908fe9cb..c679f5c364 100644
--- a/charts/celeborn/README.md
+++ b/charts/celeborn/README.md
@@ -54,6 +54,338 @@ More details in [Helm
Install](https://helm.sh/docs/helm/helm_install/).
The chart can be customized using the following [celeborn
configurations](https://celeborn.apache.org/docs/latest/configuration/#important-configurations).
Specify parameters using `--set key=value[,key=value]` argument to `helm
install`.
+## Zone-aware worker replication
+
+By default the chart deploys all workers in a single statefulset, and which
zone a worker
+lands in is whatever the scheduler decides. Pod-level spreading -
`worker.affinity`, or a
+topology spread constraint where the chart supports one - can balance that,
but the balance is
+re-derived on every scheduling decision: it holds while the constraint is
satisfiable and
+silently degrades when it is not, for example when one zone is out of capacity
for the instance
+type the workers need, or when several nodes are replaced at once.
+
+Setting `worker.zoneAwareReplication.enabled` deploys one worker statefulset
per zone
+instead, which makes zone membership structural rather than emergent:
+
+```yaml
+worker:
+ # Total across all zones; each zone gets ceil(replicas / zones) = 3.
+ replicas: 9
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ nodeSelector:
+ topology.kubernetes.io/zone: us-east-1a
+ - name: us-east-1b
+ nodeSelector:
+ topology.kubernetes.io/zone: us-east-1b
+ - name: us-east-1c
+ nodeSelector:
+ topology.kubernetes.io/zone: us-east-1c
+```
+
+This renders `celeborn-worker-us-east-1a`, `-1b` and `-1c`, each carrying a
+`celeborn.apache.org/zone` label that is part of its statefulset selector.
What it gives you:
+
+- A zone's worker count is declared, not inferred, so it survives concurrent
node churn and
+ a full recreate.
+- Each zone can be rolled, scaled or paused on its own, leaving the other
zones serving.
+- A worker's ordinal-to-zone mapping is stable, so its identity
+ (`<pod>.<service>.<namespace>.svc.<cluster>.local`) implies its zone. That
is what makes
+ zone-pinned worker selection (see below) predictable.
+
+Notes:
+
+- `worker.replicas` is the total across zones; each zone deploys
+ `ceil(worker.replicas / number of zones)`, so 5 replicas over 3 zones is 2
per zone (6
+ total). Set `zones[].replicas` to size a zone explicitly, which is also the
lever for
+ running a zone short while its capacity is constrained.
+- `zones[].nodeSelector` is merged over `worker.nodeSelector` and wins on
conflicting keys;
+ `zones[].affinity` replaces `worker.affinity` for that zone.
+- All zones share one headless service, so the service and the pod monitor
keep selecting
+ workers in every zone. Only the statefulset selectors are zone-scoped.
+- Per-zone `nodeSelector` already separates the zones physically, so a zone
+ `topologySpreadConstraints` entry becomes redundant once this is enabled. A
per-hostname
+ spread or anti-affinity rule is still meaningful within a zone.
+
+### Zone-aware replication is not zone-aware data placement
+
+This only controls where worker pods run. It does not make a client prefer a
worker in its
+own zone: Celeborn allocates slots in the master, and the client-side view of
which workers
+are eligible comes from worker tags (`celeborn.tags.tagsExpr`,
`celeborn.tags.enabled`), not
+from pod topology. To keep shuffle traffic inside one zone, tag each worker
with its zone and
+have each application select its own zone's tag. Note that
+`celeborn.client.reserveSlots.rackaware.enabled` does the opposite on purpose:
it spreads a
+partition's replicas across racks for durability.
+
+### Migrating an existing release
+
+The statefulset names change, so enabling this on a running release replaces
+`<fullname>-worker` with the per-zone statefulsets. Every worker is recreated,
and with it
+its identity, so:
+
+- Worker data on `hostPath` or `emptyDir` volumes does not survive.
+- The master holds stale worker registrations until
`celeborn.master.heartbeat.worker.timeout`
+ elapses.
+- In-flight shuffles fail.
+
+Treat it as a maintenance-window change rather than a rolling update.
+
+## Draining a worker before it is removed
+
+Deleting a worker pod - by scaling in, rolling out, or draining its node -
reaches Celeborn as
+a `SIGTERM`. That never enters the decommission path, so the worker leaves
with its shuffle data
+and any application still reading from it loses that stage. Celeborn's safe
drain is
+`DECOMMISSION`, which waits for the worker's shuffle keys to expire (up to
+`celeborn.worker.decommission.forceExitTimeout`, 6h by default) and is only
reachable through
+the worker's HTTP API.
+
+This is worth having with a fixed replica count, not only with
`worker.autoscaling` - any
+rollout removes workers too.
+
+`worker.drain.enabled` adds a `preStop` hook that calls it. The hook runs
`files/worker-drain.sh`, shipped in the chart's
+config map and mounted at `/opt/celeborn/drain/worker-drain.sh`; it takes
`POD_NAME`,
+`STS_NAME` and `WORKER_HTTP_PORT` from the environment, so the script itself
is the same in
+every zone. The hook must not decommission on a rolling update or a node
+drain, or every pod replacement would block for hours, so it distinguishes the
two: the
+statefulset controller lowers `spec.replicas` *before* deleting pods on a
scale-in, so a pod
+whose ordinal is at or above the desired count is being removed for good and
decommissions,
+while any other pod shuts down gracefully and comes back. If the desired count
cannot be read
+the hook falls back to a graceful shutdown, so a broken lookup cannot stall a
rollout.
+
+This needs three things:
+
+- `rbac.create: true`. The chart adds `get` on `statefulsets/scale` to the
role when the
+ drain is enabled; the hook reads the count with the pod's own service
account.
+- `celeborn.worker.graceful.shutdown.enabled: true`, so the non-scale-in path
actually
+ persists state and recovers on restart.
+- `worker.terminationGracePeriodSeconds` above
`celeborn.worker.decommission.forceExitTimeout`.
+ This is an upper bound, not a wait: a graceful shutdown still finishes in
+ `celeborn.worker.graceful.shutdown.timeout`, so a long grace period does not
slow rollouts
+ down.
+
+The hook calls the worker over plain HTTP and sends no credentials, so it
cannot reach a worker
+with `celeborn.worker.http.ssl.enabled` or
`celeborn.worker.http.auth.supportedSchemes` set. A
+failed exit request falls through to `SIGTERM`, which on a scale-in would
silently skip the
+drain, so the chart refuses to render `worker.drain` together with either
setting under
+`celeborn`. It can only see what is set there: enabling TLS or authentication
some other way,
+such as an extra config file, still bypasses the drain.
+
+It is off by default. Turn it on alongside `worker.autoscaling`, or leave it
off and treat
+scale-in as destructive, only letting it happen when the fleet is idle.
+
+### When the worker's storage does not outlive the pod
+
+Telling a scale-in from a rollout is only worth doing if a rollout can
recover. Graceful
+shutdown persists committed file metadata to
`celeborn.worker.graceful.shutdown.recoverPath` and
+recovers from it when the worker comes back - which needs both that path and
+`celeborn.worker.storage.dirs` to survive the pod. On an `emptyDir`, or on
instance-store disks
+that are wiped when the node is replaced, neither does: the worker returns to
empty disks, there
+is nothing to recover, and the in-flight shuffles it was holding are lost.
+
+`worker.drain.alwaysDecommission: true` drops the distinction and
decommissions on
+every termination, so a worker is never removed while an application still
needs what is on it.
+It also needs no Kubernetes API access, since it never reads the replica
count, and the chart
+leaves the `statefulsets/scale` rule out of the role.
+
+What it costs:
+
+- Rollouts and node drains wait for the drain as well, bounded by
+ `celeborn.worker.decommission.forceExitTimeout` rather than
+ `celeborn.worker.graceful.shutdown.timeout`. With `OrderedReady` a rollout
drains each worker
+ in turn, so budget accordingly.
+- Anything that evicts a pod on a deadline shorter than the drain will still
kill it mid-drain -
+ a cluster autoscaler reclaiming a node, or a spot interruption with its
two-minute notice.
+ Make sure the node pool's own grace period is at least as long as you expect
a drain to take,
+ or accept that those paths behave as they did before.
+
+#### Worked example: local NVMe through an `emptyDir`
+
+A common shape for workers is to skip network storage entirely and write to
the node's own NVMe
+- an `emptyDir` mounted at `celeborn.worker.storage.dirs`, landing on the
instance's local disks.
+On AWS that means instance-store NVMe, and on a node OS that assembles those
disks into one
+array for container storage (Bottlerocket's local-storage setup, for instance)
an `emptyDir`
+lands on that array with no extra wiring. It is a reasonable choice: shuffle
is a
+write-heavy, short-lived, reproducible workload, so there is little reason to
pay for network
+volumes, per-volume throughput ceilings, or PVCs that pin an ordinal to one
zone.
+
+The catch is lifetime, and it is shorter than people expect. An `emptyDir` is
deleted when the
+pod leaves the node - not only when the node goes away. So both
+`celeborn.worker.storage.dirs` and
`celeborn.worker.graceful.shutdown.recoverPath`, if it sits
+under the same mount, disappear on every pod replacement. Graceful shutdown
still writes its
+recovery database, and the replacement worker still starts up and
re-registers, but it comes
+back to empty disks and there is nothing to read the database against.
+
+What that costs while jobs are running: a rolling update, a chart upgrade or a
scale-in removes
+a worker that is holding shuffle data for applications still running. Those
fetches fail, and
+since `celeborn.client.push.replicate.enabled` is `false` by default there is
no second copy -
+the engine recomputes the stages that produced the lost partitions. A rollout
across a fleet of
+workers can do this repeatedly, to whatever is running at the time.
+
+So on ephemeral storage the drain is what keeps a running job's shuffle alive,
and
+`alwaysDecommission` is the setting that matters - the scale-in/rollout
distinction is pointless
+when neither can recover:
+
+```yaml
+worker:
+ # Above celeborn.worker.decommission.forceExitTimeout (6h), so kubelet does
not kill a
+ # worker part-way through draining.
+ terminationGracePeriodSeconds: 21900
+ drain:
+ enabled: true
+ alwaysDecommission: true
+ volumes:
+ - name: celeborn-data
+ emptyDir: {}
+ volumeMounts:
+ - name: celeborn-data
+ mountPath: /celeborn-data/disk0
+celeborn:
+ celeborn.worker.storage.dirs: /celeborn-data/disk0:disktype=SSD
+ celeborn.worker.graceful.shutdown.enabled: true
+```
+
+A worker then stays up until the applications holding data on it are finished,
and only then
+exits - so no running job loses its shuffle to a deployment.
+
+Note that an `emptyDir` needs no `chown` init container, unlike the `hostPath`
volumes the chart
+mounts by default: Kubernetes applies `worker.podSecurityContext.fsGroup` to
it, so the worker
+can write to the mount as it is.
+
+One thing to expect when first enabling the drain: the script is shipped in
the config map that
+master and worker share, so adding it changes the
`celeborn.apache.org/conf-hash` annotation on
+both and the masters roll as well.
+
+Two things this still does not cover. Losing the node itself - a spot
interruption, or a node
+autoscaler consolidating - destroys the disks whatever the pod does, and gives
less time than a
+drain needs; set the node pool's grace period to at least the drain you
expect, and treat the
+remainder as accepted risk. And if you want a rollout to genuinely recover
rather than wait,
+storage has to outlive the pod: a `hostPath` on the NVMe mount for both
`storage.dirs` and
+`recoverPath` survives a pod restart on the same node, though nothing survives
the node being
+replaced, and a `hostPath` gives the scheduler no reason to put the pod back
where its data is.
+
+## Autoscaling workers with KEDA
+
+`worker.autoscaling` creates a [KEDA](https://keda.sh) `ScaledObject` for each
worker
+statefulset. With zone-aware replication enabled that is one per zone, so each
zone scales
+on its own load - which is what you want when applications are pinned to a
zone, because
+their load is genuinely uneven across zones.
+
+The chart ships two triggers - disk and off-heap memory, the two resources
whose exhaustion
+takes a worker out of service - both querying metrics the worker already
exports. Point them at
+a Prometheus-compatible endpoint and they work as they are:
+
+```yaml
+worker:
+ terminationGracePeriodSeconds: 21900 # see "Draining a worker before it is
removed"
+ drain:
+ enabled: true
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus.monitoring.svc.cluster.local:9090
+ maxReplicaCount: 6
+ behavior:
+ scaleDown:
+ stabilizationWindowSeconds: 1800
+ policies:
+ - type: Pods
+ value: 1
+ periodSeconds: 900
+```
+
+**Disk usage** is the signal that bites first, since a worker that fills up
stops accepting
+pushes. It scales on the fullest worker in the statefulset, as a fraction of
Celeborn disk
+capacity:
+
+```promql
+max((1 - metrics_DeviceCelebornFreeBytes_Value{...} /
metrics_DeviceCelebornTotalBytes_Value{...})
+ and on (instance) metrics_IsDecommissioningWorker_Value{...} == 0)
+```
+
+It uses `metricType: Value`, because a ratio must not be divided across
replicas.
+
+This trigger is also the slot utilization target, which is why there is no
separate one for
+slots. Celeborn sizes a disk's slot capacity from its space -
+`maxSlots = totalSpace / estimatedPartitionSize` in `WorkerInfo` - so slots
and bytes fill in
+step, and `activeSlots / maxSlots` reduces to the disk fraction above. A slot
percentage could
+not be computed anyway: `maxSlots` lives in the master's `DiskInfo` and no
metric exports it.
+If you want the raw count as a leading signal - slots are allocated when a
stage starts, before
+its data is written - add it yourself through `worker.autoscaling.triggers`
using
+`sum(metrics_ActiveSlotsCount_Value{role="Worker"})` with `metricType:
AverageValue` and a
+per-worker target.
+
+**Off-heap memory** is the other way a worker takes itself out of service: it
stops accepting
+pushes at `celeborn.worker.directMemoryRatioToPauseReceive` (0.85) and
replication at 0.95.
+`DirectMemoryUsageRatio` is `memoryUsage / maxDirectMemory`, so it is already
a fraction of
+`CELEBORN_WORKER_OFFHEAP_MEMORY`:
+
+```promql
+max(metrics_DirectMemoryUsageRatio_Value{...} and on (instance)
metrics_IsDecommissioningWorker_Value{...} == 0)
+```
+
+Also `metricType: Value`. Keep the threshold below 0.85 so capacity arrives
before pushes
+pause; the 0.70 default leaves the gap between scaling out and throttling.
+
+Both exclude decommissioning workers. A worker that is draining still holds
its disk and its
+slots for as long as it takes, and counting it would have the fleet scale out
to replace
+capacity it has not released yet.
+
+Disable either with `diskUsage.enabled: false` / `memoryUsage.enabled: false`,
and add your own
+with `worker.autoscaling.triggers`, which are appended to the built-in ones
and rendered
+through `tpl` against the zone's context, so `{{ .zone.name }}` resolves per
zone.
+
+The built-in queries select on `role="Worker"`, the release namespace and the
pod names of the
+statefulset they scale, so several Celeborn clusters can share one Prometheus
- or one
+namespace - without scaling on each other's workers.
+
+Celeborn exports gauges as `metrics_<Name>_Value` and counters as
`metrics_<Name>_Count`. Other
+metrics worth scaling on are `ActiveShuffleSize` and `ActiveShuffleFileCount`
(data held),
+`ActiveSlotsCount` (slots allocated), and `IsHighWorkload` /
`PausePushDataStatus` (the worker
+is already in trouble).
+
+`worker.zoneAwareReplication.metricsLabel` passes the zone into
`celeborn.metrics.extraLabels`
+so the worker stamps it on everything it emits. The built-in queries do not
use it - a `zone`
+label says nothing about which release a worker belongs to - but it is there
for queries of your
+own. `role="Worker"` matters too: masters report the device gauges for
whichever volume holds
+the Ratis directory.
+
+`minReplicaCount` defaults to the statefulset's own replica count, so a zone
never scales
+below the size it was deployed at unless you set it explicitly.
+
+Scaling to zero needs a trigger that does not come from the workers. Both
built-in queries read
+metrics the statefulset's own workers emit, so once the last one is gone they
return nothing and
+KEDA has no signal to scale back up on. The chart refuses `minReplicaCount: 0`
unless
+`worker.autoscaling.triggers` has an entry - a `cron` schedule, or a query on
something outside
+the workers such as pending Spark executors. It cannot tell what that trigger
reads, so make
+sure it is one that still fires with no workers running.
+
+Enable `worker.drain` as well. Scaling in deletes worker pods, and without the
drain those
+workers leave with shuffle data an application may still need - see
+[Draining a worker before it is
removed](#draining-a-worker-before-it-is-removed).
+
+Three properties of this fleet are worth sizing for. Scaling out is slow: a
new worker
+usually needs a new node, and it starts with no shuffle data on it, so
autoscaling answers
+sustained load changes rather than bursts. Scaling in always removes the
highest ordinal,
+which is not necessarily the least busy worker. And while a worker drains, its
statefulset
+cannot grow - so a scale-in decided at low load leaves that zone one worker
short until the
+drain finishes. Pace scale-in conservatively with `behavior.scaleDown`, and
consider
+excluding draining workers from the trigger query.
+
+### Helm and the autoscaler both own `replicas`
+
+The chart keeps rendering `spec.replicas`, so a fresh install starts at the
size you asked
+for rather than at one. The cost is that anything reconciling the rendered
manifest will fight
+KEDA over that field.
+
+A plain `helm upgrade` is included in that: its three-way merge patches a
KEDA-scaled replica
+count back to the chart's value, and with `OrderedReady` the controller then
removes the excess
+ordinals one at a time, waiting for each to terminate - which is slow once
`worker.drain` is
+enabled, and slower again with `alwaysDecommission`. Keep `worker.replicas` in
step with where
+KEDA has settled, or pass `--set worker.replicas=<current>` when upgrading.
+
+For a continuous-delivery tool, tell it to ignore the field - in Argo CD,
`ignoreDifferences`
+on `/spec/replicas` for the worker statefulsets, with
`RespectIgnoreDifferences=true`.
+
## Documentation
For additional details on deploying the Celeborn Kubernetes Helm chart, please
refer to the [Celeborn on
Kubernetes](https://celeborn.apache.org/docs/latest/deploy_on_k8s/)
documentation.
diff --git a/charts/celeborn/files/worker-drain.sh
b/charts/celeborn/files/worker-drain.sh
new file mode 100644
index 0000000000..a6d9ef99c6
--- /dev/null
+++ b/charts/celeborn/files/worker-drain.sh
@@ -0,0 +1,130 @@
+#!/bin/sh
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# preStop hook for a Celeborn worker. Asks the worker to decommission when a
scale-in is
+# removing it, and to shut down gracefully otherwise. See the chart's
README.md.
+#
+# Reads POD_NAME, POD_IP, STS_NAME, WORKER_HTTP_PORT and ALWAYS_DECOMMISSION
from the
+# environment.
+
+set -u
+
+SA=/var/run/secrets/kubernetes.io/serviceaccount
+
+# Kubernetes counts preStop against terminationGracePeriodSeconds, so no
request may hang: a
+# server that accepts the connection and then stops responding would burn the
whole grace
+# period and the pod would be killed without ever draining. Only the wait
after a successful
+# exit request is allowed to take that long. `-T` is the one timeout flag both
busybox and GNU
+# wget accept.
+HTTP_TIMEOUT=10
+CURL_OPTS="--connect-timeout 5 --max-time $HTTP_TIMEOUT"
+WGET_OPTS="-T $HTTP_TIMEOUT"
+
+# preStop output is not collected, so log through the main process.
+log() {
+ _msg="celeborn preStop: $*"
+ if [ -w /proc/1/fd/1 ]; then
+ echo "$_msg" >/proc/1/fd/1
+ else
+ echo "$_msg"
+ fi
+}
+
+ORDINAL=""
+DESIRED=""
+EXIT_TYPE=GRACEFUL
+
+if [ "${ALWAYS_DECOMMISSION:-false}" = "true" ]; then
+ # Graceful shutdown persists state to recover from on restart. Where the
worker's storage
+ # does not outlive the pod there is nothing to recover, so a rollout would
drop in-flight
+ # shuffles - drain every time instead, whatever is removing this pod.
+ EXIT_TYPE=DECOMMISSION
+ log "alwaysDecommission is set - draining regardless of why this pod is
going away"
+else
+ # A scale-in must decommission, waiting for the shuffle data to expire. A
rolling update or
+ # a node drain must not, or every pod replacement would block for hours. The
statefulset
+ # controller lowers spec.replicas before deleting pods on a scale-in, so an
ordinal at or
+ # above the desired count is being removed for good.
+ ORDINAL=${POD_NAME:-}
+ ORDINAL=${ORDINAL##*-}
+
+ if [ -r "$SA/token" ]; then
+ URL="https://kubernetes.default.svc/apis/apps/v1/namespaces/$(cat
"$SA/namespace")/statefulsets/${STS_NAME:-}/scale"
+ AUTH="Authorization: Bearer $(cat "$SA/token")"
+ if command -v curl >/dev/null 2>&1; then
+ SCALE=$(curl -sS $CURL_OPTS --cacert "$SA/ca.crt" -H "$AUTH" "$URL"
2>/dev/null)
+ else
+ SCALE=$(wget -q $WGET_OPTS -O - --ca-certificate="$SA/ca.crt"
--header="$AUTH" "$URL" 2>/dev/null)
+ fi
+ DESIRED=$(printf '%s' "$SCALE" | sed 's/"status".*//' | grep -o
'"replicas":[0-9 ]*' | head -n 1 | tr -dc '0-9')
+ fi
+
+ case "$ORDINAL" in ''|*[!0-9]*) ORDINAL="" ;; esac
+ case "$DESIRED" in ''|*[!0-9]*) DESIRED="" ;; esac
+
+ # Anything unreadable falls back to a graceful shutdown, so a failed lookup
cannot stall a
+ # rollout with a drain that was never wanted.
+ if [ -n "$ORDINAL" ] && [ -n "$DESIRED" ] && [ "$ORDINAL" -ge "$DESIRED" ];
then
+ EXIT_TYPE=DECOMMISSION
+ fi
+fi
+
+if [ "${ALWAYS_DECOMMISSION:-false}" != "true" ] && [ -z "$DESIRED" ]; then
+ log "could not read ${STS_NAME:-unknown} desired replicas - a scale-in will
NOT decommission"
+fi
+log "ordinal=${ORDINAL:-unknown} desired=${DESIRED:-unknown} exit=$EXIT_TYPE"
+
+BODY='{"type":"'"$EXIT_TYPE"'"}'
+
+# The exit endpoint hands the drain to a separate thread and answers straight
away, so bounding
+# the request does not cut a drain short.
+# The worker's HTTP server binds to one address, not the wildcard:
celeborn.worker.http.host
+# defaults to <localhost>, which resolves to this pod's own address, and Jetty
is given that
+# host. So loopback is refused - target the pod IP, and keep loopback only as
a fallback for a
+# deployment that has pointed the server somewhere else, such as 0.0.0.0.
+HOSTS=""
+if [ -n "${POD_IP:-}" ]; then
+ case "$POD_IP" in
+ *:*) HOSTS="[$POD_IP]" ;;
+ *) HOSTS="$POD_IP" ;;
+ esac
+fi
+HOSTS="$HOSTS 127.0.0.1"
+
+RC=1
+for HOST in $HOSTS; do
+ EXIT_URL="http://$HOST:${WORKER_HTTP_PORT:-9096}/api/v1/workers/exit";
+ if command -v curl >/dev/null 2>&1; then
+ curl -sS -f $CURL_OPTS -X POST -H 'Content-Type: application/json' -d
"$BODY" "$EXIT_URL" >/dev/null 2>&1
+ RC=$?
+ else
+ wget -q $WGET_OPTS -O /dev/null --header='Content-Type: application/json'
--post-data="$BODY" "$EXIT_URL"
+ RC=$?
+ fi
+ if [ "$RC" -eq 0 ]; then
+ break
+ fi
+ log "exit request to $HOST failed"
+done
+
+if [ "$RC" -eq 0 ]; then
+ # Hold the pod open while the worker drains. It exits on its own once done,
and
+ # terminationGracePeriodSeconds bounds the wait.
+ while :; do sleep 5; done
+else
+ log "exit request failed, falling through to SIGTERM"
+fi
diff --git a/charts/celeborn/templates/configmap.yaml
b/charts/celeborn/templates/configmap.yaml
index 8a533f5677..fd1ee75db9 100644
--- a/charts/celeborn/templates/configmap.yaml
+++ b/charts/celeborn/templates/configmap.yaml
@@ -51,3 +51,8 @@ data:
ratis-log4j.properties: |-
{{- .Files.Get "files/conf/ratis-log4j.properties" | nindent 4 }}
+{{- if .Values.worker.drain.enabled }}
+
+ worker-drain.sh: |-
+ {{- .Files.Get "files/worker-drain.sh" | nindent 4 }}
+{{- end }}
diff --git a/charts/celeborn/templates/role.yaml
b/charts/celeborn/templates/role.yaml
index dd48c1f3d2..36a986066c 100644
--- a/charts/celeborn/templates/role.yaml
+++ b/charts/celeborn/templates/role.yaml
@@ -27,5 +27,24 @@ rules:
- apiGroups: [""]
resources: ["pods"]
verbs: ["create", "list", "delete"]
+{{- if and .Values.worker.drain.enabled (not
.Values.worker.drain.alwaysDecommission) }}
+{{- /*
+The worker's preStop hook reads its own statefulset's desired replica count,
to tell a
+scale-in from a rollout. With alwaysDecommission it never asks, so it needs no
access.
+*/}}
+{{- $root := . }}
+{{- $stsNames := list (include "celeborn.worker.statefulSet.name" .) }}
+{{- if .Values.worker.zoneAwareReplication.enabled }}
+{{- $stsNames = list }}
+{{- range .Values.worker.zoneAwareReplication.zones }}
+{{- $ctx := dict "Values" $root.Values "Release" $root.Release "Chart"
$root.Chart "zone" . }}
+{{- $stsNames = append $stsNames (include "celeborn.worker.statefulSet.name"
$ctx) }}
+{{- end }}
+{{- end }}
+- apiGroups: ["apps"]
+ resources: ["statefulsets/scale"]
+ resourceNames: [{{ range $i, $n := $stsNames }}{{ if $i }}, {{ end }}{{ $n |
quote }}{{ end }}]
+ verbs: ["get"]
+{{- end }}
{{- end }}
diff --git a/charts/celeborn/templates/worker/_helpers.tpl
b/charts/celeborn/templates/worker/_helpers.tpl
index 8d9e86266e..62239f8b6a 100644
--- a/charts/celeborn/templates/worker/_helpers.tpl
+++ b/charts/celeborn/templates/worker/_helpers.tpl
@@ -21,14 +21,22 @@ Common labels for Celeborn worker resources
{{- define "celeborn.worker.labels" -}}
{{ include "celeborn.labels" . }}
app.kubernetes.io/role: worker
+{{- if .zone }}
+celeborn.apache.org/zone: {{ .zone.name }}
+{{- end }}
{{- end }}
{{/*
-Selector labels for Celeborn worker pods
+Selector labels for Celeborn worker pods. The zone label is only added when
rendering
+within a zone context, so that cluster-wide selectors (service, pod monitor)
keep
+matching the workers of every zone.
*/}}
{{- define "celeborn.worker.selectorLabels" -}}
{{ include "celeborn.selectorLabels" . }}
app.kubernetes.io/role: worker
+{{- if .zone }}
+celeborn.apache.org/zone: {{ .zone.name }}
+{{- end }}
{{- end }}
{{/*
@@ -76,8 +84,69 @@ Create the name of the worker priority class to use
Create the name of the worker statefulset to use
*/}}
{{- define "celeborn.worker.statefulSet.name" -}}
+{{- if .zone -}}
+{{ include "celeborn.fullname" . }}-worker-{{ .zone.name }}
+{{- else -}}
{{ include "celeborn.fullname" . }}-worker
{{- end }}
+{{- end }}
+
+{{/*
+Number of replicas for a worker statefulset. Without zone-aware replication
this is
+`worker.replicas` as-is; with it, `worker.replicas` is the total across all
zones and each
+zone gets `ceil(replicas / zones)` unless the zone overrides it.
+*/}}
+{{- define "celeborn.worker.replicas" -}}
+{{- if .zone -}}
+{{- /* Key presence, not truthiness: an explicit 0 parks a zone without
removing it. */ -}}
+{{- if and (hasKey .zone "replicas") (not (kindIs "invalid" .zone.replicas))
-}}
+{{ .zone.replicas }}
+{{- else -}}
+{{ divf .Values.worker.replicas (len
.Values.worker.zoneAwareReplication.zones) | ceil | int }}
+{{- end }}
+{{- else -}}
+{{ .Values.worker.replicas }}
+{{- end }}
+{{- end }}
+
+{{/*
+Label selector used by the built-in autoscaling triggers. Scoped by pod name
to this
+statefulset's own workers, so releases sharing a namespace - or sharing a zone
name - never
+scale on each other's workers. The `zone` metric label is not enough on its
own: it says
+nothing about which release a worker belongs to.
+*/}}
+{{- define "celeborn.worker.autoscaling.selector" -}}
+{{- /* Anchored on the ordinal, so statefulset `x-a` does not also match the
pods of `x-a-y`. */ -}}
+{{ printf "role=\"Worker\",namespace=\"%s\",pod=~\"%s-[0-9]+\""
.Release.Namespace (include "celeborn.worker.statefulSet.name" .) }}
+{{- end }}
+
+{{/*
+Built-in autoscaling triggers, in front of any the user adds. Both exclude a
decommissioning
+worker, whose disk stays full and whose slots stay allocated while it drains -
counting it
+would have the fleet scale out to replace capacity it is still holding.
+*/}}
+{{- define "celeborn.worker.autoscaling.defaultTriggers" -}}
+{{- $selector := include "celeborn.worker.autoscaling.selector" . -}}
+{{- $live := printf "and on (instance)
metrics_IsDecommissioningWorker_Value{%s} == 0" $selector -}}
+{{- if .Values.worker.autoscaling.diskUsage.enabled }}
+- type: prometheus
+ {{- /* Value, not KEDA's AverageValue default: a ratio must not be divided
by the replicas. */}}
+ metricType: Value
+ metadata:
+ serverAddress: {{ required "worker.autoscaling.prometheusAddress is
required by the built-in triggers" .Values.worker.autoscaling.prometheusAddress
}}
+ query: max((1 - metrics_DeviceCelebornFreeBytes_Value{{ printf "{%s}"
$selector }} / metrics_DeviceCelebornTotalBytes_Value{{ printf "{%s}" $selector
}}) {{ $live }})
+ threshold: {{ .Values.worker.autoscaling.diskUsage.threshold | quote }}
+{{- end }}
+{{- if .Values.worker.autoscaling.memoryUsage.enabled }}
+- type: prometheus
+ {{- /* Value, not KEDA's AverageValue default: a ratio must not be divided
by the replicas. */}}
+ metricType: Value
+ metadata:
+ serverAddress: {{ required "worker.autoscaling.prometheusAddress is
required by the built-in triggers" .Values.worker.autoscaling.prometheusAddress
}}
+ query: max(metrics_DirectMemoryUsageRatio_Value{{ printf "{%s}" $selector
}} {{ $live }})
+ threshold: {{ .Values.worker.autoscaling.memoryUsage.threshold | quote }}
+{{- end }}
+{{- end }}
{{/*
Create the name of the worker podmonitor to use
diff --git a/charts/celeborn/templates/worker/scaledobject.yaml
b/charts/celeborn/templates/worker/scaledobject.yaml
new file mode 100644
index 0000000000..02b7e48c7e
--- /dev/null
+++ b/charts/celeborn/templates/worker/scaledobject.yaml
@@ -0,0 +1,77 @@
+{{- /*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/ -}}
+
+{{- if .Values.worker.autoscaling.enabled }}
+{{- if not (or .Values.worker.autoscaling.diskUsage.enabled
.Values.worker.autoscaling.memoryUsage.enabled
.Values.worker.autoscaling.triggers) }}
+{{- fail "worker.autoscaling.enabled is true but no trigger is: enable
worker.autoscaling.diskUsage or memoryUsage, or set
worker.autoscaling.triggers" }}
+{{- end }}
+{{- /* The built-in triggers read metrics from the workers themselves, so at
zero replicas nothing can wake the statefulset. */}}
+{{- $min := .Values.worker.autoscaling.minReplicaCount }}
+{{- if and (not (kindIs "invalid" $min)) (eq (toString $min) "0") (not
.Values.worker.autoscaling.triggers) }}
+{{- fail "worker.autoscaling.minReplicaCount: 0 needs a trigger in
worker.autoscaling.triggers that does not depend on the workers' own metrics:
the built-in ones go silent once the last worker is gone, and the statefulset
could never scale back up" }}
+{{- end }}
+{{- /* One ScaledObject per worker statefulset, so each zone scales on its own
load. */}}
+{{- $zones := list dict }}
+{{- if .Values.worker.zoneAwareReplication.enabled }}
+{{- $zones = .Values.worker.zoneAwareReplication.zones }}
+{{- end }}
+{{- range $index, $zone := $zones }}
+{{- if $index }}
+---
+{{- end }}
+{{- with dict "Values" $.Values "Release" $.Release "Chart" $.Chart
"Capabilities" $.Capabilities "Template" $.Template "Files" $.Files "zone"
$zone }}
+apiVersion: keda.sh/v1alpha1
+kind: ScaledObject
+metadata:
+ name: {{ include "celeborn.worker.statefulSet.name" . }}
+ namespace: {{ .Release.Namespace }}
+ labels:
+ {{- include "celeborn.worker.labels" . | nindent 4 }}
+spec:
+ scaleTargetRef:
+ apiVersion: apps/v1
+ kind: StatefulSet
+ name: {{ include "celeborn.worker.statefulSet.name" . }}
+ {{- /* Not `default`: it treats an explicit 0 as unset, which rules out
scaling to zero. */}}
+ minReplicaCount: {{ if kindIs "invalid"
.Values.worker.autoscaling.minReplicaCount }}{{ include
"celeborn.worker.replicas" . }}{{ else }}{{
.Values.worker.autoscaling.minReplicaCount }}{{ end }}
+ maxReplicaCount: {{ .Values.worker.autoscaling.maxReplicaCount }}
+ pollingInterval: {{ .Values.worker.autoscaling.pollingInterval }}
+ cooldownPeriod: {{ .Values.worker.autoscaling.cooldownPeriod }}
+ {{- with .Values.worker.autoscaling.behavior }}
+ advanced:
+ horizontalPodAutoscalerConfig:
+ behavior:
+ {{- toYaml . | nindent 8 }}
+ {{- end }}
+ triggers:
+ {{- with include "celeborn.worker.autoscaling.defaultTriggers" . | trim }}
+ {{- . | nindent 4 }}
+ {{- end }}
+ {{- $ctx := . }}
+ {{- with .Values.worker.autoscaling.triggers }}
+ {{- /*
+ Checked before rendering: without a zone `.zone.name` yields an empty
string, so a
+ `zone=""` matcher would quietly match every worker rather than failing.
+ */}}
+ {{- if and (not $ctx.Values.worker.zoneAwareReplication.enabled) (contains
".zone" (toYaml .)) }}
+ {{- fail "a worker.autoscaling.triggers entry references .zone, which only
has a value when worker.zoneAwareReplication is enabled" }}
+ {{- end }}
+ {{- tpl (toYaml .) $ctx | nindent 4 }}
+ {{- end }}
+{{- end }}
+{{- end }}
+{{- end }}
diff --git a/charts/celeborn/templates/worker/statefulset.yaml
b/charts/celeborn/templates/worker/statefulset.yaml
index 6b7e340619..06664063dc 100644
--- a/charts/celeborn/templates/worker/statefulset.yaml
+++ b/charts/celeborn/templates/worker/statefulset.yaml
@@ -15,6 +15,55 @@ See the License for the specific language governing
permissions and
limitations under the License.
*/ -}}
+{{- /*
+One statefulset per zone when zone-aware replication is enabled, a single one
otherwise.
+The no-zone case iterates once over an empty dict, which `if .zone` treats as
unset.
+*/ -}}
+{{- $zones := list dict }}
+{{- if .Values.worker.zoneAwareReplication.enabled }}
+{{- if not .Values.worker.zoneAwareReplication.zones }}
+{{- fail "worker.zoneAwareReplication.enabled is true but
worker.zoneAwareReplication.zones is empty" }}
+{{- end }}
+{{- $zones = .Values.worker.zoneAwareReplication.zones }}
+{{- $names := list }}
+{{- range $zone := $zones }}
+{{- if not $zone.name }}
+{{- fail "every entry in worker.zoneAwareReplication.zones needs a name" }}
+{{- end }}
+{{- $names = append $names $zone.name }}
+{{- end }}
+{{- if ne (len (uniq $names)) (len $names) }}
+{{- fail "worker.zoneAwareReplication.zones has duplicate names" }}
+{{- end }}
+{{- end }}
+{{- /* The drain hook's exit request is plain HTTP with no credentials. See
README.md. */}}
+{{- if .Values.worker.drain.enabled }}
+{{- if eq (get .Values.celeborn "celeborn.worker.http.ssl.enabled" | toString)
"true" }}
+{{- fail "worker.drain does not support celeborn.worker.http.ssl.enabled: the
preStop hook calls the worker over plain HTTP" }}
+{{- end }}
+{{- if get .Values.celeborn "celeborn.worker.http.auth.supportedSchemes" |
toString | trim }}
+{{- fail "worker.drain does not support
celeborn.worker.http.auth.supportedSchemes: the preStop hook sends no
credentials" }}
+{{- end }}
+{{- end }}
+{{- range $index, $zone := $zones }}
+{{- if $index }}
+---
+{{- end }}
+{{- /* Render the body against a context carrying the zone, so `.` keeps
working below. */}}
+{{- with dict "Values" $.Values "Release" $.Release "Chart" $.Chart
"Capabilities" $.Capabilities "Template" $.Template "Files" $.Files "zone"
$zone }}
+{{- $drainMount := .Values.worker.drain.enabled }}
+{{- /*
+Computed once: used both as an env var and referenced from the start command.
The value is
+never interpolated into the shell string - `$()` or a quote in a legitimate
label would
+otherwise be evaluated by /bin/sh at startup.
+*/}}
+{{- $zoneMetricLabels := "" }}
+{{- if and $zone .Values.worker.zoneAwareReplication.metricsLabel }}
+{{- $zoneMetricLabels = printf "zone=%s" $zone.name }}
+{{- with get .Values.celeborn "celeborn.metrics.extraLabels" }}
+{{- $zoneMetricLabels = printf "%s,%s" . $zoneMetricLabels }}
+{{- end }}
+{{- end }}
apiVersion: apps/v1
kind: StatefulSet
metadata:
@@ -23,7 +72,7 @@ metadata:
labels:
{{- include "celeborn.worker.labels" . | nindent 4 }}
spec:
- replicas: {{ .Values.worker.replicas }}
+ replicas: {{ include "celeborn.worker.replicas" . }}
selector:
matchLabels:
{{- include "celeborn.worker.selectorLabels" . | nindent 6 }}
@@ -103,15 +152,51 @@ spec:
true; do
echo "waiting for master";
sleep 2;
- done && exec /opt/celeborn/sbin/start-worker.sh
+ {{- /*
+ Publish the zone as a metric label, so queries select on it rather
than on pod names.
+ The conf file is loaded with setIfMissing, so a system property wins
over it, and
+ start-worker.sh builds CELEBORN_JAVA_OPTS from
CELEBORN_WORKER_JAVA_OPTS - so this
+ composes with whatever `worker.env` already sets.
+ */}}
+ {{- $zoneOpts := "" }}
+ {{- if $zoneMetricLabels }}
+ {{- $zoneOpts = "export
CELEBORN_WORKER_JAVA_OPTS=\"${CELEBORN_WORKER_JAVA_OPTS:-}
-Dceleborn.metrics.extraLabels=$CELEBORN_ZONE_METRIC_LABELS\" && " }}
+ {{- end }}
+ done && {{ $zoneOpts }}exec /opt/celeborn/sbin/start-worker.sh
ports:
- containerPort: {{ get .Values.celeborn "celeborn.worker.http.port" |
default 9096 }}
name: metrics
protocol: TCP
- {{- with .Values.worker.env }}
+ {{- $drain := $drainMount }}
+ {{- if or $drain $zoneMetricLabels .Values.worker.env }}
env:
+ {{- with $zoneMetricLabels }}
+ - name: CELEBORN_ZONE_METRIC_LABELS
+ value: {{ . | quote }}
+ {{- end }}
+ {{- if $drain }}
+ {{- /* Not $HOSTNAME: with worker.hostNetwork that is the node name,
whose trailing
+ number would be read as an ordinal and decommission the worker on
every rollout. */}}
+ - name: POD_NAME
+ valueFrom:
+ fieldRef:
+ fieldPath: metadata.name
+ {{- /* The worker's http server binds to this address, not the
wildcard. */}}
+ - name: POD_IP
+ valueFrom:
+ fieldRef:
+ fieldPath: status.podIP
+ - name: STS_NAME
+ value: {{ include "celeborn.worker.statefulSet.name" . }}
+ - name: WORKER_HTTP_PORT
+ value: {{ get .Values.celeborn "celeborn.worker.http.port" | default
9096 | quote }}
+ - name: ALWAYS_DECOMMISSION
+ value: {{ .Values.worker.drain.alwaysDecommission | quote }}
+ {{- end }}
+ {{- with .Values.worker.env }}
{{- toYaml . | nindent 8 }}
{{- end }}
+ {{- end }}
{{- with .Values.worker.envFrom }}
envFrom:
{{- toYaml . | nindent 8 }}
@@ -126,9 +211,23 @@ spec:
- name: celeborn-conf
subPath: metrics.properties
mountPath: /opt/celeborn/conf/metrics.properties
+ {{- if $drainMount }}
+ - name: celeborn-conf
+ subPath: worker-drain.sh
+ mountPath: /opt/celeborn/drain/worker-drain.sh
+ {{- end }}
{{- with .Values.worker.volumeMounts }}
{{- toYaml . | nindent 8 }}
{{- end }}
+ {{- if $drain }}
+ {{- /* The drain logic lives in files/worker-drain.sh, shipped in the
config map. */}}
+ lifecycle:
+ preStop:
+ exec:
+ command:
+ - /bin/sh
+ - /opt/celeborn/drain/worker-drain.sh
+ {{- end }}
{{- with .Values.worker.resources }}
resources:
{{- toYaml . | nindent 10 }}
@@ -153,14 +252,24 @@ spec:
path: log4j2.xml
- key: metrics.properties
path: metrics.properties
+ {{- if $drainMount }}
+ - key: worker-drain.sh
+ path: worker-drain.sh
+ {{- end }}
{{- with .Values.worker.volumes }}
{{- toYaml . | nindent 6 }}
{{- end }}
- {{- with .Values.worker.nodeSelector }}
+ {{- $nodeSelector := merge (deepCopy (default dict (and .zone
.zone.nodeSelector))) (default dict .Values.worker.nodeSelector) }}
+ {{- with $nodeSelector }}
nodeSelector:
{{- toYaml . | nindent 8 }}
{{- end }}
- {{- with .Values.worker.affinity }}
+ {{- $affinity := .Values.worker.affinity }}
+ {{- /* Key presence, not truthiness: `affinity: {}` clears the global
one for this zone. */}}
+ {{- if and .zone (hasKey .zone "affinity") }}
+ {{- $affinity = .zone.affinity }}
+ {{- end }}
+ {{- with $affinity }}
affinity:
{{- toYaml . | nindent 8 }}
{{- end }}
@@ -187,3 +296,5 @@ spec:
{{- toYaml . | nindent 8 }}
{{- end }}
terminationGracePeriodSeconds: {{
.Values.worker.terminationGracePeriodSeconds }}
+{{- end }}
+{{- end }}
diff --git a/charts/celeborn/tests/configmap_test.yaml
b/charts/celeborn/tests/configmap_test.yaml
index 397e75bd2a..2579964e04 100644
--- a/charts/celeborn/tests/configmap_test.yaml
+++ b/charts/celeborn/tests/configmap_test.yaml
@@ -57,3 +57,21 @@ tests:
- notMatchRegex:
path: data["celeborn-defaults.conf"]
pattern: "(?m)^\\s*[^#\\s][^=\\n]*\\s+=.*"
+
+ - it: Should not ship the worker drain script as default
+ asserts:
+ - notExists:
+ path: data["worker-drain.sh"]
+
+ - it: Should ship the worker drain script when the drain is enabled
+ set:
+ worker:
+ drain:
+ enabled: true
+ asserts:
+ - matchRegex:
+ path: data["worker-drain.sh"]
+ pattern: EXIT_TYPE=DECOMMISSION
+ - matchRegex:
+ path: data["worker-drain.sh"]
+ pattern: \$\{POD_NAME:-\}
diff --git a/charts/celeborn/tests/role_test.yaml
b/charts/celeborn/tests/role_test.yaml
index 8165f6c535..24f0c4fa71 100644
--- a/charts/celeborn/tests/role_test.yaml
+++ b/charts/celeborn/tests/role_test.yaml
@@ -48,3 +48,55 @@ tests:
kind: Role
name: default
namespace: celeborn
+
+ - it: Should not grant statefulset scale access as default
+ asserts:
+ - equal:
+ path: rules
+ value:
+ - apiGroups: [""]
+ resources: ["pods"]
+ verbs: ["create", "list", "delete"]
+
+ - it: Should grant statefulset scale access when the drain is enabled
+ set:
+ worker:
+ drain:
+ enabled: true
+ asserts:
+ - equal:
+ path: rules[1]
+ value:
+ apiGroups: ["apps"]
+ resources: ["statefulsets/scale"]
+ resourceNames: ["celeborn-worker"]
+ verbs: ["get"]
+
+ - it: Should not grant statefulset scale access when the drain always
decommissions
+ set:
+ worker:
+ drain:
+ enabled: true
+ alwaysDecommission: true
+ asserts:
+ - equal:
+ path: rules
+ value:
+ - apiGroups: [""]
+ resources: ["pods"]
+ verbs: ["create", "list", "delete"]
+
+ - it: Should scope the scale access to every zone statefulset
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ - name: us-east-1b
+ drain:
+ enabled: true
+ asserts:
+ - equal:
+ path: rules[1].resourceNames
+ value: ["celeborn-worker-us-east-1a", "celeborn-worker-us-east-1b"]
diff --git a/charts/celeborn/tests/worker/scaledobject_test.yaml
b/charts/celeborn/tests/worker/scaledobject_test.yaml
new file mode 100644
index 0000000000..e5faf587ca
--- /dev/null
+++ b/charts/celeborn/tests/worker/scaledobject_test.yaml
@@ -0,0 +1,361 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+suite: Test Celeborn worker scaled object
+
+templates:
+ - worker/scaledobject.yaml
+
+release:
+ name: celeborn
+
+tests:
+ - it: Should not create a scaled object as default
+ asserts:
+ - hasDocuments:
+ count: 0
+
+ - it: Should create a scaled object if `worker.autoscaling.enabled` is true
+ set:
+ worker:
+ replicas: 4
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ maxReplicaCount: 8
+ triggers:
+ - type: prometheus
+ metadata:
+ serverAddress: http://prometheus:9090
+ query: metrics_ActiveShuffleSize_Value
+ threshold: "1000"
+ asserts:
+ - hasDocuments:
+ count: 1
+ - equal:
+ path: metadata.name
+ value: celeborn-worker
+ - equal:
+ path: spec.scaleTargetRef
+ value:
+ apiVersion: apps/v1
+ kind: StatefulSet
+ name: celeborn-worker
+ - equal:
+ path: spec.minReplicaCount
+ value: 4
+ - equal:
+ path: spec.maxReplicaCount
+ value: 8
+
+ - it: Should create one scaled object per zone, each targeting its own
statefulset
+ set:
+ worker:
+ replicas: 9
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ - name: us-east-1b
+ - name: us-east-1c
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ triggers:
+ - type: prometheus
+ metadata:
+ serverAddress: http://prometheus:9090
+ query: metrics_ActiveShuffleSize_Value
+ threshold: "1000"
+ asserts:
+ - hasDocuments:
+ count: 3
+ - equal:
+ path: metadata.name
+ value: celeborn-worker-us-east-1c
+ documentIndex: 2
+ - equal:
+ path: spec.scaleTargetRef.name
+ value: celeborn-worker-us-east-1c
+ documentIndex: 2
+ - equal:
+ path: spec.minReplicaCount
+ value: 3
+ documentIndex: 2
+
+ - it: Should render the zone into the trigger query
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1b
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ triggers:
+ - type: prometheus
+ metadata:
+ serverAddress: http://prometheus:9090
+ query:
metrics_ActiveShuffleSize_Value{pod=~"celeborn-worker-{{ .zone.name }}-.*"}
+ threshold: "1000"
+ asserts:
+ - equal:
+ path: spec.triggers[2].metadata.query
+ value:
metrics_ActiveShuffleSize_Value{pod=~"celeborn-worker-us-east-1b-.*"}
+
+ - it: Should use `worker.autoscaling.minReplicaCount` when it is set
+ set:
+ worker:
+ replicas: 9
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ minReplicaCount: 2
+ triggers:
+ - type: prometheus
+ metadata:
+ serverAddress: http://prometheus:9090
+ query: metrics_ActiveShuffleSize_Value
+ threshold: "1000"
+ asserts:
+ - equal:
+ path: spec.minReplicaCount
+ value: 2
+
+ - it: Should pass scaling behavior through to the HPA
+ set:
+ worker:
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ behavior:
+ scaleDown:
+ stabilizationWindowSeconds: 1800
+ triggers:
+ - type: prometheus
+ metadata:
+ serverAddress: http://prometheus:9090
+ query: metrics_ActiveShuffleSize_Value
+ threshold: "1000"
+ asserts:
+ - equal:
+ path:
spec.advanced.horizontalPodAutoscalerConfig.behavior.scaleDown.stabilizationWindowSeconds
+ value: 1800
+
+ - it: Should add the disk and memory usage triggers by default
+ set:
+ worker:
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ asserts:
+ - lengthEqual:
+ path: spec.triggers
+ count: 2
+ - equal:
+ path: spec.triggers[0].metricType
+ value: Value
+ - matchRegex:
+ path: spec.triggers[0].metadata.query
+ pattern:
metrics_DeviceCelebornFreeBytes_Value\{role="Worker",namespace="NAMESPACE",pod=~"celeborn-worker-\[0-9\]\+"\}
+ - equal:
+ path: spec.triggers[0].metadata.threshold
+ value: "0.70"
+ - equal:
+ path: spec.triggers[1].metricType
+ value: Value
+ - matchRegex:
+ path: spec.triggers[1].metadata.query
+ pattern:
max\(metrics_DirectMemoryUsageRatio_Value\{role="Worker",namespace="NAMESPACE",pod=~"celeborn-worker-\[0-9\]\+"\}
+ - equal:
+ path: spec.triggers[1].metadata.threshold
+ value: "0.70"
+
+ - it: Should exclude decommissioning workers from both default triggers
+ set:
+ worker:
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ asserts:
+ - matchRegex:
+ path: spec.triggers[0].metadata.query
+ pattern:
metrics_IsDecommissioningWorker_Value\{role="Worker",namespace="NAMESPACE",pod=~"celeborn-worker-\[0-9\]\+"\}
== 0
+ - matchRegex:
+ path: spec.triggers[1].metadata.query
+ pattern:
metrics_IsDecommissioningWorker_Value\{role="Worker",namespace="NAMESPACE",pod=~"celeborn-worker-\[0-9\]\+"\}
== 0
+ - notMatchRegex:
+ path: spec.triggers[0].metadata.query
+ pattern: DirectMemoryUsageRatio
+
+ - it: Should scope the default triggers to the zone's own statefulset
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ asserts:
+ - matchRegex:
+ path: spec.triggers[0].metadata.query
+ pattern:
role="Worker",namespace="NAMESPACE",pod=~"celeborn-worker-us-east-1a-\[0-9\]\+"
+ - matchRegex:
+ path: spec.triggers[1].metadata.query
+ pattern:
role="Worker",namespace="NAMESPACE",pod=~"celeborn-worker-us-east-1a-\[0-9\]\+"
+
+ - it: Should scope by pod name when the zone metric label is off
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ metricsLabel: false
+ zones:
+ - name: us-east-1a
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ asserts:
+ - matchRegex:
+ path: spec.triggers[0].metadata.query
+ pattern:
role="Worker",namespace="NAMESPACE",pod=~"celeborn-worker-us-east-1a-\[0-9\]\+"
+
+ - it: Should not select another release's workers in the same namespace
+ release:
+ name: beta
+ set:
+ worker:
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ asserts:
+ - matchRegex:
+ path: spec.triggers[0].metadata.query
+ pattern: pod=~"beta-celeborn-worker-\[0-9\]\+"
+ - notMatchRegex:
+ path: spec.triggers[0].metadata.query
+ pattern: pod=~"alpha-celeborn-worker
+
+ - it: Should not select another release's workers sharing a zone name
+ release:
+ name: beta
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ asserts:
+ - matchRegex:
+ path: spec.triggers[0].metadata.query
+ pattern: pod=~"beta-celeborn-worker-us-east-1a-\[0-9\]\+"
+
+ - it: Should drop a default trigger that is disabled
+ set:
+ worker:
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ memoryUsage:
+ enabled: false
+ asserts:
+ - lengthEqual:
+ path: spec.triggers
+ count: 1
+ - equal:
+ path: spec.triggers[0].metricType
+ value: Value
+
+ - it: Should append user triggers after the built-in ones
+ set:
+ worker:
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ triggers:
+ - type: cron
+ metadata:
+ timezone: UTC
+ start: 0 6 * * *
+ end: 0 20 * * *
+ desiredReplicas: "6"
+ asserts:
+ - lengthEqual:
+ path: spec.triggers
+ count: 3
+ - equal:
+ path: spec.triggers[2].type
+ value: cron
+
+ - it: Should fail if a built-in trigger is enabled without a prometheus
address
+ set:
+ worker:
+ autoscaling:
+ enabled: true
+ asserts:
+ - failedTemplate:
+ errorMessage: worker.autoscaling.prometheusAddress is required by
the built-in triggers
+
+ - it: Should fail if autoscaling is enabled with no trigger at all
+ set:
+ worker:
+ autoscaling:
+ enabled: true
+ diskUsage:
+ enabled: false
+ memoryUsage:
+ enabled: false
+ triggers: []
+ asserts:
+ - failedTemplate:
+ errorMessage: "worker.autoscaling.enabled is true but no trigger is:
enable worker.autoscaling.diskUsage or memoryUsage, or set
worker.autoscaling.triggers"
+
+ - it: Should allow scaling to zero with a trigger of its own
+ set:
+ worker:
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ minReplicaCount: 0
+ triggers:
+ - type: cron
+ metadata:
+ timezone: UTC
+ start: 0 6 * * *
+ end: 0 20 * * *
+ desiredReplicas: "3"
+ asserts:
+ - equal:
+ path: spec.minReplicaCount
+ value: 0
+
+ - it: Should fail scaling to zero on the built-in triggers alone
+ set:
+ worker:
+ autoscaling:
+ enabled: true
+ prometheusAddress: http://prometheus:9090
+ minReplicaCount: 0
+ asserts:
+ - failedTemplate:
+ errorMessage: "worker.autoscaling.minReplicaCount: 0 needs a trigger
in worker.autoscaling.triggers that does not depend on the workers' own
metrics: the built-in ones go silent once the last worker is gone, and the
statefulset could never scale back up"
diff --git a/charts/celeborn/tests/worker/statefulset_test.yaml
b/charts/celeborn/tests/worker/statefulset_test.yaml
index ba2b23b904..ce8661be91 100644
--- a/charts/celeborn/tests/worker/statefulset_test.yaml
+++ b/charts/celeborn/tests/worker/statefulset_test.yaml
@@ -546,3 +546,424 @@ tests:
asserts:
- notExists:
path: spec.template.spec.initContainers
+
+ - it: Should create a single worker statefulset when zone-aware replication
is disabled
+ template: worker/statefulset.yaml
+ asserts:
+ - hasDocuments:
+ count: 1
+ - equal:
+ path: metadata.name
+ value: celeborn-worker
+ - notExists:
+ path: spec.selector.matchLabels["celeborn.apache.org/zone"]
+
+ - it: Should create one worker statefulset per zone if
`worker.zoneAwareReplication` is enabled
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ replicas: 9
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ - name: us-east-1b
+ - name: us-east-1c
+ asserts:
+ - hasDocuments:
+ count: 3
+ - equal:
+ path: metadata.name
+ value: celeborn-worker-us-east-1b
+ documentIndex: 1
+ - equal:
+ path: spec.selector.matchLabels["celeborn.apache.org/zone"]
+ value: us-east-1b
+ documentIndex: 1
+ - equal:
+ path: spec.template.metadata.labels["celeborn.apache.org/zone"]
+ value: us-east-1b
+ documentIndex: 1
+ - equal:
+ path: spec.serviceName
+ value: celeborn-worker-svc
+ documentIndex: 1
+
+ - it: Should split `worker.replicas` across zones, rounding up
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ replicas: 5
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ - name: us-east-1b
+ - name: us-east-1c
+ asserts:
+ - equal:
+ path: spec.replicas
+ value: 2
+
+ - it: Should use the zone replicas if `zones[].replicas` is set
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ replicas: 9
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ replicas: 7
+ asserts:
+ - equal:
+ path: spec.replicas
+ value: 7
+
+ - it: Should merge the zone node selector over `worker.nodeSelector`
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ nodeSelector:
+ workload: celeborn
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ nodeSelector:
+ topology.kubernetes.io/zone: us-east-1a
+ asserts:
+ - equal:
+ path: spec.template.spec.nodeSelector
+ value:
+ workload: celeborn
+ topology.kubernetes.io/zone: us-east-1a
+
+ - it: Should use the zone affinity if `zones[].affinity` is set
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ affinity:
+ nodeAffinity:
+ requiredDuringSchedulingIgnoredDuringExecution:
+ nodeSelectorTerms:
+ - matchExpressions:
+ - key: topology.kubernetes.io/zone
+ operator: In
+ values:
+ - us-east-1a
+ asserts:
+ - equal:
+ path:
spec.template.spec.affinity.nodeAffinity.requiredDuringSchedulingIgnoredDuringExecution.nodeSelectorTerms[0].matchExpressions[0].values[0]
+ value: us-east-1a
+
+ - it: Should fail if zone-aware replication is enabled without zones
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones: []
+ asserts:
+ - failedTemplate:
+ errorMessage: worker.zoneAwareReplication.enabled is true but
worker.zoneAwareReplication.zones is empty
+
+ - it: Should not have a preStop drain hook as default
+ template: worker/statefulset.yaml
+ asserts:
+ - notExists:
+ path: spec.template.spec.containers[0].lifecycle
+ - notContains:
+ path: spec.template.spec.volumes[0].configMap.items
+ content:
+ key: worker-drain.sh
+ path: worker-drain.sh
+
+ - it: Should run the mounted drain script from the preStop hook
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ drain:
+ enabled: true
+ asserts:
+ - equal:
+ path: spec.template.spec.containers[0].lifecycle.preStop.exec.command
+ value:
+ - /bin/sh
+ - /opt/celeborn/drain/worker-drain.sh
+ - contains:
+ path: spec.template.spec.containers[0].volumeMounts
+ content:
+ name: celeborn-conf
+ subPath: worker-drain.sh
+ mountPath: /opt/celeborn/drain/worker-drain.sh
+ - contains:
+ path: spec.template.spec.volumes[0].configMap.items
+ content:
+ key: worker-drain.sh
+ path: worker-drain.sh
+
+ - it: Should fail the drain with worker HTTP TLS enabled
+ template: worker/statefulset.yaml
+ set:
+ celeborn:
+ celeborn.worker.http.ssl.enabled: true
+ worker:
+ drain:
+ enabled: true
+ asserts:
+ - failedTemplate:
+ errorMessage: "worker.drain does not support
celeborn.worker.http.ssl.enabled: the preStop hook calls the worker over plain
HTTP"
+
+ - it: Should fail the drain with worker HTTP authentication enabled
+ template: worker/statefulset.yaml
+ set:
+ celeborn:
+ celeborn.worker.http.auth.supportedSchemes: BASIC
+ worker:
+ drain:
+ enabled: true
+ asserts:
+ - failedTemplate:
+ errorMessage: "worker.drain does not support
celeborn.worker.http.auth.supportedSchemes: the preStop hook sends no
credentials"
+
+ - it: Should allow worker HTTP TLS when the drain is disabled
+ template: worker/statefulset.yaml
+ set:
+ celeborn:
+ celeborn.worker.http.ssl.enabled: true
+ celeborn.worker.http.auth.supportedSchemes: BASIC
+ asserts:
+ - notExists:
+ path: spec.template.spec.containers[0].lifecycle
+
+ - it: Should pass the pod, statefulset and port to the drain script
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ drain:
+ enabled: true
+ asserts:
+ - equal:
+ path: spec.template.spec.containers[0].env[0]
+ value:
+ name: POD_NAME
+ valueFrom:
+ fieldRef:
+ fieldPath: metadata.name
+ - equal:
+ path: spec.template.spec.containers[0].env[1]
+ value:
+ name: POD_IP
+ valueFrom:
+ fieldRef:
+ fieldPath: status.podIP
+ - equal:
+ path: spec.template.spec.containers[0].env[2]
+ value:
+ name: STS_NAME
+ value: celeborn-worker
+ - equal:
+ path: spec.template.spec.containers[0].env[3]
+ value:
+ name: WORKER_HTTP_PORT
+ value: "9096"
+ - equal:
+ path: spec.template.spec.containers[0].env[4]
+ value:
+ name: ALWAYS_DECOMMISSION
+ value: "false"
+
+ - it: Should tell the drain script to always decommission when asked
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ drain:
+ enabled: true
+ alwaysDecommission: true
+ asserts:
+ - equal:
+ path: spec.template.spec.containers[0].env[4].value
+ value: "true"
+
+ - it: Should point the drain script at the zone statefulset
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ drain:
+ enabled: true
+ asserts:
+ - contains:
+ path: spec.template.spec.containers[0].env
+ content:
+ name: STS_NAME
+ value: celeborn-worker-us-east-1a
+
+ - it: Should not add the preStop drain hook if `worker.drain` is disabled
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ drain:
+ enabled: false
+ asserts:
+ - notExists:
+ path: spec.template.spec.containers[0].lifecycle
+
+ - it: Should not inject POD_NAME when the drain is disabled
+ template: worker/statefulset.yaml
+ asserts:
+ - equal:
+ path: spec.template.spec.containers[0].env[0].name
+ value: CELEBORN_WORKER_MEMORY
+
+ - it: Should not set the zone metric label when zone-aware replication is off
+ template: worker/statefulset.yaml
+ asserts:
+ - notMatchRegex:
+ path: spec.template.spec.containers[0].command[4]
+ pattern: extraLabels
+
+ - it: Should publish the zone as a metric label in each zone
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ - name: us-east-1b
+ asserts:
+ - contains:
+ path: spec.template.spec.containers[0].env
+ content:
+ name: CELEBORN_ZONE_METRIC_LABELS
+ value: zone=us-east-1b
+ documentIndex: 1
+ - matchRegex:
+ path: spec.template.spec.containers[0].command[4]
+ pattern: extraLabels=\$CELEBORN_ZONE_METRIC_LABELS
+ documentIndex: 1
+
+ - it: Should append the zone to any configured extra metric labels
+ template: worker/statefulset.yaml
+ set:
+ celeborn:
+ celeborn.metrics.extraLabels: env=prod
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ asserts:
+ - contains:
+ path: spec.template.spec.containers[0].env
+ content:
+ name: CELEBORN_ZONE_METRIC_LABELS
+ value: env=prod,zone=us-east-1a
+
+ - it: Should not set the zone metric label if `metricsLabel` is false
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ metricsLabel: false
+ zones:
+ - name: us-east-1a
+ asserts:
+ - notMatchRegex:
+ path: spec.template.spec.containers[0].command[4]
+ pattern: extraLabels
+
+ - it: Should fail if a zone has no name
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - nodeSelector:
+ topology.kubernetes.io/zone: us-east-1a
+ asserts:
+ - failedTemplate:
+ errorMessage: every entry in worker.zoneAwareReplication.zones needs
a name
+
+ - it: Should fail if two zones share a name
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ - name: us-east-1a
+ asserts:
+ - failedTemplate:
+ errorMessage: worker.zoneAwareReplication.zones has duplicate names
+
+ - it: "Should honour an explicit `zones[].replicas: 0`"
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ replicas: 6
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ replicas: 0
+ - name: us-east-1b
+ asserts:
+ - equal:
+ path: spec.replicas
+ value: 0
+ documentIndex: 0
+ - equal:
+ path: spec.replicas
+ value: 3
+ documentIndex: 1
+
+ - it: Should let an empty `zones[].affinity` clear the chart affinity for
that zone
+ template: worker/statefulset.yaml
+ set:
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ affinity: {}
+ - name: us-east-1b
+ asserts:
+ - notExists:
+ path: spec.template.spec.affinity
+ documentIndex: 0
+ - exists:
+ path: spec.template.spec.affinity.podAntiAffinity
+ documentIndex: 1
+
+ - it: Should not interpolate the metric label value into the start command
+ template: worker/statefulset.yaml
+ set:
+ celeborn:
+ celeborn.metrics.extraLabels: env=$(id -u)
+ worker:
+ zoneAwareReplication:
+ enabled: true
+ zones:
+ - name: us-east-1a
+ asserts:
+ - contains:
+ path: spec.template.spec.containers[0].env
+ content:
+ name: CELEBORN_ZONE_METRIC_LABELS
+ value: env=$(id -u),zone=us-east-1a
+ - notMatchRegex:
+ path: spec.template.spec.containers[0].command[4]
+ pattern: id -u
diff --git a/charts/celeborn/values.yaml b/charts/celeborn/values.yaml
index 4bdabfde66..4fb8633959 100644
--- a/charts/celeborn/values.yaml
+++ b/charts/celeborn/values.yaml
@@ -223,11 +223,114 @@ master:
worker:
# -- Number of Celeborn worker replicas to deploy, should less than node
number.
+ # With `worker.zoneAwareReplication.enabled`, this is the total across all
zones and each
+ # zone deploys `ceil(replicas / number of zones)`.
replicas: 5
- # -- Value for terminationGracePeriodSeconds for worker pods
+ # Zone-aware replication: render one worker statefulset per zone instead of
a single one
+ # for the whole cluster, so zone membership is a property of the statefulset
rather than
+ # the outcome of scheduling. See README.md.
+ zoneAwareReplication:
+ # -- Whether to deploy one Celeborn worker statefulset per zone.
+ enabled: false
+ # -- Publish the zone as a `zone` label on every metric the worker emits,
through
+ # `celeborn.metrics.extraLabels`, so autoscaling and dashboards can select
a zone without
+ # matching on pod names. Appends to any `celeborn.metrics.extraLabels`
already configured.
+ metricsLabel: true
+ # -- Zone definitions. Redefine the whole list to change any part of it,
as YAML cannot
+ # merge list entries.
+ zones: []
+ # -- Name of the zone. Used in the statefulset name and in the
+ # `celeborn.apache.org/zone` label, so it must be a valid label value.
+ # - name: us-east-1a
+ # # -- Node selector for this zone, merged over `worker.nodeSelector`
(this one wins).
+ # nodeSelector:
+ # topology.kubernetes.io/zone: us-east-1a
+ # # -- Replicas for this zone. Overrides the even split of
`worker.replicas`.
+ # replicas: 3
+ # # -- Affinity for this zone. Replaces `worker.affinity` when set.
+ # affinity: {}
+
+ # -- Value for terminationGracePeriodSeconds for worker pods. Raise it above
+ # `celeborn.worker.decommission.forceExitTimeout` when `worker.drain` is
enabled, otherwise
+ # kubelet kills a draining worker before it finishes. See README.md.
terminationGracePeriodSeconds: 30
+ # Ask a worker to drain before Kubernetes removes it. Deleting a pod only
sends SIGTERM,
+ # which never reaches Celeborn's decommission path, so the worker would
leave with its
+ # shuffle data and any application still reading it loses that stage. Useful
with a fixed
+ # replica count as well as with `worker.autoscaling`.
+ drain:
+ # -- Whether to add a preStop hook that asks the worker to decommission
when a scale-in is
+ # removing it, and to shut down gracefully otherwise. The pod tells the
two apart by
+ # comparing its own ordinal with the statefulset's desired replica count,
and falls back to
+ # a graceful shutdown when it cannot read it. Enable this alongside
`worker.autoscaling`:
+ # without it a scale-in is destructive. Needs `rbac.create` and
+ # `celeborn.worker.graceful.shutdown.enabled`. Not supported with worker
HTTP TLS or
+ # authentication. See README.md.
+ enabled: false
+ # -- Decommission on every termination, not only a scale-in. Set this when
the worker's
+ # storage does not outlive its pod: graceful shutdown persists state to
recover from on
+ # restart, and with nothing to recover a rollout drops its in-flight
shuffles. The cost is
+ # that rollouts and node drains then wait for the drain too, bounded by
+ # `celeborn.worker.decommission.forceExitTimeout`. See README.md.
+ alwaysDecommission: false
+
+ # Autoscaling for Celeborn workers, via a KEDA ScaledObject per worker
statefulset.
+ # Requires KEDA in the cluster. See README.md.
+ autoscaling:
+ # -- Whether to create a KEDA ScaledObject for each worker statefulset.
+ enabled: false
+ # -- Lower bound. Empty means the statefulset's own replica count, which is
+ # `worker.replicas` split across zones when zone-aware replication is
enabled. 0 needs a
+ # trigger in `triggers` that does not depend on the workers' own metrics.
See README.md.
+ minReplicaCount:
+ # -- Upper bound, per statefulset (so per zone, when zones are enabled).
+ maxReplicaCount: 10
+ # -- How often KEDA evaluates the triggers, in seconds.
+ pollingInterval: 30
+ # -- Seconds before scaling to zero. Only applies with `minReplicaCount:
0`; pace
+ # ordinary scale-in with `behavior.scaleDown` instead.
+ cooldownPeriod: 300
+ # -- Scaling behaviour, passed through to the underlying HPA.
+ behavior: {}
+ # -- Prometheus-compatible endpoint the built-in triggers below query.
Required unless
+ # both are disabled.
+ prometheusAddress: ""
+
+ # Scale out when the fullest worker runs out of disk. This is the signal
that bites
+ # first: a worker that fills up stops accepting pushes.
+ # This doubles as the slot utilization target: a disk's `maxSlots` is its
total space
+ # divided by the estimated partition size, so slots and disk fill in step.
See README.md.
+ diskUsage:
+ # -- Whether to add the disk usage trigger.
+ enabled: true
+ # -- Fraction of the worker's Celeborn disk capacity in use, between 0
and 1.
+ threshold: "0.70"
+
+ # Scale out when workers run short of off-heap memory. A worker stops
accepting pushes at
+ # `celeborn.worker.directMemoryRatioToPauseReceive`, so this is the other
way a worker
+ # takes itself out of service. See README.md.
+ memoryUsage:
+ # -- Whether to add the off-heap memory trigger.
+ enabled: true
+ # -- Fraction of the worker's off-heap memory in use, between 0 and 1.
Keep it below
+ # `celeborn.worker.directMemoryRatioToPauseReceive` (0.85) so capacity
arrives before
+ # pushes pause.
+ threshold: "0.70"
+
+ # -- Extra KEDA triggers, appended to the built-in ones. Rendered through
`tpl`, so
+ # `{{ .zone.name }}` resolves to the zone of the statefulset being scaled
and one
+ # definition covers every zone. Referencing it with zones off is rejected
at render time.
+ triggers: []
+ # - type: prometheus
+ # # `Value`, not KEDA's `AverageValue` default, for a saturation ratio.
See README.md.
+ # metricType: Value
+ # metadata:
+ # serverAddress: http://prometheus.monitoring.svc.cluster.local:9090
+ # query: max(1 -
metrics_DeviceCelebornFreeBytes_Value{role="Worker",zone="{{ .zone.name }}"} /
metrics_DeviceCelebornTotalBytes_Value{role="Worker",zone="{{ .zone.name }}"})
+ # threshold: "0.7"
+
# -- Annotations for Celeborn worker pods.
annotations:
# key1: value1
