Refactor and extract consts.

libbeat/beat/event.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/logp"
 )
 
 // FlagField fields used to keep information or errors when events are parsed.
@@ -54,6 +55,20 @@ func (e *Event) SetID(id string) {
 	e.Meta["_id"] = id
 }
 
+func (e *Event) GetMetaStringValue(key string) (string, error) {
+	var val string
+	if tmp, err := e.GetValue("@metadata." + key); err == nil {
+		if s, ok := tmp.(string); ok {
+			return s, nil
+		} else {
+			logp.Err("Event[%s] '%v' is no string value", key, tmp)
+			return val, nil
+		}
+	} else {
+		return val, err
+	}
+}
+
 func (e *Event) GetValue(key string) (interface{}, error) {
 	if key == "@timestamp" {
 		return e.Timestamp, nil

libbeat/outputs/elasticsearch/client.go

@@ -65,8 +65,16 @@ type bulkResultStats struct {
 
 const (
 	defaultEventType = "doc"
+	opTypeCreate     = "create"
+	opTypeDelete     = "delete"
+	opTypeIndex      = "index"
 )
 
+// opTypeKey defines the metadata key name for event operation type.
+// The key's value can be an empty string, `create`, `index`, or `delete`. If empty, the event will be `create`d if the ID
+// is known, and `index`ed otherwise.
+const opTypeKey = "@metadata.op_type"
+
 // NewClient instantiates a new client.
 func NewClient(
 	s ClientSettings,
@@ -251,20 +259,6 @@ func (client *Client) publishEvents(
 	return nil, nil
 }
 
-func eventMetaValue(event *beat.Event, key string) string {
-	var val string
-	if m := event.Meta; m != nil {
-		if tmp := m[key]; tmp != nil {
-			if s, ok := tmp.(string); ok {
-				val = s
-			} else {
-				logp.Err("Event[%s] '%v' is no string value", key, val)
-			}
-		}
-	}
-	return val
-}
-
 // bulkEncodePublishRequest encodes all bulk requests and returns slice of events
 // successfully added to the list of bulk items and the list of bulk items.
 func bulkEncodePublishRequest(
@@ -284,8 +278,7 @@ func bulkEncodePublishRequest(
 			log.Errorf("Failed to encode event meta data: %+v", err)
 			continue
 		}
-		opType := eventMetaValue(event, "op_type")
-		if opType == "delete" {
+		if opType, err := event.GetMetaStringValue(opTypeKey); err == nil && opType == opTypeDelete {
 			// We don't include the event source in a bulk DELETE
 			bulkItems = append(bulkItems, meta, nil)
 		} else {
@@ -320,8 +313,8 @@ func createEventBulkMeta(
 		return nil, err
 	}
 
-	id := eventMetaValue(event, "id")
-	opType := eventMetaValue(event, "op_type")
+	id, _ := event.GetMetaStringValue("_id")
+	opType, _ := event.GetMetaStringValue(opTypeKey)
 
 	meta := eslegclient.BulkMeta{
 		Index:    index,
@@ -330,12 +323,14 @@ func createEventBulkMeta(
 		ID:       id,
 	}
 
+	if id != "" && opType == opTypeDelete {
+		return eslegclient.BulkDeleteAction{Delete: meta}, nil
+	}
 	if id != "" || version.Major > 7 || (version.Major == 7 && version.Minor >= 5) {
-		if opType == "" || opType == "create" {
-			return eslegclient.BulkCreateAction{Create: meta}, nil
-		} else if opType == "delete" {
-			return eslegclient.BulkDeleteAction{Delete: meta}, nil
+		if opType == opTypeIndex {
+			return eslegclient.BulkIndexAction{Index: meta}, nil
 		}
+		return eslegclient.BulkCreateAction{Create: meta}, nil
 	}
 	return eslegclient.BulkIndexAction{Index: meta}, nil
 }