Rebase master, ident controller probe

example/controller.yaml

@@ -71,16 +71,16 @@ spec:
         ports:
         - containerPort: 8000
           name: http
-          livenessProbe:
-            httpGet:
-              path: /live
-              port: "8000"
-            initialDelaySeconds: 10
-            timeoutSeconds: 10
-          readinessProbe:
-            httpGet:
-              path: /ready
-              port: "8000"
-            initialDelaySeconds: 10
-            timeoutSeconds: 5
+        livenessProbe:
+          httpGet:
+            path: /live
+            port: "8000"
+          initialDelaySeconds: 10
+          timeoutSeconds: 10
+        readinessProbe:
+          httpGet:
+            path: /ready
+            port: "8000"
+          initialDelaySeconds: 10
+          timeoutSeconds: 5
       serviceAccount: elasticsearch-operator

example/example-es-cluster-minikube.yaml

@@ -5,9 +5,9 @@ metadata:
 spec:
   kibana:
     image: upmcenterprises/kibana:5.3.1
-  #cerebro:
-  #  image: upmcenterprises/cerebro:0.6.8
-  elastic-search-image: upmcenterprises/docker-elasticsearch-kubernetes:5.3.1
+  cerebro:
+   image: upmcenterprises/cerebro:0.6.8
+  elastic-search-image: upmcenterprises/docker-elasticsearch-kubernetes:5.3.1_3
   client-node-replicas: 1
   master-node-replicas: 1
   data-node-replicas: 3

pkg/k8sutil/deployments.go

