[Epic] Distributed Roboflow with Alibaba Cloud (OSS + ACK)

[zhexuany]

Overview

Transform roboflow into a distributed, fault-tolerant system using TiKV for coordination and shared-nothing compute architecture.

Design Documents

📋 Updated Roadmap: See DISTRIBUTED_DESIGN.md for the 10 Gbps throughput design.

🗺️ Issue Alignment: See ROADMAP_ALIGNMENT.md for mapping legacy phases to the new 5-phase roadmap.

Key Characteristics

• Shared-Nothing Compute: All worker pods are identical peers, no central master
• State-Externalized: All state (jobs, locks, checkpoints) in TiKV
• Reactive Streaming: Async streams with backpressure, bounded memory
• Hardware Aware: Separate CPU (parsing) and GPU (encoding) workloads
• Spot-Friendly: Checkpointing enables Spot Instance usage
• 10 Gbps Target: Designed for high-throughput processing (~1125 files/hour at 4GB each)

Architecture

┌─────────────────────────────────────────────────────────────────────────┐
│                        Alibaba Cloud Infrastructure                      │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                          │
│  ┌──────────────┐    ┌──────────────────────────────────────────────┐  │
│  │   Web UI     │───▶│              ACK/EKS Cluster                  │  │
│  │  (Monitor)   │    │                                               │  │
│  └──────────────┘    │  ┌────────────────────────────────────────┐  │  │
│                      │  │      Scanner Actor (Leader-elected)     │  │  │
│  ┌──────────────┐    │  │  - List OSS for new files               │  │  │
│  │   CLI        │───▶│  │  - Insert jobs into TiKV                │  │  │
│  │ (Submit/Mgmt)│    │  └────────────────────────────────────────┘  │  │
│  └──────────────┘    │                                               │  │
│                      │  ┌────────────────────────────────────────┐  │  │
│  ┌──────────────┐    │  │      Worker Pods (N identical peers)    │  │  │
│  │ OSS (Input)  │◀──▶│  │  - Claim jobs via TiKV CAS              │  │  │
│  │  /raw-data/  │    │  │  - Stream from OSS (range requests)     │  │  │
│  └──────────────┘    │  │  - Checkpoint progress to TiKV          │  │  │
│                      │  │  - Multipart upload to OSS              │  │  │
│  ┌──────────────┐    │  └────────────────────────────────────────┘  │  │
│  │ OSS (Output) │◀──▶│                                               │  │
│  │  /lerobot/   │    │  ┌────────────────────────────────────────┐  │  │
│  └──────────────┘    │  │           API Server                    │  │  │
│                      │  │  - REST API for job management          │  │  │
│  ┌──────────────┐    │  │  - Web UI for monitoring                │  │  │
│  │    TiKV      │◀──▶│  └────────────────────────────────────────┘  │  │
│  │   Cluster    │    │                                               │  │
│  └──────────────┘    └──────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────┘

TiKV Data Model

Key Pattern	Purpose
/jobs/{hash}	Job definition and status
/locks/{hash}	Distributed locks with TTL
/state/{hash}	Frame-level checkpoint
/heartbeat/{pod_id}	Worker liveness
/system/scanner_lock	Scanner leadership

Implementation Phases

Phase 1-3: Storage & LeRobot ✅ COMPLETE

• [Phase 1.1] Add core dependencies for storage abstraction #10, [Phase 1.2] Define Storage trait and error types #11, [Phase 1.3] Implement LocalStorage backend #23, [Phase 1.4] Implement URL/path parsing for storage backends #24, [Phase 1.5] Create StorageFactory for backend instantiation #25 - Storage Foundation
• [Phase 2.2] Implement multipart upload for large files #12, [Phase 2.1] Implement OSS/S3 backend using object_store #13, [Phase 2.3] Add retry logic and error handling for cloud operations #14, [Phase 2.4] Implement cached storage backend with local buffer #15 - Cloud Storage
• [Phase 3.1] Refactor LeRobotWriter to accept Storage backend #16, [Phase 3.2] Implement parallel episode upload with progress tracking #17 - LeRobot Writer

Phase 4: TiKV Coordination Layer ✅ COMPLETE

