Add Elasticsearch index cleaner as cron job

deploy/examples/simple-prod-with-volumes.yaml

@@ -10,8 +10,6 @@ spec:
     options:
       es:
         server-urls: http://elasticsearch:9200
-        username: elastic
-        password: changeme
   volumeMounts:
   - name: elastic-data
     mountPath: /usr/share/elasticsearch/data

deploy/examples/simple-prod.yaml

@@ -10,7 +10,5 @@ spec:
     options:
       es:
         server-urls: http://elasticsearch:9200
-        username: elastic
-        password: changeme
     dependencies:
       enabled: true

pkg/apis/io/v1alpha1/jaeger_types.go

@@ -126,6 +126,7 @@ type JaegerStorageSpec struct {
 	Options               Options                         `json:"options"`
 	CassandraCreateSchema JaegerCassandraCreateSchemaSpec `json:"cassandraCreateSchema"`
 	SparkDependencies     JaegerDependenciesSpec          `json:"dependencies"`
+	EsIndexCleaner        JaegerEsIndexCleanerSpec        `json:"esIndexCleaner"`
 }
 
 // JaegerCassandraCreateSchemaSpec holds the options related to the create-schema batch job
@@ -150,6 +151,14 @@ type JaegerDependenciesSpec struct {
 	ElasticsearchNodesWanOnly   bool   `json:"elasticsearchNodesWanOnly"`
 }
 
+// JaegerEsIndexCleanerSpec holds the options related to es-index-cleaner
+type JaegerEsIndexCleanerSpec struct {
+	Enabled      bool   `json:"enabled"`
+	NumberOfDays int    `json:"numberOfDays"`
+	Schedule     string `json:"schedule"`
+	Image        string `json:"image"`
+}
+
 func init() {
 	SchemeBuilder.Register(&Jaeger{}, &JaegerList{})
 }

pkg/cmd/start/main.go

