HinterBuild logoHinterBuild
DevOps · 12 min read

Internal Developer Platform for AI Teams

Internal Developer Platform for AI Teams guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable.

Muhammad Abdul Sami, author

Muhammad Abdul Sami

· 12 min read

  • RAG
  • Embeddings
  • Vector Databases
  • Evaluation

Table of Contents:

Why Internal Developer Platforms for AI Teams

Short answer: ML teams waste 40-60% of time on infrastructure—provisioning GPUs, configuring experiments, deploying models, debugging infrastructure. Internal Developer Platforms (IDPs) provide self-service infrastructure, golden paths, and automated workflows so data scientists ship models, not YAML.

A Series B startup had 15 data scientists filing DevOps tickets for every GPU cluster, model deployment, and experiment—average lead time: 3-5 days. We built them an IDP: self-service portal for infrastructure, one-click deployments, automated monitoring. New lead time: 15 minutes. Data scientist productivity tripled.

Key Takeaways:

  • Self-service infrastructure—data scientists provision without tickets
  • Golden paths—opinionated templates for common workflows
  • Platform abstractions—hide Kubernetes/cloud complexity
  • Built-in observability—automatic monitoring, logging, alerting
  • Developer experience—CLI, portal, IDE plugins for ML workflows

For production AI systems, IDPs eliminate infrastructure friction.


IDP Architecture Principles for ML

Platform = Abstraction + Automation + Self-Service.

┌─────────────────────────────────────────────────────────┐
│              Data Scientist Interface                   │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐  │
│  │  Web Portal  │  │  CLI Tool    │  │ IDE Plugins  │  │
│  │  (Backstage) │  │  (mlctl)     │  │ (VSCode)     │  │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘  │
│         │                 │                 │           │
└─────────┼─────────────────┼─────────────────┼───────────┘
          │                 │                 │
          ▼                 ▼                 ▼
┌─────────────────────────────────────────────────────────┐
│            Platform Orchestration Layer                 │
│                                                         │
│  ┌──────────────────────────────────────────────────┐  │
│  │            Platform API Gateway                   │  │
│  │  - Authentication (OAuth, OIDC)                   │  │
│  │  - Authorization (RBAC, ABAC)                     │  │
│  │  - Request validation & routing                   │  │
│  └──────────────────┬───────────────────────────────┘  │
│                     │                                   │
│  ┌─────────────────┴──────────────────────────────┐   │
│  │                                                 │   │
│  │  ┌──────────────┐  ┌──────────────────────┐   │   │
│  │  │ GPU Manager  │  │ Model Deployer       │   │   │
│  │  │ (Crossplane) │  │ (ArgoCD + Rollouts)  │   │   │
│  │  └──────────────┘  └──────────────────────┘   │   │
│  │                                                 │   │
│  │  ┌──────────────┐  ┌──────────────────────┐   │   │
│  │  │ Experiment   │  │ Observability Stack  │   │   │
│  │  │ Tracker      │  │ (Prometheus, Loki)   │   │   │
│  │  │ (MLflow)     │  └──────────────────────┘   │   │
│  │  └──────────────┘                              │   │
│  └────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────┘
          │                 │                 │
          ▼                 ▼                 ▼
┌─────────────────────────────────────────────────────────┐
│           Infrastructure Layer                          │
│  ┌──────────┐  ┌──────────┐  ┌──────────────────────┐  │
│  │  K8s     │  │  Vector  │  │  Object Storage      │  │
│  │  Clusters│  │  DBs     │  │  (S3, GCS)           │  │
│  └──────────┘  └──────────┘  └──────────────────────┘  │
└─────────────────────────────────────────────────────────┘

Core principles:

  1. Abstraction - Hide infrastructure complexity behind APIs
  2. Golden paths - Opinionated workflows for 80% use cases
  3. Self-service - No DevOps tickets for standard operations
  4. Standardization - Consistent patterns across teams
  5. Observability - Built-in monitoring, logging, tracing
yaml
apiVersion: ml.company.com/v1
kind: TrainingJob
metadata:
  name: fraud-model-v3
spec:
  framework: pytorch
  gpu: 1x-A100
  code: git@github.com:company/fraud-detection.git
  dataset: s3://company-data/fraud/train/
  hyperparameters:
    learning_rate: 0.001
    batch_size: 32
  
  # Platform handles everything else:
  # - GPU provisioning
  # - Environment setup
  # - Experiment tracking
  # - Model registry
  # - Monitoring

Build on Kubernetes infrastructure.


Self-Service GPU Provisioning

Data scientists request GPUs via portal or CLI—infrastructure provisions automatically.

python
# mlctl - CLI for ML platform
import click
import requests
from typing import Optional