• [Phase 4.1] Add TiKV client and define distributed schema #40 - Add TiKV client and define schema
• [Phase 4.2] Implement distributed lock manager with TTL #41 - Implement distributed lock manager
• [Phase 4.3] Implement Scanner actor with leader election #42 - Implement Scanner actor with leader election
• [Phase 4.4] Implement Worker loop with job claiming #43 - Implement Worker loop with job claiming
• [Phase 4.5] Implement heartbeat and zombie detection #44 - Implement heartbeat and zombie detection

Phase 5: Checkpointing System ✅ COMPLETE

• [Phase 5] Frame-level checkpoint with TiKV and multipart resume #19 - Frame-level checkpoint with TiKV and multipart resume

Phase 6: Storage Enhancements ✅ COMPLETE

• [Phase 6.1] Add streaming S3 reader with range requests #45 - Add streaming S3 reader with range requests
• [Phase 6.2] Add parallel multipart uploads #46 - Add parallel multipart uploads
• [Phase 5.1] Add storage support to StreamingDatasetConverter #26 - Add storage support to StreamingDatasetConverter
• [Phase 5.2] Update CLI to accept cloud URLs #27 - Update CLI for cloud URLs

Phase 7: Pipeline Integration 🚧 IN PROGRESS

• [Phase 1] Integrate LerobotWriter with Worker.process_job() #72 - Integrate LerobotWriter with Worker.process_job() ⭐ CRITICAL
• [Phase 1] Add checkpoint save during pipeline processing #73 - Add checkpoint save during pipeline processing
• [Phase 7.1] Integrate pipeline with checkpoint hooks [PARTIALLY COMPLETE] #47 - Integrate pipeline with checkpoint hooks
• [Phase 7.2] Add graceful shutdown handling [READY 🔥] #48 - Add graceful shutdown handling

Phase 8: GPU Acceleration (Optional)

• [Phase 8] Add NVENC GPU video encoding [PARALLEL ⏳] #49 - Add NVENC video encoding support

Phase 9: Kubernetes Deployment

