Support collect logs for failed agents and controller for supportbundle

When the normal supportbundle api failed for some nodes or controller,
use kubernetes api instead to collect logs. Also, in either case,
clusterinfo will always be gathered first.

Signed-off-by: Hang Yan <[email protected]>

Author: hangyan

pkg/antctl/raw/supportbundle/command.go

@@ -20,6 +20,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+
 	"net"
 	"os"
 	"path"
@@ -29,21 +30,27 @@ import (
 	"time"
 
 	"github.com/cheggaaa/pb/v3"
+	"github.com/spf13/afero"
 	"github.com/spf13/cobra"
 	"golang.org/x/sync/errgroup"
 	"golang.org/x/time/rate"
 	"gopkg.in/yaml.v2"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8sruntime "k8s.io/apimachinery/pkg/runtime"
+	utilerror "k8s.io/apimachinery/pkg/util/errors"
+	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/klog/v2"
 
 	"antrea.io/antrea/pkg/antctl/raw"
 	"antrea.io/antrea/pkg/antctl/runtime"
+	"antrea.io/antrea/pkg/apis/crd/v1beta1"
 	systemv1beta1 "antrea.io/antrea/pkg/apis/system/v1beta1"
 	antrea "antrea.io/antrea/pkg/client/clientset/versioned"
+	"antrea.io/antrea/pkg/util/compress"
 	"antrea.io/antrea/pkg/util/ip"
 	"antrea.io/antrea/pkg/util/k8s"
 )
