Refactor add_cloud_metadata to handle ECS fields easier

libbeat/processors/add_cloud_metadata/add_cloud_metadata.go

@@ -118,49 +118,32 @@ func (p *addCloudMetadata) Run(event *beat.Event) (*beat.Event, error) {
  return event, nil
  }
 
- // 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 {
- err := p.extractECSMeta(event, meta)
- if err != nil {
- return nil, err
- }
- return event, nil
- }
- }
-
- err := p.extractECSMeta(event, meta)
+ err := p.addMeta(event, meta)
  if err != nil {
  return nil, err
  }
- _, err = event.PutValue("cloud", meta)
  return event, err
 }
 
 func (p *addCloudMetadata) String() string {
  return "add_cloud_metadata=" + p.getMeta().String()
 }
 
-func (p *addCloudMetadata) extractECSMeta(event *beat.Event, meta common.MapStr) error {
- // handle ECS fields first
- if !p.initData.overwrite {
- orchestratorValue, _ := event.GetValue("orchestrator")
- if orchestratorValue != nil {
- meta.Delete("orchestrator")
- return nil
+func (p *addCloudMetadata) addMeta(event *beat.Event, meta common.MapStr) error {
+ for key, metaVal := range meta {
+ // If 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(key)
+ if cloudValue != nil {
+ continue
+ }
  }
- }
- orchestratorFields, err := meta.GetValue("orchestrator")
- if err == nil {
- _, err = event.PutValue("orchestrator", orchestratorFields)
+ _, err := event.PutValue(key, metaVal)
  if err != nil {
  return err
  }
  }
- meta.Delete("orchestrator")
-
  return nil
 }

libbeat/processors/add_cloud_metadata/http_fetcher.go

@@ -75,7 +75,7 @@ func (f *httpMetadataFetcher) fetchMetadata(ctx context.Context, client http.Cli
 
  // Apply schema.
  res.metadata = f.conv(res.metadata)
- res.metadata["provider"] = f.provider
+ res.metadata.Put("cloud.provider", f.provider)
 
  return res
 }

libbeat/processors/add_cloud_metadata/provider_alibaba_cloud.go

@@ -36,7 +36,7 @@ var alibabaCloudMetadataFetcher = provider{
  m["service"] = common.MapStr{
  "name": "ECS",
  }
- return common.MapStr(m)
+ return common.MapStr{"cloud": m}
  }
 
  urls, err := getMetadataURLs(c, ecsMetadataHost, []string{

libbeat/processors/add_cloud_metadata/provider_aws_ec2.go

@@ -45,7 +45,7 @@ var ec2MetadataFetcher = provider{
  "account": s.Object{"id": c.Str("accountId")},
  "image": s.Object{"id": c.Str("imageId")},
  }.Apply(m)
- return out
+ return common.MapStr{"cloud": out}
  }
 
  fetcher, err := newMetadataFetcher(config, "aws", nil, metadataHost, ec2Schema, ec2InstanceIdentityURI)

libbeat/processors/add_cloud_metadata/provider_azure_vm.go

@@ -50,7 +50,7 @@ var azureVMMetadataFetcher = provider{
  },
  "region": c.Str("location"),
  }.Apply(m)
- return out
+ return common.MapStr{"cloud": out}
  }
 
  fetcher, err := newMetadataFetcher(config, "azure", azHeaders, metadataHost, azSchema, azMetadataURI)

libbeat/processors/add_cloud_metadata/provider_digital_ocean.go

@@ -41,7 +41,7 @@ var doMetadataFetcher = provider{
  "name": c.Str("serviceName"),
  },
  }.Apply(m)
- return out
+ return common.MapStr{"cloud": out}
  }
  doMetadataURI := "/metadata/v1.json"
 

libbeat/processors/add_cloud_metadata/provider_google_gce.go

@@ -67,6 +67,14 @@ var gceMetadataFetcher = provider{
  out.Put(key, path.Base(p))
  }
 
+ extractECSField := func(key string, out common.MapStr, meta common.MapStr) {
+ ecs, _ := out.GetValue(key)
+ out.Delete(key)
+ if ecs != nil {
+ meta.Put(key, ecs.(common.MapStr))
+ }
+ }
+
  if instance, ok := m["instance"].(map[string]interface{}); ok {
  s.Schema{
  "instance": s.Object{
@@ -121,7 +129,12 @@ var gceMetadataFetcher = provider{
  }.ApplyTo(out, project)
  }
 
- return out
+ meta := common.MapStr{}
+ // call extractECSField for all ECS root fields like orchestrator.*
+ extractECSField("orchestrator", out, meta)
+
+ meta.DeepUpdate(common.MapStr{"cloud": out})
+ return meta
  }
 
  fetcher, err := newMetadataFetcher(config, provider, gceHeaders, metadataHost, gceSchema, gceMetadataURI)

libbeat/processors/add_cloud_metadata/provider_openstack_nova.go

@@ -49,7 +49,7 @@ func buildOpenstackNovaCreate(scheme string) func(provider string, c *common.Con
  m["service"] = common.MapStr{
  "name": "Nova",
  }
- return common.MapStr(m)
+ return common.MapStr{"cloud": m}
  }
 
  urls, err := getMetadataURLsWithScheme(c, scheme, metadataHost, []string{

libbeat/processors/add_cloud_metadata/provider_tencent_cloud.go

@@ -36,7 +36,7 @@ var qcloudMetadataFetcher = provider{
  m["service"] = common.MapStr{
  "name": "CVM",
  }
- return common.MapStr(m)
+ return common.MapStr{"cloud": m}
  }
 
  urls, err := getMetadataURLs(c, qcloudMetadataHost, []string{