Separate rabbitmqcluster_controller into smaller files (#418)

- separate according to different k8s/rmq concepts that
the Reconcile function is handling
- all broken down files have a corresponding  *_test.go
file which contains its integration tests

controllers/pod_executor.go

@@ -0,0 +1,59 @@
+package controllers
+
+import (
+	"bufio"
+	"bytes"
+	"fmt"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/remotecommand"
+)
+
+type PodExecutor interface {
+	Exec(clientset *kubernetes.Clientset, clusterConfig *rest.Config, namespace, podName, containerName string, command ...string) (string, string, error)
+}
+
+func NewPodExecutor() PodExecutor { return &rabbitmqPodExecutor{} }
+
+type rabbitmqPodExecutor struct{}
+
+func (p *rabbitmqPodExecutor) Exec(clientset *kubernetes.Clientset, clusterConfig *rest.Config, namespace, podName, containerName string, command ...string) (string, string, error) {
+	request := clientset.CoreV1().RESTClient().
+		Post().
+		Resource("pods").
+		Name(podName).
+		Namespace(namespace).
+		SubResource("exec").
+		VersionedParams(&corev1.PodExecOptions{
+			Container: containerName,
+			Command:   command,
+			Stdout:    true,
+			Stderr:    true,
+			Stdin:     false,
+		}, scheme.ParameterCodec)
+
+	exec, err := remotecommand.NewSPDYExecutor(clusterConfig, "POST", request.URL())
+	if err != nil {
+		return "", "", err
+	}
+
+	stdOut := bytes.Buffer{}
+	stdErr := bytes.Buffer{}
+
+	err = exec.Stream(remotecommand.StreamOptions{
+		Stdout: bufio.NewWriter(&stdOut),
+		Stderr: bufio.NewWriter(&stdErr),
+		Stdin:  nil,
+		Tty:    false,
+	})
+	if err != nil {
+		return stdOut.String(), stdErr.String(), err
+	}
+	if stdErr.Len() > 0 {
+		return stdOut.String(), stdErr.String(), fmt.Errorf("%v", stdErr)
+	}
+
+	return stdOut.String(), "", nil
+}