@@ -48,6 +48,9 @@ func NewStartCommand() *cobra.Command {
 	cmd.Flags().String("jaeger-spark-dependencies-image", "jaegertracing/spark-dependencies", "The Docker image for the Spark Dependencies Job")
 	viper.BindPFlag("jaeger-spark-dependencies-image", cmd.Flags().Lookup("jaeger-spark-dependencies-image"))
 
+	cmd.Flags().String("jaeger-es-index-cleaner-image", "jaegertracing/jaeger-es-index-cleaner", "The Docker image for the Jaeger Elasticsearch Index Cleaner")
+	viper.BindPFlag("jaeger-es-index-cleaner-image", cmd.Flags().Lookup("jaeger-es-index-cleaner-image"))
+
 	cmd.Flags().String("openshift-oauth-proxy-image", "openshift/oauth-proxy:latest", "The Docker image location definition for the OpenShift OAuth Proxy")
 	viper.BindPFlag("openshift-oauth-proxy-image", cmd.Flags().Lookup("openshift-oauth-proxy-image"))
 

pkg/cronjob/es_index_cleaner.go

@@ -0,0 +1,74 @@
+package cronjob
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+
+	batchv1 "k8s.io/api/batch/v1"
+	batchv1beta1 "k8s.io/api/batch/v1beta1"
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/jaegertracing/jaeger-operator/pkg/apis/io/v1alpha1"
+)
+
+func CreateEsIndexCleaner(jaeger *v1alpha1.Jaeger) *batchv1beta1.CronJob {
+	esUrls := getEsHostname(jaeger.Spec.Storage.Options.Map())
+	trueVar := true
+	name := fmt.Sprintf("%s-es-index-cleaner", jaeger.Name)
+	return &batchv1beta1.CronJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: jaeger.Namespace,
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: jaeger.APIVersion,
+					Kind:       jaeger.Kind,
+					Name:       jaeger.Name,
+					UID:        jaeger.UID,
+					Controller: &trueVar,
+				},
+			},
+		},
+		Spec: batchv1beta1.CronJobSpec{
+			Schedule: jaeger.Spec.Storage.EsIndexCleaner.Schedule,
+			JobTemplate: batchv1beta1.JobTemplateSpec{
+				Spec: batchv1.JobSpec{
+					Template: v1.PodTemplateSpec{
+						Spec: v1.PodSpec{
+							Containers: []v1.Container{
+								{
+									Image: jaeger.Spec.Storage.EsIndexCleaner.Image,
+									Name:  name,
+									Env:   []v1.EnvVar{{Name: "INDEX_PREFIX", Value: jaeger.Spec.Storage.Options.Map()["es.index-prefix"]}},
+									Args:  []string{strconv.Itoa(jaeger.Spec.Storage.EsIndexCleaner.NumberOfDays), esUrls},
+								},
+							},
+							RestartPolicy: v1.RestartPolicyNever,
+						},
+						ObjectMeta: metav1.ObjectMeta{
+							Annotations: map[string]string{
+								"prometheus.io/scrape":    "false",
+								"sidecar.istio.io/inject": "false",
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+// return first ES hostname from options map
+func getEsHostname(opts map[string]string) string {
+	urls, ok := opts["es.server-urls"]
+	if !ok {
+		return ""
+	}
+	urlArr := strings.Split(urls, ",")
+	if len(urlArr) == 0 {
+		return ""
+	}
+	return urlArr[0]
+}

pkg/cronjob/es_index_cleaner_test.go

@@ -0,0 +1,36 @@
+package cronjob
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/jaegertracing/jaeger-operator/pkg/apis/io/v1alpha1"
+)
+
+func TestGetEsHostname(t *testing.T) {
+	tests := []struct {
+		underTest map[string]string
+		hostname  string
+	}{
+		{hostname: ""},
+		{underTest: map[string]string{"": ""}, hostname: ""},
+		{underTest: map[string]string{"es.server-urls": "goo:tar"}, hostname: "goo:tar"},
+		{underTest: map[string]string{"es.server-urls": "http://es:9000,https://es2:9200"}, hostname: "http://es:9000"},
+	}
+	for _, test := range tests {
+		assert.Equal(t, test.hostname, getEsHostname(test.underTest))
+	}
+}
+
+func TestCreateEsIndexCleaner(t *testing.T) {
+	jaeger := &v1alpha1.Jaeger{Spec: v1alpha1.JaegerSpec{Storage: v1alpha1.JaegerStorageSpec{Options: v1alpha1.NewOptions(
+		map[string]interface{}{"es.index-prefix": "tenant1", "es.server-urls": "http://nowhere:666,foo"})}}}
+	cronJob := CreateEsIndexCleaner(jaeger)
+	assert.Equal(t, 2, len(cronJob.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Args))
+	// default number of days (7) is applied in normalize in controller
+	assert.Equal(t, []string{"0", "http://nowhere:666"}, cronJob.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Args)
+	assert.Equal(t, 1, len(cronJob.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Env))
+	assert.Equal(t, "INDEX_PREFIX", cronJob.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Env[0].Name)
+	assert.Equal(t, "tenant1", cronJob.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Env[0].Value)
+}

pkg/cronjob/spark_dependencies.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"strconv"
 
-	"github.com/spf13/viper"
 	batchv1 "k8s.io/api/batch/v1"
 	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	"k8s.io/api/core/v1"
@@ -20,9 +19,7 @@ func SupportedStorage(storage string) bool {
 	return supportedStorageTypes[storage]
 }
 
-func Create(jaeger *v1alpha1.Jaeger) *batchv1beta1.CronJob {
-	applyDefaults(jaeger)
-
+func CreateSparkDependencies(jaeger *v1alpha1.Jaeger) *batchv1beta1.CronJob {
 	envVars := []v1.EnvVar{
 		{Name: "STORAGE", Value: jaeger.Spec.Storage.Type},
 		{Name: "SPARK_MASTER", Value: jaeger.Spec.Storage.SparkDependencies.SparkMaster},
@@ -107,15 +104,6 @@ func getStorageEnvs(s v1alpha1.JaegerStorageSpec) []v1.EnvVar {
 	}
 }
 
-func applyDefaults(jaeger *v1alpha1.Jaeger) {
-	if jaeger.Spec.Storage.SparkDependencies.Image == "" {
-		jaeger.Spec.Storage.SparkDependencies.Image = fmt.Sprintf("%s", viper.GetString("jaeger-spark-dependencies-image"))
-	}
-	if jaeger.Spec.Storage.SparkDependencies.Schedule == "" {
-		jaeger.Spec.Storage.SparkDependencies.Schedule = "55 23 * * *"
-	}
-}
-
 func removeEmptyVars(envVars []v1.EnvVar) []v1.EnvVar {
 	var notEmpty []v1.EnvVar
 	for _, v := range envVars {

pkg/cronjob/spark_dependencies_test.go

@@ -9,24 +9,6 @@ import (
 	"github.com/jaegertracing/jaeger-operator/pkg/apis/io/v1alpha1"
 )
 
-func TestApplyDefaults(t *testing.T) {
-	tests := []struct {
-		underTest *v1alpha1.Jaeger
-		expected  *v1alpha1.Jaeger
-	}{
-		{underTest: &v1alpha1.Jaeger{}, expected: &v1alpha1.Jaeger{Spec: v1alpha1.JaegerSpec{Storage: v1alpha1.JaegerStorageSpec{
-			SparkDependencies: v1alpha1.JaegerDependenciesSpec{Schedule: "55 23 * * *"}}}}},
-		{underTest: &v1alpha1.Jaeger{Spec: v1alpha1.JaegerSpec{Storage: v1alpha1.JaegerStorageSpec{
-			SparkDependencies: v1alpha1.JaegerDependenciesSpec{Schedule: "foo"}}}},
-			expected: &v1alpha1.Jaeger{Spec: v1alpha1.JaegerSpec{Storage: v1alpha1.JaegerStorageSpec{
-				SparkDependencies: v1alpha1.JaegerDependenciesSpec{Schedule: "foo"}}}}},
-	}
-	for _, test := range tests {
-		applyDefaults(test.underTest)
-		assert.Equal(t, test.expected, test.underTest)
-	}
-}
-
 func TestRemoveEmptyVars(t *testing.T) {
 	tests := []struct {
 		underTest []v1.EnvVar
@@ -80,5 +62,5 @@ func TestStorageEnvs(t *testing.T) {
 }
 
 func TestCreate(t *testing.T) {
-	assert.NotNil(t, Create(&v1alpha1.Jaeger{Spec: v1alpha1.JaegerSpec{Storage: v1alpha1.JaegerStorageSpec{Type: "elasticsearch"}}}))
+	assert.NotNil(t, CreateSparkDependencies(&v1alpha1.Jaeger{Spec: v1alpha1.JaegerSpec{Storage: v1alpha1.JaegerStorageSpec{Type: "elasticsearch"}}}))
 }

pkg/deployment/collector_test.go

@@ -1,11 +1,12 @@
 package deployment
 
 import (
+	"testing"
+
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
-	"testing"
 
 	"github.com/jaegertracing/jaeger-operator/pkg/apis/io/v1alpha1"
 )

pkg/strategy/all-in-one.go

@@ -81,12 +81,20 @@ func (c *allInOneStrategy) Create() []runtime.Object {
 
 	if cronjob.SupportedStorage(c.jaeger.Spec.Storage.Type) {
 		if c.jaeger.Spec.Storage.SparkDependencies.Enabled {
-			os = append(os, cronjob.Create(c.jaeger))
+			os = append(os, cronjob.CreateSparkDependencies(c.jaeger))
 		} else {
 			logrus.Info("Spark dependencies are disabled - need to be enabled explicitly")
 		}
 	}
 
+	if c.jaeger.Spec.Storage.EsIndexCleaner.Enabled {
+		if c.jaeger.Spec.Storage.Type == "elasticsearch" {
+			os = append(os, cronjob.CreateEsIndexCleaner(c.jaeger))
+		} else {
+			logrus.WithField("type", c.jaeger.Spec.Storage.Type).Warn("Elasticsearch index cleaner is enabled for unsupported storage")
+		}
+	}
+
 	return os
 }
 

pkg/strategy/all-in-one_test.go

@@ -172,3 +172,42 @@ func testSparkDependencies(t *testing.T, fce func(jaeger *v1alpha1.Jaeger) S) {
 		}
 	}
 }
+
+func TestEsIndexCleanerAllInOne(t *testing.T) {
+	testEsIndexCleaner(t, func(jaeger *v1alpha1.Jaeger) S {
+		return &allInOneStrategy{jaeger: jaeger}
+	})
+}
+
+func testEsIndexCleaner(t *testing.T, fce func(jaeger *v1alpha1.Jaeger) S) {
+	tests := []struct {
+		jaeger              *v1alpha1.Jaeger
+		sparkCronJobEnabled bool
+	}{
+		{jaeger: &v1alpha1.Jaeger{Spec: v1alpha1.JaegerSpec{
+			Storage: v1alpha1.JaegerStorageSpec{Type: "elasticsearch",
+				EsIndexCleaner: v1alpha1.JaegerEsIndexCleanerSpec{Enabled: true}},
+		}}, sparkCronJobEnabled: true},
+		{jaeger: &v1alpha1.Jaeger{Spec: v1alpha1.JaegerSpec{
+			Storage: v1alpha1.JaegerStorageSpec{Type: "cassandra",
+				EsIndexCleaner: v1alpha1.JaegerEsIndexCleanerSpec{Enabled: true}},
+		}}, sparkCronJobEnabled: false},
+		{jaeger: &v1alpha1.Jaeger{Spec: v1alpha1.JaegerSpec{
+			Storage: v1alpha1.JaegerStorageSpec{Type: "kafka",
+				EsIndexCleaner: v1alpha1.JaegerEsIndexCleanerSpec{Enabled: true}},
+		}}, sparkCronJobEnabled: false},
+		{jaeger: &v1alpha1.Jaeger{Spec: v1alpha1.JaegerSpec{
+			Storage: v1alpha1.JaegerStorageSpec{Type: "elasticsearch"},
+		}}, sparkCronJobEnabled: false},
+	}
+	for _, test := range tests {
+		s := fce(test.jaeger)
+		objs := s.Create()
+		cronJobs := getTypesOf(objs, reflect.TypeOf(&batchv1beta1.CronJob{}))
+		if test.sparkCronJobEnabled {
+			assert.Equal(t, 1, len(cronJobs))
+		} else {
+			assert.Equal(t, 0, len(cronJobs))
+		}
+	}
+}

pkg/strategy/controller.go

@@ -2,6 +2,7 @@ package strategy
 
 import (
 	"context"
+	"fmt"
 	"strings"
 
 	"github.com/sirupsen/logrus"
@@ -87,6 +88,30 @@ func normalize(jaeger *v1alpha1.Jaeger) {
 		// - 'none' on any platform
 		jaeger.Spec.Ingress.Security = v1alpha1.IngressSecurityNone
 	}
+
+	normalizeSparkDependencies(&jaeger.Spec.Storage.SparkDependencies)
+	normalizeIndexCleaner(&jaeger.Spec.Storage.EsIndexCleaner)
+}
+
+func normalizeSparkDependencies(spec *v1alpha1.JaegerDependenciesSpec) {
+	if spec.Image == "" {
+		spec.Image = fmt.Sprintf("%s", viper.GetString("jaeger-spark-dependencies-image"))
+	}
+	if spec.Schedule == "" {
+		spec.Schedule = "55 23 * * *"
+	}
+}
+
+func normalizeIndexCleaner(spec *v1alpha1.JaegerEsIndexCleanerSpec) {
+	if spec.Image == "" {
+		spec.Image = fmt.Sprintf("%s", viper.GetString("jaeger-es-index-cleaner-image"))
+	}
+	if spec.Schedule == "" {
+		spec.Schedule = "55 23 * * *"
+	}
+	if spec.NumberOfDays == 0 {
+		spec.NumberOfDays = 7
+	}
 }
 
 func unknownStorage(typ string) bool {