@@ -293,12 +293,6 @@ func (k *K8sutil) CreateKibanaDeployment(baseImage, clusterName, namespace strin
 
 	component := fmt.Sprintf("elasticsearch-%s", clusterName)
 
-	elasticURL := fmt.Sprintf("https://%s:9200", component)
-
-	if !enableSSL {
-		elasticURL = fmt.Sprintf("http://%s:9200", component)
-	}
-
 	deploymentName := fmt.Sprintf("%s-%s", kibanaDeploymentName, clusterName)
 
 	// Check if deployment exists
@@ -335,7 +329,7 @@ func (k *K8sutil) CreateKibanaDeployment(baseImage, clusterName, namespace strin
 								Env: []v1.EnvVar{
 									v1.EnvVar{
 										Name:  "ELASTICSEARCH_URL",
-										Value: elasticURL,
+										Value: GetESURL(component, enableSSL),
 									},
 									v1.EnvVar{
 										Name:  "ELASTICSEARCH_SSL_CERTIFICATEAUTHORITIES",

pkg/k8sutil/k8sutil.go

@@ -354,6 +354,17 @@ func TemplateImagePullSecrets(ips []myspec.ImagePullSecrets) []v1.LocalObjectRef
 	return outSecrets
 }
 
+// GetESURL Return elasticsearch url
+func GetESURL(esHost string, enableSSL bool) string {
+
+	if !enableSSL {
+		return fmt.Sprintf("http://%s:9200", esHost)
+	}
+
+	return fmt.Sprintf("https://%s:9200", esHost)
+
+}
+
 // CreateDataNodeDeployment creates the data node deployment
 func (k *K8sutil) CreateDataNodeDeployment(deploymentType string, replicas *int32, baseImage, storageClass string, dataDiskSize string, resources myspec.Resources,
 	enableSSL bool, imagePullSecrets []myspec.ImagePullSecrets, clusterName, statsdEndpoint, networkHost, namespace, javaOptions string) error {
@@ -595,14 +606,15 @@ func (k *K8sutil) CreateDataNodeDeployment(deploymentType string, replicas *int3
 
 			if err != nil {
 				logrus.Error("Could not scale statefulSet: ", err)
+				return err
 			}
 		}
 	}
 
 	return nil
 }
 
-func (k *K8sutil) CreateCerebroConfiguration(clusterName string) map[string]string {
+func (k *K8sutil) CreateCerebroConfiguration(esHost string, enableSSL bool) map[string]string {
 
 	x := map[string]string{}
 	x["application.conf"] = fmt.Sprintf(`
@@ -633,7 +645,7 @@ hosts = [
 	host = "%s"
 	name = "%s"
 }
-]`, elasticsearchCertspath, elasticsearchCertspath, k.GetClientServiceName(clusterName), clusterName)
+]`, elasticsearchCertspath, elasticsearchCertspath, GetESURL(esHost, enableSSL), esHost)
 
 	return x
 }

pkg/k8sutil/k8sutil_test.go

@@ -0,0 +1,26 @@
+package k8sutil
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestGetESURL(t *testing.T) {
+
+	for _, v := range []struct {
+		host      string
+		expected  string
+		enableSSL bool
+	}{
+		{"es-ssl", "https://es-ssl:9200", true},
+		{"es-bla", "http://es-bla:9200", false},
+	} {
+
+		esURL := GetESURL(v.host, v.enableSSL)
+
+		if esURL != v.expected {
+			t.Errorf(fmt.Sprintf("Expected %s, got %s", v.expected, esURL))
+		}
+
+	}
+}

pkg/processor/processor.go

@@ -161,11 +161,10 @@ func (p *Processor) refreshClusters() error {
 					cluster.Spec.Snapshot.SchedulerEnabled,
 					cluster.Spec.Snapshot.Authentication.UserName,
 					cluster.Spec.Snapshot.Authentication.Password,
-					p.k8sclient.GetClientServiceNameFullDNS(cluster.ObjectMeta.Name, cluster.ObjectMeta.Namespace),
+					k8sutil.GetESURL(p.k8sclient.GetClientServiceNameFullDNS(cluster.ObjectMeta.Name, cluster.ObjectMeta.Namespace), cluster.Spec.EnableSSL),
 					cluster.ObjectMeta.Name,
 					cluster.ObjectMeta.Namespace,
 					p.k8sclient.Kclient,
-					cluster.Spec.EnableSSL,
 				),
 				Resources: myspec.Resources{
 					Limits: myspec.MemoryCPU{
@@ -344,15 +343,8 @@ func (p *Processor) processElasticSearchCluster(c *myspec.ElasticsearchCluster)
 
 	// Deploy Cerebro
 	if c.Spec.Cerebro.Image != "" {
-		cert := fmt.Sprintf("%s-%s", c.ObjectMeta.Name, "cerebro")
-
-		if err := p.k8sclient.CreateCerebroDeployment(c.Spec.Cerebro.Image, c.ObjectMeta.Name, cert, c.ObjectMeta.Namespace,
-			c.Spec.EnableSSL, c.Spec.ImagePullSecrets); err != nil {
-			logrus.Error("Error creating cerebro deployment ", err)
-			return err
-		}
-		// TODO create service
-		cerebroConf := p.k8sclient.CreateCerebroConfiguration(c.ObjectMeta.Name)
+		host := fmt.Sprintf("elasticsearch-%s", c.ObjectMeta.Name)
+		cerebroConf := p.k8sclient.CreateCerebroConfiguration(host, c.Spec.EnableSSL)
 
 		// create/update cerebro configMap
 		if p.k8sclient.ConfigmapExists(c.ObjectMeta.Namespace, fmt.Sprintf("%s-%s", c.ObjectMeta.Name, "cerebro")) {
@@ -366,6 +358,12 @@ func (p *Processor) processElasticSearchCluster(c *myspec.ElasticsearchCluster)
 				return err
 			}
 		}
+
+		if err := p.k8sclient.CreateCerebroDeployment(c.Spec.Cerebro.Image, c.ObjectMeta.Name, c.ObjectMeta.Namespace,
+			fmt.Sprintf("%s-%s", c.ObjectMeta.Name, "cerebro"), c.Spec.EnableSSL, c.Spec.ImagePullSecrets); err != nil {
+			logrus.Error("Error creating cerebro deployment ", err)
+			return err
+		}
 	}
 
 	// Setup CronSchedule

pkg/snapshot/scheduler.go

@@ -27,11 +27,10 @@ package snapshot
 import (
 	"fmt"
 
-	"k8s.io/client-go/kubernetes"
-
 	"github.com/Sirupsen/logrus"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 	apiv1 "k8s.io/client-go/pkg/api/v1"
 	batchv1 "k8s.io/client-go/pkg/apis/batch/v1"
 	batch "k8s.io/client-go/pkg/apis/batch/v2alpha1"
@@ -62,12 +61,7 @@ type Authentication struct {
 }
 
 // New creates an instance of Scheduler
-func New(bucketName, cronSchedule string, enabled bool, userName, password, svcURL, clusterName, namespace string, kc kubernetes.Interface, enableSSL bool) *Scheduler {
-	elasticURL := fmt.Sprintf("https://%s:9200", svcURL) // Internal service name of cluster
-
-	if !enableSSL {
-		elasticURL = fmt.Sprintf("http://%s:9200", svcURL) // Internal service name of cluster
-	}
+func New(bucketName, cronSchedule string, enabled bool, userName, password, elasticURL, clusterName, namespace string, kc kubernetes.Interface) *Scheduler {
 
 	return &Scheduler{
 		s3bucketName: bucketName,