Refactor Remote Host Collection (#1633)

* refactor remote collectors

* add remotecollect params struct

* remove commented checkrbac function

* removed unused function

* add temp comments

* refactor to not require RemoteCollect method per collector

* removed unneeded param

* removed unneeded param

* more refactor

* more refactor

* remove unneeded function

* remove debug print

* fix analyzer results

* move rbac to separate file

* be more specific with rbac function name

* fix imports

* fix node list file

* make k8s rest client config consistent with in cluster collection

* add ctx and otel tracing

* add test for allCollectedData

* move runHostCollectorsInPod to spec instead of metadata

* make generate

* fix broken references to supportbundle metadata

* add e2e tests

* update loader tests

* fix tests

* fix hostos remote collector spec

* update remoteHostCollectrs.yaml

---------

Co-authored-by: Dexter Yan <[email protected]>

Author: diamonwiggins

cmd/troubleshoot/cli/run.go

@@ -110,7 +110,7 @@ func runTroubleshoot(v *viper.Viper, args []string) error {
 	}
 
 	if interactive {
-		if len(mainBundle.Spec.HostCollectors) > 0 && !util.IsRunningAsRoot() {
+		if len(mainBundle.Spec.HostCollectors) > 0 && !util.IsRunningAsRoot() && !mainBundle.Spec.RunHostCollectorsInPod {
 			fmt.Print(cursor.Show())
 			if util.PromptYesNo(util.HOST_COLLECTORS_RUN_AS_ROOT_PROMPT) {
 				fmt.Println("Exiting...")
@@ -184,7 +184,7 @@ func runTroubleshoot(v *viper.Viper, args []string) error {
 		OutputPath:                v.GetString("output"),
 		Redact:                    v.GetBool("redact"),
 		FromCLI:                   true,
-		RunHostCollectorsInPod:    mainBundle.Metadata.RunHostCollectorsInPod,
+		RunHostCollectorsInPod:    mainBundle.Spec.RunHostCollectorsInPod,
 	}
 
 	nonInteractiveOutput := analysisOutput{}
@@ -199,7 +199,7 @@ func runTroubleshoot(v *viper.Viper, args []string) error {
 
 	if len(response.AnalyzerResults) > 0 {
 		if interactive {
-			if err := showInteractiveResults(mainBundle.Metadata.Name, response.AnalyzerResults, response.ArchivePath); err != nil {
+			if err := showInteractiveResults(mainBundle.Name, response.AnalyzerResults, response.ArchivePath); err != nil {
 				interactive = false
 			}
 		} else {
@@ -208,7 +208,7 @@ func runTroubleshoot(v *viper.Viper, args []string) error {
 	}
 
 	if !response.FileUploaded {
-		if appName := mainBundle.Metadata.Labels["applicationName"]; appName != "" {
+		if appName := mainBundle.Labels["applicationName"]; appName != "" {
 			f := `A support bundle for %s has been created in this directory
 named %s. Please upload it on the Troubleshoot page of
 the %s Admin Console to begin analysis.`
@@ -337,11 +337,8 @@ func loadSpecs(ctx context.Context, args []string, client kubernetes.Interface)
 			APIVersion: "troubleshoot.sh/v1beta2",
 			Kind:       "SupportBundle",
 		},
-		Metadata: troubleshootv1beta2.SupportBundleMetadata{
-			ObjectMeta: metav1.ObjectMeta{
-				Name: "merged-support-bundle-spec",
-			},
-			RunHostCollectorsInPod: false,
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "merged-support-bundle-spec",
 		},
 	}
 
@@ -351,11 +348,11 @@ func loadSpecs(ctx context.Context, args []string, client kubernetes.Interface)
 		sb := sb
 		mainBundle = supportbundle.ConcatSpec(mainBundle, &sb)
 		//check if sb has metadata and if it has RunHostCollectorsInPod set to true
-		if !reflect.DeepEqual(sb.Metadata.ObjectMeta, metav1.ObjectMeta{}) && sb.Metadata.RunHostCollectorsInPod {
-			enableRunHostCollectorsInPod = sb.Metadata.RunHostCollectorsInPod
+		if !reflect.DeepEqual(sb.ObjectMeta, metav1.ObjectMeta{}) && sb.Spec.RunHostCollectorsInPod {
+			enableRunHostCollectorsInPod = sb.Spec.RunHostCollectorsInPod
 		}
 	}
-	mainBundle.Metadata.RunHostCollectorsInPod = enableRunHostCollectorsInPod
+	mainBundle.Spec.RunHostCollectorsInPod = enableRunHostCollectorsInPod
 
 	for _, c := range kinds.CollectorsV1Beta2 {
 		mainBundle.Spec.Collectors = util.Append(mainBundle.Spec.Collectors, c.Spec.Collectors)

config/crds/troubleshoot.sh_supportbundles.yaml

@@ -2114,6 +2114,8 @@ spec:
                           type: string
                         exclude:
                           type: BoolString
+                        image:
+                          type: string
                         namespace:
                           type: string
                         podLaunchOptions:
@@ -20313,6 +20315,8 @@ spec:
                       type: object
                   type: object
                 type: array
+              runHostCollectorsInPod:
+                type: boolean
               uri:
                 description: URI optionally defines a location which is the source
                   of this spec to allow updating of the spec at runtime

pkg/apis/troubleshoot/v1beta2/remote_collector_shared.go

@@ -16,6 +16,7 @@ type RemoteCollectorMeta struct {
 type RemoteCPU struct {
 	RemoteCollectorMeta `json:",inline" yaml:",inline"`
 }
+
 type RemoteHostOS struct {
 	RemoteCollectorMeta `json:",inline" yaml:",inline"`
 }

pkg/apis/troubleshoot/v1beta2/supportbundle_types.go

@@ -20,11 +20,6 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-type SupportBundleMetadata struct {
-	metav1.ObjectMeta      `json:",inline" yaml:",inline"`
-	RunHostCollectorsInPod bool `json:"runHostCollectorsInPod,omitempty" yaml:"runHostCollectorsInPod,omitempty"`
-}
-
 // SupportBundleSpec defines the desired state of SupportBundle
 type SupportBundleSpec struct {
 	AfterCollection []*AfterCollection `json:"afterCollection,omitempty" yaml:"afterCollection,omitempty"`
@@ -33,7 +28,8 @@ type SupportBundleSpec struct {
 	Analyzers       []*Analyze         `json:"analyzers,omitempty" yaml:"analyzers,omitempty"`
 	HostAnalyzers   []*HostAnalyze     `json:"hostAnalyzers,omitempty" yaml:"hostAnalyzers,omitempty"`
 	// URI optionally defines a location which is the source of this spec to allow updating of the spec at runtime
-	Uri string `json:"uri,omitempty" yaml:"uri,omitempty"`
+	Uri                    string `json:"uri,omitempty" yaml:"uri,omitempty"`
+	RunHostCollectorsInPod bool   `json:"runHostCollectorsInPod,omitempty" yaml:"runHostCollectorsInPod,omitempty"`
 }
 
 // SupportBundleStatus defines the observed state of SupportBundle
@@ -48,8 +44,8 @@ type SupportBundleStatus struct {
 // SupportBundle is the Schema for the SupportBundles API
 // +k8s:openapi-gen=true
 type SupportBundle struct {
-	metav1.TypeMeta `json:",inline" yaml:",inline"`
-	Metadata        SupportBundleMetadata `json:"metadata,omitempty" yaml:"metadata,omitempty"`
+	metav1.TypeMeta   `json:",inline" yaml:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty" yaml:"metadata,omitempty"`
 
 	Spec   SupportBundleSpec   `json:"spec,omitempty" yaml:"spec,omitempty"`
 	Status SupportBundleStatus `json:"status,omitempty"`

pkg/apis/troubleshoot/v1beta2/zz_generated.deepcopy.go

@@ -481,11 +481,9 @@ func TestCollectClusterResources_CustomResource(t *testing.T) {
 
 	// Create a CR
 	sbObject := troubleshootv1beta2.SupportBundle{
-		Metadata: troubleshootv1beta2.SupportBundleMetadata{
-			ObjectMeta: metav1.ObjectMeta{
-				Name:      "supportbundle",
-				Namespace: "default",
-			},
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "supportbundle",
+			Namespace: "default",
 		},
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "SupportBundle",

pkg/client/troubleshootclientset/typed/troubleshoot/v1beta2/fake/fake_supportbundle.go

@@ -1,14 +1,43 @@
 package collect
 
 import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/pkg/errors"
 	troubleshootv1beta2 "github.com/replicatedhq/troubleshoot/pkg/apis/troubleshoot/v1beta2"
+	"github.com/replicatedhq/troubleshoot/pkg/constants"
+	"golang.org/x/sync/errgroup"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apiserver/pkg/storage/names"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 )
 
 type HostCollector interface {
 	Title() string
 	IsExcluded() (bool, error)
 	Collect(progressChan chan<- interface{}) (map[string][]byte, error)
-	RemoteCollect(progressChan chan<- interface{}) (map[string][]byte, error) // RemoteCollect is used to priviledge pods to collect data from different nodes
+}
+
+type RemoteCollectParams struct {
+	ProgressChan  chan<- interface{}
+	HostCollector *troubleshootv1beta2.HostCollect
+	BundlePath    string
+	ClientConfig  *rest.Config // specify actual type
+	Image         string
+	PullPolicy    string        // specify actual type if needed
+	Timeout       time.Duration // specify duration type if needed
+	LabelSelector string
+	NamePrefix    string
+	Namespace     string
+	Title         string
 }
 
 func GetHostCollector(collector *troubleshootv1beta2.HostCollect, bundlePath string) (HostCollector, bool) {
@@ -81,3 +110,131 @@ func hostCollectorTitleOrDefault(meta troubleshootv1beta2.HostCollectorMeta, def
 	}
 	return defaultTitle
 }
+
+func RemoteHostCollect(ctx context.Context, params RemoteCollectParams) (map[string][]byte, error) {
+	scheme := runtime.NewScheme()
+	if err := corev1.AddToScheme(scheme); err != nil {
+		return nil, errors.Wrap(err, "failed to add runtime scheme")
+	}
+
+	client, err := kubernetes.NewForConfig(params.ClientConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	runner := &podRunner{
+		client:       client,
+		scheme:       scheme,
+		image:        params.Image,
+		pullPolicy:   params.PullPolicy,
+		waitInterval: remoteCollectorDefaultInterval,
+	}
+
+	// Get all the nodes where we should run.
+	nodes, err := listNodesNamesInSelector(ctx, client, params.LabelSelector)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get the list of nodes matching a nodeSelector")
+	}
+
+	if params.NamePrefix == "" {
+		params.NamePrefix = remoteCollectorNamePrefix
+	}
+
+	result, err := runRemote(ctx, runner, nodes, params.HostCollector, names.SimpleNameGenerator, params.NamePrefix, params.Namespace)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to run collector remotely")
+	}
+
+	allCollectedData := mapCollectorResultToOutput(result, params)
+	output := NewResult()
+
+	// save the first result we find in the node and save it
+	for node, result := range allCollectedData {
+		var nodeResult map[string]string
+		if err := json.Unmarshal(result, &nodeResult); err != nil {
+			return nil, errors.Wrap(err, "failed to marshal node results")
+		}
+
+		for file, collectorResult := range nodeResult {
+			directory := filepath.Dir(file)
+			fileName := filepath.Base(file)
+			// expected file name for remote collectors will be the normal path separated by / and the node name
+			output.SaveResult(params.BundlePath, fmt.Sprintf("%s/%s/%s", directory, node, fileName), bytes.NewBufferString(collectorResult))
+		}
+	}
+
+	// check if NODE_LIST_FILE exists
+	_, err = os.Stat(constants.NODE_LIST_FILE)
+	// if it not exists, save the nodes list
+	if err != nil {
+		nodesBytes, err := json.MarshalIndent(HostOSInfoNodes{Nodes: nodes}, "", " ")
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to marshal host os info nodes")
+		}
+		output.SaveResult(params.BundlePath, constants.NODE_LIST_FILE, bytes.NewBuffer(nodesBytes))
+	}
+	return output, nil
+}
+
+func runRemote(ctx context.Context, runner runner, nodes []string, collector *troubleshootv1beta2.HostCollect, nameGenerator names.NameGenerator, namePrefix string, namespace string) (map[string][]byte, error) {
+	g, ctx := errgroup.WithContext(ctx)
+	results := make(chan map[string][]byte, len(nodes))
+
+	for _, node := range nodes {
+		node := node
+		g.Go(func() error {
+			// May need to evaluate error and log warning.  Otherwise any error
+			// here will cancel the context of other goroutines and no results
+			// will be returned.
+			return runner.run(ctx, collector, namespace, nameGenerator.GenerateName(namePrefix+"-"), node, results)
+		})
+	}
+
+	// Wait for all collectors to complete or return the first error.
+	if err := g.Wait(); err != nil {
+		return nil, errors.Wrap(err, "failed remote collection")
+	}
+	close(results)
+
+	output := make(map[string][]byte)
+	for result := range results {
+		r := result
+		for k, v := range r {
+			output[k] = v
+		}
+	}
+
+	return output, nil
+}
+
+func mapCollectorResultToOutput(result map[string][]byte, params RemoteCollectParams) map[string][]byte {
+	allCollectedData := make(map[string][]byte)
+
+	for k, v := range result {
+		if curBytes, ok := allCollectedData[k]; ok {
+			var curResults map[string]string
+			if err := json.Unmarshal(curBytes, &curResults); err != nil {
+				params.ProgressChan <- errors.Errorf("failed to read existing results for collector %s: %v\n", params.Title, err)
+				continue
+			}
+			var newResults map[string]string
+			if err := json.Unmarshal(v, &newResults); err != nil {
+				params.ProgressChan <- errors.Errorf("failed to read new results for collector %s: %v\n", params.Title, err)
+				continue
+			}
+			for file, data := range newResults {
+				curResults[file] = data
+			}
+			combinedResults, err := json.Marshal(curResults)
+			if err != nil {
+				params.ProgressChan <- errors.Errorf("failed to combine results for collector %s: %v\n", params.Title, err)
+				continue
+			}
+			allCollectedData[k] = combinedResults
+		} else {
+			allCollectedData[k] = v
+		}
+
+	}
+	return allCollectedData
+}