Remote host collectors (#392)

* Add collect command and remote host collectors

Adds the ability to run a host collector on a set of remote k8s nodes.
Target nodes can be filtered using the --selector flag, with the same
syntax as kubectl.  Existing flags for --collector-image,
--collector-pullpolicy and --request-timeout are used.  To run on a
specified node, --selector="kubernetes.io/hostname=kind-worker2" could
be used.

The collect command is used by the remote collector to output the
results using a "raw" format, which uses the filename as the key, and
the value the output as a escaped json string.  When run manually it
defaults to fully decoded json. The existing block devices,
ipv4interfaces and services host collectors don't decode properly - the
fix is to convert their slice output to a map (fix not included as
unsure what depends on the existing format).

The collect command is also useful for troubleshooting preflight issues.

Examples are included to show remote collector usage.

```
bin/collect --collector-image=croomes/troubleshoot:latest  examples/collect/remote/memory.yaml --namespace test
{
  "kind-control-plane": {
    "system/memory.json": {
      "total": 1304207360
    }
  },
  "kind-worker": {
    "system/memory.json": {
      "total": 1695780864
    }
  },
  "kind-worker2": {
    "system/memory.json": {
      "total": 1726353408
    }
  }
}
```

The preflight command has been updated to run remote collectors.  To run
a host collector remotely it must be specified in the spec as a
`remoteCollector`:

```
apiVersion: troubleshoot.sh/v1beta2
kind: HostPreflight
metadata:
  name: memory
spec:
  remoteCollectors:
    - memory:
        collectorName: memory
  analyzers:
    - memory:
        outcomes:
          - fail:
              when: "< 8Gi"
              message: At least 8Gi of memory is required
          - warn:
              when: "< 32Gi"
              message: At least 32Gi of memory is recommended
          - pass:
              message: The system has as sufficient memory
```

Results for each node are analyzed separately, with the node name
appended to the title:

```
bin/preflight --interactive=false --collector-image=croomes/troubleshoot:latest examples/preflight/remote/memory.yaml --format=json
{memory running 0 1}
{memory completed 1 1}
{
  "fail": [
    {
      "title": "Amount of Memory (kind-worker2)",
      "message": "At least 8Gi of memory is required"
    },
    {
      "title": "Amount of Memory (kind-worker)",
      "message": "At least 8Gi of memory is required"
    },
    {
      "title": "Amount of Memory (kind-control-plane)",
      "message": "At least 8Gi of memory is required"
    }
  ]
}
```

Also added a host collector to allow preflight checks of required kernel
modules, which is the main driver for this change.

Author: croomes

Makefile

@@ -58,6 +58,10 @@ preflight: generate fmt vet
 .PHONY: analyze
 analyze: generate fmt vet
 	go build ${BUILDFLAGS} ${LDFLAGS} -o bin/analyze github.com/replicatedhq/troubleshoot/cmd/analyze
+	
+.PHONY: collect
+collect: generate fmt vet
+	go build ${BUILDFLAGS} ${LDFLAGS} -o bin/collect github.com/replicatedhq/troubleshoot/cmd/collect	
 
 .PHONY: fmt
 fmt:

cmd/collect/cli/root.go

@@ -0,0 +1,65 @@
+package cli
+
+import (
+	"os"
+	"strings"
+
+	"github.com/replicatedhq/troubleshoot/pkg/k8sutil"
+	"github.com/replicatedhq/troubleshoot/pkg/logger"
+	"github.com/spf13/cobra"
+	"github.com/spf13/viper"
+)
+
+func RootCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:          "collect [url]",
+		Args:         cobra.MinimumNArgs(1),
+		Short:        "Run a collector",
+		Long:         `Run a collector and output the results.`,
+		SilenceUsage: true,
+		PreRun: func(cmd *cobra.Command, args []string) {
+			viper.BindPFlags(cmd.Flags())
+		},
+		RunE: func(cmd *cobra.Command, args []string) error {
+			v := viper.GetViper()
+
+			logger.SetQuiet(v.GetBool("quiet"))
+			return runCollect(v, args[0])
+		},
+	}
+
+	cobra.OnInitialize(initConfig)
+
+	cmd.AddCommand(VersionCmd())
+
+	cmd.Flags().StringSlice("redactors", []string{}, "names of the additional redactors to use")
+	cmd.Flags().Bool("redact", true, "enable/disable default redactions")
+	cmd.Flags().String("format", "json", "output format, one of json or raw.")
+	cmd.Flags().String("collector-image", "", "the full name of the collector image to use")
+	cmd.Flags().String("collector-pull-policy", "", "the pull policy of the collector image")
+	cmd.Flags().String("selector", "", "selector (label query) to filter remote collection nodes on.")
+	cmd.Flags().Bool("collect-without-permissions", false, "always generate a support bundle, even if it some require additional permissions")
+
+	// hidden in favor of the `insecure-skip-tls-verify` flag
+	cmd.Flags().Bool("allow-insecure-connections", false, "when set, do not verify TLS certs when retrieving spec and reporting results")
+	cmd.Flags().MarkHidden("allow-insecure-connections")
+
+	viper.BindPFlags(cmd.Flags())
+
+	viper.SetEnvKeyReplacer(strings.NewReplacer("-", "_"))
+
+	k8sutil.AddFlags(cmd.Flags())
+
+	return cmd
+}
+
+func InitAndExecute() {
+	if err := RootCmd().Execute(); err != nil {
+		os.Exit(1)
+	}
+}
+
+func initConfig() {
+	viper.SetEnvPrefix("TROUBLESHOOT")
+	viper.AutomaticEnv()
+}