• [Phase 9.1] Implement long-running Worker Deployment [READY TO START] #18 - Long-running Worker Deployment
• [Phase 6.2] Create container image and Helm chart [BLOCKED by #18] #20 - Create container image and Helm chart

Phase 10: CLI & Web UI

• [Phase 10.1] Add CLI for job submission [READY TO START] #50 - CLI for job submission and management
• [Phase 10.2] Add web UI for job monitoring and management #51 - Web UI for job monitoring and management

Phase 11: Observability

• [Phase 7.1] Add Prometheus metrics for monitoring #21 - Add Prometheus metrics
• [Phase 7.2] Add structured logging with SLS integration #22 - Add structured logging

New 5-Phase Roadmap (from DISTRIBUTED_DESIGN.md)

New Phase	Description	Key Issues
Phase 1	Pipeline Integration	#72, #73, #47, #48
Phase 2	Prefetch Pipeline	(Future issues)
Phase 3	GPU Acceleration	#49
Phase 4	Production Hardening	#20, #21, #22
Phase 5	Multi-Format Support	(KPS already implemented)

Dependency Graph

Phase 1-6: ✅ COMPLETE

Phase 7: Pipeline Integration (CURRENT PRIORITY)
#72 (LerobotWriter) ─► #73 (Checkpoint Save) ─► #47 (Pipeline Hooks)
                                              └─► #48 (Graceful Shutdown)

Phase 8: GPU (parallel, optional)
#47 ─► #49 (NVENC)

Phase 9: Kubernetes
#47 + #48 ─► #18 ─► #20

Phase 10: CLI & Web UI
#40 ─► #50 (CLI) ─► #51 (Web UI)

Phase 11: Observability
#47 ─► #21 (Metrics)
#18 ─► #22 (Logging)

User Interaction

CLI Commands

# Submit jobs
roboflow submit oss://bucket/path/*.mcap --output oss://bucket/output/

# Manage jobs
roboflow jobs list --status failed
roboflow jobs retry <job-id>
roboflow jobs cancel <job-id>

# Start worker
roboflow worker

# Start UI server
roboflow ui --port 8080

Web UI Features

• Dashboard with job status overview
• Job list with filtering and search
• Job detail with progress and logs
• Retry/Cancel/Delete actions
• Worker status monitoring

Success Criteria

• Process 1000+ files from OSS without data loss
• Survive random pod terminations (Spot Instance friendly)
• Horizontal scaling with N worker pods
• Resume interrupted jobs from frame-level checkpoint
• CLI for job submission and management
• Web UI for monitoring and operations
• < 10% overhead compared to local processing
• Full observability with metrics and logs
• 10 Gbps throughput with 20-24 GPU workers

Architecture Benefits

1. Resilience: Unplug half the cluster, remaining nodes resume work
2. Scalability: Change replicas from 20 to 100, no config changes
3. Cost Efficiency: Spot Instances + GPU acceleration
4. Simplicity: One Rust binary + TiKV, no Hadoop/Spark
5. Operability: CLI and Web UI for easy management

[zhexuany]

Progress Update: Phase 7 (Pipeline Integration) ✅ COMPLETED

Phase 1-6 are complete, and Phase 7 (Pipeline Integration) is now COMPLETED:

Recently Completed (Feb 2025)

• ✅ [Phase 1] Integrate LerobotWriter with Worker.process_job() #72 - LerobotWriter integrated with Worker.process_job()

  • Streaming reader for bag/MCAP files
  • Frame processing through alignment pipeline
  • Cloud storage support (S3/OSS) via EpisodeUploadCoordinator
  • Parquet and video uploads in parallel

• ✅ [Phase 1] Add checkpoint save during pipeline processing #73 - Checkpoint save during pipeline processing

  • ProgressCallback trait for periodic checkpoint saves
  • Checkpoints every 1000 frames (configurable)
  • Checkpoints every 30 seconds (configurable)
  • Episode index and byte offset tracking

Updated Implementation Status

Phase	Status	Notes
Phase 1-3	✅ Complete	Storage & LeRobot foundation
Phase 4	✅ Complete	TiKV coordination (client, locks, scanner, worker, heartbeat)
Phase 5	✅ Complete	Checkpointing system
Phase 6	✅ Complete	Storage enhancements (streaming S3, parallel uploads)
Phase 7	✅ COMPLETE	Pipeline Integration (#72, #73)
Phase 8	⏳ TODO	GPU Acceleration (NVENC) - #49
Phase 9	⏳ TODO	Kubernetes Deployment - #18, #20
Phase 10	⏳ TODO	CLI & Web UI - #50, #51
Phase 11	⏳ TODO	Observability - #21, #22

Remaining Work for Full Distributed System

The following issues remain open and can be tackled in any order (mostly independent):

1. [Phase 7.2] Add graceful shutdown handling [READY 🔥] #48 - Graceful shutdown handling (enables K8s deployment)
2. [Phase 8] Add NVENC GPU video encoding [PARALLEL ⏳] #49 - NVENC GPU video encoding (optional performance boost)
3. [Phase 10.1] Add CLI for job submission [READY TO START] #50 - CLI for job submission and management
4. [Phase 10.2] Add web UI for job monitoring and management #51 - Web UI for job monitoring
5. [Phase 9.1] Implement long-running Worker Deployment [READY TO START] #18 - Long-running Worker Deployment (K8s)
6. [Phase 6.2] Create container image and Helm chart [BLOCKED by #18] #20 - Container image and Helm chart
7. [Phase 7.1] Add Prometheus metrics for monitoring #21 - Prometheus metrics
8. [Phase 7.2] Add structured logging with SLS integration #22 - Structured logging with SLS integration
9. [Cleanup] Remove deprecated code after distributed architecture migration #55 - Code cleanup (deferred until after distributed features complete)

Recommended Next Steps

For production deployment, prioritize:

1. [Phase 7.2] Add graceful shutdown handling [READY 🔥] #48 (Graceful shutdown) → [Phase 9.1] Implement long-running Worker Deployment [READY TO START] #18 (K8s Deployment) → [Phase 6.2] Create container image and Helm chart [BLOCKED by #18] #20 (Helm chart)
2. In parallel: [Phase 10.1] Add CLI for job submission [READY TO START] #50 (CLI) and/or [Phase 10.2] Add web UI for job monitoring and management #51 (Web UI) for operations
3. Later: [Phase 8] Add NVENC GPU video encoding [PARALLEL ⏳] #49 (NVENC) for performance optimization
4. Finally: [Phase 7.1] Add Prometheus metrics for monitoring #21, [Phase 7.2] Add structured logging with SLS integration #22 (observability) and [Cleanup] Remove deprecated code after distributed architecture migration #55 (cleanup)

[zhexuany]

🚀 MAJOR UPDATE: Gap Analysis & Roadmap Refresh

Implementation Status as of Feb 2025

After comprehensive gap analysis between DISTRIBUTED_DESIGN.md and current codebase:

✅ FULLY COMPLETED (Phase 1-7)

Component	Status	Note
Phase 1-3	✅ COMPLETE	Storage & LeRobot foundation
Phase 4	✅ COMPLETE	TiKV coordination (client, locks, scanner, worker, heartbeat)
Phase 5	✅ COMPLETE	Checkpointing system
Phase 6	✅ COMPLETE	Storage enhancements (streaming S3, parallel uploads)
Phase 7	✅ COMPLETE	Pipeline Integration (#72, #73, #47)

🔥 READY TO START (Critical Path)

#48 Graceful Shutdown [READY] ─────► #18 K8s Worker Deployment ─────► #20 Container + Helm
                                              │
                                              │
#50 CLI Commands [PARALLEL] ──────────────┘

⏳ PARALLEL / OPTIONAL

Issue	Title	Status	Priority
#76	PrefetchQueue (NEW!)	Ready	HIGH - 40% throughput gain
#49	NVENC GPU Encoding	Ready	Medium - 10x speedup
#51	Web UI	Ready	Medium - UX enhancement

⏸️ BLOCKED

Issue	Title	Blocked By
#20	Container + Helm	Waiting for #18
#21, #22	Observability	After deployment
#55	Code Cleanup	After all features

📝 CLOSED / PARTIAL

Issue	Status
#72	LerobotWriter Integration
#73	Checkpoint Save
#47	Pipeline Hooks

---

New Issue Created

#76 - [Phase 2] Implement PrefetchQueue for I/O overlap

This is a critical missing component from the design document. The PrefetchQueue:

• Downloads next job while processing current one
• Uses 2 prefetch slots per worker
• Implements parallel range-request downloader (16 connections)
• Target: 40% throughput improvement from I/O overlap

---

Recommended Execution Order

Path A: Production Deployment (Recommended)

1. [Phase 7.2] Add graceful shutdown handling [READY 🔥] #48 - Graceful Shutdown (1-2 days)
2. [Phase 9.1] Implement long-running Worker Deployment [READY TO START] #18 - K8s Worker Deployment (1 week)
3. [Phase 6.2] Create container image and Helm chart [BLOCKED by #18] #20 - Container + Helm Chart (1 week)
4. [Phase 10.1] Add CLI for job submission [READY TO START] #50 - CLI Commands (3-5 days, can be parallel)
5. [Phase 7.1] Add Prometheus metrics for monitoring #21, [Phase 7.2] Add structured logging with SLS integration #22 - Observability (after deployment)

Path B: Performance First (Alternative)

1. [Phase 2] Implement PrefetchQueue for I/O overlap #76 - PrefetchQueue (1 week) ← HIGH IMPACT
2. [Phase 8] Add NVENC GPU video encoding [PARALLEL ⏳] #49 - NVENC GPU Encoding (1-2 weeks)
3. Then follow Path A

Path C: Full Stack (Aggressive)

• Run [Phase 7.2] Add graceful shutdown handling [READY 🔥] #48, [Phase 10.1] Add CLI for job submission [READY TO START] #50, [Phase 2] Implement PrefetchQueue for I/O overlap #76 in parallel
• Follow with [Phase 9.1] Implement long-running Worker Deployment [READY TO START] #18, [Phase 6.2] Create container image and Helm chart [BLOCKED by #18] #20, [Phase 8] Add NVENC GPU video encoding [PARALLEL ⏳] #49

---

Key Architectural Notes

1. Frame-level vs Episode-level Checkpoints: Current implementation uses frame-level checkpointing. Design document suggested episode-level, but frame-level works and provides better granularity. This is an intentional architectural deviation.

2. 10 Gbps Throughput Target: Not yet validated. Requires:

   • PrefetchQueue ([Phase 2] Implement PrefetchQueue for I/O overlap #76) - for I/O overlap
   • NVENC ([Phase 8] Add NVENC GPU video encoding [PARALLEL ⏳] #49) - for GPU encoding speed
   • Multi-worker scaling testing

3. Dependencies Met: All core infrastructure is complete. Remaining work is primarily operational (K8s deployment, CLI, UI) and performance optimization (Prefetch, NVENC).