Add rollover support

deploy/examples/simple-prod.yaml

@@ -10,3 +10,4 @@ spec:
     options:
       es:
         server-urls: http://elasticsearch:9200
+        use-aliases: true

pkg/apis/jaegertracing/v1/jaeger_types.go

@@ -146,6 +146,7 @@ type JaegerStorageSpec struct {
 	CassandraCreateSchema JaegerCassandraCreateSchemaSpec `json:"cassandraCreateSchema"`
 	SparkDependencies     JaegerDependenciesSpec          `json:"dependencies"`
 	EsIndexCleaner        JaegerEsIndexCleanerSpec        `json:"esIndexCleaner"`
+	Rollover              JaegerEsRolloverSpec            `json: esRollover`
 	Elasticsearch         ElasticsearchSpec               `json:"elasticsearch"`
 }
 
@@ -188,6 +189,15 @@ type JaegerEsIndexCleanerSpec struct {
 	Image        string `json:"image"`
 }
 
+// JaegerEsRolloverSpec holds the options related to es-rollover
+type JaegerEsRolloverSpec struct {
+	Image      string `json:"image"`
+	Schedule   string `json:"schedule"`
+	Conditions string `json:"schedule"`
+	Unit       string `json:"unit"`
+	UnitCount  *int   `json:"unitCount"`
+}
+
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
 // JaegerList contains a list of Jaeger

pkg/cmd/start/main.go

