基于K8S API管理工作负载

下载kubeconf文件

基于K8S API管理工作负载，首先需要获取kubeconf文件，具体参考这里。

我们将文件保存为：~/myconf-eb-kubeconfig

代码示例

以下给出通过api查看default names下所有pod的示例代码。

Python代码

首先安装 kubernetes pip包：

pip install kubernetes

示例代码如下：

from kubernetes import client, config
from kubernetes.client.rest import ApiException
import os

def list_pods_in_default_namespace(kubeconfig_path):
    """
    查看 default namespace 中的 pods
    """
    try:
	# 加载指定的 kubeconfig 文件
	config.load_kube_config(config_file=kubeconfig_path)

	# 创建 CoreV1Api 实例
	v1 = client.CoreV1Api()

	# 获取 default namespace 中的 pods
	print(f"正在从 kubeconfig: {kubeconfig_path} 获取 default namespace 中的 pods...")
	pods = v1.list_namespaced_pod(namespace="default")

	print(f"\n在 default namespace 中找到 {len(pods.items)} 个 pod:")
	print("=" * 80)

	# 打印 pod 信息
	for i, pod in enumerate(pods.items, 1):
	    print(f"{i}. Pod名称: {pod.metadata.name}")
	    print(f"   命名空间: {pod.metadata.namespace}")
	    print(f"   状态: {pod.status.phase}")
	    print(f"   节点: {pod.spec.node_name if pod.spec.node_name else '未调度'}")
	    print(f"   IP地址: {pod.status.pod_ip}")
	    print(f"   创建时间: {pod.metadata.creation_timestamp}")
	    print(f"   标签: {pod.metadata.labels}")
	    print("-" * 80)

    except ApiException as e:
	print(f"Kubernetes API 错误: {e}")
    except Exception as e:
	print(f"发生错误: {e}")

def main():
    # 指定 kubeconfig 文件路径
    kubeconfig_path = os.path.expanduser("~/myconf-eb-kubeconfig")

    # 检查文件是否存在
    if not os.path.exists(kubeconfig_path):
	print(f"错误: kubeconfig 文件不存在: {kubeconfig_path}")
	return

    # 列出 default namespace 中的 pods
    list_pods_in_default_namespace(kubeconfig_path)

if __name__ == "__main__":
    main()

Go代码

示例代码如下：

package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"k8s.io/client-go/util/homedir"
)

func main() {
	// 设置 kubeconfig 文件路径
	kubeconfigPath := filepath.Join(homedir.HomeDir(), "myconf-eb-kubeconfig")

	// 检查文件是否存在
	if _, err := os.Stat(kubeconfigPath); os.IsNotExist(err) {
		log.Fatalf("错误: kubeconfig 文件不存在: %s", kubeconfigPath)
	}

	// 加载 kubeconfig 配置
	config, err := clientcmd.BuildConfigFromFlags("", kubeconfigPath)
	if err != nil {
		log.Fatalf("加载 kubeconfig 失败: %v", err)
	}

	// 创建 Kubernetes 客户端
	clientset, err := kubernetes.NewForConfig(config)
	if err != nil {
		log.Fatalf("创建 Kubernetes 客户端失败: %v", err)
	}

	// 获取 default namespace 中的 pods
	fmt.Printf("正在从 kubeconfig: %s 获取 default namespace 中的 pods...\n", kubeconfigPath)

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	pods, err := clientset.CoreV1().Pods("default").List(ctx, metav1.ListOptions{})
	if err != nil {
		log.Fatalf("获取 pods 失败: %v", err)
	}

	fmt.Printf("\n在 default namespace 中找到 %d 个 pod:\n", len(pods.Items))
	fmt.Println("================================================================================")

	for i, pod := range pods.Items {
		fmt.Printf("%d. Pod名称: %s\n", i+1, pod.Name)
		fmt.Printf("   命名空间: %s\n", pod.Namespace)
		fmt.Printf("   状态: %s\n", pod.Status.Phase)
		fmt.Printf("   节点: %s\n", pod.Spec.NodeName)
		fmt.Printf("   IP地址: %s\n", pod.Status.PodIP)
		fmt.Printf("   创建时间: %s\n", pod.CreationTimestamp.Format("2006-01-02 15:04:05"))
		fmt.Printf("   标签: %v\n", pod.Labels)
		fmt.Println("--------------------------------------------------------------------------------")
	}
}