class MLPlatformClient:
    """Client for ML platform API."""
    
    def __init__(self, api_url: str, token: str):
        self.api_url = api_url
        self.headers = {"Authorization": f"Bearer {token}"}
    
    def create_gpu_cluster(
        self,
        name: str,
        gpu_type: str,
        count: int,
        max_count: Optional[int] = None,
    ) -> dict:
        """Request GPU cluster provisioning."""
        payload = {
            "name": name,
            "gpu_type": gpu_type,
            "min_nodes": count,
            "max_nodes": max_count or count * 2,
        }
        
        response = requests.post(
            f"{self.api_url}/v1/gpu-clusters",
            json=payload,
            headers=self.headers,
        )
        response.raise_for_status()
        return response.json()
    
    def get_cluster_status(self, name: str) -> dict:
        """Check cluster provisioning status."""
        response = requests.get(
            f"{self.api_url}/v1/gpu-clusters/{name}",
            headers=self.headers,
        )
        response.raise_for_status()
        return response.json()
    
    def list_available_gpus(self) -> list:
        """List available GPU types."""
        response = requests.get(
            f"{self.api_url}/v1/gpu-types",
            headers=self.headers,
        )
        response.raise_for_status()
        return response.json()

@click.group()
def cli():
    """ML Platform CLI."""
    pass

@cli.command()
@click.option('--name', required=True, help='Cluster name')
@click.option('--gpu', required=True, type=click.Choice(['t4', 'v100', 'a100']))
@click.option('--count', default=1, help='Number of GPUs')
def create_cluster(name: str, gpu: str, count: int):
    """Create GPU cluster for training."""
    client = MLPlatformClient(
        api_url="https://ml-platform.company.com",
        token=get_token(),
    )
    
    click.echo(f"Creating GPU cluster '{name}' with {count}x {gpu.upper()}...")
    
    result = client.create_gpu_cluster(
        name=name,
        gpu_type=gpu,
        count=count,
    )
    
    click.echo(f"✓ Cluster requested: {result['id']}")
    click.echo(f"Status: {result['status']}")
    click.echo(f"Estimated ready time: {result['eta']} minutes")
    
    # Wait for provisioning
    import time
    while True:
        status = client.get_cluster_status(name)
        
        if status['status'] == 'ready':
            click.echo(f"✓ Cluster ready!")
            click.echo(f"Connect: kubectl config use-context {status['kubeconfig']}")
            break
        elif status['status'] == 'failed':
            click.echo(f"✗ Provisioning failed: {status['error']}")
            break
        
        click.echo(f"  Status: {status['status']}... ({status['progress']}%)")
        time.sleep(30)

@cli.command()
def list_gpus():
    """List available GPU types."""
    client = MLPlatformClient(
        api_url="https://ml-platform.company.com",
        token=get_token(),
    )
    
    gpus = client.list_available_gpus()
    
    click.echo("Available GPU types:")
    for gpu in gpus:
        click.echo(f"  {gpu['name']:10} - ${gpu['cost_per_hour']:.2f}/hr - {gpu['memory']} VRAM")

def get_token() -> str:
    """Get auth token from config."""
    import os
    return os.environ.get("ML_PLATFORM_TOKEN", "")

if __name__ == "__main__":
    cli()

Usage:

bash
# List available GPUs
mlctl list-gpus
# Available GPU types:
#   t4         - $0.35/hr - 16GB VRAM
#   v100       - $2.48/hr - 32GB VRAM
#   a100       - $3.67/hr - 80GB VRAM

# Create training cluster
mlctl create-cluster --name fraud-training --gpu a100 --count 4
# Creating GPU cluster 'fraud-training' with 4x A100...
# ✓ Cluster requested: cls-7f3d9a
# Status: provisioning
# Estimated ready time: 12 minutes

# Cluster provisions via Crossplane in background

Backend API with Crossplane integration.


Model Deployment Pipeline

One-command model deployment from registry to production.

python
# platform_api/deployments.py
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import Optional
import kubernetes
from jinja2 import Template

app = FastAPI(title="ML Platform API")

class ModelDeploymentRequest(BaseModel):
    """Model deployment specification."""
    model_name: str
    model_version: str
    replicas: int = 3
    resources: dict = {
        "cpu": "2000m",
        "memory": "4Gi",
    }
    autoscaling: Optional[dict] = {
        "min_replicas": 2,
        "max_replicas": 10,
        "target_cpu": 70,
    }
    environment: str = "staging"  # or 'production'

