Fine-tuning 70B+ parameter Large Language Models (LLMs) is no longer confined to isolated, homogeneous supercomputing clusters. As GPU availability fluctuates and cloud costs balloon, infrastructure teams are increasingly forced to embrace hybrid Kubernetes topographies—blending fixed on-premises bare-metal GPUs (e.g., NVIDIA H100/A100 nodes connected via NVLink) with dynamic, ephemeral cloud spot instances (e.g., AWS p4d.24xlarge or GCP a2-megagpu-16g).
However, running distributed PyTorch training jobs across hybrid cloud boundaries introduces severe failure modes:
- Cloud Spot Preemptions: Spot instances can be reclaimed with as little as 30 to 120 seconds of notice.
- Network Heterogeneity: Latency variations across hybrid interconnects (e.g., AWS DirectConnect vs. internal Infiniband) can cause Distributed Data Parallel (DDP) or Fully Sharded Data Parallel (FSDP) gradient synchronization bottlenecks.
- State Corruption: Interrupted gradient steps without atomic checkpointing lead to corrupted model states, lost compute time, and sky-rocketing cloud bills.
In this deep dive, we will architect a zero-downtime, fault-tolerant LLM fine-tuning pipeline using Ray on Kubernetes (KubeRay). We will cover the infrastructure topology, dynamic KubeRay dynamic cluster management, fault-tolerant state handling via asynchronous multi-tier checkpointing, and spot instance lifecycle orchestration.
1. Network & Compute Topology Architecture
To make hybrid distributed training viable, you must decouple your high-bandwidth pipeline dependencies from your stateless data-parallel bursts.
Attempting to run Tensor Parallelism (TP) or Pipeline Parallelism (PP) across a Hybrid Cloud WAN will destroy throughput due to inter-node communication delays over NCCL. Tensor Parallelism requires intra-node NVLink speeds (up to 900 GB/s on H100s).
The golden rule of hybrid LLM fine-tuning is: Pin TP and PP inside localized, homogeneous node pools, and scale Data Parallelism (DP) / Fully Sharded Data Parallelism (FSDP) across dynamic cloud boundaries.
+-----------------------------------+
| Control Plane / Storage |
| - Kubernetes API Server |
| - Ray Head Node (On-Prem Pod) |
| - Object Store (MinIO / S3 API) |
+-----------------+-----------------+
|
+------------------------+------------------------+
| |
v v
+------------------------------+ +------------------------------+
| On-Premises Dedicated Pool | | Cloud Spot Burst Pool |
| (Low Latency, Fixed Cost) | | (Dynamic, Preemptible) |
| | | |
| +--------------------------+ | DirectConnect | +--------------------------+ |
| | Worker Group: Static | | <=============> | | Worker Group: Spot | |
| | - Node 1: 8x A100 (NVLink)| | 10-100 Gbps | | - Node 3: 8x A100 (Spot) | |
| | - Node 2: 8x A100 (NVLink)| | Low Latency | | - Node 4: 8x A100 (Spot) | |
| +--------------------------+ | | +--------------------------+ |
| | | |
| - FSDP / DP Sync | | - FSDP / DP Sync |
| - Local FastNVMe Storage | | - Ephemeral Fast NVMe |
+------------------------------+ +------------------------------+
Key Architectural Components
- Ray Head Node: Executed as a highly available deployment on the on-premises control plane (guaranteeing state preservation during cloud network partitions).
- Static Worker Pool (On-Prem): High-priority Kubernetes nodes running permanent workloads. These nodes handle the baseline fine-tuning capacity.
- Burst Worker Pool (Cloud Spot): Ephemeral nodes added via the Ray Autoscaler. Managed via specialized Kubernetes node groups with taint/toleration configs to capture spot instances.
- Network Fabric: AWS DirectConnect or Azure ExpressRoute providing a flat, overlay networking model (e.g., Cilium or Calico with WireGuard encryption) permitting pod-to-pod communication across environments.
2. Infrastructure as Code: KubeRay Hybrid Cluster Spec
To deploy this setup, we utilize the KubeRay Operator. The manifest below defines a hybrid RayCluster that leverages node selectors, tolerations, and custom worker group configurations to isolate static bare-metal resources from ephemeral cloud spot capacity.
apiVersion: ray.io/v1
kind: RayCluster
metadata:
name: llm-finetune-hybrid-cluster
namespace: ray-system
spec:
rayVersion: '2.35.0'
headGroupSpec:
rayStartParams:
dashboard-host: '0.0.0.0'
num-gpus: '0' # Control plane head node uses CPU only
block: 'true'
template:
spec:
nodeSelector:
topology.ecstaticloud.com/zone: on-prem-control
containers:
- name: ray-head
image: rayproject/ray-ml:2.35.0-py310-gpu
resources:
limits:
cpu: "8"
memory: "32Gi"
requests:
cpu: "4"
memory: "16Gi"
volumeMounts:
- mountPath: /tmp/ray
name: ray-logs
volumes:
- name: ray-logs
emptyDir: {}
workerGroupSpecs:
# --- STATIC ON-PREM WORKER GROUP ---
- groupName: static-onprem-gpu
replicas: 2
minReplicas: 2
maxReplicas: 4
rayStartParams:
block: 'true'
template:
metadata:
labels:
capacity-type: static
spec:
nodeSelector:
environment: on-prem
accelerator: nvidia-a100
containers:
- name: ray-worker
image: rayproject/ray-ml:2.35.0-py310-gpu
securityContext:
capabilities:
add: ["SYS_PTRACE"] # Required for efficient NCCL shared memory ops
resources:
limits:
nvidia.com/gpu: "8"
cpu: "64"
memory: "256Gi"
requests:
nvidia.com/gpu: "8"
cpu: "32"
memory: "128Gi"
# --- DYNAMIC CLOUD SPOT WORKER GROUP ---
- groupName: dynamic-cloud-spot-gpu
replicas: 0
minReplicas: 0
maxReplicas: 8
rayStartParams:
block: 'true'
template:
metadata:
labels:
capacity-type: spot
spec:
tolerations:
- key: "aws.amazon.com/spot"
operator: "Exists"
effect: "NoSchedule"
nodeSelector:
node.kubernetes.io/instance-type: p4d.24xlarge
containers:
- name: ray-worker
image: rayproject/ray-ml:2.35.0-py310-gpu
resources:
limits:
nvidia.com/gpu: "8"
cpu: "96"
memory: "1152Gi"
requests:
nvidia.com/gpu: "8"
cpu: "64"
memory: "512Gi"
volumeMounts:
- mountPath: /tmp/scratch
name: fast-nvme
volumes:
- name: fast-nvme
hostPath:
path: /mnt/nvme-drive
3. Asynchronous Multi-Tier Fault-Tolerant Checkpointing
When training large models (e.g., Llama 3 70B), standard synchronous checkpointing to object storage (like AWS S3) blocks the execution pipeline. If a 140GB model snapshot takes 3 minutes to upload, uploading checkpoints every 100 steps will degrade overall training efficiency by up to 30%.
To achieve zero-downtime tolerance against spot preemption, we implement a two-tier asynchronous checkpointing pipeline:
- Tier 1 (Synchronous, Low Latency): Write state dicts directly to host-local NVMe scratch space (
/mnt/nvme-drive). Execution resumes immediately (takes seconds). - Tier 2 (Asynchronous Thread): A background thread streams the NVMe state to distributed persistent object storage (Ceph/MinIO/S3).
If a cloud spot instance is preempted mid-stream, the training job rolls back to the latest globally verified checkpoint in S3, while unaffected nodes pull locally cached states.
+-------------------------------------------------------------------+
| Ray Training Loop |
| |
| [Step N Completed] |
| | |
| v |
| [Tier 1: Synchronous Save] ----> Writes to Local NVMe Drives |
| | (Blocking: ~2-5 Seconds) |
| v |
| [Resume Step N+1] <--- Pipeline Continues Immediately |
| |
+---------|---------------------------------------------------------+
|
| (Spawns Non-Blocking Thread)
v
+-------------------------------------------------------------------+
| Background Process |
| |
| [Tier 2: Async Upload] --------> Streams NVMe Payload to Cloud |
| Object Storage (S3/MinIO) |
| (Non-blocking to main loop) |
+-------------------------------------------------------------------+
Here is how you write a resilient Ray Train script implementing PyTorch FSDP with dynamic checkpointing and automatic recovery.
import os
import time
import threading
import torch
import torch.distributed as dist
import ray
import ray.train
from ray.train import ScalingConfig, FailureConfig, CheckpointConfig
from ray.train.torch import TorchTrainer, TorchConfig
from torch.distributed.fsdp import (
FullyShardedDataParallel as FSDP,
FullStateDictConfig,
StateDictType,
)
# Multi-tier async uploader class
class AsyncCheckpointUploader:
def __init__(self, bucket_name):
import boto3
self.s3 = boto3.client("s3")
self.bucket = bucket_name
def upload_async(self, local_path, remote_prefix):
def _upload():
for root, _, files in os.walk(local_path):
for file in files:
full_local_path = os.path.join(root, file)
relative_path = os.path.relpath(full_local_path, local_path)
s3_key = os.path.join(remote_prefix, relative_path)
self.s3.upload_file(full_local_path, self.bucket, s3_key)
print(f"Successfully synced checkpoint to s3://{self.bucket}/{remote_prefix}")
thread = threading.Thread(target=_upload)
thread.daemon = True
thread.start()
def train_loop_per_worker(config: dict):
# Initialize Distributed Training
world_size = ray.train.get_context().get_world_size()
rank = ray.train.get_context().get_world_rank()
local_rank = ray.train.get_context().get_local_rank()
# Configure device and environment
torch.cuda.set_device(local_rank)
# Mock Large Language Model instantiation
model = torch.nn.Sequential(
torch.nn.Linear(8192, 8192),
torch.nn.ReLU(),
torch.nn.Linear(8192, 8192)
).cuda(local_rank)
# Wrap with Fully Sharded Data Parallel (FSDP)
model = FSDP(model)
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)
# Resume capability check
checkpoint = ray.train.get_checkpoint()
start_epoch = 0
if checkpoint:
with checkpoint.as_directory() as ckpt_dir:
state_dict = torch.load(os.path.join(ckpt_dir, "model.pt"))
model.load_state_dict(state_dict)
print(f"Rank {rank}: Resumed state successfully from checkpoint.")
uploader = AsyncCheckpointUploader(bucket_name=config["s3_bucket"])
for epoch in range(start_epoch, config["epochs"]):
# Simulated forward/backward pass
inputs = torch.randn(32, 8192, device=local_rank)
outputs = model(inputs)
loss = outputs.sum()
loss.backward()
optimizer.step()
optimizer.zero_grad()
# Checkpointing frequency condition
if epoch % config["checkpoint_freq"] == 0:
local_ckpt_dir = f"/tmp/scratch/ckpt_epoch_{epoch}"
os.makedirs(local_ckpt_dir, exist_ok=True)
# Extract state dict using FSDP state management
save_policy = FullStateDictConfig(offload_to_cpu=True, rank0_only=True)
with FSDP.state_dict_type(model, StateDictType.FULL_STATE_DICT, save_policy):
cpu_state = model.state_dict()
if rank == 0:
# Tier 1: Local NVMe Save (Synchronous, ultra-fast)
file_path = os.path.join(local_ckpt_dir, "model.pt")
torch.save(cpu_state, file_path)
# Tier 2: Cloud Object Storage Upload (Asynchronous, non-blocking)
uploader.upload_async(
local_path=local_ckpt_dir,
remote_prefix=f"llm-checkpoints/run-01/epoch_{epoch}"
)
# Report status to Ray Train Engine
ray.train.report(
metrics={"loss": loss.item(), "epoch": epoch},
checkpoint=ray.train.Checkpoint.from_directory(local_ckpt_dir)
)
def run_distributed_fine_tuning():
# Setup resilient cluster scaling strategy
scaling_config = ScalingConfig(
num_workers=4, # Dynamic worker allocation
use_gpu=True,
resources_per_worker={"CPU": 8, "GPU": 1},
)
# Configure auto-reboot mechanics for handling preemptions
failure_config = FailureConfig(
max_failures=10, # Auto-retry job up to 10 times upon node dropouts
fail_fast=False
)
checkpoint_config = CheckpointConfig(
num_to_keep=3,
checkpoint_score_attribute="epoch",
checkpoint_score_order="max"
)
trainer = TorchTrainer(
train_loop_per_worker=train_loop_per_worker,
train_loop_config={
"epochs": 1000,
"checkpoint_freq": 10,
"s3_bucket": "ecstaticloud-ml-checkpoints"
},
torch_config=TorchConfig(backend="nccl"),
scaling_config=scaling_config,
failure_config=failure_config,
checkpoint_config=checkpoint_config
)
print("Initiating Distributed Fault-Tolerant Fine-Tuning...")
results = trainer.fit()
print("Training finished successfully:", results)
if __name__ == "__main__":
ray.init()
run_distributed_fine_tuning()
4. Graceful Preemption Handling via Spot Instance Lifecycle Hooks
When cloud providers reclaim spot instances, the Kubernetes node receives a termination notice. For AWS EC2 Spot instances, this notice is exposed via the Metadata Service (IMDS) 120 seconds before termination.
To prevent deadlocks in NCCL Ring AllReduce communications when a spot node disappears, we must catch this signal, remove the dying worker from the active Ray cluster, trigger an emergency state save, and allow remaining nodes to continue.
Cloud Provider Node Termination Handler KubeRay / Ray Engine
| | |
|--- (120s Spot Interruption) ---->| |
| |--- (Intercepts SIGTERM) -------->|
| | |--- (Drains Worker Node)
| | |--- (Flushes Local State)
| | |--- (Fails Over Gracefully)
| | |
Implementing the Termination Handler DaemonSet
Deploy the AWS Node Termination Handler or a custom sidecar daemon in Kubernetes that listens for metadata termination events.
apiVersion: apps/v1
kind: DaemonSet
metadata:
name: spot-termination-handler
namespace: ray-system
spec:
selector:
matchLabels:
app: spot-termination-handler
template:
metadata:
labels:
app: spot-termination-handler
spec:
containers:
- name: handler
image: amazon/aws-node-termination-handler:v1.21.0
env:
- name: NODE_NAME
valueFrom:
fieldRef:
fieldPath: spec.nodeName
- name: DELETE_LOCAL_DATA
value: "true"
- name: IGNORE_DAEMON_SETS
value: "true"
- name: POD_TERMINATION_GRACE_PERIOD
value: "110" # Give Ray 110s to gracefully offload and flush state
NCCL Health Checks & Re-Initialization Strategy
Standard PyTorch FSDP crashes the entire process tree if a single node disappears during communication, throwing an unrecoverable NCCL error: unhandled system error.
To achieve true dynamic resiliency without crashing the entire training run, enable NCCL's network fault recovery mechanisms via environment variables inside the KubeRay Worker specs:
# Enable NCCL Non-blocking operations and automatic failure detection
export NCCL_ASYNC_ERROR_HANDLING=1
export NCCL_IB_DISABLE=0
export NCCL_CROSS_NIC=1
# Set aggressive timeouts for fast failure identification during spot drops
export TORCH_NCCL_HEARTBEAT_TIMEOUT_SEC=60
export TORCH_DISTRIBUTED_DEBUG=INFO
When NCCL_ASYNC_ERROR_HANDLING=1 is set, Ray Train captures worker crashes gracefully. Rather than hanging indefinitely, the cluster head node detaches the terminated worker group, triggers FailureConfig mechanics, scales down the world size or spins up a replacement spot worker, restores from the last valid S3 checkpoint, and resumes training automatically.
5. Cost-Performance Matrix: Static vs. Hybrid Spot Topology
The table below demonstrates the actual impact of deploying this architecture for a 100-hour Llama-3-70B fine-tuning job across 32 GPUs:
| Operational Metric | Pure Cloud On-Demand | Pure On-Premises Bare-Metal | Hybrid Kubernetes with Ray (Spot Burst) |
| :--- | :--- | :--- | :--- |
| Compute Topology | 4x p4d.24xlarge (AWS) | 4x 8-GPU Local Chassis | 2x Local Chassis + 2x p4d.24xlarge (Spot) |
| Hourly Compute Cost | ~$131.12 / hr | ~$18.00 / hr (Power/Colo) | ~$48.20 / hr (Blended) |
| Total Pipeline Cost | $13,112 | $1,800 | $4,820 |
| Preemption Risk | 0% | 0% | High (Mitigated by Ray) |
| Down-time / Lost Steps| None | None | < 1.2% Total Loss |
| Storage Overhead | Standard EBS | Local Storage | NVMe + Asynchronous S3 Sync |
Architectural Principles for Production Pipelines
- Strict Parallelism Separation: Always bound Tensor Parallelism (TP) inside intra-node NVLink layers. Restrict cross-cloud hybrid boundaries strictly to Data Parallel (DP/FSDP) communication across encrypted high-throughput links.
- Multi-Tier Checkpointing: Never let cloud object storage latency block GPU training loops. Write synchronously to local host NVMe drives and offload S3 sync operations to background threads.
- Decouple the Control Plane: Keep the Kubernetes Control Plane and KubeRay Head Node inside stable, non-preemptible hardware (on-premise or high-availability cloud control instances).
- Resilient NCCL Configurations: Force PyTorch and NCCL to fail fast using
NCCL_ASYNC_ERROR_HANDLING=1. AllowRay Trainto catch orchestrations failures and rebuild distributed process groups automatically.
By leveraging KubeRay as a dynamic orchestration layer over a Hybrid Kubernetes Cluster, you can scale LLM fine-tuning capacity elastically, withstand spot preemptions gracefully, and drastically reduce training infrastructure costs.