k8s-maestro

A Kubernetes workflow orchestrator for Rust with minimal requirements and full power.

k8s-maestro is a Rust library that provides a high-level, type-safe API for orchestrating complex workflows on Kubernetes. It abstracts away the complexity of managing Kubernetes resources while giving you full control when you need it.

What k8s-maestro Does

k8s-maestro helps you:

• Run Jobs on Kubernetes - Execute containerized workloads as Kubernetes Jobs with automatic retries, resource limits, and cleanup
• Build Multi-step Workflows - Chain multiple jobs together with dependencies, parallel execution, and conditional logic
• Expose Services - Create Services and Ingress resources to make your workloads accessible
• Manage Configuration - Handle ConfigMaps, Secrets, and environment variables seamlessly
• Add Sidecars - Attach logging, monitoring, or utility containers to your workflows
• Persist Data - Use PersistentVolumeClaims for stateful workloads

Key Features

Installation

Add to your Cargo.toml:

[dependencies]
k8s-maestro = "1.0"

Feature Flags

k8s-maestro uses feature flags to minimize dependencies and ensure compatibility with your Kubernetes cluster.

Kubernetes Version Features (Required)

You must enable one Kubernetes version feature matching your cluster version:

k8s-maestro = { version = "1.0", features = ["k8s_v1_28"] }

Note: These features enable the kube and k8s-openapi crates with the correct API version. Using a lower version (e.g., k8s_v1_28) on a newer cluster works fine for common resources.

Optional Features

Example with exec-steps:

k8s-maestro = { version = "1.0", features = ["k8s_v1_28", "exec-steps"] }

Default Features

By default, k8s-maestro enables:

• k8s_v1_28 - Kubernetes 1.28 API support
• exec-steps - Local command execution

To disable defaults and select only what you need:

k8s-maestro = { version = "1.0", default-features = false, features = ["k8s_v1_30"] }

Migrating from v0.3.0?

See the Migration Guide for updating to the new workflow-centric API.

Client Cloning

MaestroK8sClient implements Clone, making it cheap and easy to share across multiple workflows:

let k8s_client = MaestroK8sClient::new().await?;

// Clone is lightweight - just increments an Arc counter
let workflow1 = WorkflowBuilder::new()
    .add_step(KubeJobStep::new("job1", "nginx:latest", k8s_client.clone()))
    .build();

let workflow2 = WorkflowBuilder::new()
    .add_step(KubeJobStep::new("job2", "postgres:16", k8s_client.clone()))
    .build();

// Get the underlying Client when needed
let underlying_client = k8s_client.into_inner();

Key benefits:

• Zero-cost cloning - Only increments an Arc reference counter
• Thread-safe - Share across async tasks without synchronization
• Flexible - Use into_inner() to extract the underlying Client when needed

Quick Start

Simple Job

use k8s_maestro::{MaestroClientBuilder, WorkflowBuilder};
use k8s_maestro::steps::KubeJobStep;
use k8s_maestro::clients::MaestroK8sClient;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Create the Kubernetes client - handles cluster authentication
    let k8s_client = MaestroK8sClient::new().await?;

    // Build the maestro client with a cloned k8s_client
    let client = MaestroClientBuilder::new()
        .with_namespace("default")
        .with_client(k8s_client.clone())
        .build()?;

    // Create workflow using k8s_client (moved into the step)
    let workflow = WorkflowBuilder::new()
        .with_name("my-workflow")
        .add_step(KubeJobStep::new("my-job", "nginx:latest", k8s_client))
        .build()?;

    let execution = client.execute_workflow(&workflow).await?;
    println!("Workflow executed: {:?}", execution);

    Ok(())
}

ETL Pipeline with Dependencies

use k8s_maestro::{WorkflowBuilder, MaestroClientBuilder};
use k8s_maestro::workflows::DependencyChain;
use k8s_maestro::steps::KubeJobStep;
use k8s_maestro::clients::MaestroK8sClient;

