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¶
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