class DeploymentManager:
    """Manage model deployments."""
    
    def __init__(self):
        kubernetes.config.load_incluster_config()
        self.k8s_apps = kubernetes.client.AppsV1Api()
        self.k8s_core = kubernetes.client.CoreV1Api()
    
    def deploy_model(
        self,
        request: ModelDeploymentRequest,
        user: str,
    ) -> dict:
        """Deploy model to Kubernetes."""
        # Generate manifests from template
        manifests = self._generate_manifests(request, user)
        
        # Apply via Kubernetes API
        namespace = f"ml-{request.environment}"
        
        # Create deployment
        deployment = kubernetes.client.V1Deployment(
            **manifests['deployment']
        )
        
        try:
            self.k8s_apps.create_namespaced_deployment(
                namespace=namespace,
                body=deployment,
            )
        except kubernetes.client.exceptions.ApiException as e:
            if e.status == 409:  # Already exists
                self.k8s_apps.patch_namespaced_deployment(
                    name=deployment.metadata.name,
                    namespace=namespace,
                    body=deployment,
                )
        
        # Create service
        service = kubernetes.client.V1Service(
            **manifests['service']
        )
        
        try:
            self.k8s_core.create_namespaced_service(
                namespace=namespace,
                body=service,
            )
        except kubernetes.client.exceptions.ApiException as e:
            if e.status != 409:
                raise
        
        # Create HPA if autoscaling enabled
        if request.autoscaling:
            hpa = self._create_hpa(request, namespace)
            # Apply HPA...
        
        return {
            "deployment_id": f"{request.model_name}-{request.model_version}",
            "namespace": namespace,
            "status": "deploying",
            "endpoint": f"https://{request.model_name}.{request.environment}.company.com",
        }
    
    def _generate_manifests(
        self,
        request: ModelDeploymentRequest,
        user: str,
    ) -> dict:
        """Generate Kubernetes manifests."""
        deployment_template = Template("""
apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ model_name }}-{{ model_version }}
  labels:
    app: {{ model_name }}
    version: {{ model_version }}
    deployed-by: {{ user }}
spec:
  replicas: {{ replicas }}
  selector:
    matchLabels:
      app: {{ model_name }}
      version: {{ model_version }}
  template:
    metadata:
      labels:
        app: {{ model_name }}
        version: {{ model_version }}
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9090"
    spec:
      containers:
      - name: model-server
        image: company/model-server:latest
        env:
        - name: MODEL_NAME
          value: {{ model_name }}
        - name: MODEL_VERSION
          value: "{{ model_version }}"
        - name: MODEL_URI
          value: "s3://company-models/{{ model_name }}/{{ model_version }}"
        
        ports:
        - containerPort: 8080
          name: http
        - containerPort: 9090
          name: metrics
        
        resources:
          requests:
            cpu: {{ resources.cpu }}
            memory: {{ resources.memory }}
          limits:
            cpu: {{ resources.cpu }}
            memory: {{ resources.memory }}
        
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
        
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 10
""")
        
        return {
            'deployment': deployment_template.render(
                model_name=request.model_name,
                model_version=request.model_version,
                replicas=request.replicas,
                resources=request.resources,
                user=user,
            ),
            'service': self._generate_service_manifest(request),
        }

manager = DeploymentManager()

@app.post("/v1/deployments")
async def deploy_model(
    request: ModelDeploymentRequest,
    user: str = Depends(get_current_user),
):
    """Deploy model endpoint."""
    # Validate model exists in registry
    if not model_exists(request.model_name, request.model_version):
        raise HTTPException(404, "Model not found in registry")
    
    # Check user permissions
    if not user_can_deploy(user, request.environment):
        raise HTTPException(403, "Insufficient permissions")
    
    # Deploy
    result = manager.deploy_model(request, user)
    
    return result

@app.get("/v1/deployments/{deployment_id}")
async def get_deployment_status(deployment_id: str):
    """Get deployment status."""
    # Query Kubernetes for deployment status
    # Return health, metrics, etc.
    pass

CLI wrapper:

bash
# Deploy from CLI
mlctl deploy \
  --model fraud-detection \
  --version v2.3.0 \
  --environment staging \
  --replicas 3

# Platform handles:
# - Pulls model from MLflow registry
# - Generates Kubernetes manifests
# - Deploys via ArgoCD
# - Configures monitoring
# - Sets up autoscaling
# - Creates ingress

# Output:
# ✓ Model deployed to staging
# Endpoint: https://fraud-detection.staging.company.com
# Monitoring: https://grafana.company.com/d/model-fraud-detection

Integrate with ArgoCD.


Experiment Tracking Portal

Centralized experiment tracking with MLflow backend.

python
# platform_portal/experiments.py
from fastapi import FastAPI, Request
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
import mlflow
from mlflow.tracking import MlflowClient

app = FastAPI()
templates = Jinja2Templates(directory="templates")

mlflow.set_tracking_uri("https://mlflow.company.com")
client = MlflowClient()

@app.get("/experiments", response_class=HTMLResponse)
async def list_experiments(request: Request):
    """List all experiments."""
    experiments = client.search_experiments()
    
    # Enrich with latest run info
    for exp in experiments:
        runs = client.search_runs(
            experiment_ids=[exp.experiment_id],
            order_by=["start_time DESC"],
            max_results=5,
        )
        exp.latest_runs = runs
    
    return templates.TemplateResponse(
        "experiments.html",
        {
            "request": request,
            "experiments": experiments,
        },
    )

@app.get("/experiments/{experiment_id}")
async def experiment_detail(request: Request, experiment_id: str):
    """Experiment detail view."""
    experiment = client.get_experiment(experiment_id)
    
    # Get all runs
    runs = client.search_runs(
        experiment_ids=[experiment_id],
        order_by=["metrics.f1_score DESC"],
    )
    
    # Extract metrics for comparison
    metrics_comparison = []
    for run in runs:
        metrics_comparison.append({
            "run_id": run.info.run_id,
            "run_name": run.data.tags.get("mlflow.runName", run.info.run_id[:8]),
            "accuracy": run.data.metrics.get("accuracy", 0),
            "f1_score": run.data.metrics.get("f1_score", 0),
            "precision": run.data.metrics.get("precision", 0),
            "recall": run.data.metrics.get("recall", 0),
            "params": run.data.params,
            "status": run.info.status,
            "start_time": run.info.start_time,
        })
    
    return templates.TemplateResponse(
        "experiment_detail.html",
        {
            "request": request,
            "experiment": experiment,
            "runs": metrics_comparison,
        },
    )

