Exec collector

config/crds/troubleshoot.replicated.com_collectors.yaml

@@ -413,6 +413,33 @@ spec:
                   - namespace
                   - containerPath
                   type: object
+                exec:
+                  properties:
+                    args:
+                      items:
+                        type: string
+                      type: array
+                    command:
+                      items:
+                        type: string
+                      type: array
+                    containerName:
+                      type: string
+                    name:
+                      type: string
+                    namespace:
+                      type: string
+                    selector:
+                      items:
+                        type: string
+                      type: array
+                    timeout:
+                      type: string
+                  required:
+                  - name
+                  - selector
+                  - namespace
+                  type: object
                 http:
                   properties:
                     get:

config/crds/troubleshoot.replicated.com_preflights.yaml

@@ -635,6 +635,33 @@ spec:
                       - namespace
                       - containerPath
                       type: object
+                    exec:
+                      properties:
+                        args:
+                          items:
+                            type: string
+                          type: array
+                        command:
+                          items:
+                            type: string
+                          type: array
+                        containerName:
+                          type: string
+                        name:
+                          type: string
+                        namespace:
+                          type: string
+                        selector:
+                          items:
+                            type: string
+                          type: array
+                        timeout:
+                          type: string
+                      required:
+                      - name
+                      - selector
+                      - namespace
+                      type: object
                     http:
                       properties:
                         get:

config/crds/zz_generated.deepcopy.go

