feat: Support elasticsearch data stream

Signed-off-by: zzzk1 <[email protected]>

cmd/es-rollover/app/flags.go

@@ -47,7 +47,7 @@ func AddFlags(flags *flag.FlagSet) {
 	flags.Bool(archive, false, "Handle archive indices")
 	flags.String(username, "", "The username required by storage")
 	flags.String(password, "", "The password required by storage")
-	flags.Bool(useILM, false, "Use ILM to manage jaeger indices")
+	flags.Bool(useILM, true, "Use ILM to manage jaeger indices")
 	flags.String(ilmPolicyName, "jaeger-ilm-policy", "The name of the ILM policy to use if ILM is active")
 	flags.Int(timeout, 120, "Number of seconds to wait for master node response")
 	flags.Bool(skipDependencies, false, "Disable rollover for dependencies index")

cmd/es-rollover/app/init/action.go

@@ -10,6 +10,8 @@ import (
 	"net/http"
 	"strings"
 
+	"github.com/elastic/go-elasticsearch/v8/typedapi/types"
+
 	"github.com/jaegertracing/jaeger/cmd/es-rollover/app"
 	"github.com/jaegertracing/jaeger/pkg/es"
 	"github.com/jaegertracing/jaeger/pkg/es/client"
@@ -56,7 +58,10 @@ func (c Action) Do() error {
 			return err
 		}
 		if !policyExist {
-			return fmt.Errorf("ILM policy %s doesn't exist in Elasticsearch. Please create it and re-run init", c.Config.ILMPolicyName)
+			err := createDefaultILMPolicy(c.ILMClient, c.Config.ILMPolicyName)
+			if err != nil {
+				return fmt.Errorf("ILM policy %s Create/Update failed %w", c.Config.ILMPolicyName, err)
+			}
 		}
 	}
 	rolloverIndices := app.RolloverIndices(c.Config.Archive, c.Config.SkipDependencies, c.Config.AdaptiveSampling, c.Config.Config.IndexPrefix)
@@ -68,6 +73,64 @@ func (c Action) Do() error {
 	return nil
 }
 
+const esILMPolicy = `
+{
+  "policy": {
+    "phases": {
+      "hot": {
+        "min_age": "9999ms",
+        "actions": {
+          "rollover": {
+            "max_age": "1m"
+          },
+          "set_priority": {
+            "priority": 9999
+          }
+        }
+      },
+      "delete": {
+        "min_age": "9999m",
+        "actions": {
+          "delete": {}
+        }
+      }
+    }
+  }
+}
+`
+
+func createDefaultILMPolicy(c client.IndexManagementLifecycleAPI, policy string) error {
+	ilmPolicy := types.NewIlmPolicy()
+	err := json.Unmarshal([]byte(esILMPolicy), ilmPolicy)
+	if err != nil {
+		return fmt.Errorf("error unmarshalling ILM policy: %w", err)
+	}
+	err = c.CreateOrUpdate(policy, client.Policy{ILMPolicy: *ilmPolicy})
+	if err != nil {
+		var esErr client.ResponseError
+		if errors.As(err, &esErr) {
+			if esErr.StatusCode != http.StatusBadRequest || esErr.Body == nil {
+				return esErr.Err
+			}
+			// check for the reason of the error
+			jsonError := map[string]any{}
+			err := json.Unmarshal(esErr.Body, &jsonError)
+			if err != nil {
+				// return unmarshal error
+				return err
+			}
+			errorMap := jsonError["error"].(map[string]any)
+			// check for reason, ignore already exist error
+			if strings.Contains(errorMap["type"].(string), "resource_already_exists_exception") {
+				return nil
+			}
+		}
+		// Return any other error unrelated to the response
+		return err
+	}
+	return nil
+}
+
 func createIndexIfNotExist(c client.IndexAPI, index string) error {
 	err := c.CreateIndex(index)
 	if err != nil {

pkg/es/client/ilm_client.go

@@ -4,9 +4,12 @@
 package client
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
+
+	"github.com/elastic/go-elasticsearch/v8/typedapi/types"
 )
 
 var _ IndexManagementLifecycleAPI = (*ILMClient)(nil)
@@ -17,6 +20,35 @@ type ILMClient struct {
 	MasterTimeoutSeconds int
 }
 
+type Policy struct {
+	ILMPolicy types.IlmPolicy `json:"policy"`
+}
+
+// CreateOrUpdate Add or update a ILMPolicy
+func (i ILMClient) CreateOrUpdate(name string, ilmPolicy Policy) error {
+	body, err := json.Marshal(ilmPolicy)
+	if err != nil {
+		return err
+	}
+	_, err = i.request(elasticRequest{
+		endpoint: "_ilm/policy/" + name,
+		body:     body,
+		method:   http.MethodPut,
+	})
+
+	var respError ResponseError
+	if errors.As(err, &respError) {
+		if respError.StatusCode == http.StatusNotFound {
+			return nil
+		}
+	}
+
+	if err != nil {
+		return fmt.Errorf("failed to create/update ILM policy: %s, %w", name, err)
+	}
+	return nil
+}
+
 // Exists verify if a ILM policy exists
 func (i ILMClient) Exists(name string) (bool, error) {
 	_, err := i.request(elasticRequest{

pkg/es/client/ilm_client_test.go

@@ -4,11 +4,14 @@
 package client
 
 import (
+	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"strings"
 	"testing"
 
+	"github.com/elastic/go-elasticsearch/v8/typedapi/types"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -64,3 +67,77 @@ func TestExists(t *testing.T) {
 		})
 	}
 }
+
+const esILMPolicy = `
+{
+  "policy": {
+    "phases": {
+      "hot": {
+        "min_age": "0ms",
+        "actions": {
+          "rollover": {
+            "max_age": "1m"
+          },
+          "set_priority": {
+            "priority": 100
+          }
+        }
+      },
+      "delete": {
+        "min_age": "2m",
+        "actions": {
+          "delete": {}
+        }
+      }
+    }
+  }
+}
+`
+
+func TestCreate(t *testing.T) {
+	policy := "jaeger-ilm-policy"
+	tests := []struct {
+		name         string
+		responseCode int
+		response     string
+		errContains  string
+	}{
+		{
+			name:         "successful",
+			responseCode: http.StatusOK,
+		},
+		{
+			name:         "client error",
+			responseCode: http.StatusBadRequest,
+			response:     esErrResponse,
+			errContains:  "failed to create/update ILM policy: jaeger-ilm-policy",
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			testServer := httptest.NewServer(http.HandlerFunc(func(res http.ResponseWriter, req *http.Request) {
+				assert.True(t, strings.HasSuffix(req.URL.String(), "_ilm/policy/jaeger-ilm-policy"))
+				assert.Equal(t, http.MethodPut, req.Method)
+				res.WriteHeader(test.responseCode)
+				res.Write([]byte(test.response))
+			}))
+			defer testServer.Close()
+
+			c := &ILMClient{
+				Client: Client{
+					Client:    testServer.Client(),
+					Endpoint:  testServer.URL,
+					BasicAuth: "foobar",
+				},
+			}
+			ilmPolicy := types.NewIlmPolicy()
+			json.Unmarshal([]byte(esILMPolicy), ilmPolicy)
+			fmt.Printf("%+v\n", *ilmPolicy)
+
+			err := c.CreateOrUpdate(policy, Policy{ILMPolicy: *ilmPolicy})
+			if test.errContains != "" {
+				require.ErrorContains(t, err, test.errContains)
+			}
+		})
+	}
+}

pkg/es/client/interfaces.go

@@ -19,4 +19,5 @@ type ClusterAPI interface {
 
 type IndexManagementLifecycleAPI interface {
 	Exists(name string) (bool, error)
+	CreateOrUpdate(policy string, ilmPolicy Policy) error
 }

plugin/storage/integration/elasticsearch_test.go

@@ -44,6 +44,7 @@ const (
 	dependenciesTemplateName = "jaeger-dependencies"
 	primaryNamespace         = "es"
 	archiveNamespace         = "es-archive"
+	ilmPolicy                = "jaeger-ilm-policy"
 )
 
 type ESStorageIntegration struct {
@@ -110,7 +111,7 @@ func (*ESStorageIntegration) initializeESFactory(t *testing.T, allTagsAsFields b
 		fmt.Sprintf("--es.num-shards=%v", 5),
 		fmt.Sprintf("--es.num-replicas=%v", 1),
 		fmt.Sprintf("--es.index-prefix=%v", indexPrefix),
-		fmt.Sprintf("--es.use-ilm=%v", false),
+		fmt.Sprintf("--es.use-ilm=%v", true),
 		fmt.Sprintf("--es.service-cache-ttl=%v", 1*time.Second),
 		fmt.Sprintf("--es.tags-as-fields.all=%v", allTagsAsFields),
 		fmt.Sprintf("--es.bulk.actions=%v", 1),
@@ -245,3 +246,37 @@ func (s *ESStorageIntegration) cleanESIndexTemplates(t *testing.T, prefix string
 	}
 	return nil
 }
+
+func TestElasticsearchStorage_ILMPolicy(t *testing.T) {
+	SkipUnlessEnv(t, "elasticsearch", "opensearch")
+	t.Cleanup(func() {
+		testutils.VerifyGoLeaksOnceForES(t)
+	})
+	c := getESHttpClient(t)
+	require.NoError(t, healthCheck(c))
+	s := &ESStorageIntegration{}
+	s.initializeES(t, c, true)
+	esVersion, err := s.getVersion()
+	require.NoError(t, err)
+	if esVersion == 8 {
+		request := s.v8Client.ILM.GetLifecycle.WithPolicy(ilmPolicy)
+		ilmPolicyExistsResponse, err := s.v8Client.ILM.GetLifecycle(request)
+		require.NoError(t, err)
+		assert.Equal(t, 200, ilmPolicyExistsResponse.StatusCode)
+	} else {
+		// nothing
+	}
+	s.cleanEsILMPolicy(t, ilmPolicy)
+}
+
+func (s *ESStorageIntegration) cleanEsILMPolicy(t *testing.T, policy string) error {
+	version, err := s.getVersion()
+	require.NoError(t, err)
+	if version == 8 {
+		_, err := s.v8Client.ILM.RemovePolicy(ilmPolicy)
+		require.NoError(t, err)
+	} else {
+		// nothing
+	}
+	return nil
+}