How to Upskill for 2026 Trends: Learn Rust 1.85 and Kubernetes 1.32 for High-Demand Roles

// Operator main module for Redis cluster scaling
// Requires: rustc 1.85+, kube-rs 2.0, Kubernetes 1.32 cluster with metrics-server
use kube::{
    Client, Api,
    error::Error as KubeError,
    api::{PostParams, PatchParams, Patch, ListParams},
};
use kube::runtime::{Controller, watcher, WatchStreamExt};
use serde::{Deserialize, Serialize};
use serde_json::json;
use prometheus::{
    register_counter, Counter, register_histogram, Histogram, HistogramOpts,
    Encoder, TextEncoder,
};
use std::error::Error;
use std::sync::Arc;
use tokio::time::sleep;
use tracing::{info, error, warn};
use tracing_subscriber::{fmt, EnvFilter};

// Custom resource definition for RedisCluster
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct RedisClusterSpec {
    pub replicas: i32,
    pub shard_count: i32,
    pub scaling_threshold: f64, // CPU utilization % to trigger scaling
}

#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct RedisClusterStatus {
    pub ready_replicas: i32,
    pub current_cpu: f64,
}

// Metrics definitions
lazy_static::lazy_static! {
    static ref SCALING_EVENTS: Counter = register_counter!(
        "redis_operator_scaling_events_total",
        "Total number of scaling events triggered by the operator"
    ).unwrap();
    static ref RECONCILE_DURATION: Histogram = register_histogram!(
        HistogramOpts::new(
            "redis_operator_reconcile_duration_seconds",
            "Time spent reconciling RedisCluster resources"
        )
        .buckets(vec![0.01, 0.05, 0.1, 0.5, 1.0, 5.0])
    ).unwrap();
}

#[tokio::main]
async fn main() -> Result<(), Box> {
    // Initialize tracing with env filter (RUST_LOG=info by default)
    tracing_subscriber::fmt()
        .with(EnvFilter::from_default_env())
        .init();

    // Load kubeconfig or use in-cluster config
    let client = match Client::try_default().await {
        Ok(c) => {
            info!("Initialized Kubernetes client for cluster version 1.32+");
            c
        }
        Err(e) => {
            error!(error = %e, "Failed to initialize Kubernetes client");
            return Err(e.into());
        }
    };

    // Verify cluster version compatibility
    let server_version = client.apiserver_version().await?;
    let major: i32 = server_version.major.parse()?;
    let minor: i32 = server_version.minor.split('.').next().unwrap().parse()?;
    if major < 1 || minor < 32 {
        error!(
            version = %format!("{}.{}", major, minor),
            "Kubernetes cluster version must be 1.32 or higher"
        );
        return Err("Incompatible Kubernetes version".into());
    }

    // Set up API for RedisCluster custom resources
    let redis_clusters: Api = Api::all(client.clone());
    let metrics_client: Api = Api::all(client.clone());

    // Start Prometheus metrics server on port 8080
    let metrics_client_clone = client.clone();
    tokio::spawn(async move {
        let addr = "0.0.0.0:8080".parse().unwrap();
        info!(addr = %addr, "Starting metrics server");
        warp::serve(
            warp::path!("metrics")
                .map(|| {
                    let encoder = TextEncoder::new();
                    let mut buffer = vec![];
                    let metrics = prometheus::gather();
                    encoder.encode(&metrics, &mut buffer).unwrap();
                    String::from_utf8(buffer).unwrap()
                })
        )
        .run(addr)
        .await;
    });

    // Run controller
    info!("Starting RedisCluster controller");
    Controller::new(redis_clusters, watcher::Config::default())
        .run(reconcile, error_policy, Arc::new(Context { client }))
        .for_each(|res| async move {
            match res {
                Ok(o) => info!(obj = ?o, "Reconciliation successful"),
                Err(e) => error!(error = %e, "Reconciliation failed"),
            }
        })
        .await;

    Ok(())
}

// Context struct to pass client to reconcile functions
struct Context {
    client: Client,
}