@app.post("/experiments/{experiment_id}/promote")
async def promote_to_production(experiment_id: str, run_id: str):
    """Promote best run to production."""
    # Register model in MLflow
    run = client.get_run(run_id)
    model_uri = f"runs:/{run_id}/model"
    
    model_details = mlflow.register_model(
        model_uri,
        name=run.data.tags.get("model_name", "unknown"),
    )
    
    # Transition to Production stage
    client.transition_model_version_stage(
        name=model_details.name,
        version=model_details.version,
        stage="Production",
    )
    
    # Trigger deployment via platform API
    # (calls deployment API from previous section)
    
    return {
        "status": "promoted",
        "model": model_details.name,
        "version": model_details.version,
    }

HTML template (excerpt):

html
<!-- templates/experiment_detail.html -->
<div class="experiment-detail">
  <h1>{{ experiment.name }}</h1>
  
  <div class="metrics-comparison">
    <table>
      <thead>
        <tr>
          <th>Run</th>
          <th>Accuracy</th>
          <th>F1 Score</th>
          <th>Precision</th>
          <th>Recall</th>
          <th>Status</th>
          <th>Actions</th>
        </tr>
      </thead>
      <tbody>
        {% for run in runs %}
        <tr>
          <td><a href="/runs/{{ run.run_id }}">{{ run.run_name }}</a></td>
          <td>{{ "%.4f"|format(run.accuracy) }}</td>
          <td>{{ "%.4f"|format(run.f1_score) }}</td>
          <td>{{ "%.4f"|format(run.precision) }}</td>
          <td>{{ "%.4f"|format(run.recall) }}</td>
          <td>{{ run.status }}</td>
          <td>
            <button onclick="promoteRun('{{ run.run_id }}')">
              Promote to Production
            </button>
          </td>
        </tr>
        {% endfor %}
      </tbody>
    </table>
  </div>
</div>

Connect to data pipelines.


Golden Paths & Templates

Opinionated workflows for 80% of use cases—reduce decisions, increase consistency.

yaml
# templates/training-job-pytorch.yaml
apiVersion: ml.company.com/v1
kind: TrainingJob
metadata:
  name: {{ .Values.name }}
  labels:
    framework: pytorch
    team: {{ .Values.team }}
spec:
  # Pre-configured GPU setup
  compute:
    gpu: {{ .Values.gpu | default "1x-T4" }}
    cpu: {{ .Values.cpu | default "4000m" }}
    memory: {{ .Values.memory | default "16Gi" }}
  
  # Golden path: Git + Docker
  source:
    git:
      repo: {{ .Values.git_repo }}
      branch: {{ .Values.git_branch | default "main" }}
    
    docker:
      image: company/pytorch-training:latest
      command: ["python", "train.py"]
      args: {{ .Values.training_args | toYaml | nindent 8 }}
  
  # Automatic experiment tracking
  tracking:
    mlflow:
      experiment: {{ .Values.experiment_name }}
      auto_log: true
  
  # Automatic model registry
  outputs:
    model:
      registry: mlflow
      name: {{ .Values.model_name }}
      auto_register: true
  
  # Built-in monitoring
  monitoring:
    prometheus: true
    grafana_dashboard: true
    alerts:
      - name: training-failure
        condition: status == "failed"
      - name: low-gpu-utilization
        condition: gpu_utilization < 50

CLI scaffolding:

bash
# Create new training job from template
mlctl new training-job \
  --name fraud-v3 \
  --framework pytorch \
  --template classification

# Generates:
# - Training job YAML
# - Training script template
# - Config file structure
# - Documentation

# Output:
# Created training job scaffold:
#   fraud-v3/
#     ├── job.yaml           # Training job config
#     ├── train.py           # Training script template
#     ├── config.yaml        # Hyperparameters
#     ├── requirements.txt   # Dependencies
#     └── README.md          # Usage guide

# Edit files, then submit:
mlctl submit fraud-v3/

Template library:

python
# platform/templates.py
TEMPLATES = {
    "pytorch-classification": {
        "description": "Image/text classification with PyTorch",
        "files": {
            "train.py": pytorch_classification_template,
            "job.yaml": training_job_template,
            "config.yaml": hyperparameter_template,
        },
    },
    "pytorch-regression": {
        "description": "Regression models with PyTorch",
        "files": {...},
    },
    "tensorflow-nlp": {
        "description": "NLP models with TensorFlow",
        "files": {...},
    },
    "inference-api": {
        "description": "FastAPI model serving endpoint",
        "files": {...},
    },
}

