Enable require_alias for Bulk requests for all actions when target is a write alias

CHANGELOG.next.asciidoc

@@ -36,6 +36,8 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 *Affecting all Beats*
 
+- Enable `require_alias` for Bulk requests for all actions when target is a write alias. {issue}27874[27874] {pull}29879[29879]
+
 
 *Auditbeat*
 

libbeat/esleg/eslegclient/bulkapi.go

@@ -51,10 +51,11 @@ type BulkDeleteAction struct {
 }
 
 type BulkMeta struct {
-	Index    string `json:"_index" struct:"_index"`
-	DocType  string `json:"_type,omitempty" struct:"_type,omitempty"`
-	Pipeline string `json:"pipeline,omitempty" struct:"pipeline,omitempty"`
-	ID       string `json:"_id,omitempty" struct:"_id,omitempty"`
+	Index        string `json:"_index" struct:"_index"`
+	DocType      string `json:"_type,omitempty" struct:"_type,omitempty"`
+	Pipeline     string `json:"pipeline,omitempty" struct:"pipeline,omitempty"`
+	ID           string `json:"_id,omitempty" struct:"_id,omitempty"`
+	RequireAlias bool   `json:"require_alias,omitempty" struct:"require_alias,omitempty"`
 }
 
 type bulkRequest struct {

libbeat/idxmgmt/std.go

@@ -337,13 +337,17 @@ func (s *ilmIndexSelector) Select(evt *beat.Event) (string, error) {
 	return idx, err
 }
 
+func (s ilmIndexSelector) IsAlias() bool { return true }
+
 func (s indexSelector) Select(evt *beat.Event) (string, error) {
 	if idx := getEventCustomIndex(evt, s.beatInfo); idx != "" {
 		return idx, nil
 	}
 	return s.sel.Select(evt)
 }
 
+func (s indexSelector) IsAlias() bool { return false }
+
 func getEventCustomIndex(evt *beat.Event, beatInfo beat.Info) string {
 	if len(evt.Meta) == 0 {
 		return ""

libbeat/outputs/elasticsearch/client.go

@@ -317,6 +317,10 @@ func (client *Client) createEventBulkMeta(version common.Version, event *beat.Ev
 		ID:       id,
 	}
 
+	if isRequireAliasSupported(version) {
+		meta.RequireAlias = client.index.IsAlias()
+	}
+
 	if opType == events.OpTypeDelete {
 		if id != "" {
 			return eslegclient.BulkDeleteAction{Delete: meta}, nil
@@ -333,6 +337,10 @@ func (client *Client) createEventBulkMeta(version common.Version, event *beat.Ev
 	return eslegclient.BulkIndexAction{Index: meta}, nil
 }
 
+func isRequireAliasSupported(version common.Version) bool {
+	return !version.LessThan(common.MustNewVersion("7.10.0"))
+}
+
 func (client *Client) getPipeline(event *beat.Event) (string, error) {
 	if event.Meta != nil {
 		pipeline, err := events.GetMetaStringValue(*event, events.FieldMetaPipeline)

libbeat/outputs/elasticsearch/client_integration_test.go

@@ -420,7 +420,8 @@ func connectTestEs(t *testing.T, cfg interface{}, stats outputs.Observer) (outpu
 	}
 
 	info := beat.Info{Beat: "libbeat"}
-	im, _ := idxmgmt.DefaultSupport(nil, info, nil)
+	// ILM must be disabled otherwise custom index settings are ignored.
+	im, _ := idxmgmt.DefaultSupport(nil, info, disabledILMConfig())
 	output, err := makeES(im, info, stats, config)
 	if err != nil {
 		t.Fatal(err)
@@ -438,6 +439,10 @@ func connectTestEs(t *testing.T, cfg interface{}, stats outputs.Observer) (outpu
 	return client, client
 }
 
+func disabledILMConfig() *common.Config {
+	return common.MustNewConfigFrom(map[string]interface{}{"setup": map[string]interface{}{"ilm": map[string]interface{}{"enabled": false}}})
+}
+
 // setupRoleMapping sets up role mapping for the Kerberos user beats@ELASTIC
 func setupRoleMapping(t *testing.T, host string) error {
 	_, client := connectTestEsWithoutStats(t, map[string]interface{}{

libbeat/outputs/elasticsearch/client_test.go

@@ -409,22 +409,49 @@ func TestClientWithHeaders(t *testing.T) {
 
 func TestBulkEncodeEvents(t *testing.T) {
 	cases := map[string]struct {
-		version string
-		docType string
-		config  common.MapStr
-		events  []common.MapStr
+		version   string
+		docType   string
+		config    common.MapStr
+		ilmConfig *common.Config
+		isAlias   bool
+		events    []common.MapStr
 	}{
 		"6.x": {
-			version: "6.8.0",
-			docType: "doc",
-			config:  common.MapStr{},
-			events:  []common.MapStr{{"message": "test"}},
+			version:   "6.8.0",
+			docType:   "doc",
+			config:    common.MapStr{},
+			ilmConfig: common.NewConfig(),
+			events:    []common.MapStr{{"message": "test"}},
 		},
-		"latest": {
-			version: version.GetDefaultVersion(),
-			docType: "",
-			config:  common.MapStr{},
-			events:  []common.MapStr{{"message": "test"}},
+		"require_alias not supported": {
+			version:   "7.9.0",
+			docType:   "",
+			config:    common.MapStr{},
+			ilmConfig: common.NewConfig(),
+			events:    []common.MapStr{{"message": "test"}},
+		},
+		"require_alias is supported": {
+			version:   "7.10.0",
+			docType:   "",
+			config:    common.MapStr{},
+			ilmConfig: common.NewConfig(),
+			isAlias:   true,
+			events:    []common.MapStr{{"message": "test"}},
+		},
+		"latest with ILM": {
+			version:   version.GetDefaultVersion(),
+			docType:   "",
+			config:    common.MapStr{},
+			ilmConfig: common.NewConfig(),
+			isAlias:   true,
+			events:    []common.MapStr{{"message": "test"}},
+		},
+		"latest without ILM": {
+			version:   version.GetDefaultVersion(),
+			docType:   "",
+			config:    common.MapStr{},
+			ilmConfig: disabledILMConfig(),
+			events:    []common.MapStr{{"message": "test"}},
 		},
 	}
 
@@ -437,7 +464,7 @@ func TestBulkEncodeEvents(t *testing.T) {
 				Version:     test.version,
 			}
 
-			im, err := idxmgmt.DefaultSupport(nil, info, common.NewConfig())
+			im, err := idxmgmt.DefaultSupport(nil, info, test.ilmConfig)
 			require.NoError(t, err)
 
 			index, pipeline, err := buildSelectors(im, info, cfg)
@@ -479,6 +506,7 @@ func TestBulkEncodeEvents(t *testing.T) {
 				}
 
 				assert.NotEqual(t, "", meta.Index)
+				assert.Equal(t, test.isAlias, meta.RequireAlias)
 				assert.Equal(t, test.docType, meta.DocType)
 			}
 
@@ -487,6 +515,10 @@ func TestBulkEncodeEvents(t *testing.T) {
 	}
 }
 
+func disabledILMConfig() *common.Config {
+	return common.MustNewConfigFrom(map[string]interface{}{"setup": map[string]interface{}{"ilm": map[string]interface{}{"enabled": false}}})
+}
+
 func TestBulkEncodeEventsWithOpType(t *testing.T) {
 	cases := []common.MapStr{
 		{"_id": "111", "op_type": e.OpTypeIndex, "message": "test 1", "bulkIndex": 0},

libbeat/outputs/elasticsearch/death_letter_selector.go

@@ -34,3 +34,5 @@ func (d DeadLetterSelector) Select(event *beat.Event) (string, error) {
 	}
 	return d.Selector.Select(event)
 }
+
+func (d DeadLetterSelector) IsAlias() bool { return false }

libbeat/outputs/outil/select.go

@@ -87,6 +87,8 @@ func (s Selector) Select(evt *beat.Event) (string, error) {
 	return s.sel.sel(evt)
 }
 
+func (s Selector) IsAlias() bool { return false }
+
 // IsEmpty checks if the selector is not configured and will always return an empty string.
 func (s Selector) IsEmpty() bool {
 	return s.sel == nilSelector || s.sel == nil

libbeat/outputs/output_reg.go

@@ -43,8 +43,10 @@ type IndexManager interface {
 }
 
 // IndexSelector is used to find the index name an event shall be indexed to.
+// It also used to check if during indexing required_alias should be set.
 type IndexSelector interface {
 	Select(event *beat.Event) (string, error)
+	IsAlias() bool
 }
 
 // Group configures and combines multiple clients into load-balanced group of clients