Skip to content

Kubernetes Tools

The Kubernetes tools collection provides a comprehensive set of tools and workflows for interacting with Kubernetes clusters, managing deployments, and troubleshooting issues.

Installation

Bash
pip install kubiya-community-tools

Available Tools

Cluster Management

Tool Description
get_cluster_info Get information about the current cluster
get_cluster_resources List all resources in the cluster
get_namespaces List all namespaces in the cluster
check_pod_status Check the status of pods in a namespace
check_node_status Check the status of nodes in the cluster

Deployment Management

Tool Description
deploy_from_yaml Deploy resources from a YAML file
rollout_restart Restart a deployment
scale_deployment Scale a deployment to a specified number of replicas
update_image Update the image of a deployment
get_deployment_history Get the rollout history of a deployment

Resource Troubleshooting

Tool Description
get_pod_logs Get logs from a pod
describe_pod Get detailed information about a pod
exec_command Execute a command in a container
check_resource_usage Check CPU and memory usage of pods
identify_problems Identify common problems in a namespace

Example Tools

Python
from kubiya_sdk import tool
from kubernetes import client, config

@tool(
    image="bitnami/kubectl:latest",
    requirements=["kubernetes"]
)
def get_cluster_info() -> dict:
    """Get information about the current Kubernetes cluster"""
    config.load_kube_config()
    v1 = client.CoreV1Api()

    # Get nodes
    nodes = v1.list_node().items
    node_info = [{
        "name": node.metadata.name,
        "status": node.status.conditions[-1].type,
        "kubelet_version": node.status.node_info.kubelet_version
    } for node in nodes]

    # Get namespaces
    namespaces = v1.list_namespace().items
    namespace_info = [{
        "name": ns.metadata.name,
        "status": ns.status.phase
    } for ns in namespaces]

    return {
        "nodes": node_info,
        "node_count": len(nodes),
        "namespaces": namespace_info,
        "namespace_count": len(namespaces)
    }

Available Workflows

Cluster Health Check

The kubernetes_cluster_health workflow provides a comprehensive health check of your Kubernetes cluster.

Python
from community_tools.kubernetes.workflows import create_cluster_health_workflow

# Create the workflow
cluster_health_workflow = create_cluster_health_workflow()

# Execute the workflow
result = cluster_health_workflow.execute({
    "NAMESPACE": "default",
    "DETAILED": True
})

# Print the health summary
print(result["health_summary"])

Deployment Management

The kubernetes_deployment_rollout_restart workflow provides a way to safely restart deployments.

Python
from community_tools.kubernetes.workflows import create_deployment_rollout_restart_workflow

# Create the workflow
restart_workflow = create_deployment_rollout_restart_workflow()

# Execute the workflow
result = restart_workflow.execute({
    "NAMESPACE": "default",
    "DEPLOYMENT_NAME": "my-app",
    "TIMEOUT_SECONDS": 300
})

# Print the restart report
print(result["deployment_restart_report"])

Integration with Workflows

You can easily incorporate Kubernetes tools into your own workflows:

Python
from kubiya_sdk.workflows.workflow import Workflow, WorkflowNode
from kubiya_sdk.workflows.node_types import NodeType
from community_tools.kubernetes.tools import register_kubernetes_tools

# Register the Kubernetes tools
register_kubernetes_tools()

# Create a workflow that uses Kubernetes tools
deployment_workflow = Workflow(
    id="app-deployment",
    name="Application Deployment",
    description="Deploy and verify an application",
    nodes=[
        WorkflowNode(
            name="deploy_app",
            description="Deploy the application",
            node_type=NodeType.TOOL,
            tool_config={
                "tool_name": "deploy_from_yaml",
                "input_mapping": {
                    "yaml_content": "$params.deployment_yaml",
                    "namespace": "$params.namespace"
                }
            }
        ),
        WorkflowNode(
            name="verify_deployment",
            description="Verify deployment status",
            node_type=NodeType.TOOL,
            tool_config={
                "tool_name": "check_pod_status",
                "input_mapping": {
                    "namespace": "$params.namespace",
                    "label_selector": "app=$params.app_name"
                }
            }
        )
    ]
)

Configuration

The Kubernetes tools require configuration for accessing your cluster. You can set this up using dynamic configuration:

Python
from kubiya_sdk import config_model
from pydantic import BaseModel

@config_model(name="kubernetes_config", description="Kubernetes Configuration")
class KubernetesConfig(BaseModel):
    """Kubernetes Configuration Schema"""
    kubeconfig_path: str = None  # Path to kubeconfig file, or None to use default
    context: str = None  # Kubernetes context to use, or None to use current context
    in_cluster: bool = False  # Whether to use in-cluster configuration