@@ -353,6 +353,11 @@ func (in *Collect) DeepCopyInto(out *Collect) {
 		*out = new(Run)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Exec != nil {
+		in, out := &in.Exec, &out.Exec
+		*out = new(Exec)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.Copy != nil {
 		in, out := &in.Copy, &out.Copy
 		*out = new(Copy)
@@ -626,6 +631,36 @@ func (in *CustomResourceDefinition) DeepCopy() *CustomResourceDefinition {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Exec) DeepCopyInto(out *Exec) {
+	*out = *in
+	if in.Selector != nil {
+		in, out := &in.Selector, &out.Selector
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.Command != nil {
+		in, out := &in.Command, &out.Command
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.Args != nil {
+		in, out := &in.Args, &out.Args
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Exec.
+func (in *Exec) DeepCopy() *Exec {
+	if in == nil {
+		return nil
+	}
+	out := new(Exec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Get) DeepCopyInto(out *Get) {
 	*out = *in

config/samples/troubleshoot_v1beta1_collector.yaml

@@ -23,6 +23,30 @@ spec:
   #     command: ["ping"]
   #     args: ["www.google.com"]
   #     timeout: 5s
+  - exec:
+      name: mysql-vars
+      selector:
+        - app=mysql
+      namespace: default
+      command: ["mysql"]
+      args: ["-ureplicated", "-ppassword", "-e", "show processlist"]
+      timeout: 60m
+  - exec:
+      name: hosts
+      selector:
+        - app=graphql-api
+      namespace: default
+      command: ["cat"]
+      args: ["/etc/hosts"]
+      timeout: 60m
+  - exec:
+      name: broken
+      selector:
+        - app=graphql-api
+      namespace: default
+      command: ["cat"]
+      args: ["/etc/hostdasddsda"]
+      timeout: 60m
   # - copy:
   #     selector:
   #       - app=illmannered-cricket-mysql

pkg/apis/troubleshoot/v1beta1/collector_shared.go

@@ -34,6 +34,16 @@ type Run struct {
 	ImagePullPolicy string   `json:"imagePullPolicy,omitempty" yaml:"imagePullPolicy,omitempty"`
 }
 
+type Exec struct {
+	Name          string   `json:"name" yaml:"name"`
+	Selector      []string `json:"selector" yaml:"selector"`
+	Namespace     string   `json:"namespace" yaml:"namespace"`
+	ContainerName string   `json:"containerName,omitempty" yaml:"containerName,omitempty"`
+	Command       []string `json:"command,omitempty" yaml:"command,omitempty"`
+	Args          []string `json:"args,omitempty" yaml:"args,omitempty"`
+	Timeout       string   `json:"timeout,omitempty" yaml:"timeout,omitempty"`
+}
+
 type Copy struct {
 	Selector      []string `json:"selector" yaml:"selector"`
 	Namespace     string   `json:"namespace" yaml:"namespace"`
@@ -74,6 +84,7 @@ type Collect struct {
 	Secret           *Secret           `json:"secret,omitempty" yaml:"secret,omitempty"`
 	Logs             *Logs             `json:"logs,omitempty" yaml:"logs,omitempty"`
 	Run              *Run              `json:"run,omitempty" yaml:"run,omitempty"`
+	Exec             *Exec             `json:"exec,omitempty" yaml:"exec,omitempty"`
 	Copy             *Copy             `json:"copy,omitempty" yaml:"copy,omitempty"`
 	HTTP             *HTTP             `json:"http,omitempty" yaml:"http,omitempty"`
 }

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

@@ -369,6 +369,11 @@ func (in *Collect) DeepCopyInto(out *Collect) {
 		*out = new(Run)
 		(*in).DeepCopyInto(*out)
 	}
+	if in.Exec != nil {
+		in, out := &in.Exec, &out.Exec
+		*out = new(Exec)
+		(*in).DeepCopyInto(*out)
+	}
 	if in.Copy != nil {
 		in, out := &in.Copy, &out.Copy
 		*out = new(Copy)
@@ -642,6 +647,36 @@ func (in *CustomResourceDefinition) DeepCopy() *CustomResourceDefinition {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *Exec) DeepCopyInto(out *Exec) {
+	*out = *in
+	if in.Selector != nil {
+		in, out := &in.Selector, &out.Selector
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.Command != nil {
+		in, out := &in.Command, &out.Command
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+	if in.Args != nil {
+		in, out := &in.Args, &out.Args
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Exec.
+func (in *Exec) DeepCopy() *Exec {
+	if in == nil {
+		return nil
+	}
+	out := new(Exec)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Get) DeepCopyInto(out *Get) {
 	*out = *in

pkg/collect/collector.go

@@ -33,6 +33,9 @@ func (c *Collector) RunCollectorSync() error {
 	if collect.Run != nil {
 		return Run(collect.Run, c.Redact)
 	}
+	if collect.Exec != nil {
+		return Exec(collect.Exec, c.Redact)
+	}
 	if collect.Copy != nil {
 		return Copy(collect.Copy, c.Redact)
 	}

pkg/collect/exec.go

@@ -0,0 +1,160 @@
+package collect
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	troubleshootv1beta1 "github.com/replicatedhq/troubleshoot/pkg/apis/troubleshoot/v1beta1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/remotecommand"
+	"sigs.k8s.io/controller-runtime/pkg/client/config"
+)
+
+type ExecOutput struct {
+	Results map[string][]byte `json:"exec/,omitempty"`
+}
+
+type execResult struct {
+	Stdout string `json:"stdout,omitempty"`
+	Stderr string `json:"stderr,omitempty"`
+	Error  error  `json:"error,omitempty"`
+}
+
+func Exec(execCollector *troubleshootv1beta1.Exec, redact bool) error {
+	if execCollector.Timeout == "" {
+		return execWithoutTimeout(execCollector, redact)
+	}
+
+	timeout, err := time.ParseDuration(execCollector.Timeout)
+	if err != nil {
+		return err
+	}
+
+	execChan := make(chan error, 1)
+	go func() {
+		execChan <- execWithoutTimeout(execCollector, redact)
+	}()
+
+	select {
+	case <-time.After(timeout):
+		return errors.New("timeout")
+	case err := <-execChan:
+		return err
+	}
+}
+
+func execWithoutTimeout(execCollector *troubleshootv1beta1.Exec, redact bool) error {
+	cfg, err := config.GetConfig()
+	if err != nil {
+		return err
+	}
+
+	client, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return err
+	}
+
+	pods, err := listPodsInSelectors(client, execCollector.Namespace, execCollector.Selector)
+	if err != nil {
+		return err
+	}
+
+	execOutput := &ExecOutput{
+		Results: make(map[string][]byte),
+	}
+
+	for _, pod := range pods {
+		output, err := getExecOutputs(client, pod, execCollector, redact)
+		if err != nil {
+			return err
+		}
+
+		b, err := json.MarshalIndent(output, "", "  ")
+		if err != nil {
+			return err
+		}
+
+		execOutput.Results[fmt.Sprintf("%s/%s/%s", pod.Namespace, pod.Name, execCollector.Name)] = b
+	}
+
+	if redact {
+		execOutput, err = execOutput.Redact()
+		if err != nil {
+			return err
+		}
+	}
+
+	b, err := json.MarshalIndent(execOutput, "", "  ")
+	if err != nil {
+		return err
+	}
+
+	fmt.Printf("%s\n", b)
+
+	return nil
+}
+
+func getExecOutputs(client *kubernetes.Clientset, pod corev1.Pod, execCollector *troubleshootv1beta1.Exec, doRedact bool) (*execResult, error) {
+	cfg, err := config.GetConfig()
+	if err != nil {
+		return nil, err
+	}
+
+	container := pod.Spec.Containers[0].Name
+	if execCollector.ContainerName != "" {
+		container = execCollector.ContainerName
+	}
+
+	req := client.CoreV1().RESTClient().Post().Resource("pods").Name(pod.Name).Namespace(pod.Namespace).SubResource("exec")
+	scheme := runtime.NewScheme()
+	if err := corev1.AddToScheme(scheme); err != nil {
+		return nil, err
+	}
+
+	parameterCodec := runtime.NewParameterCodec(scheme)
+	req.VersionedParams(&corev1.PodExecOptions{
+		Command:   append(execCollector.Command, execCollector.Args...),
+		Container: container,
+		Stdin:     true,
+		Stdout:    false,
+		Stderr:    true,
+		TTY:       false,
+	}, parameterCodec)
+
+	exec, err := remotecommand.NewSPDYExecutor(cfg, "POST", req.URL())
+	if err != nil {
+		return nil, err
+	}
+
+	stdout := new(bytes.Buffer)
+	stderr := new(bytes.Buffer)
+
+	err = exec.Stream(remotecommand.StreamOptions{
+		Stdin:  nil,
+		Stdout: stdout,
+		Stderr: stderr,
+		Tty:    false,
+	})
+
+	return &execResult{
+		Stdout: stdout.String(),
+		Stderr: stderr.String(),
+		Error:  err,
+	}, nil
+}
+
+func (r *ExecOutput) Redact() (*ExecOutput, error) {
+	results, err := redactMap(r.Results)
+	if err != nil {
+		return nil, err
+	}
+
+	return &ExecOutput{
+		Results: results,
+	}, nil
+}