// Reconcile function for RedisCluster resources
async fn reconcile(
    redis_cluster: RedisCluster,
    ctx: Arc,
) -> Result<(), KubeError> {
    let _timer = RECONCILE_DURATION.start_timer();
    let client = &ctx.client;
    let name = redis_cluster.metadata.name.as_deref().unwrap_or("unknown");
    let namespace = redis_cluster.metadata.namespace.as_deref().unwrap_or("default");

    info!(
        name = %name,
        namespace = %namespace,
        "Reconciling RedisCluster resource"
    );

    // Get current metrics for the Redis cluster
    let metrics_api: Api = Api::namespaced(client.clone(), namespace);
    let metrics = match metrics_api.get(&format!("{}-redis", name)).await {
        Ok(m) => m,
        Err(KubeError::NotFound { .. }) => {
            warn!(name = %name, "No metrics found for Redis cluster, skipping scaling");
            return Ok(());
        }
        Err(e) => return Err(e),
    };

    // Calculate average CPU utilization
    let cpu_usage: f64 = metrics.usage.cpu.parse().unwrap_or(0.0);
    let cpu_percent = cpu_usage / 1_000_000_000.0 * 100.0; // Convert nanoseconds to percent

    // Check if scaling is needed
    let spec = &redis_cluster.spec;
    if cpu_percent > spec.scaling_threshold {
        SCALING_EVENTS.inc();
        let new_replicas = spec.replicas + 1;
        info!(
            cpu = %cpu_percent,
            old_replicas = %spec.replicas,
            new_replicas = %new_replicas,
            "Scaling up Redis cluster"
        );

        // Patch the resource with new replica count
        let patch = json!({
            "spec": { "replicas": new_replicas }
        });
        let params = PatchParams::apply("redis-operator");
        let redis_clusters: Api = Api::namespaced(client.clone(), namespace);
        redis_clusters.patch(&name, ¶ms, &Patch::Merge(&patch)).await?;
    }

    Ok(())
}

// Error policy for controller
fn error_policy(error: &KubeError, _ctx: Arc) -> kube::runtime::controller::Action {
    error!(error = %error, "Reconciliation error, retrying in 5s");
    kube::runtime::controller::Action::requeue(Duration::from_secs(5))
}

// Mock RedisCluster struct for compilation (full CRD would be generated via kube-derive)
#[derive(Serialize, Deserialize, Clone, Debug)]
struct RedisCluster {
    metadata: kube::api::ObjectMeta,
    spec: RedisClusterSpec,
    status: Option,
}

impl kube::Resource for RedisCluster {
    type DynamicType = ();
    fn kind() -> &'static str { "RedisCluster" }
    fn group() -> &'static str { "redis.example.com" }
    fn version() -> &'static str { "v1alpha1" }
    fn plural() -> &'static str { "redisclusters" }
}

# Kubernetes 1.32 manifests for Redis Operator deployment
# Requires: Kubernetes 1.32+ with sidecar containers GA enabled
# Apply with: kubectl apply -f operator-manifests.yaml

---
# CustomResourceDefinition for RedisCluster (v1beta1 CRD for K8s 1.32+)
apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: redisclusters.redis.example.com
spec:
  group: redis.example.com
  versions:
    - name: v1alpha1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
              properties:
                replicas:
                  type: integer
                  minimum: 1
                  maximum: 10
                  description: "Number of Redis replicas in the cluster"
                shard_count:
                  type: integer
                  minimum: 1
                  maximum: 5
                  description: "Number of shards for Redis Cluster mode"
                scaling_threshold:
                  type: number
                  minimum: 0.0
                  maximum: 100.0
                  description: "CPU utilization % threshold to trigger scaling"
              required: ["replicas", "shard_count", "scaling_threshold"]
            status:
              type: object
              properties:
                ready_replicas:
                  type: integer
                  description: "Number of ready Redis replicas"
                current_cpu:
                  type: number
                  description: "Current average CPU utilization %"
  scope: Namespaced
  names:
    plural: redisclusters
    singular: rediscluster
    kind: RedisCluster
    shortNames:
      - rc

---
# ServiceAccount for the operator
apiVersion: v1
kind: ServiceAccount
metadata:
  name: redis-operator
  namespace: redis-operator
  labels:
    app: redis-operator
    version: "1.85.0"

---
# Role for operator permissions (K8s 1.32+ uses RBAC v1)
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: redis-operator-role
  namespace: redis-operator
  labels:
    app: redis-operator
rules:
  # Permissions for RedisCluster CRDs
  - apiGroups: ["redis.example.com"]
    resources: ["redisclusters", "redisclusters/status"]
    verbs: ["get", "list", "watch", "patch", "update"]
  # Permissions for pods (to read metrics)
  - apiGroups: [""]
    resources: ["pods", "pods/metrics"]
    verbs: ["get", "list", "watch"]
  # Permissions for events (to emit operator events)
  - apiGroups: [""]
    resources: ["events"]
    verbs: ["create", "patch"]

