client-go使用之获取pod状态

本文主要介绍如何使用client-go对k8s集群中的Pod进行相关操作，也是自己在工作和学习中使用client-go的相关经验总结

一、初始化Pod连接客户端

本质上是对创建与k8s交互客户端的二次封装，可以基于返回的结构体对象扩展不同的方法，每个方法对应着pod不同的操作，既提高了代码的可读性又避免了在对pod进行不同操作时需要反复初始化客户端的问题

type PodBox struct {
    Clientset    kubernetes.Interface
    Config    *restclient.Config
}

func NewPodBox() (*PodBox, error) {
    kubeconfig := "k8s集群config配置文件"

    //kubecofig转换为clientConfig
    clientConfig,err := clientcmd.NewClientConfigFromBytes([]byte(kubeconfig))
    if err != nil {
        return nil,tracerr.Wrap(err)
    }
    //clientConfig转换为rest.config
    config,err := clientConfig.ClientConfig()
    if err != nil {
        return nil,tracerr.Wrap(err)
    }
    //创建k8s客户端
    cs,err := kubernetes.NewForConfig(config)
    if err != nil {
        return nil,tracerr.Wrap(err)
    }

    podBox := &PodBox{
        Clientset: cs,
        Config: config,
    }
    return podBox,nil
}

二、获取Pod实例

func (box *PodBox) Get(podName, namespace string) (*coreV1.Pod,error) {
    return box.Clientset.CoreV1().Pods(namespace).Get(context.TODO(),podName,metav1.GetOptions{})
}

2.1 获取Pod状态

const (
    ContainersReady string = "ContainersReady"
    PodInitialized  string = "Initialized"
    PodReady   string = "Ready"
    PodScheduled  string = "PodScheduled"
)

const (
    ConditionTrue    string = "True"
    ConditionFalse   string = "False"
    ConditionUnknown string = "Unknown"
)

// GetPodStatus
// 获取Pod状态
// Pending
// Running
// Successed
// Unavailable
// Initializing
// Scheduling
// NotReady
// Failed
func GetPodStatus(pod *coreV1.Pod) string {
    for _,cond := range pod.Status.Conditions {
        if string(cond.Type) == ContainersReady {
            if string(cond.Status) != ConditionTrue {
                return "Unavailable"
            }
        } else if string(cond.Type) == PodInitialized && string(cond.Status) != ConditionTrue {
            return "Initializing"
        } else if string(cond.Type) == PodReady {
            if string(cond.Status) != ConditionTrue {
                return "Unavailable"
            }
            for _,containerState := range pod.Status.ContainerStatuses {
                if !containerState.Ready {
                    return "Unavailable"
                }
            }
        } else if string(cond.Type) == PodScheduled && string(cond.Status) != ConditionTrue {
            return "Scheduling"
        }
    }
    return string(pod.Status.Phase)
}

2.2 获取Pod内容器最大重启次数

func MaxContainerRestarts(pod *coreV1.Pod) int {
    maxRestarts := 0
    for _, c := range pod.Status.ContainerStatuses {
        maxRestarts = integer.IntMax(maxRestarts, int(c.RestartCount))
    }
    return maxRestarts
}

2.3 获取pod内容器日志

// WatchContainerLogWithPodNameAndContainerName
// 根据命名空间和pod名称以及容器名称，从k8s-apiserver读取容器日志流
//入参：
//Container:容器名称
//Follow:跟踪Pod的日志流，默认为false（关闭）对应kubectl logs命令中的 -f 参数
//TailLines:如果设置，则显示从日志末尾开始的行数。如果未指定，则从容器的创建开始或从秒开始或从时间开始显示日志
//Previous:返回以前终止的容器日志。默认为false（关闭）
func WatchContainerLogWithPodNameAndContainerName(clientset *kubernetes.Clientset, namespace, podName,
    containerName string,tailLine int64, previous bool) (io.ReadCloser, error) {
    logOpt := &coreV1.PodLogOptions{
        Container: containerName,
        Follow: true,
        TailLines: &tailLine,
        Previous: previous,
    }
    req := clientset.CoreV1().Pods(namespace).GetLogs(podName,logOpt)
    return req.Stream(context.TODO())
}

三、根据deployment名称获取pod列表

// PodsGetWithDeploymentNameAndNS
// 根据命名空间和deployment名称，从k8s处获取deployment拥有的pod列表
func PodsGetWithDeploymentNameAndNS(clientset *kubernetes.Clientset, namespace,
    deploymentName string) ([]coreV1.Pod, error) {

    label := "app="+deploymentName
    podList,err := clientset.CoreV1().Pods(namespace).List(context.TODO(),metav1.ListOptions{LabelSelector: label})
    if err != nil {
        return nil,tracerr.Wrap(err)
    }

    return podList.Items,nil
}

四、删除Pod

func (box *PodBox) Delete(podName, namespace string) error {
    return box.Clientset.CoreV1().Pods(namespace).Delete(context.TODO(),podName,metav1.DeleteOptions{})
}

五、在Pod的容器中执行命令

func (box *PodBox) Exec(cmd []string, ptyHandler PtyHandler, namespace, podName, containerName string) error {
    defer func() {
        ptyHandler.Done()
    }()
    req := box.Clientset.CoreV1().RESTClient().Post().
        Resource("pods").Name(podName).
        Namespace(namespace).SubResource("exec").
        VersionedParams(&coreV1.PodExecOptions{
            Container: containerName,
            Command: cmd,
            Stdin: !(ptyHandler.Stdin() == nil),
            Stdout: !(ptyHandler.Stdout() == nil),
            Stderr: !(ptyHandler.Stderr() == nil),
            TTY: ptyHandler.Tty(),
        },scheme.ParameterCodec)

    exec,err := remotecommand.NewSPDYExecutor(box.Config,"POST",req.URL())
    if err != nil {
        return tracerr.Wrap(err)
    }

    err = exec.Stream(remotecommand.StreamOptions{
        Stdin: ptyHandler.Stdin(),
        Stdout: ptyHandler.Stdout(),
        Stderr: ptyHandler.Stderr(),
        TerminalSizeQueue: ptyHandler,
        Tty: ptyHandler.Tty(),
    })
    if err != nil {
        return tracerr.Wrap(err)
    }
    return nil
}

六、Pod内文件下载

func (box *PodBox) CopyFromPod(namespace, podName, containerName, filePath string) (io.Reader,error) {
    filedir,filename := filepath.Split(filePath)
    cmd := []string {"sh", "-c", fmt.Sprintf("cd %s && if [ ! -f ./%s ]; then  exit 100; fi && tar cf - %s", filedir,filename,filename)}
    reader,outstream := io.Pipe()
    ts := NewTerminalSession(nil,outstream,outstream)
    go func() {
        defer outstream.Close()
        err := box.Exec(cmd,ts,namespace,podName,containerName)
        if err != nil {
            tracerr.Print(err)
        }
    }()

    return reader,nil
}

type sizeQueue struct {
    resizeChan chan remotecommand.TerminalSize
    doneChan   chan struct{}
}

type TerminalSession struct {
    stdin    io.Reader
    stdout    io.Writer
    stderr    io.Writer
    sizeQueue
    tty    bool
}

func NewTerminalSession(stdin io.Reader,stdout io.Writer,stderr io.Writer) *TerminalSession {
    return &TerminalSession{
        stdin: stdin,
        stdout: stdout,
        stderr: stderr,
        tty: false,
        sizeQueue: sizeQueue{
            resizeChan: make(chan remotecommand.TerminalSize),
            doneChan:   make(chan struct{}),
        },
    }
}