Cherry-pick to 7.0: Change add_cloud_metadata to not overwrite cloud field (#11612)

CHANGELOG.next.asciidoc

@@ -40,6 +40,8 @@ https://github.com/elastic/beats/compare/v7.0.0-...7.0[Check the HEAD diff]
 
 *Metricbeat*
 
+- Change `add_cloud_metadata` processor to not overwrite `cloud` field when it already exist in the event. {pull}11612[11612] {issue}11305[11305]
+
 *Packetbeat*
 
 *Winlogbeat*

libbeat/docs/processors-using.asciidoc

@@ -428,15 +428,19 @@ processors:
 - add_cloud_metadata: ~
 -------------------------------------------------------------------------------
 
-The `add_cloud_metadata` processor has one optional configuration setting named
-`timeout` that specifies the maximum amount of time to wait for a successful
-response when detecting the hosting provider. The default timeout value is
-`3s`.
+The `add_cloud_metadata` processor has two optional configuration settings.
+The first one is `timeout` which specifies the maximum amount of time to wait
+for a successful response when detecting the hosting provider. The default
+timeout value is `3s`.
 
 If a timeout occurs then no instance metadata will be added to the events. This
 makes it possible to enable this processor for all your deployments (in the
 cloud or on-premise).
 
+The second optional configuration setting is `overwrite`. When `overwrite` is
+`true`, `add_cloud_metadata` overwrites existing `cloud.*` fields (`false` by
+default).
+
 The metadata that is added to events varies by hosting provider. Below are
 examples for each of the supported providers.
 

libbeat/processors/add_cloud_metadata/add_cloud_metadata.go

@@ -43,6 +43,9 @@ const (
 
  // Default config
  defaultTimeOut = 3 * time.Second
+
+ // Default overwrite
+ defaultOverwrite = false
 )
 
 var debugf = logp.MakeDebug("filters")
@@ -301,9 +304,11 @@ func setupFetchers(c *common.Config) ([]*metadataFetcher, error) {
 
 func newCloudMetadata(c *common.Config) (processors.Processor, error) {
  config := struct {
- Timeout time.Duration `config:"timeout"` // Amount of time to wait for responses from the metadata services.
+ Timeout time.Duration `config:"timeout"` // Amount of time to wait for responses from the metadata services.
+ Overwrite bool `config:"overwrite"` // Overwrite if cloud.* fields already exist.
  }{
- Timeout: defaultTimeOut,
+ Timeout: defaultTimeOut,
+ Overwrite: defaultOverwrite,
  }
  err := c.Unpack(&config)
  if err != nil {
@@ -316,16 +321,17 @@ func newCloudMetadata(c *common.Config) (processors.Processor, error) {
  }
 
  p := &addCloudMetadata{
- initData: &initData{fetchers, config.Timeout},
+ initData: &initData{fetchers, config.Timeout, config.Overwrite},
  }
 
  go p.initOnce.Do(p.init)
  return p, nil
 }
 
 type initData struct {
- fetchers []*metadataFetcher
- timeout time.Duration
+ fetchers []*metadataFetcher
+ timeout time.Duration
+ overwrite bool
 }
 
 type addCloudMetadata struct {
@@ -341,7 +347,6 @@ func (p *addCloudMetadata) init() {
  return
  }
  p.metadata = result.metadata
- p.initData = nil
  logp.Info("add_cloud_metadata: hosting provider type detected as %v, metadata=%v",
  result.provider, result.metadata.String())
 }
@@ -357,8 +362,16 @@ func (p *addCloudMetadata) Run(event *beat.Event) (*beat.Event, error) {
  return event, nil
  }
 
- // This overwrites the meta.cloud if it exists. But the cloud key should be
- // reserved for this processor so this should happen.
+ // If cloud key exists in event already and overwrite flag is set to false, this processor will not overwrite the
+ // cloud fields. For example aws module writes cloud.instance.* to events already, with overwrite=false,
+ // add_cloud_metadata should not overwrite these fields with new values.
+ if !p.initData.overwrite {
+ cloudValue, _ := event.GetValue("cloud")
+ if cloudValue != nil {
+ return event, nil
+ }
+ }
+
  _, err := event.PutValue("cloud", meta)
 
  return event, err

libbeat/processors/add_cloud_metadata/provider_aws_ec2_test.go

@@ -64,7 +64,8 @@ func TestRetrieveAWSMetadata(t *testing.T) {
  defer server.Close()
 
  config, err := common.NewConfigFrom(map[string]interface{}{
- "host": server.Listener.Addr().String(),
+ "host": server.Listener.Addr().String(),
+ "overwrite": false,
  })
  if err != nil {
  t.Fatal(err)
@@ -75,23 +76,181 @@ func TestRetrieveAWSMetadata(t *testing.T) {
  t.Fatal(err)
  }
 
- actual, err := p.Run(&beat.Event{Fields: common.MapStr{}})
+ cases := []struct {
+ fields common.MapStr
+ expectedResults common.MapStr
+ }{
+ {
+ common.MapStr{},
+ common.MapStr{
+ "cloud": common.MapStr{
+ "provider": "ec2",
+ "instance": common.MapStr{
+ "id": "i-11111111",
+ },
+ "machine": common.MapStr{
+ "type": "t2.medium",
+ },
+ "region": "us-east-1",
+ "availability_zone": "us-east-1c",
+ },
+ },
+ },
+ {
+ common.MapStr{
+ "cloud": common.MapStr{
+ "instance": common.MapStr{
+ "id": "i-000",
+ },
+ },
+ },
+ common.MapStr{
+ "cloud": common.MapStr{
+ "instance": common.MapStr{
+ "id": "i-000",
+ },
+ },
+ },
+ },
+ {
+ common.MapStr{
+ "provider": "ec2",
+ },
+ common.MapStr{
+ "provider": "ec2",
+ "cloud": common.MapStr{
+ "provider": "ec2",
+ "instance": common.MapStr{
+ "id": "i-11111111",
+ },
+ "machine": common.MapStr{
+ "type": "t2.medium",
+ },
+ "region": "us-east-1",
+ "availability_zone": "us-east-1c",
+ },
+ },
+ },
+ {
+ common.MapStr{
+ "cloud.provider": "ec2",
+ },
+ // NOTE: In this case, add_cloud_metadata will overwrite cloud fields because
+ // it won't detect cloud.provider as a cloud field. This is not the behavior we
+ // expect and will find a better solution later in issue 11697.
+ common.MapStr{
+ "cloud.provider": "ec2",
+ "cloud": common.MapStr{
+ "provider": "ec2",
+ "instance": common.MapStr{
+ "id": "i-11111111",
+ },
+ "machine": common.MapStr{
+ "type": "t2.medium",
+ },
+ "region": "us-east-1",
+ "availability_zone": "us-east-1c",
+ },
+ },
+ },
+ }
+
+ for _, c := range cases {
+ actual, err := p.Run(&beat.Event{Fields: c.fields})
+ if err != nil {
+ t.Fatal(err)
+ }
+ assert.Equal(t, c.expectedResults, actual.Fields)
+ }
+}
+
+func TestRetrieveAWSMetadataOverwriteTrue(t *testing.T) {
+ logp.TestingSetup()
+
+ server := initEC2TestServer()
+ defer server.Close()
+
+ config, err := common.NewConfigFrom(map[string]interface{}{
+ "host": server.Listener.Addr().String(),
+ "overwrite": true,
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ p, err := newCloudMetadata(config)
  if err != nil {
  t.Fatal(err)
  }
 
- expected := common.MapStr{
- "cloud": common.MapStr{
- "provider": "ec2",
- "instance": common.MapStr{
- "id": "i-11111111",
+ cases := []struct {
+ fields common.MapStr
+ expectedResults common.MapStr
+ }{
+ {
+ common.MapStr{},
+ common.MapStr{
+ "cloud": common.MapStr{
+ "provider": "ec2",
+ "instance": common.MapStr{
+ "id": "i-11111111",
+ },
+ "machine": common.MapStr{
+ "type": "t2.medium",
+ },
+ "region": "us-east-1",
+ "availability_zone": "us-east-1c",
+ },
+ },
+ },
+ {
+ common.MapStr{
+ "cloud": common.MapStr{
+ "instance": common.MapStr{
+ "id": "i-000",
+ },
+ },
  },
- "machine": common.MapStr{
- "type": "t2.medium",
+ common.MapStr{
+ "cloud": common.MapStr{
+ "provider": "ec2",
+ "instance": common.MapStr{
+ "id": "i-11111111",
+ },
+ "machine": common.MapStr{
+ "type": "t2.medium",
+ },
+ "region": "us-east-1",
+ "availability_zone": "us-east-1c",
+ },
  },
- "region": "us-east-1",
- "availability_zone": "us-east-1c",
  },
+ {
+ common.MapStr{
+ "cloud.provider": "ec2",
+ },
+ common.MapStr{
+ "cloud.provider": "ec2",
+ "cloud": common.MapStr{
+ "provider": "ec2",
+ "instance": common.MapStr{
+ "id": "i-11111111",
+ },
+ "machine": common.MapStr{
+ "type": "t2.medium",
+ },
+ "region": "us-east-1",
+ "availability_zone": "us-east-1c",
+ },
+ },
+ },
+ }
+
+ for _, c := range cases {
+ actual, err := p.Run(&beat.Event{Fields: c.fields})
+ if err != nil {
+ t.Fatal(err)
+ }
+ assert.Equal(t, c.expectedResults, actual.Fields)
  }
- assert.Equal(t, expected, actual.Fields)
 }