@@ -59,6 +66,10 @@ const (
 // Command is the support bundle command implementation.
 var Command *cobra.Command
 
+var (
+	defaultFS = afero.NewOsFs()
+)
+
 var option = &struct {
 	dir            string
 	labelSelector  string
@@ -533,6 +544,20 @@ func controllerRemoteRunE(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("failed to create clientset: %w", err)
 	}
 
+	if err := os.MkdirAll(option.dir, 0700|os.ModeDir); err != nil {
+		return fmt.Errorf("error when creating output dir: %w", err)
+	}
+
+	f, err := os.Create(filepath.Join(option.dir, "clusterinfo"))
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	err = getClusterInfo(f, k8sClientset)
+	if err != nil {
+		return err
+	}
+
 	var controllerClient *rest.RESTClient
 	var agentClients map[string]*rest.RESTClient
 
@@ -577,29 +602,17 @@ func controllerRemoteRunE(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("no matched Nodes found to collect agent bundles")
 	}
 
-	if err := os.MkdirAll(option.dir, 0700|os.ModeDir); err != nil {
-		return fmt.Errorf("error when creating output dir: %w", err)
-	}
 	amount := len(agentClients) * 2
 	if controllerClient != nil {
 		amount += 2
 	}
 	bar := barTmpl.Start(amount)
 	defer bar.Finish()
 	defer bar.Set("prefix", "Finish ")
-	f, err := os.Create(filepath.Join(option.dir, "clusterinfo"))
-	if err != nil {
-		return err
-	}
-	defer f.Close()
-	err = getClusterInfo(f, k8sClientset)
-	if err != nil {
-		return err
-	}
 
 	results := requestAll(agentClients, controllerClient, bar)
 	results = downloadAll(agentClients, controllerClient, dir, bar, results)
-	return processResults(results, dir)
+	return processResults(antreaClientset, k8sClientset, results, dir)
 }
 
 func genErrorMsg(resultMap map[string]error) string {
@@ -611,8 +624,9 @@ func genErrorMsg(resultMap map[string]error) string {
 }
 
 // processResults will output the failed nodes and their reasons if any. If no data was collected,
-// error is returned, otherwise will return nil.
-func processResults(resultMap map[string]error, dir string) error {
+// error is returned, otherwise will return nil. For failed nodes and controller, will also try to get logs from
+// kubernetes api.
+func processResults(antreaClientset antrea.Interface, k8sClient kubernetes.Interface, resultMap map[string]error, dir string) error {
 	resultStr := ""
 	var failedNodes []string
 	allFailed := true
@@ -628,7 +642,8 @@ func processResults(resultMap map[string]error, dir string) error {
 		}
 	}
 
-	if resultMap[""] != nil {
+	controllerFail := resultMap[""] != nil
+	if controllerFail {
 		fmt.Println("Controller Info Failed Reason: " + resultMap[""].Error())
 	}
 
@@ -641,9 +656,184 @@ func processResults(resultMap map[string]error, dir string) error {
 		err = writeFailedNodes(dir, failedNodes)
 	}
 
+	if controllerFail {
+		err := downloadControllerBundleFromKubernetes(antreaClientset, k8sClient, dir)
+		if err != nil {
+			fmt.Println("Failed to download controller logs from kubernetes api: " + err.Error())
+		} else {
+			allFailed = false
+		}
+	}
+
+	// download logs from kubernetes api
+	if failedNodes != nil {
+		err := downloadAgentBundleFromKubernetes(antreaClientset, k8sClient, failedNodes, dir)
+		if err != nil {
+			fmt.Println("Failed to download agent logs from kubernetes api: " + err.Error())
+		} else {
+			allFailed = false
+		}
+	}
+
 	if allFailed {
 		return fmt.Errorf("no data was collected: %s", genErrorMsg(resultMap))
 	} else {
 		return err
 	}
 }
+
+func downloadControllerBundleFromKubernetes(antreaClientset antrea.Interface, k8sClient kubernetes.Interface, dir string) error {
+	var errors []error
+	controllerInfo, err := antreaClientset.CrdV1beta1().AntreaControllerInfos().Get(context.TODO(), "antrea-controller", metav1.GetOptions{})
+	if err != nil {
+		errors = append(errors, err)
+	}
+
+	pods, err := k8sClient.CoreV1().Pods("kube-system").List(context.TODO(), metav1.ListOptions{
+		ResourceVersion: "0",
+		LabelSelector:   "app=antrea,component=antrea-controller",
+	})
+	if err != nil {
+		return err
+	}
+
+	for _, pod := range pods.Items {
+		tmpDir, err := afero.TempDir(defaultFS, "", "bundle_tmp_")
+		if err != nil {
+			errors = append(errors, err)
+			continue
+		}
+		defer defaultFS.RemoveAll(tmpDir)
+
+		if controllerInfo != nil {
+			data, err := yaml.Marshal(controllerInfo)
+			if err != nil {
+				errors = append(errors, err)
+			} else {
+				err = afero.WriteFile(defaultFS, filepath.Join(tmpDir, "controllerinfo"), data, 0644)
+				errors = append(errors, err)
+			}
+		}
+
+		err = downloadPodLogs(k8sClient, "controller", pod.Namespace, pod.Name, []string{"antrea-controller"}, dir, tmpDir)
+		errors = append(errors, err)
+	}
+	return utilerror.NewAggregate(errors)
+}
+
+func downloadAgentBundleFromKubernetes(antreaClientset antrea.Interface, k8sClient kubernetes.Interface, failedNodes []string, dir string) error {
+	var errors []error
+	agentInfoList, err := antreaClientset.CrdV1beta1().AntreaAgentInfos().List(context.TODO(), metav1.ListOptions{ResourceVersion: "0"})
+	if err != nil {
+		// in case when there are no agentInfo exist, but there are still pods running. we can ignore this error
+		// and get pod logs
+		errors = append(errors, err)
+	}
+
+	agentInfoMap := map[string]v1beta1.AntreaAgentInfo{}
+	for _, agentInfo := range agentInfoList.Items {
+		agentInfoMap[agentInfo.Name] = agentInfo
+	}
+
+	pods, err := k8sClient.CoreV1().Pods("kube-system").List(context.TODO(), metav1.ListOptions{
+		ResourceVersion: "0",
+		LabelSelector:   "app=antrea,component=antrea-agent",
+	})
+	if err != nil {
+		return err
+	}
+
+	failedNodesSet := sets.NewString(failedNodes...)
+
+	for _, pod := range pods.Items {
+
+		tmpDir, err := afero.TempDir(defaultFS, "", "bundle_tmp_")
+		if err != nil {
+			errors = append(errors, err)
+			continue
+		}
+		defer defaultFS.RemoveAll(tmpDir)
+
+		if !failedNodesSet.Has(pod.Spec.NodeName) {
+			continue
+		}
+
+		if agentInfo, ok := agentInfoMap[pod.Spec.NodeName]; ok {
+			data, err := yaml.Marshal(agentInfo)
+			if err != nil {
+				errors = append(errors, err)
+			}
+			err = afero.WriteFile(defaultFS, filepath.Join(tmpDir, "agentinfo"), data, 0644)
+			errors = append(errors, err)
+		}
+
+		var containerNames []string
+		for _, container := range pod.Spec.Containers {
+			containerNames = append(containerNames, container.Name)
+		}
+		for _, container := range pod.Spec.InitContainers {
+			containerNames = append(containerNames, container.Name)
+		}
+
+		err = downloadPodLogs(k8sClient, "agent_"+pod.Spec.NodeName, pod.Namespace, pod.Name, containerNames, dir, tmpDir)
+		errors = append(errors, err)
+
+	}
+	return utilerror.NewAggregate(errors)
+
+}
+
+func downloadPodLogs(k8sClient kubernetes.Interface, comp string, namespace string, podName string, containers []string, dir string, tmpDir string) error {
+	var errors []error
+
+	for _, containerName := range containers {
+		containerDirName := containerName
+		if strings.HasPrefix(containerName, "antrea-") {
+			containerDirName = strings.ReplaceAll(containerName, "antrea-", "")
+		}
+
+		podLogDir := filepath.Join(tmpDir, "logs", containerDirName)
+		err := os.MkdirAll(podLogDir, 0755)
+		if err != nil {
+			return err
+		}
+
+		fileName := filepath.Join(podLogDir, containerName+".log")
+		f, err := os.Create(fileName)
+		if err != nil {
+			errors = append(errors, err)
+			continue
+		}
+
+		logOption := &corev1.PodLogOptions{
+			Container: containerName,
+		}
+		logs := k8sClient.CoreV1().Pods(namespace).GetLogs(podName, logOption)
+		logStream, err := logs.Stream(context.TODO())
+		if err != nil {
+			errors = append(errors, err)
+			f.Close()
+			continue
+		}
+
+		_, err = io.Copy(f, logStream)
+		errors = append(errors, err)
+		err = logStream.Close()
+		errors = append(errors, err)
+		f.Close()
+	}
+
+	gzFileName := filepath.Join(dir, comp+".tar.gz")
+
+	f, err := os.Create(gzFileName)
+	if err != nil {
+		errors = append(errors, err)
+	} else {
+		_, err := compress.PackDir(tmpDir, f)
+		errors = append(errors, err)
+		f.Close()
+	}
+
+	return utilerror.NewAggregate(errors)
+
+}

pkg/util/compress/compress.go

@@ -0,0 +1,69 @@
+// Copyright 2022 Antrea Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package compress
+
+import (
+	"archive/tar"
+	"compress/gzip"
+	"crypto/sha256"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/spf13/afero"
+)
+
+var (
+	defaultFS = afero.NewOsFs()
+)
+
+// PackDir will compress a dir to a tar.gz file
+func PackDir(dir string, writer io.Writer) ([]byte, error) {
+	hash := sha256.New()
+	gzWriter := gzip.NewWriter(io.MultiWriter(hash, writer))
+	defer gzWriter.Close()
+	targzWriter := tar.NewWriter(gzWriter)
+	defer targzWriter.Close()
+	err := afero.Walk(defaultFS, dir, func(filePath string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if !info.Mode().IsRegular() || info.IsDir() {
+			return nil
+		}
+		header, err := tar.FileInfoHeader(info, info.Name())
+		if err != nil {
+			return err
+		}
+		header.Name = strings.TrimPrefix(strings.ReplaceAll(filePath, dir, ""), string(filepath.Separator))
+		err = targzWriter.WriteHeader(header)
+		if err != nil {
+			return err
+		}
+		f, err := defaultFS.Open(filePath)
+		if err != nil {
+			return err
+		}
+		defer f.Close()
+		_, err = io.Copy(targzWriter, f)
+		return err
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return hash.Sum(nil), nil
+}