Add group_measurements_by_instance_label flag to perfmon configuration

metricbeat/metricbeat.reference.yml

@@ -661,6 +661,7 @@ metricbeat.modules:
   enabled: true
   period: 10s
   perfmon.ignore_non_existent_counters: true
+  perfmon.group_measurements_by_instance: true
   perfmon.counters:
   #  - instance_label: processor.name
   #    instance_name: total
@@ -798,7 +799,7 @@ metricbeat.modules:
 #
 #   event -> filter1 -> event1 -> filter2 ->event2 ...
 #
-# The supported processors are drop_fields, drop_event, include_fields, 
+# The supported processors are drop_fields, drop_event, include_fields,
 # decode_json_fields, and add_cloud_metadata.
 #
 # For example, you can use the following processors to keep the fields that
@@ -888,7 +889,7 @@ metricbeat.modules:
 #    match_pids: ["system.process.ppid"]
 #    target: system.process.parent
 #
-# The following example decodes fields containing JSON strings 
+# The following example decodes fields containing JSON strings
 # and replaces the strings with valid JSON objects.
 #
 #processors:

metricbeat/module/windows/_meta/config.reference.yml

@@ -3,6 +3,7 @@
   enabled: true
   period: 10s
   perfmon.ignore_non_existent_counters: true
+  perfmon.group_measurements_by_instance: true
   perfmon.counters:
   #  - instance_label: processor.name
   #    instance_name: total

metricbeat/module/windows/perfmon/_meta/docs.asciidoc

@@ -14,6 +14,7 @@ to collect. The example below collects processor time and disk writes every
   metricsets: [perfmon]
   period: 10s
   perfmon.ignore_non_existent_counters: true
+  perfmon.group_measurements_by_instance: true
   perfmon.counters:
     - instance_label: processor.name
       instance_name: total
@@ -34,6 +35,12 @@ metricset to ignore errors caused by counters that do not exist when set to
 true. Instead of an error, a message will be logged at the info level stating
 that the counter does not exist.
 
+*`group_measurements_by_instance`*:: A boolean option that causes metricbeat
+to send all measurements with a matching perfmon instance as part of a single
+event. In the above example, this will cause the physical_disk.write.per_sec
+and physical_disk.write.time.pct measurements to be sent as a single event.
+The default behaviour is for all measurements to be sent as separate events.
+
 *`counters`*:: Counters specifies a list of queries to perform. Each individual
 counter requires three config options - `instance_label`, `measurement_label`,
 and `query`.

metricbeat/module/windows/perfmon/pdh_windows.go

@@ -326,11 +326,12 @@ func (q *Query) Close() error {
 }
 
 type PerfmonReader struct {
-	query         *Query            // PDH Query
-	instanceLabel map[string]string // Mapping of counter path to key used for the label (e.g. processor.name)
-	measurement   map[string]string // Mapping of counter path to key used for the value (e.g. processor.cpu_time).
-	executed      bool              // Indicates if the query has been executed.
-	log           *logp.Logger
+	query             *Query            // PDH Query
+	instanceLabel     map[string]string // Mapping of counter path to key used for the label (e.g. processor.name)
+	measurement       map[string]string // Mapping of counter path to key used for the value (e.g. processor.cpu_time).
+	executed          bool              // Indicates if the query has been executed.
+	log               *logp.Logger      //
+	groupMeasurements bool              // Indicates if measurements with the same instance label should be sent in the same event
 }
 
 // NewPerfmonReader creates a new instance of PerfmonReader.
@@ -341,10 +342,11 @@ func NewPerfmonReader(config Config) (*PerfmonReader, error) {
 	}
 
 	r := &PerfmonReader{
-		query:         query,
-		instanceLabel: map[string]string{},
-		measurement:   map[string]string{},
-		log:           logp.NewLogger("perfmon"),
+		query:             query,
+		instanceLabel:     map[string]string{},
+		measurement:       map[string]string{},
+		log:               logp.NewLogger("perfmon"),
+		groupMeasurements: config.GroupMeasurements,
 	}
 
 	for _, counter := range config.CounterConfig {
@@ -388,36 +390,53 @@ func (r *PerfmonReader) Read() ([]mb.Event, error) {
 		return nil, errors.Wrap(err, "failed formatting counter values")
 	}
 
-	// Write the values into the map.
-	events := make([]mb.Event, 0, len(values))
+	eventMap := make(map[string]*mb.Event)
 
 	for counterPath, values := range values {
-		for _, val := range values {
+		for ind, val := range values {
 			if val.Err != nil && !r.executed {
 				r.log.Debugw("Ignoring the first measurement because the data isn't ready",
 					"error", val.Err, logp.Namespace("perfmon"), "query", counterPath)
 				continue
 			}
 
-			event := mb.Event{
-				MetricSetFields: common.MapStr{},
-				Error:           errors.Wrapf(val.Err, "failed on query=%v", counterPath),
+			var eventKey string
+			if r.groupMeasurements {
+				// Send measurements with the same instance label as part of the same event
+				eventKey = val.Instance
+			} else {
+				// Send every measurement as an individual event
+				eventKey = counterPath + str(ind)
 			}
 
-			if val.Instance != "" {
-				event.MetricSetFields.Put(r.instanceLabel[counterPath], val.Instance)
+			// Create a new event if the key doesn't exist in the map
+			if _, ok := eventMap[eventKey]; !ok {
+				eventMap[eventKey] = &mb.Event{
+					MetricSetFields: common.MapStr{},
+					Error:           errors.Wrapf(val.Err, "failed on query=%v", counterPath),
+				}
+
+				if val.Instance != "" {
+					eventMap[eventKey].MetricSetFields.Put(r.instanceLabel[counterPath], val.Instance)
+				}
 			}
 
+			event := eventMap[eventKey]
+
 			if val.Measurement != nil {
 				event.MetricSetFields.Put(r.measurement[counterPath], val.Measurement)
 			} else {
 				event.MetricSetFields.Put(r.measurement[counterPath], 0)
 			}
-
-			events = append(events, event)
 		}
 	}
 
+	// Write the values into the map.
+	events := make([]mb.Event, 0, len(eventMap))
+	for _, val := range eventMap {
+		events = append(events, *val)
+	}
+
 	r.executed = true
 	return events, nil
 }

metricbeat/module/windows/perfmon/perfmon.go

@@ -40,8 +40,9 @@ type CounterConfig struct {
 
 // Config for the windows perfmon metricset.
 type Config struct {
-	IgnoreNECounters bool            `config:"perfmon.ignore_non_existent_counters"`
-	CounterConfig    []CounterConfig `config:"perfmon.counters" validate:"required"`
+	IgnoreNECounters  bool            `config:"perfmon.ignore_non_existent_counters"`
+	GroupMeasurements bool            `config:"perfmon.group_measurements_by_instance"`
+	CounterConfig     []CounterConfig `config:"perfmon.counters" validate:"required"`
 }
 
 func init() {