tinker with DSL config name

elastic · Oct 3, 2023 · 247ef03 · 247ef03
1 parent 4bb819c
commit 247ef03
Show file tree

Hide file tree

Showing 5 changed files with 40 additions and 11 deletions.
diff --git a/libbeat/idxmgmt/lifecycle/client_handler_integration_test.go b/libbeat/idxmgmt/lifecycle/client_handler_integration_test.go
@@ -49,7 +49,7 @@ func TestESClientHandler_CheckILMEnabled(t *testing.T) {
 	t.Run("no ilm if disabled", func(t *testing.T) {
 		cfg := RawConfig{
 			ILM: config.MustNewConfigFrom(mapstr.M{"enabled": false, "policy_name": "test", "check_exists": true}),
-			DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "policy_name": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
+			DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "data_stream_pattern": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
 		}
 		h, err := newESClientHandler(t, cfg)
 		require.NoError(t, err)
@@ -61,7 +61,7 @@ func TestESClientHandler_CheckILMEnabled(t *testing.T) {
 	t.Run("with ilm if enabled", func(t *testing.T) {
 		cfg := RawConfig{
 			ILM: config.MustNewConfigFrom(mapstr.M{"enabled": true, "policy_name": "test", "check_exists": true}),
-			DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "policy_name": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
+			DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "data_stream_pattern": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
 		}
 		h, err := newESClientHandler(t, cfg)
 		require.NoError(t, err)
@@ -80,7 +80,7 @@ func TestESClientHandler_ILMPolicy(t *testing.T) {
 		}
 		cfg := RawConfig{
 			ILM: config.MustNewConfigFrom(mapstr.M{"enabled": true, "policy_name": "test", "check_exists": true}),
-			DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "policy_name": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
+			DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "data_stream_pattern": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
 		}
 		rawClient := newRawESClient(t)
 		h, err := NewESClientHandler(rawClient, beat.Info{Beat: "testbeat"}, cfg)
@@ -102,7 +102,7 @@ func TestESClientHandler_ILMPolicy(t *testing.T) {
 		}
 		cfg := RawConfig{
 			ILM: config.MustNewConfigFrom(mapstr.M{"enabled": true, "policy_name": "test", "check_exists": true}),
-			DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "policy_name": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
+			DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "data_stream_pattern": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
 		}
 		rawClient := newRawESClient(t)
 		h, err := NewESClientHandler(rawClient, beat.Info{Beat: "testbeat"}, cfg)
@@ -186,7 +186,7 @@ func getEnv(name, def string) string {
 func TestFileClientHandler_CheckILMEnabled(t *testing.T) {
 	defaultCfg := RawConfig{
 		ILM: config.MustNewConfigFrom(mapstr.M{"enabled": true, "policy_name": "test", "check_exists": true}),
-		DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "policy_name": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
+		DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "data_stream_pattern": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
 	}
 	for name, test := range map[string]struct {
 		version    string

diff --git a/libbeat/idxmgmt/lifecycle/config.go b/libbeat/idxmgmt/lifecycle/config.go
@@ -26,7 +26,8 @@ import (
 
 // Config is used for unpacking a config.C.
 type Config struct {
-	Enabled    bool                     `config:"enabled"`
+	Enabled bool `config:"enabled"`
+	// PolicyName, used by ILM
 	PolicyName fmtstr.EventFormatString `config:"policy_name"`
 	PolicyFile string                   `config:"policy_file"`
 	// used only for testing
@@ -41,6 +42,18 @@ type Config struct {
 	Overwrite bool `config:"overwrite"`
 }
 
+// DSLNameConfig just stores the datastream name for the DSL policy
+// as this is the only config value that differs between ILM and DSL
+type DSLNameConfig struct {
+	DataStreamPattern fmtstr.EventFormatString `config:"data_stream_pattern"`
+}
+
+func DefaultDSLName() DSLNameConfig {
+	return DSLNameConfig{
+		DataStreamPattern: *fmtstr.MustCompileEvent("%{[beat.name]}-%{[beat.version]}"),
+	}
+}
+
 // LifecycleConfig maps all possible ILM/DSL config values present in a config
 type LifecycleConfig struct {
 	ILM Config `config:"setup.ilm"`

diff --git a/libbeat/idxmgmt/lifecycle/es_client_handler.go b/libbeat/idxmgmt/lifecycle/es_client_handler.go
@@ -62,6 +62,14 @@ func NewESClientHandler(c ESClient, info beat.Info, cfg RawConfig) (*ESClientHan
 			err = cfg.DSL.Unpack(&lifecycleCfg)
 		}
 
+		// unpack name value separately
+		dsName := DefaultDSLName()
+		err := cfg.DSL.Unpack(&dsName)
+		if err != nil {
+			return nil, fmt.Errorf("error unpacking DSL data stream name: %w", err)
+		}
+		lifecycleCfg.PolicyName = dsName.DataStreamPattern
+
 	} else {
 		lifecycleCfg = DefaultILMConfig(info).ILM
 		if cfg.ILM != nil {

diff --git a/libbeat/idxmgmt/lifecycle/es_client_test.go b/libbeat/idxmgmt/lifecycle/es_client_test.go
@@ -64,22 +64,22 @@ func TestESSetup(t *testing.T) {
 
 	defaultILMCfg := RawConfig{
 		ILM: config.MustNewConfigFrom(mapstr.M{"enabled": true, "policy_name": "test", "check_exists": true}),
-		DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "policy_name": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
+		DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "data_stream_pattern": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
 	}
 
 	defaultDSLCfg := RawConfig{
 		ILM: config.MustNewConfigFrom(mapstr.M{"enabled": false, "policy_name": "test", "check_exists": true}),
-		DSL: config.MustNewConfigFrom(mapstr.M{"enabled": true, "policy_name": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
+		DSL: config.MustNewConfigFrom(mapstr.M{"enabled": true, "data_stream_pattern": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
 	}
 
 	bothDisabledConfig := RawConfig{
 		ILM: config.MustNewConfigFrom(mapstr.M{"enabled": false, "policy_name": "test", "check_exists": true}),
-		DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "policy_name": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
+		DSL: config.MustNewConfigFrom(mapstr.M{"enabled": false, "data_stream_pattern": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
 	}
 
 	bothEnabledConfig := RawConfig{
 		ILM: config.MustNewConfigFrom(mapstr.M{"enabled": true, "policy_name": "test", "check_exists": true}),
-		DSL: config.MustNewConfigFrom(mapstr.M{"enabled": true, "policy_name": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
+		DSL: config.MustNewConfigFrom(mapstr.M{"enabled": true, "data_stream_pattern": "%{[beat.name]}-%{[beat.version]}", "check_exists": true}),
 	}
 
 	cases := map[string]struct {
@@ -158,7 +158,7 @@ func TestESSetup(t *testing.T) {
 			serverless: true,
 			cfg: RawConfig{
 				DSL: config.MustNewConfigFrom(mapstr.M{"enabled": true, "overwrite": true,
-					"check_exists": true, "policy_name": "test"}),
+					"check_exists": true, "data_stream_pattern": "test"}),
 			},
 			err:            false,
 			existingPolicy: mapstr.M{"existing": "policy"},

diff --git a/libbeat/idxmgmt/lifecycle/file_client_handler.go b/libbeat/idxmgmt/lifecycle/file_client_handler.go
@@ -51,6 +51,14 @@ func NewFileClientHandler(c FileClient, info beat.Info, cfg RawConfig) (*FileCli
 	if cfg.DSL.Enabled() {
 		lifecycleCfg = DefaultDSLConfig(info).DSL
 		err = cfg.DSL.Unpack(&lifecycleCfg)
+
+		// unpack name value separately
+		dsName := DefaultDSLName()
+		err := cfg.DSL.Unpack(&dsName)
+		if err != nil {
+			return nil, fmt.Errorf("error unpacking DSL data stream name: %w", err)
+		}
+		lifecycleCfg.PolicyName = dsName.DataStreamPattern
 	} else if cfg.ILM.Enabled() {
 		lifecycleCfg = DefaultILMConfig(info).ILM
 		err = cfg.ILM.Unpack(&lifecycleCfg)