Add replace_fields config option in add_host_metadata for replacing host fields

CHANGELOG.next.asciidoc

@@ -385,7 +385,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Set index.max_docvalue_fields_search in index template to increase value to 200 fields. {issue}20215[20215]
 - Add leader election for Kubernetes autodiscover. {pull}20281[20281]
 - Add capability of enriching process metadata with contianer id also for non-privileged containers in `add_process_metadata` processor. {pull}19767[19767]
-
+- Add replace_fields config option in add_host_metadata for replacing host fields. {pull}20490[20490] {issue}20464[20464]
 
 *Auditbeat*
 

libbeat/processors/add_host_metadata/add_host_metadata.go

@@ -81,6 +81,11 @@ func New(cfg *common.Config) (processors.Processor, error) {
 
 // Run enriches the given event with the host meta data
 func (p *addHostMetadata) Run(event *beat.Event) (*beat.Event, error) {
+	// check replace_host_fields field
+	if !p.config.ReplaceFields && skipAddingHostMetadata(event) {
+		return event, nil
+	}
+
 	err := p.loadData()
 	if err != nil {
 		return nil, err
@@ -146,3 +151,22 @@ func (p *addHostMetadata) String() string {
 	return fmt.Sprintf("%v=[netinfo.enabled=[%v], cache.ttl=[%v]]",
 		processorName, p.config.NetInfoEnabled, p.config.CacheTTL)
 }
+
+func skipAddingHostMetadata(event *beat.Event) bool {
+	// If host fields exist(besides host.name added by libbeat) in event, skip add_host_metadata.
+	hostFields, err := event.Fields.GetValue("host")
+
+	// Don't skip if there are no fields
+	if err != nil || hostFields == nil {
+		return false
+	}
+
+	hostFieldsMap := hostFields.(common.MapStr)
+	// or if "name" is the only field, don't skip
+	hasName, _ := hostFieldsMap.HasKey("name")
+	if hasName && len(hostFieldsMap) == 1 {
+		return false
+	}
+
+	return true
+}

libbeat/processors/add_host_metadata/add_host_metadata_test.go

@@ -31,6 +31,11 @@ import (
 	"github.com/elastic/go-sysinfo/types"
 )
 
+var (
+	hostName = "testHost"
+	hostID   = "9C7FAB7B"
+)
+
 func TestConfigDefault(t *testing.T) {
 	event := &beat.Event{
 		Fields:    common.MapStr{},
@@ -196,3 +201,213 @@ func TestConfigGeoDisabled(t *testing.T) {
 	assert.Error(t, err)
 	assert.Equal(t, nil, eventGeoField)
 }
+
+func TestEventWithReplaceFieldsFalse(t *testing.T) {
+	cfg := map[string]interface{}{}
+	cfg["replace_fields"] = false
+	testConfig, err := common.NewConfigFrom(cfg)
+	assert.NoError(t, err)
+
+	p, err := New(testConfig)
+	switch runtime.GOOS {
+	case "windows", "darwin", "linux":
+		assert.NoError(t, err)
+	default:
+		assert.IsType(t, types.ErrNotImplemented, err)
+		return
+	}
+
+	cases := []struct {
+		title                   string
+		event                   beat.Event
+		hostLengthLargerThanOne bool
+		hostLengthEqualsToOne   bool
+		expectedHostFieldLength int
+	}{
+		{
+			"replace_fields=false with only host.name",
+			beat.Event{
+				Fields: common.MapStr{
+					"host": common.MapStr{
+						"name": hostName,
+					},
+				},
+			},
+			true,
+			false,
+			-1,
+		},
+		{
+			"replace_fields=false with only host.id",
+			beat.Event{
+				Fields: common.MapStr{
+					"host": common.MapStr{
+						"id": hostID,
+					},
+				},
+			},
+			false,
+			true,
+			1,
+		},
+		{
+			"replace_fields=false with host.name and host.id",
+			beat.Event{
+				Fields: common.MapStr{
+					"host": common.MapStr{
+						"name": hostName,
+						"id":   hostID,
+					},
+				},
+			},
+			true,
+			false,
+			2,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.title, func(t *testing.T) {
+			newEvent, err := p.Run(&c.event)
+			assert.NoError(t, err)
+
+			v, err := newEvent.GetValue("host")
+			assert.NoError(t, err)
+			assert.Equal(t, c.hostLengthLargerThanOne, len(v.(common.MapStr)) > 1)
+			assert.Equal(t, c.hostLengthEqualsToOne, len(v.(common.MapStr)) == 1)
+			if c.expectedHostFieldLength != -1 {
+				assert.Equal(t, c.expectedHostFieldLength, len(v.(common.MapStr)))
+			}
+		})
+	}
+}
+
+func TestEventWithReplaceFieldsTrue(t *testing.T) {
+	cfg := map[string]interface{}{}
+	cfg["replace_fields"] = true
+	testConfig, err := common.NewConfigFrom(cfg)
+	assert.NoError(t, err)
+
+	p, err := New(testConfig)
+	switch runtime.GOOS {
+	case "windows", "darwin", "linux":
+		assert.NoError(t, err)
+	default:
+		assert.IsType(t, types.ErrNotImplemented, err)
+		return
+	}
+
+	cases := []struct {
+		title                   string
+		event                   beat.Event
+		hostLengthLargerThanOne bool
+		hostLengthEqualsToOne   bool
+	}{
+		{
+			"replace_fields=true with host.name",
+			beat.Event{
+				Fields: common.MapStr{
+					"host": common.MapStr{
+						"name": hostName,
+					},
+				},
+			},
+			true,
+			false,
+		},
+		{
+			"replace_fields=true with host.id",
+			beat.Event{
+				Fields: common.MapStr{
+					"host": common.MapStr{
+						"id": hostID,
+					},
+				},
+			},
+			true,
+			false,
+		},
+		{
+			"replace_fields=true with host.name and host.id",
+			beat.Event{
+				Fields: common.MapStr{
+					"host": common.MapStr{
+						"name": hostName,
+						"id":   hostID,
+					},
+				},
+			},
+			true,
+			false,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.title, func(t *testing.T) {
+			newEvent, err := p.Run(&c.event)
+			assert.NoError(t, err)
+
+			v, err := newEvent.GetValue("host")
+			assert.NoError(t, err)
+			assert.Equal(t, c.hostLengthLargerThanOne, len(v.(common.MapStr)) > 1)
+			assert.Equal(t, c.hostLengthEqualsToOne, len(v.(common.MapStr)) == 1)
+		})
+	}
+}
+
+func TestSkipAddingHostMetadata(t *testing.T) {
+	cases := []struct {
+		title        string
+		event        beat.Event
+		expectedSkip bool
+	}{
+		{
+			"event only with host.name",
+			beat.Event{
+				Fields: common.MapStr{
+					"host": common.MapStr{
+						"name": hostName,
+					},
+				},
+			},
+			false,
+		},
+		{
+			"event only with host.id",
+			beat.Event{
+				Fields: common.MapStr{
+					"host": common.MapStr{
+						"id": hostID,
+					},
+				},
+			},
+			true,
+		},
+		{
+			"event with host.name and host.id",
+			beat.Event{
+				Fields: common.MapStr{
+					"host": common.MapStr{
+						"name": hostName,
+						"id":   hostID,
+					},
+				},
+			},
+			true,
+		},
+		{
+			"event without host field",
+			beat.Event{
+				Fields: common.MapStr{},
+			},
+			false,
+		},
+	}
+
+	for _, c := range cases {
+		t.Run(c.title, func(t *testing.T) {
+			skip := skipAddingHostMetadata(&c.event)
+			assert.Equal(t, c.expectedSkip, skip)
+		})
+	}
+}

libbeat/processors/add_host_metadata/config.go

@@ -29,11 +29,13 @@ type Config struct {
 	CacheTTL       time.Duration   `config:"cache.ttl"`
 	Geo            *util.GeoConfig `config:"geo"`
 	Name           string          `config:"name"`
+	ReplaceFields  bool            `config:"replace_fields"` // replace existing host fields with add_host_metadata
 }
 
 func defaultConfig() Config {
 	return Config{
 		NetInfoEnabled: true,
 		CacheTTL:       5 * time.Minute,
+		ReplaceFields:  true,
 	}
 }

libbeat/processors/add_host_metadata/docs/add_host_metadata.asciidoc

@@ -42,6 +42,8 @@ It has the following settings:
 
 `geo.region_iso_code`:: (Optional) ISO region code.
 
+`replace_fields`:: (Optional) Default true. If set to false, original host
+fields from the event will not be replaced by host fields from `add_host_metadata`.
 
 The `add_host_metadata` processor annotates each event with relevant metadata from the host machine.
 The fields added to the event look like the following:
@@ -75,3 +77,9 @@ The fields added to the event look like the following:
    }
 }
 -------------------------------------------------------------------------------
+
+Note: `add_host_metadata` processor will overwrite host fields if `host.*`
+fields already exist in the event from Beats by default with `replace_fields`
+equals to `true`.
+Please use `add_observer_metadata` if the beat is being used to monitor external
+systems.