#[tokio::main]
async fn example() -> anyhow::Result<()> {
    // Create a single k8s_client to share across all steps
    let k8s_client = MaestroK8sClient::new().await?;

    // Build maestro client with a clone
    let client = MaestroClientBuilder::new()
        .with_client(k8s_client.clone())
        .build()?;

    // Clone k8s_client for each step - lightweight and efficient
    let workflow = WorkflowBuilder::new()
        .with_name("etl-pipeline")
        .add_step(KubeJobStep::new("extract", "python:3.11", k8s_client.clone()))
        .add_step(KubeJobStep::new("transform", "python:3.11", k8s_client.clone()))
        .add_step(KubeJobStep::new("load", "postgres:16", k8s_client.clone()))
        .with_parallelism(2)
        .build()?;

    client.execute_workflow(&workflow).await?;
    Ok(())
}

Service with Ingress

use k8s_maestro::{WorkflowBuilder, ServiceBuilder, ServiceType, IngressBuilder, MaestroClientBuilder};
use k8s_maestro::steps::KubeJobStep;
use k8s_maestro::clients::MaestroK8sClient;
use std::collections::BTreeMap;

#[tokio::main]
async fn example() -> anyhow::Result<()> {
    // Create and clone k8s_client for both service and workflow
    let k8s_client = MaestroK8sClient::new().await?;

    let client = MaestroClientBuilder::new()
        .with_client(k8s_client.clone())
        .build()?;

    let mut selector = BTreeMap::new();
    selector.insert("app".to_string(), "my-app".to_string());

    let service = ServiceBuilder::new()
        .with_name("my-service")
        .with_port(80, 8080, "TCP")
        .with_selector(selector)
        .with_type(ServiceType::ClusterIP)
        .build()?;

    // Use k8s_client for the workflow step
    let workflow = WorkflowBuilder::new()
        .with_name("web-workflow")
        .add_step(KubeJobStep::new("web-app", "nginx:latest", k8s_client))
        .build()?;

    client.create_service(&service).await?;
    client.execute_workflow(&workflow).await?;
    Ok(())
}

API Documentation

Examples

The examples directory contains comprehensive examples:

AI Assistant Integration

k8s-maestro includes a dedicated AI assistant skill to help you integrate the library into your projects. The k8s-maestro-integrator skill provides:

• Smart Resource Selection - Guidance on when to use Jobs, Pods, or Workflows
• Integration Patterns - API vs channel integration strategies
• Feature Detection - Automatic detection of required features based on cluster version
• Code Generation - Generate workflow and test code from descriptions
• Testing Patterns - Unit, integration, and E2E test patterns with Kind

Installation

# Using skills.sh (recommended)
npx skills add https://github.com/neurono-ml/k8s-maestro --skill k8s-maestro-integrator

Example Usage

You: I need to create an ETL pipeline that extracts data from an API,
     transforms it with Python, and loads it into PostgreSQL.

AI: [Provides complete Rust implementation with WorkflowBuilder,
     KubeJobStep, dependency chains, and proper error handling]

See the skill documentation for details.

Contributing

We welcome contributions! Please follow these steps:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Write tests for your changes (TDD approach)
4. Ensure all tests pass (cargo test --verbose)
5. Run clippy (cargo clippy)
6. Format your code (cargo fmt)
7. Commit and create a Pull Request

Testing

# Unit tests (fast, no cluster needed)
cargo test --lib

# Integration tests (requires Docker)
cargo test --test '*' -- --ignored

# Specific test
cargo test integration_test_kubernetes -- --exact

License

Dual-licensed under:

• MIT License (LICENSE-MIT)
• Apache License, Version 2.0 (LICENSE-APACHE)

Contact

• André Claudino - @andreclaudino
• Pedro Braga - @braga-rp
• Romulo Tavares - @tavaresrft

Documentation

• GitHub Pages
• Getting Started
• API Reference