def scaffold_from_template(template_name: str, output_dir: str, **kwargs):
    """Generate project from template."""
    template = TEMPLATES[template_name]
    
    for filename, template_content in template["files"].items():
        rendered = Template(template_content).render(**kwargs)
        
        output_path = Path(output_dir) / filename
        output_path.write_text(rendered)
    
    print(f"✓ Created project from '{template_name}' template")

Developer Portal Implementation

Backstage-based portal for unified ML platform access.

yaml
# backstage/app-config.yaml
app:
  title: ML Platform
  baseUrl: https://portal.company.com

backend:
  baseUrl: https://portal.company.com
  listen:
    port: 7007
  
  database:
    client: pg
    connection:
      host: postgres.platform.svc
      port: 5432
      user: backstage
      password: ${POSTGRES_PASSWORD}

catalog:
  providers:
    github:
      company:
        organization: 'company'
        catalogPath: '/catalog-info.yaml'
        filters:
          branch: 'main'
          repository: '.*-ml$'  # Only ML repos
  
  locations:
    # Register ML platform components
    - type: file
      target: ./catalog/ml-platform.yaml

# ML Platform plugin config
mlPlatform:
  apiUrl: https://ml-api.company.com
  mlflowUrl: https://mlflow.company.com
  grafanaUrl: https://grafana.company.com

Custom Backstage plugin for ML workflows:

typescript
// plugins/ml-platform/src/components/GPUClusterPage.tsx
import React, { useState, useEffect } from 'react';
import { useApi } from '@backstage/core-plugin-api';
import { mlPlatformApiRef } from '../api';

export const GPUClusterPage = () => {
  const api = useApi(mlPlatformApiRef);
  const [clusters, setClusters] = useState([]);
  
  useEffect(() => {
    api.listGPUClusters().then(setClusters);
  }, [api]);
  
  const handleCreateCluster = async (config: ClusterConfig) => {
    await api.createGPUCluster(config);
    // Refresh list
  };
  
  return (
    <div>
      <h1>GPU Clusters</h1>
      
      <ClusterList clusters={clusters} />
      
      <CreateClusterForm onSubmit={handleCreateCluster} />
    </div>
  );
};

// API client
class MLPlatformApiClient implements MLPlatformApi {
  async listGPUClusters(): Promise<GPUCluster[]> {
    const response = await fetch(`${baseUrl}/v1/gpu-clusters`);
    return response.json();
  }
  
  async createGPUCluster(config: ClusterConfig): Promise<GPUCluster> {
    const response = await fetch(`${baseUrl}/v1/gpu-clusters`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(config),
    });
    return response.json();
  }
  
  // More methods...
}

Catalog entities for ML components:

yaml
# catalog/ml-models.yaml
apiVersion: backstage.io/v1alpha1
kind: Component
metadata:
  name: fraud-detection-model
  description: Real-time fraud detection ML model
  tags:
    - ml-model
    - production
    - pytorch
  annotations:
    mlflow.io/model-uri: models:/fraud-detection/production
    github.com/project-slug: company/fraud-detection
spec:
  type: ml-model
  lifecycle: production
  owner: ml-team
  
  # Custom ML fields
  model:
    framework: pytorch
    version: v2.3.0
    accuracy: 0.947
    latency_p95: 120ms
  
  # Links
  links:
    - url: https://mlflow.company.com/models/fraud-detection
      title: MLflow Registry
    - url: https://grafana.company.com/d/fraud-detection
      title: Monitoring Dashboard
    - url: https://fraud-detection.prod.company.com
      title: Production Endpoint

Build with backend API engineering.


Production Patterns & Best Practices

Pattern 1: Multi-Tenancy with Namespaces

yaml
# Each team gets isolated namespace
apiVersion: v1
kind: Namespace
metadata:
  name: ml-team-fraud
  labels:
    team: fraud
    platform-managed: "true"

---
# Resource quotas per team
apiVersion: v1
kind: ResourceQuota
metadata:
  name: team-quota
  namespace: ml-team-fraud
spec:
  hard:
    requests.nvidia.com/gpu: "8"      # Max 8 GPUs
    requests.cpu: "32"                # Max 32 CPUs
    requests.memory: "128Gi"          # Max 128GB RAM
    persistentvolumeclaims: "10"
    services.loadbalancers: "2"

---
# RBAC for team
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: fraud-team-admin
  namespace: ml-team-fraud
subjects:
- kind: Group
  name: ml-team-fraud
  apiGroup: rbac.authorization.k8s.io
roleRef:
  kind: ClusterRole
  name: ml-team-role
  apiGroup: rbac.authorization.k8s.io

Pattern 2: Cost Tracking & Chargeback

python
# platform/cost_tracking.py
from dataclasses import dataclass
from datetime import datetime, timezone
import prometheus_client

@dataclass
class ResourceUsage:
    """Track resource usage for billing."""
    team: str
    namespace: str
    gpu_hours: float
    cpu_hours: float
    storage_gb_hours: float
    timestamp: datetime