@@ -59,6 +59,9 @@ func NewStartCommand() *cobra.Command {
 	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("jaeger-es-rollover-image", "jaegertracing/jaeger-es-rollover", "The Docker image for the Jaeger Elasticsearch Rollover")
+	viper.BindPFlag("jaeger-es-rollover-image", cmd.Flags().Lookup("jaeger-es-rollover-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/controller/jaeger/jaeger_controller.go

@@ -7,9 +7,10 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/viper"
-	"k8s.io/apimachinery/pkg/api/errors"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -80,7 +81,7 @@ func (r *ReconcileJaeger) Reconcile(request reconcile.Request) (reconcile.Result
 	instance := &v1.Jaeger{}
 	err := r.client.Get(context.Background(), request.NamespacedName, instance)
 	if err != nil {
-		if errors.IsNotFound(err) {
+		if k8serrors.IsNotFound(err) {
 			// Request object not found, could have been deleted after reconcile request.
 			// Owned objects are automatically garbage collected. For additional cleanup logic use finalizers.
 			// Return and don't requeue
@@ -101,12 +102,6 @@ func (r *ReconcileJaeger) Reconcile(request reconcile.Request) (reconcile.Result
 
 	logFields := instance.Logger().WithField("execution", execution)
 
-	// wait for all the dependencies to succeed
-	if err := r.handleDependencies(str); err != nil {
-		logFields.WithError(err).Error("failed to handle the dependencies")
-		return reconcile.Result{}, err
-	}
-
 	if err := r.apply(*instance, str); err != nil {
 		logFields.WithError(err).Error("failed to apply the changes")
 		return reconcile.Result{}, err
@@ -156,6 +151,11 @@ func (r *ReconcileJaeger) apply(jaeger v1.Jaeger, str strategy.S) error {
 		}).Warn("An Elasticsearch cluster should be provisioned, but provisioning is disabled for this Jaeger Operator")
 	}
 
+	// storage dependencies have to be deployed after ES is ready
+	if err := r.handleDependencies(str); err != nil {
+		return errors.Wrap(err, "failed to handler dependencies")
+	}
+
 	if err := r.applyRoles(jaeger, str.Roles()); err != nil {
 		return err
 	}

pkg/cronjob/es_index_cleaner.go

@@ -3,19 +3,21 @@ package cronjob
 import (
 	"fmt"
 	"strconv"
-	"strings"
 
 	batchv1 "k8s.io/api/batch/v1"
 	batchv1beta1 "k8s.io/api/batch/v1beta1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/jaegertracing/jaeger-operator/pkg/apis/jaegertracing/v1"
+	"github.com/jaegertracing/jaeger-operator/pkg/util"
 )
 
+// CreateEsIndexCleaner returns a new cronjob for the Elasticsearch Index Cleaner operation
+
 // CreateEsIndexCleaner returns a new cronjob for the Elasticsearch Index Cleaner operation
 func CreateEsIndexCleaner(jaeger *v1.Jaeger) *batchv1beta1.CronJob {
-	esUrls := getEsHostname(jaeger.Spec.Storage.Options.Map())
+	esUrls := util.GetEsHostname(jaeger.Spec.Storage.Options.Map())
 	trueVar := true
 	one := int32(1)
 	name := fmt.Sprintf("%s-es-index-cleaner", jaeger.Name)
@@ -62,10 +64,10 @@ func CreateEsIndexCleaner(jaeger *v1.Jaeger) *batchv1beta1.CronJob {
 						Spec: corev1.PodSpec{
 							Containers: []corev1.Container{
 								{
-									Image:   jaeger.Spec.Storage.EsIndexCleaner.Image,
 									Name:    name,
-									Env:     removeEmptyVars([]corev1.EnvVar{{Name: "INDEX_PREFIX", Value: jaeger.Spec.Storage.Options.Map()["es.index-prefix"]}}),
+									Image:   jaeger.Spec.Storage.EsIndexCleaner.Image,
 									Args:    []string{strconv.Itoa(jaeger.Spec.Storage.EsIndexCleaner.NumberOfDays), esUrls},
+									Env:     esScriptEnvVars(jaeger.Spec.Storage.Options),
 									EnvFrom: envFromSource,
 								},
 							},
@@ -83,16 +85,3 @@ func CreateEsIndexCleaner(jaeger *v1.Jaeger) *batchv1beta1.CronJob {
 		},
 	}
 }
-
-// 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

@@ -8,21 +8,6 @@ import (
 	"github.com/jaegertracing/jaeger-operator/pkg/apis/jaegertracing/v1"
 )
 
-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 := &v1.Jaeger{Spec: v1.JaegerSpec{Storage: v1.JaegerStorageSpec{Options: v1.NewOptions(
 		map[string]interface{}{"es.index-prefix": "tenant1", "es.server-urls": "http://nowhere:666,foo"})}}}

pkg/cronjob/es_rollover.go

@@ -0,0 +1,128 @@
+package cronjob
+
+import (
+	"fmt"
+	"strconv"
+
+	batchv1 "k8s.io/api/batch/v1"
+	batchv1beta1 "k8s.io/api/batch/v1beta1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/jaegertracing/jaeger-operator/pkg/apis/jaegertracing/v1"
+	"github.com/jaegertracing/jaeger-operator/pkg/util"
+)
+
+// CreateRollover returns objects which are necessary to run rolover actions for indices
+func CreateRollover(jaeger *v1.Jaeger) []batchv1beta1.CronJob {
+	return []batchv1beta1.CronJob{rollover(jaeger), lookback(jaeger)}
+}
+
+func rollover(jaeger *v1.Jaeger) batchv1beta1.CronJob {
+	name := fmt.Sprintf("%s-es-rollover", jaeger.Name)
+	envs := esScriptEnvVars(jaeger.Spec.Storage.Options)
+	if jaeger.Spec.Storage.Rollover.Conditions != "" {
+		envs = append(envs, corev1.EnvVar{Name: "CONDITIONS", Value: jaeger.Spec.Storage.Rollover.Conditions})
+	}
+	one := int32(1)
+	ttlHourInSec := int32(60 * 60)
+	return batchv1beta1.CronJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            name,
+			Namespace:       jaeger.Namespace,
+			Labels:          util.Labels(name, "cronjob-es-rollover", *jaeger),
+			OwnerReferences: []metav1.OwnerReference{util.AsOwner(jaeger)},
+		},
+		Spec: batchv1beta1.CronJobSpec{
+			ConcurrencyPolicy: batchv1beta1.ForbidConcurrent,
+			Schedule:          jaeger.Spec.Storage.Rollover.Schedule,
+			JobTemplate: batchv1beta1.JobTemplateSpec{
+				Spec: batchv1.JobSpec{
+					TTLSecondsAfterFinished: &ttlHourInSec,
+					Parallelism:             &one,
+					Template: corev1.PodTemplateSpec{
+						ObjectMeta: metav1.ObjectMeta{
+							Annotations: map[string]string{
+								"prometheus.io/scrape":    "false",
+								"sidecar.istio.io/inject": "false",
+							},
+						},
+						Spec: corev1.PodSpec{
+							RestartPolicy: corev1.RestartPolicyOnFailure,
+							Containers: []corev1.Container{
+								{
+									Name:  name,
+									Image: jaeger.Spec.Storage.Rollover.Image,
+									Args:  []string{"rollover", util.GetEsHostname(jaeger.Spec.Storage.Options.Map())},
+									Env:   envs,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func lookback(jaeger *v1.Jaeger) batchv1beta1.CronJob {
+	name := fmt.Sprintf("%s-es-lookback", jaeger.Name)
+	envs := esScriptEnvVars(jaeger.Spec.Storage.Options)
+	if jaeger.Spec.Storage.Rollover.Unit != "" {
+		envs = append(envs, corev1.EnvVar{Name: "UNIT", Value: jaeger.Spec.Storage.Rollover.Unit})
+	}
+	if jaeger.Spec.Storage.Rollover.UnitCount != nil {
+		envs = append(envs, corev1.EnvVar{Name: "UNIT_COUNT", Value: strconv.Itoa(*jaeger.Spec.Storage.Rollover.UnitCount)})
+	}
+	ttlHourInSec := int32(60 * 60)
+	return batchv1beta1.CronJob{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            name,
+			Namespace:       jaeger.Namespace,
+			Labels:          util.Labels(name, "cronjob-es-lookback", *jaeger),
+			OwnerReferences: []metav1.OwnerReference{util.AsOwner(jaeger)},
+		},
+		Spec: batchv1beta1.CronJobSpec{
+			ConcurrencyPolicy: batchv1beta1.ForbidConcurrent,
+			Schedule:          jaeger.Spec.Storage.Rollover.Schedule,
+			JobTemplate: batchv1beta1.JobTemplateSpec{
+				Spec: batchv1.JobSpec{
+					TTLSecondsAfterFinished: &ttlHourInSec,
+					Template: corev1.PodTemplateSpec{
+						ObjectMeta: metav1.ObjectMeta{
+							Annotations: map[string]string{
+								"prometheus.io/scrape":    "false",
+								"sidecar.istio.io/inject": "false",
+							},
+						},
+						Spec: corev1.PodSpec{
+							RestartPolicy: corev1.RestartPolicyOnFailure,
+							Containers: []corev1.Container{
+								{
+									Name:  name,
+									Image: jaeger.Spec.Storage.Rollover.Image,
+									Args:  []string{"lookback", util.GetEsHostname(jaeger.Spec.Storage.Options.Map())},
+									Env:   envs,
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+}
+
+func esScriptEnvVars(opts v1.Options) []corev1.EnvVar {
+	var envs []corev1.EnvVar
+	if val, ok := opts.Map()["es.index-prefix"]; ok {
+		envs = append(envs, corev1.EnvVar{Name: "INDEX_PREFIX", Value: val})
+	}
+	if val, ok := opts.Map()["es.username"]; ok {
+		envs = append(envs, corev1.EnvVar{Name: "ES_USERNAME", Value: val})
+	}
+	if val, ok := opts.Map()["es.password"]; ok {
+		envs = append(envs, corev1.EnvVar{Name: "ES_PASSWORD", Value: val})
+	}
+	return envs
+}