cmd/collect/cli/run.go

@@ -0,0 +1,184 @@
+package cli
+
+import (
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"os/signal"
+	"strings"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/replicatedhq/troubleshoot/cmd/util"
+	troubleshootv1beta2 "github.com/replicatedhq/troubleshoot/pkg/apis/troubleshoot/v1beta2"
+	"github.com/replicatedhq/troubleshoot/pkg/client/troubleshootclientset/scheme"
+	troubleshootclientsetscheme "github.com/replicatedhq/troubleshoot/pkg/client/troubleshootclientset/scheme"
+	"github.com/replicatedhq/troubleshoot/pkg/collect"
+	"github.com/replicatedhq/troubleshoot/pkg/docrewrite"
+	"github.com/replicatedhq/troubleshoot/pkg/k8sutil"
+	"github.com/replicatedhq/troubleshoot/pkg/specs"
+	"github.com/replicatedhq/troubleshoot/pkg/supportbundle"
+	"github.com/spf13/viper"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+const (
+	defaultTimeout = 30 * time.Second
+)
+
+func runCollect(v *viper.Viper, arg string) error {
+	go func() {
+		signalChan := make(chan os.Signal, 1)
+		signal.Notify(signalChan, os.Interrupt)
+		<-signalChan
+		os.Exit(0)
+	}()
+
+	var collectorContent []byte
+	var err error
+	if strings.HasPrefix(arg, "secret/") {
+		// format secret/namespace-name/secret-name
+		pathParts := strings.Split(arg, "/")
+		if len(pathParts) != 3 {
+			return errors.Errorf("path %s must have 3 components", arg)
+		}
+
+		spec, err := specs.LoadFromSecret(pathParts[1], pathParts[2], "collect-spec")
+		if err != nil {
+			return errors.Wrap(err, "failed to get spec from secret")
+		}
+
+		collectorContent = spec
+	} else if _, err = os.Stat(arg); err == nil {
+		b, err := ioutil.ReadFile(arg)
+		if err != nil {
+			return err
+		}
+
+		collectorContent = b
+	} else {
+		if !util.IsURL(arg) {
+			return fmt.Errorf("%s is not a URL and was not found (err %s)", arg, err)
+		}
+
+		req, err := http.NewRequest("GET", arg, nil)
+		if err != nil {
+			return err
+		}
+		req.Header.Set("User-Agent", "Replicated_Collect/v1beta2")
+		resp, err := http.DefaultClient.Do(req)
+		if err != nil {
+			return err
+		}
+		defer resp.Body.Close()
+
+		body, err := ioutil.ReadAll(resp.Body)
+		if err != nil {
+			return err
+		}
+
+		collectorContent = body
+	}
+
+	collectorContent, err = docrewrite.ConvertToV1Beta2(collectorContent)
+	if err != nil {
+		return errors.Wrap(err, "failed to convert to v1beta2")
+	}
+
+	multidocs := strings.Split(string(collectorContent), "\n---\n")
+
+	troubleshootclientsetscheme.AddToScheme(scheme.Scheme)
+	decode := scheme.Codecs.UniversalDeserializer().Decode
+
+	additionalRedactors := &troubleshootv1beta2.Redactor{}
+	for idx, redactor := range v.GetStringSlice("redactors") {
+		redactorObj, err := supportbundle.GetRedactorFromURI(redactor)
+		if err != nil {
+			return errors.Wrapf(err, "failed to get redactor spec %s, #%d", redactor, idx)
+		}
+
+		if redactorObj != nil {
+			additionalRedactors.Spec.Redactors = append(additionalRedactors.Spec.Redactors, redactorObj.Spec.Redactors...)
+		}
+	}
+
+	for i, additionalDoc := range multidocs {
+		if i == 0 {
+			continue
+		}
+		additionalDoc, err := docrewrite.ConvertToV1Beta2([]byte(additionalDoc))
+		if err != nil {
+			return errors.Wrap(err, "failed to convert to v1beta2")
+		}
+		obj, _, err := decode(additionalDoc, nil, nil)
+		if err != nil {
+			return errors.Wrapf(err, "failed to parse additional doc %d", i)
+		}
+		multidocRedactors, ok := obj.(*troubleshootv1beta2.Redactor)
+		if !ok {
+			continue
+		}
+		additionalRedactors.Spec.Redactors = append(additionalRedactors.Spec.Redactors, multidocRedactors.Spec.Redactors...)
+	}
+
+	// make sure we don't block any senders
+	progressCh := make(chan interface{})
+	defer close(progressCh)
+	go func() {
+		for range progressCh {
+		}
+	}()
+
+	restConfig, err := k8sutil.GetRESTConfig()
+	if err != nil {
+		return errors.Wrap(err, "failed to convert kube flags to rest config")
+	}
+
+	labelSelector, err := labels.Parse(v.GetString("selector"))
+	if err != nil {
+		return errors.Wrap(err, "unable to parse selector")
+	}
+
+	namespace := v.GetString("namespace")
+	if namespace == "" {
+		namespace = "default"
+	}
+
+	timeout := v.GetDuration("request-timeout")
+	if timeout == 0 {
+		timeout = defaultTimeout
+	}
+
+	createOpts := collect.CollectorRunOpts{
+		CollectWithoutPermissions: v.GetBool("collect-without-permissions"),
+		KubernetesRestConfig:      restConfig,
+		Image:                     v.GetString("collector-image"),
+		PullPolicy:                v.GetString("collector-pullpolicy"),
+		LabelSelector:             labelSelector.String(),
+		Namespace:                 namespace,
+		Timeout:                   timeout,
+		ProgressChan:              progressCh,
+	}
+
+	// we only support HostCollector or RemoteCollector kinds.
+	hostCollector, err := collect.ParseHostCollectorFromDoc([]byte(multidocs[0]))
+	if err == nil {
+		results, err := collect.CollectHost(hostCollector, additionalRedactors, createOpts)
+		if err != nil {
+			return errors.Wrap(err, "failed to collect from host")
+		}
+		return showHostStdoutResults(v.GetString("format"), hostCollector.Name, results)
+	}
+
+	remoteCollector, err := collect.ParseRemoteCollectorFromDoc([]byte(multidocs[0]))
+	if err == nil {
+		results, err := collect.CollectRemote(remoteCollector, additionalRedactors, createOpts)
+		if err != nil {
+			return errors.Wrap(err, "failed to collect from remote host(s)")
+		}
+		return showRemoteStdoutResults(v.GetString("format"), remoteCollector.Name, results)
+	}
+
+	return errors.New("failed to parse hostCollector or remoteCollector")
+}

cmd/collect/cli/stdout_results.go

@@ -0,0 +1,103 @@
+package cli
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/pkg/errors"
+	"github.com/replicatedhq/troubleshoot/pkg/collect"
+)
+
+const (
+	// FormatJSON is intended for CLI output.
+	FormatJSON = "json"
+
+	// FormatRaw is intended for consumption by a remote collector.  Output is a
+	// string of quoted JSON.
+	FormatRaw = "raw"
+)
+
+func showHostStdoutResults(format string, collectName string, results *collect.HostCollectResult) error {
+	switch format {
+	case FormatJSON:
+		return showHostStdoutResultsJSON(collectName, results.AllCollectedData)
+	case FormatRaw:
+		return showHostStdoutResultsRaw(collectName, results.AllCollectedData)
+	default:
+		return errors.Errorf("unknown output format: %q", format)
+	}
+}
+
+func showRemoteStdoutResults(format string, collectName string, results *collect.RemoteCollectResult) error {
+	switch format {
+	case FormatJSON:
+		return showRemoteStdoutResultsJSON(collectName, results.AllCollectedData)
+	case FormatRaw:
+		return errors.Errorf("raw format not supported for remote collectors")
+	default:
+		return errors.Errorf("unknown output format: %q", format)
+	}
+}
+
+func showHostStdoutResultsJSON(collectName string, results map[string][]byte) error {
+	output := make(map[string]interface{})
+	for file, collectorResult := range results {
+		var collectedItems map[string]interface{}
+		if err := json.Unmarshal([]byte(collectorResult), &collectedItems); err != nil {
+			return errors.Wrap(err, "failed to marshal collector results")
+		}
+		output[file] = collectedItems
+	}
+
+	formatted, err := json.MarshalIndent(output, "", "  ")
+	if err != nil {
+		return errors.Wrap(err, "failed to convert output to json")
+	}
+
+	fmt.Print(string(formatted))
+	return nil
+}
+
+// showHostStdoutResultsRaw outputs the collector output as a string of quoted json.
+func showHostStdoutResultsRaw(collectName string, results map[string][]byte) error {
+	strData := map[string]string{}
+	for k, v := range results {
+		strData[k] = string(v)
+	}
+	formatted, err := json.MarshalIndent(strData, "", "    ")
+	if err != nil {
+		return errors.Wrap(err, "failed to convert output to json")
+	}
+	fmt.Print(string(formatted))
+	return nil
+}
+
+func showRemoteStdoutResultsJSON(collectName string, results map[string][]byte) error {
+	type CollectorResult map[string]interface{}
+	type NodeResult map[string]CollectorResult
+
+	var output = make(map[string]NodeResult)
+
+	for node, result := range results {
+		var nodeResult map[string]string
+		if err := json.Unmarshal(result, &nodeResult); err != nil {
+			return errors.Wrap(err, "failed to marshal node results")
+		}
+		nr := make(NodeResult)
+		for file, collectorResult := range nodeResult {
+			var collectedItems map[string]interface{}
+			if err := json.Unmarshal([]byte(collectorResult), &collectedItems); err != nil {
+				return errors.Wrap(err, "failed to marshal collector results")
+			}
+			nr[file] = collectedItems
+		}
+		output[node] = nr
+	}
+
+	formatted, err := json.MarshalIndent(output, "", "  ")
+	if err != nil {
+		return errors.Wrap(err, "failed to convert output to json")
+	}
+	fmt.Print(string(formatted))
+	return nil
+}

cmd/collect/cli/version.go

@@ -0,0 +1,22 @@
+package cli
+
+import (
+	"fmt"
+
+	"github.com/replicatedhq/troubleshoot/pkg/version"
+	"github.com/spf13/cobra"
+)
+
+func VersionCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "version",
+		Short: "Print the current version and exit",
+		Long:  `Print the current version and exit`,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			fmt.Printf("Replicated Collect %s\n", version.Version())
+
+			return nil
+		},
+	}
+	return cmd
+}

cmd/collect/main.go

@@ -0,0 +1,10 @@
+package main
+
+import (
+	"github.com/replicatedhq/troubleshoot/cmd/collect/cli"
+	_ "k8s.io/client-go/plugin/pkg/client/auth"
+)
+
+func main() {
+	cli.InitAndExecute()
+}