class CostTracker:
    """Track and report infrastructure costs."""
    
    # GPU pricing
    GPU_COSTS = {
        "t4": 0.35,      # $/hour
        "v100": 2.48,
        "a100": 3.67,
    }
    
    # CPU/memory pricing
    CPU_COST = 0.03    # $/vCPU/hour
    MEMORY_COST = 0.004  # $/GB/hour
    
    def calculate_team_cost(
        self,
        team: str,
        start_date: datetime,
        end_date: datetime,
    ) -> dict:
        """Calculate team's infrastructure cost."""
        usage = self._query_prometheus(team, start_date, end_date)
        
        # Calculate costs
        gpu_cost = sum(
            usage['gpu_hours'][gpu_type] * self.GPU_COSTS[gpu_type]
            for gpu_type in usage['gpu_hours']
        )
        
        cpu_cost = usage['cpu_hours'] * self.CPU_COST
        memory_cost = usage['memory_gb_hours'] * self.MEMORY_COST
        
        total = gpu_cost + cpu_cost + memory_cost
        
        return {
            "team": team,
            "period": {
                "start": start_date.isoformat(),
                "end": end_date.isoformat(),
            },
            "breakdown": {
                "gpu": gpu_cost,
                "cpu": cpu_cost,
                "memory": memory_cost,
            },
            "total": total,
        }
    
    def generate_monthly_report(self) -> list[dict]:
        """Generate cost report for all teams."""
        teams = self._list_teams()
        
        # Current month
        start = datetime.now(timezone.utc).replace(day=1, hour=0, minute=0)
        end = datetime.now(timezone.utc)
        
        return [
            self.calculate_team_cost(team, start, end)
            for team in teams
        ]

Pattern 3: Compliance & Audit Trail

python
# platform/audit.py
from typing import Any
import json

class AuditLogger:
    """Audit all platform actions."""
    
    async def log_action(
        self,
        user: str,
        action: str,
        resource_type: str,
        resource_id: str,
        details: dict[str, Any],
    ) -> None:
        """Log platform action for compliance."""
        entry = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "user": user,
            "action": action,
            "resource": {
                "type": resource_type,
                "id": resource_id,
            },
            "details": details,
        }
        
        # Write to audit log
        await self._write_to_storage(entry)
        
        # Send to SIEM if critical action
        if self._is_critical(action):
            await self._send_to_siem(entry)
    
    def _is_critical(self, action: str) -> bool:
        """Check if action requires SIEM notification."""
        return action in [
            "deploy_production",
            "delete_model",
            "modify_permissions",
            "access_pii_data",
        ]

# Usage in API endpoints
@app.post("/v1/deployments")
async def deploy_model(
    request: ModelDeploymentRequest,
    user: str = Depends(get_current_user),
    audit: AuditLogger = Depends(get_audit_logger),
):
    # Deploy model
    result = await manager.deploy_model(request, user)
    
    # Audit log
    await audit.log_action(
        user=user,
        action="deploy_model",
        resource_type="model_deployment",
        resource_id=result['deployment_id'],
        details={
            "model": request.model_name,
            "version": request.model_version,
            "environment": request.environment,
        },
    )
    
    return result

Deploy on cloud infrastructure.


Related implementation guides:

Primary references: official documentation, official documentation, official documentation, official documentation.

Internal Developer Platform for AI Teams Decision Table

DecisionPrefer the simpler path whenAdd operational complexity when
ArchitectureOne component can own the contract and stateIndependent scaling or fault isolation is required
RolloutOffline replay covers the meaningful casesLive behavior requires shadow traffic and a canary
RecoveryA failed operation is safe to repeatPartial effects require idempotency or reconciliation
MeasurementOne service objective represents user impactQuality, latency, and cost need separate gates

Operating Internal Developer Platform for AI Teams as a System

The implementation is only one part of Internal Developer Platform for AI Teams. A production design also needs an explicit contract for inputs, outputs, ownership, and failure behavior. Write that contract before selecting a library. It should identify which component validates input, where state lives, what may be retried, and which result is authoritative when two components disagree. This prevents a convenient prototype boundary from silently becoming the long-term architecture.

Start with a representative baseline. Capture request shape, traffic distribution, dependency latency, error classes, and the quality signal users actually care about. Averages hide the cases that cause incidents, so keep percentiles and segment measurements by workload type. Record the configuration and dataset version beside every result. Without that context, a faster or more accurate run cannot be reproduced and should not be used to approve a rollout.

Define the failure model

List failures by where they originate: invalid input, capacity exhaustion, dependency timeout, partial state change, malformed output, and semantically wrong output. Each class needs a different response. Validation errors should fail immediately. Transient dependency failures may be retried with a budget and jitter. An operation that may have committed must use an idempotency key or reconciliation step before retrying. A syntactically valid but incorrect result belongs in evaluation and review, not a blind retry loop.

Set a deadline for the complete operation and derive smaller budgets for each dependency. Local timeouts that add up to more than the caller's deadline merely create abandoned work. Propagate cancellation where the protocol supports it. Bound every queue, retry loop, context buffer, and concurrency pool; an unbounded safety mechanism becomes a second outage during overload.