---
# RoleBinding to attach Role to ServiceAccount
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: redis-operator-rolebinding
  namespace: redis-operator
  labels:
    app: redis-operator
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: Role
  name: redis-operator-role
subjects:
  - kind: ServiceAccount
    name: redis-operator
    namespace: redis-operator

---
# Deployment for the Redis Operator (uses K8s 1.32 sidecar GA for metrics)
apiVersion: apps/v1
kind: Deployment
metadata:
  name: redis-operator
  namespace: redis-operator
  labels:
    app: redis-operator
    version: "1.85.0"
spec:
  replicas: 1
  selector:
    matchLabels:
      app: redis-operator
  template:
    metadata:
      labels:
        app: redis-operator
      annotations:
        # K8s 1.32 sidecar container annotation (GA)
        sidecar.inject: "true"
    spec:
      serviceAccountName: redis-operator
      containers:
        # Main operator container (Rust 1.85 build)
        - name: operator
          image: redis-operator:v1.85.0
          ports:
            - containerPort: 8080
              name: metrics
              protocol: TCP
          env:
            - name: RUST_LOG
              value: "info"
            - name: KUBERNETES_PORT_443_TCP_ADDR
              value: "kubernetes.default.svc"
          resources:
            requests:
              cpu: "100m"
              memory: "128Mi"
            limits:
              cpu: "500m"
              memory: "512Mi"
          livenessProbe:
            httpGet:
              path: /healthz
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 10
          readinessProbe:
            httpGet:
              path: /readyz
              port: 8080
            initialDelaySeconds: 3
            periodSeconds: 5
        # K8s 1.32 sidecar for metrics collection (no more init containers needed)
        - name: metrics-sidecar
          image: prometheus/node-exporter:v1.6.0
          args:
            - "--path.rootfs=/host"
            - "--collector.filesystem.mount-points-exclude=^/(sys|proc|dev|host|etc)($$|/)"
          ports:
            - containerPort: 9100
              name: sidecar-metrics
          resources:
            requests:
              cpu: "50m"
              memory: "32Mi"
            limits:
              cpu: "100m"
              memory: "64Mi"
          volumeMounts:
            - name: host-root
              mountPath: /host
              readOnly: true
      volumes:
        - name: host-root
          hostPath:
            path: /
            type: Directory

// k6 benchmark script to compare Rust 1.85 vs Go Kubernetes operators
// Run with: k6 run --vus 100 --duration 30s benchmark.js
// Measures reconciliation latency, memory usage, and CPU overhead

import http from 'k6/http';
import { check, sleep } from 'k6';
import { Trend, Rate } from 'k6/metrics';
import { randomString } from 'https://jslib.k6.io/k6-utils/1.4.0/index.js';

// Custom metrics
const reconciliationLatency = new Trend('operator_reconcile_latency_ms');
const scalingSuccessRate = new Rate('operator_scaling_success_rate');
const operatorType = __ENV.OPERATOR_TYPE || 'rust'; // 'rust' or 'go'

// Test configuration
const BASE_URL = `http://${operatorType}-operator.redis-operator.svc:8080`;
const REDIS_CLUSTER_URL = 'http://kubernetes.default.svc/api/v1/namespaces/benchmark/redisclusters';

// Setup function: create test namespace and initial RedisCluster
export function setup() {
  const namespace = 'benchmark';
  // Create namespace
  const nsRes = http.post(
    'http://kubernetes.default.svc/api/v1/namespaces',
    JSON.stringify({ metadata: { name: namespace } }),
    { headers: { 'Content-Type': 'application/json' } }
  );
  check(nsRes, { 'namespace created': (r) => r.status === 201 || r.status === 409 });

  // Apply CRD if not exists
  const crdRes = http.get(
    'http://kubernetes.default.svc/apis/apiextensions.k8s.io/v1/customresourcedefinitions/redisclusters.redis.example.com'
  );
  if (crdRes.status === 404) {
    const crdManifest = open('./crd.yaml');
    http.post(
      'http://kubernetes.default.svc/apis/apiextensions.k8s.io/v1/customresourcedefinitions',
      crdManifest,
      { headers: { 'Content-Type': 'application/yaml' } }
    );
  }

  // Create initial RedisCluster resource
  const clusterName = `bench-cluster-${randomString(8)}`;
  const clusterManifest = JSON.stringify({
    apiVersion: 'redis.example.com/v1alpha1',
    kind: 'RedisCluster',
    metadata: { name: clusterName, namespace },
    spec: {
      replicas: 1,
      shard_count: 1,
      scaling_threshold: 70.0,
    },
  });

  const clusterRes = http.post(
    REDIS_CLUSTER_URL,
    clusterManifest,
    { headers: { 'Content-Type': 'application/json' } }
  );
  check(clusterRes, { 'redis cluster created': (r) => r.status === 201 });

  return { namespace, clusterName };
}

