tessl/pypi-kubernetes

Python client library for interacting with Kubernetes clusters through the Kubernetes API

—

Pending

Overview

Eval results

Files

Autoscaling

Name: tessl/pypi-kubernetes
Author: tessl

Horizontal Pod Autoscaler (HPA) management for automatic scaling of deployments, replica sets, and other scalable resources based on CPU, memory, or custom metrics.

Capabilities

Horizontal Pod Autoscaler V1

Basic horizontal pod autoscaling based on CPU utilization metrics.

class AutoscalingV1Api:
    def create_namespaced_horizontal_pod_autoscaler(
        self,
        namespace: str,
        body: V1HorizontalPodAutoscaler,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1HorizontalPodAutoscaler:
        """Create a horizontal pod autoscaler."""
    
    def list_namespaced_horizontal_pod_autoscaler(
        self,
        namespace: str,
        pretty: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1HorizontalPodAutoscalerList:
        """List horizontal pod autoscalers in specified namespace."""
    
    def list_horizontal_pod_autoscaler_for_all_namespaces(
        self,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        pretty: str = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V1HorizontalPodAutoscalerList:
        """List horizontal pod autoscalers across all namespaces."""
    
    def read_namespaced_horizontal_pod_autoscaler(
        self,
        name: str,
        namespace: str,
        pretty: str = None
    ) -> V1HorizontalPodAutoscaler:
        """Read specified horizontal pod autoscaler."""
    
    def read_namespaced_horizontal_pod_autoscaler_status(
        self,
        name: str,
        namespace: str,
        pretty: str = None
    ) -> V1HorizontalPodAutoscaler:
        """Read status of specified horizontal pod autoscaler."""
    
    def patch_namespaced_horizontal_pod_autoscaler(
        self,
        name: str,
        namespace: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1HorizontalPodAutoscaler:
        """Update specified horizontal pod autoscaler."""
    
    def replace_namespaced_horizontal_pod_autoscaler(
        self,
        name: str,
        namespace: str,
        body: V1HorizontalPodAutoscaler,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V1HorizontalPodAutoscaler:
        """Replace specified horizontal pod autoscaler."""
    
    def delete_namespaced_horizontal_pod_autoscaler(
        self,
        name: str,
        namespace: str,
        dry_run: str = None,
        grace_period_seconds: int = None,
        pretty: str = None
    ) -> V1Status:
        """Delete specified horizontal pod autoscaler."""

Horizontal Pod Autoscaler V2

Advanced horizontal pod autoscaling with support for multiple metrics including CPU, memory, and custom metrics.

class AutoscalingV2Api:
    def create_namespaced_horizontal_pod_autoscaler(
        self,
        namespace: str,
        body: V2HorizontalPodAutoscaler,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V2HorizontalPodAutoscaler:
        """Create a horizontal pod autoscaler with advanced metrics."""
    
    def list_namespaced_horizontal_pod_autoscaler(
        self,
        namespace: str,
        pretty: str = None,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V2HorizontalPodAutoscalerList:
        """List horizontal pod autoscalers in specified namespace."""
    
    def list_horizontal_pod_autoscaler_for_all_namespaces(
        self,
        field_selector: str = None,
        label_selector: str = None,
        limit: int = None,
        pretty: str = None,
        timeout_seconds: int = None,
        watch: bool = None
    ) -> V2HorizontalPodAutoscalerList:
        """List horizontal pod autoscalers across all namespaces."""
    
    def read_namespaced_horizontal_pod_autoscaler(
        self,
        name: str,
        namespace: str,
        pretty: str = None
    ) -> V2HorizontalPodAutoscaler:
        """Read specified horizontal pod autoscaler."""
    
    def read_namespaced_horizontal_pod_autoscaler_status(
        self,
        name: str,
        namespace: str,
        pretty: str = None
    ) -> V2HorizontalPodAutoscaler:
        """Read status of specified horizontal pod autoscaler."""
    
    def patch_namespaced_horizontal_pod_autoscaler(
        self,
        name: str,
        namespace: str,
        body: object,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V2HorizontalPodAutoscaler:
        """Update specified horizontal pod autoscaler."""
    
    def replace_namespaced_horizontal_pod_autoscaler(
        self,
        name: str,
        namespace: str,
        body: V2HorizontalPodAutoscaler,
        dry_run: str = None,
        field_manager: str = None,
        pretty: str = None
    ) -> V2HorizontalPodAutoscaler:
        """Replace specified horizontal pod autoscaler."""
    
    def delete_namespaced_horizontal_pod_autoscaler(
        self,
        name: str,
        namespace: str,
        dry_run: str = None,
        grace_period_seconds: int = None,
        pretty: str = None
    ) -> V1Status:
        """Delete specified horizontal pod autoscaler."""

Resource Types

V1HorizontalPodAutoscaler (V1 API)

class V1HorizontalPodAutoscaler:
    api_version: str
    kind: str
    metadata: V1ObjectMeta
    spec: V1HorizontalPodAutoscalerSpec
    status: V1HorizontalPodAutoscalerStatus

class V1HorizontalPodAutoscalerSpec:
    max_replicas: int
    min_replicas: int
    scale_target_ref: V1CrossVersionObjectReference
    target_cpu_utilization_percentage: int

V2HorizontalPodAutoscaler (V2 API)

class V2HorizontalPodAutoscaler:
    api_version: str
    kind: str
    metadata: V1ObjectMeta
    spec: V2HorizontalPodAutoscalerSpec
    status: V2HorizontalPodAutoscalerStatus

class V2HorizontalPodAutoscalerSpec:
    max_replicas: int
    min_replicas: int
    scale_target_ref: V2CrossVersionObjectReference
    metrics: list[V2MetricSpec]
    behavior: V2HorizontalPodAutoscalerBehavior

class V2MetricSpec:
    type: str  # "Resource", "Pods", "Object", "External"
    resource: V2ResourceMetricSource
    pods: V2PodsMetricSource
    object: V2ObjectMetricSource
    external: V2ExternalMetricSource

Status Types

class V1HorizontalPodAutoscalerStatus:
    current_cpu_utilization_percentage: int
    current_replicas: int
    desired_replicas: int
    last_scale_time: datetime

class V2HorizontalPodAutoscalerStatus:
    conditions: list[V2HorizontalPodAutoscalerCondition]
    current_metrics: list[V2MetricStatus]
    current_replicas: int
    desired_replicas: int
    last_scale_time: datetime

Usage Examples

Basic CPU-based Autoscaling (V1)

from kubernetes import client, config

# Load configuration
config.load_kube_config()

# Create autoscaling V1 API client
autoscaling_v1 = client.AutoscalingV1Api()

# Create basic HPA for CPU scaling
hpa_spec = client.V1HorizontalPodAutoscalerSpec(
    scale_target_ref=client.V1CrossVersionObjectReference(
        api_version="apps/v1",
        kind="Deployment",
        name="web-app"
    ),
    min_replicas=2,
    max_replicas=10,
    target_cpu_utilization_percentage=70
)

hpa = client.V1HorizontalPodAutoscaler(
    api_version="autoscaling/v1",
    kind="HorizontalPodAutoscaler",
    metadata=client.V1ObjectMeta(name="web-app-hpa"),
    spec=hpa_spec
)

# Create HPA
result = autoscaling_v1.create_namespaced_horizontal_pod_autoscaler(
    namespace="default",
    body=hpa
)
print(f"HPA created: {result.metadata.name}")

Advanced Multi-Metric Autoscaling (V2)

# Create autoscaling V2 API client
autoscaling_v2 = client.AutoscalingV2Api()

# Define multiple metrics for scaling decisions
metrics = [
    # CPU utilization metric
    client.V2MetricSpec(
        type="Resource",
        resource=client.V2ResourceMetricSource(
            name="cpu",
            target=client.V2MetricTarget(
                type="Utilization",
                average_utilization=70
            )
        )
    ),
    # Memory utilization metric
    client.V2MetricSpec(
        type="Resource",
        resource=client.V2ResourceMetricSource(
            name="memory",
            target=client.V2MetricTarget(
                type="Utilization",
                average_utilization=80
            )
        )
    ),
    # Custom metric example
    client.V2MetricSpec(
        type="Pods",
        pods=client.V2PodsMetricSource(
            metric=client.V2MetricIdentifier(
                name="requests_per_second"
            ),
            target=client.V2MetricTarget(
                type="AverageValue",
                average_value="100"
            )
        )
    )
]

# Define scaling behavior
behavior = client.V2HorizontalPodAutoscalerBehavior(
    scale_up=client.V2HPAScalingRules(
        stabilization_window_seconds=60,
        policies=[
            client.V2HPAScalingPolicy(
                type="Percent",
                value=100,
                period_seconds=60
            )
        ]
    ),
    scale_down=client.V2HPAScalingRules(
        stabilization_window_seconds=300,
        policies=[
            client.V2HPAScalingPolicy(
                type="Percent",
                value=10,
                period_seconds=60
            )
        ]
    )
)

# Create advanced HPA specification
hpa_v2_spec = client.V2HorizontalPodAutoscalerSpec(
    scale_target_ref=client.V2CrossVersionObjectReference(
        api_version="apps/v1",
        kind="Deployment",
        name="web-app"
    ),
    min_replicas=3,
    max_replicas=50,
    metrics=metrics,
    behavior=behavior
)

hpa_v2 = client.V2HorizontalPodAutoscaler(
    api_version="autoscaling/v2",
    kind="HorizontalPodAutoscaler",
    metadata=client.V1ObjectMeta(name="web-app-advanced-hpa"),
    spec=hpa_v2_spec
)

# Create advanced HPA
result = autoscaling_v2.create_namespaced_horizontal_pod_autoscaler(
    namespace="default",
    body=hpa_v2
)
print(f"Advanced HPA created: {result.metadata.name}")

Monitoring HPA Status

# Get HPA status
hpa_status = autoscaling_v2.read_namespaced_horizontal_pod_autoscaler_status(
    name="web-app-advanced-hpa",
    namespace="default"
)

print(f"Current replicas: {hpa_status.status.current_replicas}")
print(f"Desired replicas: {hpa_status.status.desired_replicas}")

# Check current metrics
if hpa_status.status.current_metrics:
    for metric in hpa_status.status.current_metrics:
        if metric.type == "Resource":
            resource_name = metric.resource.name
            current_value = metric.resource.current.average_utilization
            print(f"{resource_name} utilization: {current_value}%")

# Check conditions
if hpa_status.status.conditions:
    for condition in hpa_status.status.conditions:
        print(f"Condition {condition.type}: {condition.status}")
        if condition.reason:
            print(f"  Reason: {condition.reason}")
        if condition.message:
            print(f"  Message: {condition.message}")

List and Manage HPAs

# List all HPAs in namespace
hpas = autoscaling_v2.list_namespaced_horizontal_pod_autoscaler(
    namespace="default"
)

for hpa in hpas.items:
    print(f"HPA: {hpa.metadata.name}")
    print(f"  Target: {hpa.spec.scale_target_ref.kind}/{hpa.spec.scale_target_ref.name}")
    print(f"  Replicas: {hpa.spec.min_replicas}-{hpa.spec.max_replicas}")
    if hpa.status:
        print(f"  Current: {hpa.status.current_replicas} replicas")

# Update HPA
hpa.spec.max_replicas = 20
updated_hpa = autoscaling_v2.replace_namespaced_horizontal_pod_autoscaler(
    name="web-app-advanced-hpa",
    namespace="default",
    body=hpa
)
print(f"HPA updated, new max replicas: {updated_hpa.spec.max_replicas}")

# Delete HPA
autoscaling_v2.delete_namespaced_horizontal_pod_autoscaler(
    name="web-app-advanced-hpa",
    namespace="default"
)
print("HPA deleted")

Import Statements

from kubernetes import client
from kubernetes.client import AutoscalingV1Api, AutoscalingV2Api
from kubernetes.client import V1HorizontalPodAutoscaler, V1HorizontalPodAutoscalerSpec
from kubernetes.client import V2HorizontalPodAutoscaler, V2HorizontalPodAutoscalerSpec
from kubernetes.client import V2MetricSpec, V2ResourceMetricSource, V2MetricTarget
from kubernetes.client import V2HorizontalPodAutoscalerBehavior, V2HPAScalingRules

Install with Tessl CLI