kubernetes.utils.metrics module

Metrics utilities for Kubernetes resource monitoring.

Provides helpers for fetching and processing resource usage data from the metrics.k8s.io API endpoint, enabling monitoring and autoscaling workflows.

kubernetes.utils.metrics.get_nodes_metrics(api_client)

Fetch current resource usage for all cluster nodes.

Retrieves CPU and memory consumption metrics from the metrics-server for every node in the cluster.

Parameters:

api_client: An initialized kubernetes.client.ApiClient instance

Returns:

A dictionary containing the metrics response with structure: {

‘kind’: ‘NodeMetricsList’, ‘apiVersion’: ‘metrics.k8s.io/v1beta1’, ‘metadata’: {…}, ‘items’: [

{

‘metadata’: {‘name’: ‘node-1’, …}, ‘timestamp’: ‘2024-01-01T00:00:00Z’, ‘window’: ’30s’, ‘usage’: {‘cpu’: ‘100m’, ‘memory’: ‘1024Mi’}

]

}

Raises:

ApiException: If the metrics server is not available or request fails

Example:

>>> from kubernetes import client, config
>>> config.load_kube_config()
>>> api_client = client.ApiClient()
>>> metrics = get_nodes_metrics(api_client)
>>> for node in metrics['items']:
...     name = node['metadata']['name']
...     cpu = node['usage']['cpu']
...     mem = node['usage']['memory']
...     print(f"Node {name}: CPU={cpu}, Memory={mem}")

kubernetes.utils.metrics.get_pods_metrics(api_client, namespace, label_selector=None)

Fetch current resource usage for pods in a namespace.

Retrieves CPU and memory consumption metrics from the metrics-server for pods in the specified namespace, with optional label filtering.

Parameters:

api_client: An initialized kubernetes.client.ApiClient instance namespace: The namespace name to query (required) label_selector: Optional label query to filter pods (e.g., ‘app=web,env=prod’)

Returns:

A dictionary containing the metrics response with structure: {

‘kind’: ‘PodMetricsList’, ‘apiVersion’: ‘metrics.k8s.io/v1beta1’, ‘metadata’: {…}, ‘items’: [

{

‘metadata’: {‘name’: ‘pod-1’, ‘namespace’: ‘default’, …}, ‘timestamp’: ‘2024-01-01T00:00:00Z’, ‘window’: ’30s’, ‘containers’: [

{

‘name’: ‘container-1’, ‘usage’: {‘cpu’: ‘50m’, ‘memory’: ‘512Mi’}

]

]

}

Raises:

ValueError: If namespace is None or empty ApiException: If the metrics server is not available or request fails

Example:

>>> from kubernetes import client, config
>>> config.load_kube_config()
>>> api_client = client.ApiClient()
>>> 
>>> # Get all pods in namespace
>>> metrics = get_pods_metrics(api_client, 'default')
>>> 
>>> # Get pods with specific labels
>>> metrics = get_pods_metrics(api_client, 'default', 'app=nginx')
>>> 
>>> for pod in metrics['items']:
...     pod_name = pod['metadata']['name']
...     print(f"Pod: {pod_name}")
...     for container in pod['containers']:
...         cname = container['name']
...         cpu = container['usage']['cpu']
...         mem = container['usage']['memory']
...         print(f"  Container {cname}: CPU={cpu}, Memory={mem}")

kubernetes.utils.metrics.get_pods_metrics_in_all_namespaces(api_client, namespaces, label_selector=None)

Fetch pod metrics across multiple namespaces.

Queries pod metrics in each specified namespace and returns an aggregated result. If a namespace query fails, the error is captured in the result rather than raising an exception.

Parameters:

api_client: An initialized kubernetes.client.ApiClient instance namespaces: A list of namespace names to query label_selector: Optional label query applied to all namespaces

Returns:

A dictionary mapping namespace names to their metrics or error info: {

‘namespace-1’: {

‘items’: […], ‘kind’: ‘PodMetricsList’, …

}, ‘namespace-2’: {

‘error’: ‘error message’, ‘kind’: ‘Error’

}

Example:

>>> from kubernetes import client, config
>>> config.load_kube_config()
>>> api_client = client.ApiClient()
>>> 
>>> namespaces = ['default', 'kube-system', 'monitoring']
>>> all_metrics = get_pods_metrics_in_all_namespaces(api_client, namespaces)
>>> 
>>> for ns, result in all_metrics.items():
...     if 'error' in result:
...         print(f"{ns}: ERROR - {result['error']}")
...     else:
...         pod_count = len(result.get('items', []))
...         print(f"{ns}: {pod_count} pods")