Merge pull request #437 from replicatedhq/divolgin/memory

Save collector data to disk directly

Author: divolgin

pkg/apis/troubleshoot/v1beta2/collector_shared.go

@@ -424,6 +424,10 @@ func (c *Collect) GetName() string {
 		name = c.Exec.CollectorName
 		selector = strings.Join(c.Exec.Selector, ",")
 	}
+	if c.Data != nil {
+		collector = "data"
+		name = c.Data.CollectorName
+	}
 	if c.Copy != nil {
 		collector = "copy"
 		name = c.Copy.CollectorName

pkg/collect/ceph.go

@@ -1,6 +1,7 @@
 package collect
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"path"
@@ -89,7 +90,7 @@ var CephCommands = []CephCommand{
 	},
 }
 
-func Ceph(c *Collector, cephCollector *troubleshootv1beta2.Ceph) (map[string][]byte, error) {
+func Ceph(c *Collector, cephCollector *troubleshootv1beta2.Ceph) (CollectorResult, error) {
 	ctx := context.TODO()
 
 	if cephCollector.Namespace == "" {
@@ -101,16 +102,16 @@ func Ceph(c *Collector, cephCollector *troubleshootv1beta2.Ceph) (map[string][]b
 		return nil, err
 	}
 
-	final := map[string][]byte{}
+	output := NewResult()
 	var multiErr *multierror.Error
 	for _, command := range CephCommands {
 		results, err := cephCommandExec(ctx, c, cephCollector, pod, command)
 		multiErr = multierror.Append(multiErr, errors.Wrapf(err, "failed to exec command %s", command.ID))
-		for fileName, output := range results {
-			final[fileName] = output
+		for fileName, r := range results {
+			output.SaveResult(c.BundlePath, fileName, bytes.NewBuffer(r))
 		}
 	}
-	return final, nil
+	return output, nil
 }
 
 func cephCommandExec(ctx context.Context, c *Collector, cephCollector *troubleshootv1beta2.Ceph, pod *corev1.Pod, command CephCommand) (map[string][]byte, error) {

pkg/collect/cluster_info.go

@@ -1,6 +1,7 @@
 package collect
 
 import (
+	"bytes"
 	"encoding/json"
 	"path/filepath"
 
@@ -14,23 +15,20 @@ type ClusterVersion struct {
 	String string        `json:"string"`
 }
 
-func ClusterInfo(c *Collector) (map[string][]byte, error) {
+func ClusterInfo(c *Collector) (CollectorResult, error) {
 	client, err := kubernetes.NewForConfig(c.ClientConfig)
 	if err != nil {
 		return nil, errors.Wrap(err, "Failed to create kubernetes clientset")
 	}
 
-	clusterInfoOutput := map[string][]byte{}
+	output := NewResult()
 
-	// cluster version
 	clusterVersion, clusterErrors := clusterVersion(client)
-	clusterInfoOutput[filepath.Join("cluster-info", "cluster_version.json")] = clusterVersion
-	clusterInfoOutput[filepath.Join("cluster-info", "errors.json")], err = marshalNonNil(clusterErrors)
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to marshal errors")
-	}
 
-	return clusterInfoOutput, nil
+	output.SaveResult(c.BundlePath, filepath.Join("cluster-info", "cluster_version.json"), bytes.NewBuffer(clusterVersion))
+	output.SaveResult(c.BundlePath, filepath.Join("cluster-info", "errors.json"), marshalErrors(clusterErrors))
+
+	return output, nil
 }
 
 func clusterVersion(client *kubernetes.Clientset) ([]byte, []string) {

pkg/collect/cluster_resources.go

@@ -1,6 +1,7 @@
 package collect
 
 import (
+	"bytes"
 	"context"
 	"encoding/base64"
 	"encoding/json"
@@ -17,40 +18,31 @@ import (
 	"k8s.io/client-go/kubernetes"
 )
 
-func ClusterResources(c *Collector, clusterResourcesCollector *troubleshootv1beta2.ClusterResources) (map[string][]byte, error) {
+func ClusterResources(c *Collector, clusterResourcesCollector *troubleshootv1beta2.ClusterResources) (CollectorResult, error) {
 	client, err := kubernetes.NewForConfig(c.ClientConfig)
 	if err != nil {
 		return nil, err
 	}
 
 	ctx := context.Background()
-	clusterResourcesOutput := map[string][]byte{}
+	output := NewResult()
 
 	// namespaces
 	var namespaceNames []string
 	if len(clusterResourcesCollector.Namespaces) > 0 {
 		namespaces, namespaceErrors := getNamespaces(ctx, client, clusterResourcesCollector.Namespaces)
 		namespaceNames = clusterResourcesCollector.Namespaces
-		clusterResourcesOutput["cluster-resources/namespaces.json"] = namespaces
-		clusterResourcesOutput["cluster-resources/namespaces-errors.json"], err = marshalNonNil(namespaceErrors)
-		if err != nil {
-			return nil, err
-		}
+		output.SaveResult(c.BundlePath, "cluster-resources/namespaces.json", bytes.NewBuffer(namespaces))
+		output.SaveResult(c.BundlePath, "cluster-resources/namespaces-errors.json", marshalErrors(namespaceErrors))
 	} else if c.Namespace != "" {
 		namespace, namespaceErrors := getNamespace(ctx, client, c.Namespace)
-		clusterResourcesOutput["cluster-resources/namespaces.json"] = namespace
-		clusterResourcesOutput["cluster-resources/namespaces-errors.json"], err = marshalNonNil(namespaceErrors)
-		if err != nil {
-			return nil, err
-		}
+		output.SaveResult(c.BundlePath, "cluster-resources/namespaces.json", bytes.NewBuffer(namespace))
+		output.SaveResult(c.BundlePath, "cluster-resources/namespaces-errors.json", marshalErrors(namespaceErrors))
 		namespaceNames = append(namespaceNames, c.Namespace)
 	} else {
 		namespaces, namespaceList, namespaceErrors := getAllNamespaces(ctx, client)
-		clusterResourcesOutput["cluster-resources/namespaces.json"] = namespaces
-		clusterResourcesOutput["cluster-resources/namespaces-errors.json"], err = marshalNonNil(namespaceErrors)
-		if err != nil {
-			return nil, err
-		}
+		output.SaveResult(c.BundlePath, "cluster-resources/namespaces.json", bytes.NewBuffer(namespaces))
+		output.SaveResult(c.BundlePath, "cluster-resources/namespaces-errors.json", marshalErrors(namespaceErrors))
 		if namespaceList != nil {
 			for _, namespace := range namespaceList.Items {
 				namespaceNames = append(namespaceNames, namespace.Name)
@@ -61,168 +53,117 @@ func ClusterResources(c *Collector, clusterResourcesCollector *troubleshootv1bet
 	// pods
 	pods, podErrors := pods(ctx, client, namespaceNames)
 	for k, v := range pods {
-		clusterResourcesOutput[path.Join("cluster-resources/pods", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/pods-errors.json"], err = marshalNonNil(podErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/pods", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/pods-errors.json", marshalErrors(podErrors))
 
 	// services
 	services, servicesErrors := services(ctx, client, namespaceNames)
 	for k, v := range services {
-		clusterResourcesOutput[path.Join("cluster-resources/services", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/services-errors.json"], err = marshalNonNil(servicesErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/services", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/services-errors.json", marshalErrors(servicesErrors))
 
 	// deployments
 	deployments, deploymentsErrors := deployments(ctx, client, namespaceNames)
 	for k, v := range deployments {
-		clusterResourcesOutput[path.Join("cluster-resources/deployments", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/deployments-errors.json"], err = marshalNonNil(deploymentsErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/deployments", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/deployments-errors.json", marshalErrors(deploymentsErrors))
 
 	// statefulsets
 	statefulsets, statefulsetsErrors := statefulsets(ctx, client, namespaceNames)
 	for k, v := range statefulsets {
-		clusterResourcesOutput[path.Join("cluster-resources/statefulsets", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/statefulsets-errors.json"], err = marshalNonNil(statefulsetsErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/statefulsets", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/statefulsets-errors.json", marshalErrors(statefulsetsErrors))
 
 	// jobs
 	jobs, jobsErrors := jobs(ctx, client, namespaceNames)
 	for k, v := range jobs {
-		clusterResourcesOutput[path.Join("cluster-resources/jobs", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/jobs-errors.json"], err = marshalNonNil(jobsErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/jobs", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/jobs-errors.json", marshalErrors(jobsErrors))
 
 	// cronJobs
 	cronJobs, cronJobsErrors := cronJobs(ctx, client, namespaceNames)
 	for k, v := range cronJobs {
-		clusterResourcesOutput[path.Join("cluster-resources/cronjobs", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/cronjobs-errors.json"], err = marshalNonNil(cronJobsErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/cronjobs", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/cronjobs-errors.json", marshalErrors(cronJobsErrors))
 
 	// ingress
 	ingress, ingressErrors := ingress(ctx, client, namespaceNames)
 	for k, v := range ingress {
-		clusterResourcesOutput[path.Join("cluster-resources/ingress", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/ingress-errors.json"], err = marshalNonNil(ingressErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/ingress", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/ingress-errors.json", marshalErrors(ingressErrors))
 
 	// storage classes
 	storageClasses, storageErrors := storageClasses(ctx, client)
-	clusterResourcesOutput["cluster-resources/storage-classes.json"] = storageClasses
-	clusterResourcesOutput["cluster-resources/storage-errors.json"], err = marshalNonNil(storageErrors)
-	if err != nil {
-		return nil, err
-	}
+	output.SaveResult(c.BundlePath, "cluster-resources/storage-classes.json", bytes.NewBuffer(storageClasses))
+	output.SaveResult(c.BundlePath, "cluster-resources/storage-errors.json", marshalErrors(storageErrors))
 
 	// crds
 	crdClient, err := apiextensionsv1beta1clientset.NewForConfig(c.ClientConfig)
 	if err != nil {
 		return nil, err
 	}
 	customResourceDefinitions, crdErrors := crds(ctx, crdClient)
-	clusterResourcesOutput["cluster-resources/custom-resource-definitions.json"] = customResourceDefinitions
-	clusterResourcesOutput["cluster-resources/custom-resource-definitions-errors.json"], err = marshalNonNil(crdErrors)
-	if err != nil {
-		return nil, err
-	}
+	output.SaveResult(c.BundlePath, "cluster-resources/custom-resource-definitions.json", bytes.NewBuffer(customResourceDefinitions))
+	output.SaveResult(c.BundlePath, "cluster-resources/custom-resource-definitions-errors.json", marshalErrors(crdErrors))
 
 	// imagepullsecrets
 	imagePullSecrets, pullSecretsErrors := imagePullSecrets(ctx, client, namespaceNames)
 	for k, v := range imagePullSecrets {
-		clusterResourcesOutput[path.Join("cluster-resources/image-pull-secrets", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/image-pull-secrets-errors.json"], err = marshalNonNil(pullSecretsErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/image-pull-secrets", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/image-pull-secrets-errors.json", marshalErrors(pullSecretsErrors))
 
 	// nodes
 	nodes, nodeErrors := nodes(ctx, client)
-	clusterResourcesOutput["cluster-resources/nodes.json"] = nodes
-	clusterResourcesOutput["cluster-resources/nodes-errors.json"], err = marshalNonNil(nodeErrors)
-	if err != nil {
-		return nil, err
-	}
+	output.SaveResult(c.BundlePath, "cluster-resources/nodes.json", bytes.NewBuffer(nodes))
+	output.SaveResult(c.BundlePath, "cluster-resources/nodes-errors.json", marshalErrors(nodeErrors))
 
 	groups, resources, groupsResourcesErrors := apiResources(ctx, client)
-	clusterResourcesOutput["cluster-resources/groups.json"] = groups
-	clusterResourcesOutput["cluster-resources/resources.json"] = resources
-	clusterResourcesOutput["cluster-resources/groups-resources-errors.json"], err = marshalNonNil(groupsResourcesErrors)
-	if err != nil {
-		return nil, err
-	}
+	output.SaveResult(c.BundlePath, "cluster-resources/groups.json", bytes.NewBuffer(groups))
+	output.SaveResult(c.BundlePath, "cluster-resources/resources.json", bytes.NewBuffer(resources))
+	output.SaveResult(c.BundlePath, "cluster-resources/groups-resources-errors.json", marshalErrors(groupsResourcesErrors))
 
 	// limit ranges
 	limitRanges, limitRangesErrors := limitRanges(ctx, client, namespaceNames)
 	for k, v := range limitRanges {
-		clusterResourcesOutput[path.Join("cluster-resources/limitranges", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/limitranges-errors.json"], err = marshalNonNil(limitRangesErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/limitranges", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/limitranges-errors.json", marshalErrors(limitRangesErrors))
 
 	// auth cani
 	authCanI, authCanIErrors := authCanI(ctx, client, namespaceNames)
 	for k, v := range authCanI {
-		clusterResourcesOutput[path.Join("cluster-resources/auth-cani-list", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/auth-cani-list-errors.json"], err = marshalNonNil(authCanIErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/auth-cani-list", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/auth-cani-list-errors.json", marshalErrors(authCanIErrors))
 
 	//Events
 	events, eventsErrors := events(ctx, client, namespaceNames)
 	for k, v := range events {
-		clusterResourcesOutput[path.Join("cluster-resources/events", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/events-errors.json"], err = marshalNonNil(eventsErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/events", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/events-errors.json", marshalErrors(eventsErrors))
 
 	//Persistent Volumes
 	pvs, pvsErrors := pvs(ctx, client)
-	clusterResourcesOutput["cluster-resources/pvs.json"] = pvs
-	clusterResourcesOutput["cluster-resources/pvs-errors.json"], err = marshalNonNil(pvsErrors)
-	if err != nil {
-		return nil, err
-	}
+	output.SaveResult(c.BundlePath, "cluster-resources/pvs.json", bytes.NewBuffer(pvs))
+	output.SaveResult(c.BundlePath, "cluster-resources/pvs-errors.json", marshalErrors(pvsErrors))
 
 	//Persistent Volume Claims
 	pvcs, pvcsErrors := pvcs(ctx, client, namespaceNames)
 	for k, v := range pvcs {
-		clusterResourcesOutput[path.Join("cluster-resources/pvcs", k)] = v
-	}
-	clusterResourcesOutput["cluster-resources/pvcs-errors.json"], err = marshalNonNil(pvcsErrors)
-	if err != nil {
-		return nil, err
+		output.SaveResult(c.BundlePath, path.Join("cluster-resources/pvcs", k), bytes.NewBuffer(v))
 	}
+	output.SaveResult(c.BundlePath, "cluster-resources/pvcs-errors.json", marshalErrors(pvcsErrors))
 
-	return clusterResourcesOutput, nil
+	return output, nil
 }
 
 func getAllNamespaces(ctx context.Context, client *kubernetes.Clientset) ([]byte, *corev1.NamespaceList, []string) {

pkg/collect/collectd.go

@@ -8,8 +8,8 @@ import (
 	restclient "k8s.io/client-go/rest"
 )
 
-func Collectd(ctx context.Context, namespace string, clientConfig *restclient.Config, client kubernetes.Interface, collector *troubleshootv1beta2.Collectd) (map[string][]byte, error) {
-	return CopyFromHost(ctx, namespace, clientConfig, client, &troubleshootv1beta2.CopyFromHost{
+func Collectd(ctx context.Context, c *Collector, collector *troubleshootv1beta2.Collectd, namespace string, clientConfig *restclient.Config, client kubernetes.Interface) (CollectorResult, error) {
+	copyFromHost := &troubleshootv1beta2.CopyFromHost{
 		CollectorMeta:   collector.CollectorMeta,
 		Name:            "collectd/rrd",
 		Namespace:       collector.Namespace,
@@ -18,5 +18,6 @@ func Collectd(ctx context.Context, namespace string, clientConfig *restclient.Co
 		ImagePullSecret: collector.ImagePullSecret,
 		Timeout:         collector.Timeout,
 		HostPath:        collector.HostPath,
-	})
+	}
+	return CopyFromHost(ctx, c, copyFromHost, namespace, clientConfig, client)
 }