Design a degraded mode before it is needed. Depending on the workload, that can mean returning a cached answer, selecting a simpler path, placing work in a durable queue, or asking for human review. The degraded response must be visible in telemetry and, where it changes meaning, visible to the caller. Silent fallback makes quality regressions almost impossible to diagnose.

Measure the decision, not just the component

Use three layers of signals. System metrics cover latency, throughput, saturation, and errors. Correctness metrics measure whether the result satisfies its contract. Business or user metrics show whether the system solved the intended problem. Improving only one layer can move the others backward, so release criteria should name acceptable movement for all three.

Attach a reason code to every route, rejection, fallback, and retry. Include version identifiers for configuration, code, model, schema, and data when relevant. Logs should let an engineer reconstruct a decision without storing secrets or raw personal data. Traces should cross process boundaries, while metrics should remain low-cardinality enough to operate reliably.

Alert on symptoms that require action, not every internal anomaly. A useful alert names the affected service objective, links to a runbook, and distinguishes a customer-visible incident from exhausted headroom. Dashboards serve a different purpose: they support diagnosis and capacity planning. Treating a dashboard as an alerting strategy leaves failures undiscovered until someone happens to look.

Roll out with reversible steps

Ship Internal Developer Platform for AI Teams behind a versioned interface and a kill switch. Begin with offline replay using production-shaped, privacy-safe samples. Then use shadow execution when duplicate work has acceptable cost and side effects can be suppressed. A small canary should exercise the real dependency graph before traffic expands. Compare the canary with the baseline by cohort rather than mixing both populations into one aggregate.

Promotion gates should be written before the rollout. Include a minimum sample size or observation window, maximum regression in tail latency and error rate, and a correctness threshold. Roll back automatically when a hard safety boundary is crossed; use manual review for ambiguous quality movement. Preserve enough evidence from both paths to explain why the gate passed or failed.

Configuration deserves the same discipline as code. Review changes, validate them before activation, keep an immutable history, and make rollback a single operation. If a deployment changes code and configuration together, record both versions. Otherwise an incident responder may roll back the binary while leaving the triggering configuration active.

Capacity and cost controls

Model capacity in units the bottleneck understands: concurrent connections, tokens, queue jobs, database transactions, GPU memory, or bytes in flight. Convert the expected traffic distribution into those units and include burst behavior. Then load-test the first constrained dependency, not merely the public endpoint. A system that accepts more work than it can finish within its deadline is overloaded even if CPU utilization looks comfortable.

Cost is also a reliability limit. Add per-request attribution, tenant or workflow budgets, and a global circuit breaker for unexpectedly expensive paths. Review unit economics at the same granularity as performance; a cheap median can conceal a small class of requests responsible for most spend. Optimize only after measuring, because reducing context, replicas, validation, or redundancy can trade visible cost for less visible risk.

Production readiness review

Before launch, ask an engineer who did not build the feature to follow the runbook through one simulated failure. Verify backups or checkpoints by restoring them, not by checking that a job reported success. Exercise credential rotation, dependency unavailability, bad configuration, and rollback. Assign an owner for each alarm and a date for reviewing thresholds after real traffic arrives.

The final architecture document should be short enough to remain current. Keep the decision, rejected alternatives, invariants, dependency contracts, dashboards, and rollback procedure. Link detailed experiments rather than pasting them into the document. Teams that need help turning this review into an operable service can use our Internal Developer Platform for AI Teams engineering support.

Operating Internal Developer Platform for AI Teams as a System

The implementation is only one part of Internal Developer Platform for AI Teams. A production design also needs an explicit contract for inputs, outputs, ownership, and failure behavior. Write that contract before selecting a library. It should identify which component validates input, where state lives, what may be retried, and which result is authoritative when two components disagree. This prevents a convenient prototype boundary from silently becoming the long-term architecture.

Start with a representative baseline. Capture request shape, traffic distribution, dependency latency, error classes, and the quality signal users actually care about. Averages hide the cases that cause incidents, so keep percentiles and segment measurements by workload type. Record the configuration and dataset version beside every result. Without that context, a faster or more accurate run cannot be reproduced and should not be used to approve a rollout.

Define the failure model

List failures by where they originate: invalid input, capacity exhaustion, dependency timeout, partial state change, malformed output, and semantically wrong output. Each class needs a different response. Validation errors should fail immediately. Transient dependency failures may be retried with a budget and jitter. An operation that may have committed must use an idempotency key or reconciliation step before retrying. A syntactically valid but incorrect result belongs in evaluation and review, not a blind retry loop.

Set a deadline for the complete operation and derive smaller budgets for each dependency. Local timeouts that add up to more than the caller's deadline merely create abandoned work. Propagate cancellation where the protocol supports it. Bound every queue, retry loop, context buffer, and concurrency pool; an unbounded safety mechanism becomes a second outage during overload.

Design a degraded mode before it is needed. Depending on the workload, that can mean returning a cached answer, selecting a simpler path, placing work in a durable queue, or asking for human review. The degraded response must be visible in telemetry and, where it changes meaning, visible to the caller. Silent fallback makes quality regressions almost impossible to diagnose.

