Add Cloud Foundry Processor

CHANGELOG.next.asciidoc

@@ -125,6 +125,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add support for multiple password in redis output. {issue}16058[16058] {pull}16206[16206]
 - Windows .exe files now have embedded file version info. {issue}15232[15232]t
 - Remove experimental flag from `setup.template.append_fields` {pull}16576[16576]
+- Add `add_cloudfoundry_metadata` processor to annotate events with Cloud Foundry application data. {pull}16621[16621
 
 *Auditbeat*
 
@@ -148,6 +149,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Improve ECS categorization field mappings in googlecloud module. {issue}16030[16030] {pull}16500[16500]
 - Improve ECS field mappings in haproxy module. {issue}16162[16162] {pull}16529[16529]
 - Improve the decode_cef processor by reducing the number of memory allocations. {pull}16587[16587]
+- Add `cloudfoundry` input to send events from Cloud Foundry. {pull}16586[16586]
 
 *Heartbeat*
 

libbeat/docs/processors-list.asciidoc

@@ -20,6 +20,9 @@ endif::[]
 ifndef::no_add_kubernetes_metadata_processor[]
 * <<add-kubernetes-metadata,`add_kubernetes_metadata`>>
 endif::[]
+ifndef::no_add_cloudfoundry_metadata_processor[]
+* <<add-cloudfoundry-metadata,`add_cloudfoundry_metadata`>>
+endif::[]
 ifndef::no_add_labels_processor[]
 * <<add-labels, `add_labels`>>
 endif::[]
@@ -119,6 +122,9 @@ endif::[]
 ifndef::no_add_kubernetes_metadata_processor[]
 include::{libbeat-processors-dir}/add_kubernetes_metadata/docs/add_kubernetes_metadata.asciidoc[]
 endif::[]
+ifndef::no_add_cloudfoundry_metadata_processor[]
+include::{x-libbeat-processors-dir}/add_cloudfoundry_metadata/docs/add_cloudfoundry_metadata.asciidoc[]
+endif::[]
 ifndef::no_add_labels_processor[]
 include::{libbeat-processors-dir}/actions/docs/add_labels.asciidoc[]
 endif::[]

libbeat/docs/shared-beats-attributes.asciidoc

@@ -4,6 +4,7 @@
 :dockerconfig: https://raw.githubusercontent.com/elastic/beats/{branch}/deploy/docker/{beatname_lc}.docker.yml
 :downloads: https://artifacts.elastic.co/downloads/beats
 :libbeat-processors-dir: {beats-root}/libbeat/processors
+:x-libbeat-processors-dir: {beats-root}/x-pack/libbeat/processors
 :libbeat-outputs-dir: {beats-root}/libbeat/outputs
 :x-filebeat-processors-dir: {beats-root}/x-pack/filebeat/processors
 :winlogbeat-processors-dir: {beats-root}/winlogbeat/processors

x-pack/libbeat/cmd/inject.go

@@ -11,6 +11,9 @@ import (
 	"github.com/elastic/beats/x-pack/libbeat/licenser"
 	_ "github.com/elastic/beats/x-pack/libbeat/management"
 
+	// register processors
+	_ "github.com/elastic/beats/x-pack/libbeat/processors/add_cloudfoundry_metadata"
+
 	// register autodiscover providers
 	_ "github.com/elastic/beats/x-pack/libbeat/autodiscover/providers/aws/ec2"
 	_ "github.com/elastic/beats/x-pack/libbeat/autodiscover/providers/aws/elb"

x-pack/libbeat/common/cloudfoundry/events.go

@@ -199,7 +199,7 @@ func (e *EventLog) SourceID() string                 { return e.sourceID }
 func (e *EventLog) ToFields() common.MapStr {
 	fields := baseMapWithApp(e)
 	fields.DeepUpdate(common.MapStr{
-		"cf": common.MapStr{
+		"cloudfoundry": common.MapStr{
 			e.String(): common.MapStr{
 				"source": common.MapStr{
 					"instance": e.SourceID(),
@@ -208,6 +208,7 @@ func (e *EventLog) ToFields() common.MapStr {
 			},
 		},
 		"message": e.Message(),
+		"stream":  e.MessageType().String(),
 	})
 	return fields
 }
@@ -236,7 +237,7 @@ func (e *EventCounter) Total() uint64           { return e.total }
 func (e *EventCounter) ToFields() common.MapStr {
 	fields := baseMap(e)
 	fields.DeepUpdate(common.MapStr{
-		"cf": common.MapStr{
+		"cloudfoundry": common.MapStr{
 			e.String(): common.MapStr{
 				"name":  e.Name(),
 				"delta": e.Delta(),
@@ -271,7 +272,7 @@ func (e *EventValueMetric) Unit() string            { return e.unit }
 func (e *EventValueMetric) ToFields() common.MapStr {
 	fields := baseMap(e)
 	fields.DeepUpdate(common.MapStr{
-		"cf": common.MapStr{
+		"cloudfoundry": common.MapStr{
 			e.String(): common.MapStr{
 				"name":  e.Name(),
 				"unit":  e.Unit(),
@@ -313,7 +314,7 @@ func (e *EventContainerMetric) DiskBytesQuota() uint64   { return e.diskBytesQuo
 func (e *EventContainerMetric) ToFields() common.MapStr {
 	fields := baseMapWithApp(e)
 	fields.DeepUpdate(common.MapStr{
-		"cf": common.MapStr{
+		"cloudfoundry": common.MapStr{
 			e.String(): common.MapStr{
 				"instance_index":     e.InstanceIndex(),
 				"cpu.pct":            e.CPUPercentage(),
@@ -351,7 +352,7 @@ func (e *EventError) Source() string          { return e.source }
 func (e *EventError) ToFields() common.MapStr {
 	fields := baseMap(e)
 	fields.DeepUpdate(common.MapStr{
-		"cf": common.MapStr{
+		"cloudfoundry": common.MapStr{
 			e.String(): common.MapStr{
 				"source": e.Source(),
 			},
@@ -472,6 +473,8 @@ func envelopeToEvent(env *events.Envelope) Event {
 		return newEventValueMetric(env)
 	case events.Envelope_ContainerMetric:
 		return newEventContainerMetric(env)
+	case events.Envelope_Error:
+		return newEventError(env)
 	}
 	return nil
 }
@@ -488,9 +491,7 @@ func envelopMap(evt Event) common.MapStr {
 
 func baseMap(evt Event) common.MapStr {
 	return common.MapStr{
-		"module":  "cf",
-		"dataset": fmt.Sprintf("cf.%s", evt),
-		"cf": common.MapStr{
+		"cloudfoundry": common.MapStr{
 			evt.String(): common.MapStr{
 				"timestamp": evt.Timestamp(),
 				"type":      evt.String(),
@@ -504,7 +505,7 @@ func baseMapWithApp(evt EventWithAppID) common.MapStr {
 	base := baseMap(evt)
 	appID := evt.AppGuid()
 	if appID != "" {
-		base.Put("cf.app.id", appID)
+		base.Put("cloudfoundry.app.id", appID)
 	}
 	return base
 }

x-pack/libbeat/processors/add_cloudfoundry_metadata/_meta/fields.yml

@@ -0,0 +1,39 @@
+- key: cloudfoundry
+  title: Cloud Foundry
+  description: >
+    Cloud Foundry information collected from Cloud Foundry.
+  short_config: false
+  anchor: cloudfoundry-processor
+  fields:
+    - name: cloudfoundry
+      type: group
+      fields:
+        - name: app.id
+          type: keyword
+          description: >
+            Cloud Foundry application ID
+
+        - name: app.name
+          type: keyword
+          description: >
+            Cloud Foundry application name
+
+        - name: space.id
+          type: keyword
+          description: >
+            Cloud Foundry space name
+
+        - name: space.name
+          type: keyword
+          description: >
+            Cloud Foundry space name
+
+        - name: org.id
+          type: keyword
+          description: >
+            Cloud Foundry organization ID
+
+        - name: org.name
+          type: keyword
+          description: >
+            Cloud Foundry organization name

x-pack/libbeat/processors/add_cloudfoundry_metadata/add_cloudfoundry_metadata.go

@@ -0,0 +1,99 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+// +build linux darwin windows
+
+package add_cloudfoundry_metadata
+
+import (
+	"time"
+
+	"github.com/elastic/beats/x-pack/libbeat/common/cloudfoundry"
+
+	"github.com/pkg/errors"
+
+	"github.com/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/libbeat/processors"
+)
+
+const (
+	processorName = "add_cloudfoundry_metadata"
+)
+
+func init() {
+	processors.RegisterPlugin(processorName, New)
+}
+
+type addCloudFoundryMetadata struct {
+	log    *logp.Logger
+	client cloudfoundry.Client
+}
+
+const selector = "add_cloudfoundry_metadata"
+
+// New constructs a new add_cloudfoundry_metadata processor.
+func New(cfg *common.Config) (processors.Processor, error) {
+	var config cloudfoundry.Config
+	if err := cfg.Unpack(&config); err != nil {
+		return nil, errors.Wrapf(err, "fail to unpack the %v configuration", processorName)
+	}
+
+	log := logp.NewLogger(selector)
+	hub := cloudfoundry.NewHub(&config, "add_cloudfoundry_metadata", log)
+	client, err := hub.Client()
+	if err != nil {
+		log.Debugf("%s: failed to created cloudfoundry client: %+v", processorName, err)
+	}
+
+	// Janitor run every 5 minutes to clean up the client cache.
+	client.StartJanitor(5 * time.Minute)
+
+	return &addCloudFoundryMetadata{
+		log:    log,
+		client: client,
+	}, nil
+}
+
+func (d *addCloudFoundryMetadata) Run(event *beat.Event) (*beat.Event, error) {
+	if d.client == nil {
+		return event, nil
+	}
+	valI, err := event.GetValue("cloudfoundry.app.id")
+	if err != nil {
+		// doesn't have the required cf.app.id value to add more information
+		return event, nil
+	}
+	val, _ := valI.(string)
+	if val == "" {
+		// wrong type or not set
+		return event, nil
+	}
+	app, err := d.client.GetAppByGuid(val)
+	if err != nil {
+		d.log.Warnf("failed to get application info for GUID(%s): %v", val, err)
+		return event, nil
+	}
+	event.Fields.DeepUpdate(common.MapStr{
+		"cloudfoundry": common.MapStr{
+			"app": common.MapStr{
+				"name": app.Name,
+			},
+			"space": common.MapStr{
+				"id":   app.SpaceData.Meta.Guid,
+				"name": app.SpaceData.Entity.Name,
+			},
+			"org": common.MapStr{
+				"id":   app.SpaceData.Entity.OrgData.Meta.Guid,
+				"name": app.SpaceData.Entity.OrgData.Entity.Name,
+			},
+		},
+	})
+	return event, nil
+}
+
+func (d *addCloudFoundryMetadata) String() string {
+	return processorName
+}