// Main test function
export default function (data) {
  const { namespace, clusterName } = data;
  const url = `${REDIS_CLUSTER_URL}/${clusterName}`;

  // Trigger scaling by updating scaling threshold to 50%
  const payload = JSON.stringify({
    spec: { scaling_threshold: 50.0 },
  });

  const params = {
    headers: {
      'Content-Type': 'application/json-patch+json',
    },
  };

  const startTime = new Date().getTime();
  const res = http.patch(url, payload, params);
  const endTime = new Date().getTime();

  // Record metrics
  reconciliationLatency.add(endTime - startTime);
  scalingSuccessRate.add(res.status === 200);

  check(res, {
    'patch successful': (r) => r.status === 200,
    'response time < 500ms': (r) => endTime - startTime < 500,
  });

  // Simulate load: get metrics endpoint
  const metricsRes = http.get(`${BASE_URL}/metrics`);
  check(metricsRes, {
    'metrics endpoint returns 200': (r) => r.status === 200,
    'metrics contains operator events': (r) => r.body.includes('redis_operator_scaling_events_total'),
  });

  sleep(1);
}

// Teardown function: clean up resources
export function teardown(data) {
  const { namespace, clusterName } = data;
  // Delete RedisCluster
  http.del(`${REDIS_CLUSTER_URL}/${clusterName}`);
  // Delete namespace
  http.del(`http://kubernetes.default.svc/api/v1/namespaces/${namespace}`);
}

// Options for the benchmark
export const options = {
  stages: [
    { duration: '30s', target: 50 }, // Ramp up to 50 VUs
    { duration: '1m', target: 50 }, // Stay at 50 VUs
    { duration: '30s', target: 0 }, // Ramp down
  ],
  thresholds: {
    'operator_reconcile_latency_ms': ['p(95)<300'], // 95% of requests under 300ms
    'operator_scaling_success_rate': ['rate>0.99'], // 99% success rate
  },
};

// Native async trait in Rust 1.85 (no async-trait crate needed)
pub trait AsyncReconciler {
    async fn reconcile(&self, resource: RedisCluster) -> Result<(), KubeError>;
}

impl AsyncReconciler for RedisReconciler {
    async fn reconcile(&self, resource: RedisCluster) -> Result<(), KubeError> {
        // Reconciliation logic here
        Ok(())
    }
}

# K8s 1.32 GA sidecar annotation (no beta annotations needed)
apiVersion: apps/v1
kind: Deployment
spec:
  template:
    metadata:
      annotations:
        sidecar.inject: "true" # GA annotation, works with 1.32+
    spec:
      containers:
        - name: operator
          image: redis-operator:v1.85.0
        - name: otel-collector
          image: otel/opentelemetry-collector:0.88.0

// Compile-time CRD validation with kube-derive 2.0 and Rust 1.85
use kube_derive::CustomResource;
use serde::{Deserialize, Serialize};

#[derive(CustomResource, Serialize, Deserialize, Clone, Debug)]
#[kube(
    group = "redis.example.com",
    version = "v1alpha1",
    kind = "RedisCluster",
    plural = "redisclusters"
)]
pub struct RedisClusterSpec {
    pub replicas: i32, // Compile-time error if missing in CRD YAML
    pub shard_count: i32,
    pub scaling_threshold: f64,
}

rust-k8s-operator-2026/
├── Cargo.toml
├── src/
│   ├── main.rs
│   ├── reconcile.rs
│   ├── crd.rs
│   └── metrics.rs
├── k8s-manifests/
│   ├── crd.yaml
│   ├── role.yaml
│   ├── rolebinding.yaml
│   ├── serviceaccount.yaml
│   └── deployment.yaml
├── benchmarks/
│   ├── k6-benchmark.js
│   └── results/
│       ├── rust-1.85.json
│       └── go-1.23.json
├── tests/
│   ├── integration.rs
│   └── unit.rs
└── README.md