Measure the decision, not just the component

Use three layers of signals. System metrics cover latency, throughput, saturation, and errors. Correctness metrics measure whether the result satisfies its contract. Business or user metrics show whether the system solved the intended problem. Improving only one layer can move the others backward, so release criteria should name acceptable movement for all three.

Attach a reason code to every route, rejection, fallback, and retry. Include version identifiers for configuration, code, model, schema, and data when relevant. Logs should let an engineer reconstruct a decision without storing secrets or raw personal data. Traces should cross process boundaries, while metrics should remain low-cardinality enough to operate reliably.

Alert on symptoms that require action, not every internal anomaly. A useful alert names the affected service objective, links to a runbook, and distinguishes a customer-visible incident from exhausted headroom. Dashboards serve a different purpose: they support diagnosis and capacity planning. Treating a dashboard as an alerting strategy leaves failures undiscovered until someone happens to look.

Roll out with reversible steps

Ship Internal Developer Platform for AI Teams behind a versioned interface and a kill switch. Begin with offline replay using production-shaped, privacy-safe samples. Then use shadow execution when duplicate work has acceptable cost and side effects can be suppressed. A small canary should exercise the real dependency graph before traffic expands. Compare the canary with the baseline by cohort rather than mixing both populations into one aggregate.

Promotion gates should be written before the rollout. Include a minimum sample size or observation window, maximum regression in tail latency and error rate, and a correctness threshold. Roll back automatically when a hard safety boundary is crossed; use manual review for ambiguous quality movement. Preserve enough evidence from both paths to explain why the gate passed or failed.

Configuration deserves the same discipline as code. Review changes, validate them before activation, keep an immutable history, and make rollback a single operation. If a deployment changes code and configuration together, record both versions. Otherwise an incident responder may roll back the binary while leaving the triggering configuration active.

Capacity and cost controls

Model capacity in units the bottleneck understands: concurrent connections, tokens, queue jobs, database transactions, GPU memory, or bytes in flight. Convert the expected traffic distribution into those units and include burst behavior. Then load-test the first constrained dependency, not merely the public endpoint. A system that accepts more work than it can finish within its deadline is overloaded even if CPU utilization looks comfortable.

Cost is also a reliability limit. Add per-request attribution, tenant or workflow budgets, and a global circuit breaker for unexpectedly expensive paths. Review unit economics at the same granularity as performance; a cheap median can conceal a small class of requests responsible for most spend. Optimize only after measuring, because reducing context, replicas, validation, or redundancy can trade visible cost for less visible risk.

Production readiness review

Before launch, ask an engineer who did not build the feature to follow the runbook through one simulated failure. Verify backups or checkpoints by restoring them, not by checking that a job reported success. Exercise credential rotation, dependency unavailability, bad configuration, and rollback. Assign an owner for each alarm and a date for reviewing thresholds after real traffic arrives.

The final architecture document should be short enough to remain current. Keep the decision, rejected alternatives, invariants, dependency contracts, dashboards, and rollback procedure. Link detailed experiments rather than pasting them into the document. Teams that need help turning this review into an operable service can use our Internal Developer Platform for AI Teams engineering support.

Frequently Asked Questions

How is an IDP different from Kubernetes?

IDP abstracts Kubernetes with ML-specific APIs (TrainingJob, ModelDeployment) so data scientists don't write YAML or understand pods/services. IDP = opinionated platform on top of K8s.

Should we build or buy an IDP?

Build if you have unique ML workflows and 20+ data scientists. Buy/extend (Backstage, Kubeflow) if standard workflows fit. Most companies customize open-source tools rather than building from scratch.

How do we handle model governance?

Built into platform: approval workflows for production deployments, audit logs for compliance, model registry for versioning, automated bias/fairness testing in CI/CD.

What about data access control?

Role-Based Access Control (RBAC) enforced at platform API level. Data scientists request dataset access via portal → approval workflow → credentials provisioned with least privilege.

How do we prevent GPU waste?

Automatic shutdown of idle resources, budget alerts per team, cost dashboards showing real usage, and approval gates for expensive GPU types (A100s).

Can we integrate with existing tools?

Yes—IDP integrates with existing infrastructure (MLflow, S3, PostgreSQL). Use platform as orchestration layer, not replacement for specialized tools.


Conclusion

Internal Developer Platforms eliminate infrastructure friction for ML teams:

  • Self-service provisioning—GPUs, databases, storage without DevOps tickets
  • Golden paths—opinionated templates for common workflows
  • Unified portal—single interface for experiments, deployments, monitoring
  • Built-in governance—audit logs, cost tracking, compliance automation
  • Developer experience—CLI, portal, IDE plugins for seamless workflows
  • Platform abstractions—hide Kubernetes complexity from data scientists

For high-velocity ML organizations, IDPs are non-negotiable.

At HinterBuild, we build production ML platforms:

Contact us for platform engineering consulting.

Free consultation

Book a free consultation call on internal developer platforms for AI

30-minute call with the HinterBuild team. Discuss your project, architecture questions, or next steps — no obligation.

Book a meeting

Keep reading