如何使用 Go 从 Kubernetes 获取日志？

Question

我正在寻找如何使用 Go 从 Kubernetes 集群中的 pod 获取日志的解决方案。我看过“https://github.com/kubernetes/client-go”和“https://godoc.org/sigs.k8s.io/controller-runtime/pkg/client”，但不明白如何将它们用于此目的。除了日志之外，我在 K8S 中获取 pod 或任何其他对象的信息都没有问题。

例如，我使用“https://godoc.org/sigs.k8s.io/controller-runtime/pkg/client#example-Client--Get”中的 Get() 来获取 K8S 作业信息：

found := &batchv1.Job{}
err = r.client.Get(context.TODO(), types.NamespacedName{Name: job.Name, Namespace: job.Namespace}, found)

请分享您现在如何获取 pod 的日志。 任何建议将不胜感激！

更新： Kubernetes go client api for log of a particular pod 中提供的解决方案已过时。它有一些提示，但与当前的库相比不是最新的。

Answer 1

控制器运行时客户端库尚不支持 /status 以外的子资源，因此您必须使用 client-go，如另一个问题所示。

Answer 2

这是我们最终使用 client-go 库得出的结论：

func getPodLogs(pod corev1.Pod) string {
    podLogOpts := corev1.PodLogOptions{}
    config, err := rest.InClusterConfig()
    if err != nil {
        return "error in getting config"
    }
    // creates the clientset
    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        return "error in getting access to K8S"
    }
    req := clientset.CoreV1().Pods(pod.Namespace).GetLogs(pod.Name, &podLogOpts)
    podLogs, err := req.Stream()
    if err != nil {
        return "error in opening stream"
    }
    defer podLogs.Close()

    buf := new(bytes.Buffer)
    _, err = io.Copy(buf, podLogs)
    if err != nil {
        return "error in copy information from podLogs to buf"
    }
    str := buf.String()

    return str
}

我希望它会对某人有所帮助。请分享您对如何从 Kubernetes 中的 pod 获取日志的想法或解决方案。

Answer 3

@Emixam23

我相信你会发现这个 sn-p 很有用。

如何获取 Pod 的动态名称？

import  metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

    labelSelector := metav1.LabelSelector{MatchLabels: map[string]string{<LABEL_KEY>: <LABEL_VALUE>}}
    listOptions := metav1.ListOptions{
        LabelSelector: labels.Set(labelSelector.MatchLabels).String(),
    }
    pod, err := k8sClient.CoreV1().Pods(<NAMESPACE>).List(listOptions)
    podName := pod.Items[0].ObjectMeta.Name

Answer 4

如果你想在client-go v11.0.0+读流，代码是这样的，你可以自己创建clientset：

func GetPodLogs(namespace string, podName string, containerName string, follow bool) error {
    count := int64(100)
    podLogOptions := v1.PodLogOptions{
        Container: containerName,
        Follow:    follow,
        TailLines: &count,
    }

    podLogRequest := clientSet.CoreV1().
        Pods(namespace).
        GetLogs(podName, &podLogOptions)
    stream, err := podLogRequest.Stream(context.TODO())
    if err != nil {
        return err
    }
    defer stream.Close()

    for {
        buf := make([]byte, 2000)
        numBytes, err := stream.Read(buf)
        if numBytes == 0 {
            continue
        }
        if err == io.EOF {
            break
        }
        if err != nil {
            return err
        }
        message := string(buf[:numBytes])
        fmt.Print(message)
    }
    return nil
}

Answer 5

结合在其他地方和此处找到的一些答案以流式传输（拖尾）所有容器（包括 init）的日志：

func GetPodLogs(namespace string, podName string) {
    pod, err := clientSet.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{})
    if err != nil {
        return err
    }
    wg := &sync.WaitGroup{}
    functionList := []func(){}
    for _, container := range append(pod.Spec.InitContainers, pod.Spec.Containers...) {
        podLogOpts := v1.PodLogOptions{}
        podLogOpts.Follow = true
        podLogOpts.TailLines = &[]int64{int64(100)}[0]
        podLogOpts.Container = container.Name
        podLogs, err := clientSet.CoreV1().Pods(namespace).GetLogs(podName, &podLogOpts).Stream(ctx)
        if err != nil {
            return err
        }
        defer podLogs.Close()
        functionList = append(functionList, func() {
            defer wg.Done()
            reader := bufio.NewScanner(podLogs)
            for reader.Scan() {
                select {
                case <-ctx.Done():
                    return
                default:
                    line := reader.Text()
                    fmt.Println(worker+"/"+podLogOpts.Container, line)
                }
            }
            log.Printf("INFO log EOF " + reader.Err().Error() + ": " + worker + "/" + podLogOpts.Container)
        })
    }

    wg.Add(len(functionList))
    for _, f := range functionList {
        go f()
    }
    wg.Wait()
    return nil
}