Tail Kubernetes pod logs in Golang, reconnect if the stream times out/fails

tail.go

package main

import (
	"bufio"
	"context"
	"fmt"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/tools/clientcmd"
	"time"
)

func main() {
	cli := buildK8sCli()

	podNs := "default"
	podName := "loggen"

	stopCh := make(chan bool)
	go tailLogsWithRetry(cli, podNs, podName, stopCh)

	// sleep for some time before stopping
	time.Sleep(10 * time.Second)

	// stop tailing logs after sleeping
	stopCh <- true

	// sleep for some more time, verify that log tailing is stopped
	time.Sleep(5 * time.Second)
}

func buildK8sCli() *kubernetes.Clientset {
	r := clientcmd.NewDefaultClientConfigLoadingRules()
	c, _ := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(r,
		&clientcmd.ConfigOverrides{}).ClientConfig()
	cli, _ := kubernetes.NewForConfig(c)
	return cli
}

func tailLogsWithRetry(cli *kubernetes.Clientset, ns string, name string, stopCh <-chan bool) {
	for {
		retry, _ := tailLogs(cli, ns, name, stopCh)
		if !retry {
			return
		}
	}
}

func tailLogs(cli *kubernetes.Clientset, ns string, name string, stopCh <-chan bool) (bool, error) {
	tailLines := int64(10)
	s, err := cli.CoreV1().Pods(ns).
		GetLogs(name, &corev1.PodLogOptions{
			Follow: true, TailLines: &tailLines,
		}).Stream(context.TODO())
	if err != nil {
		fmt.Println("error, will retry")
		return true, err
	}

	defer func() { _ = s.Close() }()

	sc := bufio.NewScanner(s)
	for {
		select {
		case <-stopCh:
			fmt.Println("stopped, will not retry")
			return false, nil
		default:
			if !sc.Scan() {
				fmt.Println("error, will retry")
				return true, nil
			}
			fmt.Println(sc.Text())
		}
	}
}