concentriq-compute-sdk 0.1.1

Python SDK for the Concentriq Compute Platform - submit and manage distributed compute jobs

Concentriq Compute SDK

Python SDK for the Concentriq Compute Platform - submit and manage distributed compute jobs with ease.

Installation

pip install concentriq-compute-sdk

Or with uv:

uv add concentriq-compute-sdk

Quick Start

Submitting Jobs

from concentriq_compute_sdk import ComputeClient, JobSpec, ResourceSpec

# Initialize client (auto-discovers Kubernetes config)
client = ComputeClient()

# Create job specification
job_spec = JobSpec(
    job_data={"input": "your-data"},
    resources=ResourceSpec(
        cpu="2000m",      # 2 CPU cores
        memory="4Gi",     # 4GB RAM
        gpu_count=1       # Optional: 1 GPU
    ),
    customer_image="your-org/processor:latest",
    timeout_seconds=3600,
    priority="high"
)

# Submit job
job_id = client.submit_job(job_spec)
print(f"Job submitted: {job_id}")

NodePool Targeting (Advanced)

Control where your jobs run by specifying architecture, capacity type, storage, and GPU requirements:

from concentriq_compute_sdk import ComputeClient, JobSpec, ResourceSpec, NodePoolSpec, GPUVRAM

# Request by VRAM tier (flexible, cost-efficient)
job_spec = JobSpec(
    job_data={"model": "llama-13b"},
    resources=ResourceSpec(cpu="8", memory="32Gi", gpu_count=1),
    nodepool=NodePoolSpec(
        gpu_memory_min_mib=GPUVRAM.STANDARD  # Gets A10G or L4 (22GB)
    )
)

# Request specific GPU model
job_spec = JobSpec(
    job_data={"model": "llama-70b-fp8"},
    resources=ResourceSpec(cpu="16", memory="64Gi", gpu_count=1),
    nodepool=NodePoolSpec(
        capacity_type="on-demand",  # Spot (default) or on-demand
        gpu_model="l40s"            # Need L40S for fp8 precision
    )
)

# ARM workload with large storage
job_spec = JobSpec(
    job_data={"task": "cpu-intensive"},
    resources=ResourceSpec(cpu="8", memory="16Gi"),
    nodepool=NodePoolSpec(
        architecture="arm64",       # amd64 (default) or arm64
        storage_tier="large"        # default (200GB), large (320GB), xlarge (500GB)
    )
)

GPU VRAM Reference:

• GPUVRAM.STANDARD (22,888 MiB / ~22GB): A10G or L4
• GPUVRAM.LARGE (45,776 MiB / ~45GB): L40S

Defaults (if nodepool not specified): amd64 + spot + default storage

Building a Controller

Controllers poll for work and submit jobs. You own the control flow.

import time
from concentriq_compute_sdk import BaseController, JobSpec, ResourceSpec

class MyController(BaseController):
    def run(self):
        while True:
            # Poll your API for work
            work_items = self.fetch_work_from_api()

            if work_items:
                # Create job specs
                job_specs = [
                    JobSpec(
                        job_data=item,
                        resources=ResourceSpec(cpu="1000m", memory="2Gi")
                    )
                    for item in work_items
                ]

                # Submit with automatic logging/tracing
                job_ids = self.submit_jobs(job_specs)
                print(f"Submitted {len(job_ids)} jobs")

            time.sleep(60)

    def fetch_work_from_api(self):
        # Your logic here
        return []

# Run controller
controller = MyController()
controller.run()

Building a Processor

Processors execute jobs. You own the execution flow.

from concentriq_compute_sdk import BaseProcessor

class MyProcessor(BaseProcessor):
    def run(self):
        try:
            # Load job data (automatic from ConfigMap)
            job_data = self.load_job_data()
            self.on_job_started(self.job_id, job_data)

            # Your processing logic
            results = self.process(job_data)

            # Report completion
            self.report_complete(results)
            self.on_job_completed(self.job_id, results)

        except Exception as e:
            self.on_job_error(self.job_id, e)
            raise

    def process(self, job_data):
        # Your business logic here
        input_data = job_data.get("input")

        # Process...

        return {"status": "completed", "output": "result"}

# Run processor
processor = MyProcessor()
processor.run()

Features

• Simple API: Only 3 methods to implement (controller) or 1 method (processor)
• Type-Safe: Pydantic models with validation
• Versioned: v1 API for backward compatibility
• Observable: Built-in OpenTelemetry tracing and structured logging
• Flexible Resources: Specify exact CPU/memory/GPU per job
• Automatic Cleanup: Jobs and ConfigMaps cleaned up via TTL and owner references

Architecture

Job Flow:

1. Your Controller runs in a loop, creating JobSpecs from your data source
2. ComputeClient submits Kubernetes Job with ConfigMap containing job data
3. Kueue schedules job when resources available
4. Your Processor loads job data from ConfigMap mount and executes
5. Your Processor reports results to your chosen destination

Versioning

The SDK uses semantic versioning with a versioned API following Kubernetes API evolution:

# Explicit v1alpha1 import (recommended for libraries)
from concentriq_compute_sdk.v1alpha1 import ComputeClient, JobSpec

# Convenience import (maps to current API, currently v1alpha1)
from concentriq_compute_sdk import ComputeClient, JobSpec

Alpha API (v1alpha1): This is the first implementation and may have breaking changes in future releases. When the API stabilizes after production use:

• Graduate to v1beta1 (feature-complete, needs validation)
• Then to v1 (battle-tested, backward compatible)
• Keep v1alpha1 for experimental features

Requirements

• Python 3.10+
• Kubernetes cluster with Kueue installed
• Valid kubeconfig or in-cluster configuration

Development

# Clone repository
git clone https://github.com/Proscia/concentriq-compute-sdk
cd concentriq-compute-sdk

# Install with dev dependencies
uv sync --all-extras

# Run tests
uv run pytest

# Lint
uv run ruff check src tests

# Type check
uv run mypy src

Environment Variables

ComputeClient Configuration

Variable	Default	Description
NAMESPACE	default	Kubernetes namespace for jobs
QUEUE_NAME	customer-queue	Kueue queue name
JOB_IMAGE	customer-job-processor:latest	Default processor image
SERVICE_ACCOUNT	default	Kubernetes service account
JOB_TTL_SECONDS	3600	Job cleanup TTL (1 hour)
LOG_LEVEL	INFO	Logging level

Processor Configuration

Variable	Description
JOB_ID	Auto-set by platform
CONFIG_MAP_NAME	Auto-set by platform

License

MIT - See LICENSE for details.

Support

• Documentation: https://github.com/Proscia/concentriq-compute-sdk
• Issues: https://github.com/Proscia/concentriq-compute-sdk/issues
• Changelog: CHANGELOG.md