Fixed deployment of Elasticsearch via its operator (#234)

Fixed deployment of Elasticsearch via its operator

Signed-off-by: Juraci Paixão Kröhling <[email protected]>

README.adoc

@@ -242,14 +242,13 @@ spec:
 
 == Elasticsearch storage
 
-If no `es.server-urls` are provided Jaeger operator creates Elasticsearch CR based on the configuration
-provided in storage section. Make sure link:https://github.com/openshift/elasticsearch-operator[elasticsearch-operator]
-is running in your cluster otherwise Elasticsearch deployment will not be created. The Elasticsearch is meant
-to be dedicated for a single Jaeger instance.
+Under some circumstances, the Jaeger Operator can make use of the Elasticsearch operator to provision a suitable Elasticsearch cluster.
 
-At the moment there can be only one Jaeger with Elasticsearch instance in a namespace.
+IMPORTANT: this feature is experimental and currently works only on OpenShift clusters. Elasticsearch also requires the memory setting to be configured like `minishift ssh -- 'sudo sysctl -w vm.max_map_count=262144'`
 
-Note that Elasticsearch requires virtual memory settings: `minikube ssh -- 'sudo sysctl -w vm.max_map_count=262144'`
+When there are no `es.server-urls` options as part of a Jaeger `production` instance, the Jaeger Operator creates an Elasticsearch cluster via the Elasticsearch Operator by creating a CR based on the configuration provided in storage section. Make sure link:https://github.com/openshift/elasticsearch-operator[elasticsearch-operator] is running in your cluster otherwise the Elasticsearch deployment will not be created. The Elasticsearch cluster is meant to be dedicated for a single Jaeger instance.
+
+IMPORTANT: At the moment there can be only one Jaeger with self-provisioned Elasticsearch instance per namespace.
 
 == Accessing the UI
 

pkg/apis/addtoscheme_io_v1alpha1.go

@@ -2,9 +2,10 @@ package apis
 
 import (
 	"github.com/jaegertracing/jaeger-operator/pkg/apis/io/v1alpha1"
+	esv1alpha1 "github.com/jaegertracing/jaeger-operator/pkg/storage/elasticsearch/v1alpha1"
 )
 
 func init() {
 	// Register the types with the Scheme so the components can map objects to GroupVersionKinds and back
-	AddToSchemes = append(AddToSchemes, v1alpha1.SchemeBuilder.AddToScheme)
+	AddToSchemes = append(AddToSchemes, v1alpha1.SchemeBuilder.AddToScheme, esv1alpha1.SchemeBuilder.AddToScheme)
 }

pkg/controller/controller.go

@@ -2,10 +2,7 @@ package controller
 
 import (
 	routev1 "github.com/openshift/api/route/v1"
-	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
-
-	esv1alpha1 "github.com/jaegertracing/jaeger-operator/pkg/storage/elasticsearch/v1alpha1"
 )
 
 // AddToManagerFuncs is a list of functions to add all Controllers to the Manager
@@ -16,9 +13,6 @@ func AddToManager(m manager.Manager) error {
 	if err := routev1.AddToScheme(m.GetScheme()); err != nil {
 		return err
 	}
-	// TODO temporal fix https://github.com/jaegertracing/jaeger-operator/issues/206
-	gv := schema.GroupVersion{Group: "logging.openshift.io", Version: "v1alpha1"}
-	m.GetScheme().AddKnownTypes(gv, &esv1alpha1.Elasticsearch{})
 
 	for _, f := range AddToManagerFuncs {
 		if err := f(m); err != nil {

pkg/controller/jaeger/elasticsearch.go

@@ -0,0 +1,52 @@
+package jaeger
+
+import (
+	"context"
+
+	log "github.com/sirupsen/logrus"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"github.com/jaegertracing/jaeger-operator/pkg/apis/io/v1alpha1"
+	"github.com/jaegertracing/jaeger-operator/pkg/inventory"
+	esv1alpha1 "github.com/jaegertracing/jaeger-operator/pkg/storage/elasticsearch/v1alpha1"
+)
+
+func (r *ReconcileJaeger) applyElasticsearches(jaeger v1alpha1.Jaeger, desired []esv1alpha1.Elasticsearch) error {
+	opts := client.MatchingLabels(map[string]string{
+		"app.kubernetes.io/instance":   jaeger.Name,
+		"app.kubernetes.io/managed-by": "jaeger-operator",
+	})
+	list := &esv1alpha1.ElasticsearchList{}
+	if err := r.client.List(context.Background(), opts, list); err != nil {
+		return err
+	}
+
+	logFields := log.WithFields(log.Fields{
+		"namespace": jaeger.Namespace,
+		"instance":  jaeger.Name,
+	})
+
+	inv := inventory.ForElasticsearches(list.Items, desired)
+	for _, d := range inv.Create {
+		logFields.WithField("elasticsearch", d.Name).Debug("creating elasticsearch")
+		if err := r.client.Create(context.Background(), &d); err != nil {
+			return err
+		}
+	}
+
+	for _, d := range inv.Update {
+		logFields.WithField("elasticsearch", d.Name).Debug("updating elasticsearch")
+		if err := r.client.Update(context.Background(), &d); err != nil {
+			return err
+		}
+	}
+
+	for _, d := range inv.Delete {
+		logFields.WithField("elasticsearch", d.Name).Debug("deleting elasticsearch")
+		if err := r.client.Delete(context.Background(), &d); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

pkg/controller/jaeger/elasticsearch_test.go

@@ -0,0 +1,131 @@
+package jaeger
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	"github.com/jaegertracing/jaeger-operator/pkg/apis/io/v1alpha1"
+	esv1alpha1 "github.com/jaegertracing/jaeger-operator/pkg/storage/elasticsearch/v1alpha1"
+	"github.com/jaegertracing/jaeger-operator/pkg/strategy"
+)
+
+func TestElasticsearchesCreate(t *testing.T) {
+	// prepare
+	nsn := types.NamespacedName{
+		Name: "TestElasticsearchesCreate",
+	}
+
+	objs := []runtime.Object{
+		v1alpha1.NewJaeger(nsn.Name),
+	}
+
+	req := reconcile.Request{
+		NamespacedName: nsn,
+	}
+
+	r, cl := getReconciler(objs)
+	r.strategyChooser = func(jaeger *v1alpha1.Jaeger) strategy.S {
+		s := strategy.New().WithElasticsearches([]esv1alpha1.Elasticsearch{{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: nsn.Name,
+			},
+		}})
+		return s
+	}
+
+	// test
+	res, err := r.Reconcile(req)
+
+	// verify
+	assert.NoError(t, err)
+	assert.False(t, res.Requeue, "We don't requeue for now")
+
+	persisted := &esv1alpha1.Elasticsearch{}
+	persistedName := types.NamespacedName{
+		Name:      nsn.Name,
+		Namespace: nsn.Namespace,
+	}
+	err = cl.Get(context.Background(), persistedName, persisted)
+	assert.Equal(t, persistedName.Name, persisted.Name)
+	assert.NoError(t, err)
+}
+
+func TestElasticsearchesUpdate(t *testing.T) {
+	// prepare
+	nsn := types.NamespacedName{
+		Name: "TestElasticsearchesUpdate",
+	}
+
+	orig := esv1alpha1.Elasticsearch{}
+	orig.Name = nsn.Name
+	orig.Annotations = map[string]string{"key": "value"}
+
+	objs := []runtime.Object{
+		v1alpha1.NewJaeger(nsn.Name),
+		&orig,
+	}
+
+	r, cl := getReconciler(objs)
+	r.strategyChooser = func(jaeger *v1alpha1.Jaeger) strategy.S {
+		updated := esv1alpha1.Elasticsearch{}
+		updated.Name = orig.Name
+		updated.Annotations = map[string]string{"key": "new-value"}
+
+		s := strategy.New().WithElasticsearches([]esv1alpha1.Elasticsearch{updated})
+		return s
+	}
+
+	// test
+	_, err := r.Reconcile(reconcile.Request{NamespacedName: nsn})
+	assert.NoError(t, err)
+
+	// verify
+	persisted := &esv1alpha1.Elasticsearch{}
+	persistedName := types.NamespacedName{
+		Name:      orig.Name,
+		Namespace: orig.Namespace,
+	}
+	err = cl.Get(context.Background(), persistedName, persisted)
+	assert.Equal(t, "new-value", persisted.Annotations["key"])
+	assert.NoError(t, err)
+}
+
+func TestElasticsearchesDelete(t *testing.T) {
+	// prepare
+	nsn := types.NamespacedName{
+		Name: "TestElasticsearchesDelete",
+	}
+
+	orig := esv1alpha1.Elasticsearch{}
+	orig.Name = nsn.Name
+
+	objs := []runtime.Object{
+		v1alpha1.NewJaeger(nsn.Name),
+		&orig,
+	}
+
+	r, cl := getReconciler(objs)
+	r.strategyChooser = func(jaeger *v1alpha1.Jaeger) strategy.S {
+		return strategy.S{}
+	}
+
+	// test
+	_, err := r.Reconcile(reconcile.Request{NamespacedName: nsn})
+	assert.NoError(t, err)
+
+	// verify
+	persisted := &esv1alpha1.Elasticsearch{}
+	persistedName := types.NamespacedName{
+		Name:      orig.Name,
+		Namespace: orig.Namespace,
+	}
+	err = cl.Get(context.Background(), persistedName, persisted)
+	assert.Empty(t, persisted.Name)
+	assert.Error(t, err) // not found
+}

pkg/controller/jaeger/jaeger_controller.go

@@ -132,6 +132,10 @@ func defaultStrategyChooser(instance *v1alpha1.Jaeger) strategy.S {
 }
 
 func (r *ReconcileJaeger) apply(jaeger v1alpha1.Jaeger, str strategy.S) error {
+	if err := r.applyElasticsearches(jaeger, str.Elasticsearches()); err != nil {
+		return err
+	}
+
 	if err := r.applyRoles(jaeger, str.Roles()); err != nil {
 		return err
 	}

pkg/controller/jaeger/jaeger_controller_test.go

@@ -14,6 +14,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	"github.com/jaegertracing/jaeger-operator/pkg/apis/io/v1alpha1"
+	esv1alpha1 "github.com/jaegertracing/jaeger-operator/pkg/storage/elasticsearch/v1alpha1"
 	"github.com/jaegertracing/jaeger-operator/pkg/strategy"
 )
 
@@ -90,8 +91,16 @@ func TestDeletedInstance(t *testing.T) {
 
 func getReconciler(objs []runtime.Object) (*ReconcileJaeger, client.Client) {
 	s := scheme.Scheme
+
+	// OpenShift Route
 	osv1.Install(s)
+
+	// Jaeger
 	s.AddKnownTypes(v1alpha1.SchemeGroupVersion, &v1alpha1.Jaeger{})
+
+	// Jaeger's Elasticsearch
+	s.AddKnownTypes(v1alpha1.SchemeGroupVersion, &esv1alpha1.Elasticsearch{}, &esv1alpha1.ElasticsearchList{})
+
 	cl := fake.NewFakeClient(objs...)
 	return &ReconcileJaeger{client: cl, scheme: s}, cl
 }

pkg/inventory/elasticsearch.go

@@ -0,0 +1,62 @@
+package inventory
+
+import (
+	esv1alpha1 "github.com/jaegertracing/jaeger-operator/pkg/storage/elasticsearch/v1alpha1"
+)
+
+// Elasticsearch represents the elastic search inventory based on the current and desired states
+type Elasticsearch struct {
+	Create []esv1alpha1.Elasticsearch
+	Update []esv1alpha1.Elasticsearch
+	Delete []esv1alpha1.Elasticsearch
+}
+
+// ForElasticsearches builds a new elastic search inventory based on the existing and desired states
+func ForElasticsearches(existing []esv1alpha1.Elasticsearch, desired []esv1alpha1.Elasticsearch) Elasticsearch {
+	update := []esv1alpha1.Elasticsearch{}
+	mcreate := esMap(desired)
+	mdelete := esMap(existing)
+
+	for k, v := range mcreate {
+		if t, ok := mdelete[k]; ok {
+			tp := t.DeepCopy()
+
+			tp.Spec = v.Spec
+			tp.ObjectMeta.OwnerReferences = v.ObjectMeta.OwnerReferences
+
+			for k, v := range v.ObjectMeta.Annotations {
+				tp.ObjectMeta.Annotations[k] = v
+			}
+
+			for k, v := range v.ObjectMeta.Labels {
+				tp.ObjectMeta.Labels[k] = v
+			}
+
+			update = append(update, *tp)
+			delete(mcreate, k)
+			delete(mdelete, k)
+		}
+	}
+
+	return Elasticsearch{
+		Create: esList(mcreate),
+		Update: update,
+		Delete: esList(mdelete),
+	}
+}
+
+func esMap(deps []esv1alpha1.Elasticsearch) map[string]esv1alpha1.Elasticsearch {
+	m := map[string]esv1alpha1.Elasticsearch{}
+	for _, d := range deps {
+		m[d.Name] = d
+	}
+	return m
+}
+
+func esList(m map[string]esv1alpha1.Elasticsearch) []esv1alpha1.Elasticsearch {
+	l := []esv1alpha1.Elasticsearch{}
+	for _, v := range m {
+		l = append(l, v)
+	}
+	return l
+}

pkg/inventory/elasticsearch_test.go

@@ -0,0 +1,53 @@
+package inventory
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	esv1alpha1 "github.com/jaegertracing/jaeger-operator/pkg/storage/elasticsearch/v1alpha1"
+)
+
+func TestElasticsearchInventory(t *testing.T) {
+	toCreate := esv1alpha1.Elasticsearch{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "to-create",
+		},
+	}
+	toUpdate := esv1alpha1.Elasticsearch{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "to-update",
+		},
+		Spec: esv1alpha1.ElasticsearchSpec{
+			ManagementState: esv1alpha1.ManagementStateManaged,
+		},
+	}
+	updated := esv1alpha1.Elasticsearch{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "to-update",
+		},
+		Spec: esv1alpha1.ElasticsearchSpec{
+			ManagementState: esv1alpha1.ManagementStateUnmanaged,
+		},
+	}
+	toDelete := esv1alpha1.Elasticsearch{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "to-delete",
+		},
+	}
+
+	existing := []esv1alpha1.Elasticsearch{toUpdate, toDelete}
+	desired := []esv1alpha1.Elasticsearch{updated, toCreate}
+
+	inv := ForElasticsearches(existing, desired)
+	assert.Len(t, inv.Create, 1)
+	assert.Equal(t, "to-create", inv.Create[0].Name)
+
+	assert.Len(t, inv.Update, 1)
+	assert.Equal(t, "to-update", inv.Update[0].Name)
+	assert.Equal(t, esv1alpha1.ManagementStateUnmanaged, inv.Update[0].Spec.ManagementState)
+
+	assert.Len(t, inv.Delete, 1)
+	assert.Equal(t, "to-delete", inv.Delete[0].Name)
+}