[metricbeat] Add a new 'match_by_parent_instance' option to 'perfmon'…

… module (#43002)

* Add a new 'match_by_parent_instance' option to 'perfmon' module

* Add type assertion check

CHANGELOG.next.asciidoc

@@ -473,6 +473,7 @@ otherwise no tag is added. {issue}42208[42208] {pull}42403[42403]
 - Add new metricset wmi for the windows module. {pull}42017[42017]
 - Update beat module with apm-server tail sampling monitoring metrics fields {pull}42569[42569]
 - Log every 401 response from Kubernetes API Server {pull}42714[42714]
+- Add a new `match_by_parent_instance` option to `perfmon` module. {pull}43002[43002]
 
 *Metricbeat*
 - Add benchmark module {pull}41801[41801]

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

@@ -49,6 +49,13 @@ The default behaviour is for all measurements to be sent as separate events.
 
 *`refresh_wildcard_counters`*:: A boolean option to refresh the counter list at each fetch. By default, the counter list will be retrieved at the starting time, to refresh the list at each fetch, users will have to enable this setting.
 
+*`match_by_parent_instance`*:: A boolean option that causes all instances
+of the same parent to have the same instance value. In the above example,
+this will cause metrics for `svchost`, `svchost#1`, etc. to have an `instance`
+value of `svchost`. If set to `false` they will keep the original values
+(`svchost`, `svchost#1`, etc.).
+It defaults to `true`.
+
 
 [float]
 ==== Query Configuration

metricbeat/module/windows/perfmon/config.go

@@ -35,6 +35,7 @@ type Config struct {
 	RefreshWildcardCounters bool          `config:"perfmon.refresh_wildcard_counters"`
 	Queries                 []Query       `config:"perfmon.queries"`
 	GroupAllCountersTo      string        `config:"perfmon.group_all_counter"`
+	MatchByParentInstance   *bool         `config:"perfmon.match_by_parent_instance"`
 }
 
 // QueryConfig for perfmon queries. This will be used as the new configuration format
@@ -77,6 +78,10 @@ func (conf *Config) Validate() error {
 	return nil
 }
 
+func (conf *Config) ShouldMatchByParentInstance() bool {
+	return conf.MatchByParentInstance == nil || *conf.MatchByParentInstance
+}
+
 func isValidFormat(format string) bool {
 	for _, form := range allowedFormats {
 		if form == format {

metricbeat/module/windows/perfmon/data.go

@@ -75,12 +75,14 @@ func (re *Reader) groupToEvents(counters map[string][]pdh.CounterValue) []mb.Eve
 					eventMap[eventKey].Error = fmt.Errorf("failed on query=%v: %w", counterPath, val.Err.Error)
 				}
 				if val.Instance != "" {
-					// will ignore instance index
-					if ok, match := matchesParentProcess(val.Instance); ok {
-						eventMap[eventKey].MetricSetFields.Put(counter.InstanceField, match)
-					} else {
-						eventMap[eventKey].MetricSetFields.Put(counter.InstanceField, val.Instance)
+					instanceName := val.Instance
+					if re.config.ShouldMatchByParentInstance() {
+						if ok, match := matchesParentProcess(val.Instance); ok {
+							// will ignore instance index
+							instanceName = match
+						}
 					}
+					eventMap[eventKey].MetricSetFields.Put(counter.InstanceField, instanceName)
 				}
 			}
 
@@ -126,13 +128,13 @@ func (re *Reader) groupToSingleEvent(counters map[string][]pdh.CounterValue) mb.
 					continue
 				}
 				var counterVal float64
-				switch val.Measurement.(type) {
+				switch m := val.Measurement.(type) {
 				case int64:
-					counterVal = float64(val.Measurement.(int64))
+					counterVal = float64(m)
 				case int:
-					counterVal = float64(val.Measurement.(int))
+					counterVal = float64(m)
 				default:
-					counterVal = val.Measurement.(float64)
+					counterVal, _ = val.Measurement.(float64)
 				}
 				if _, ok := measurements[readerCounter.QueryField]; !ok {
 					measurements[readerCounter.QueryField] = counterVal

metricbeat/module/windows/perfmon/data_test.go

@@ -292,6 +292,82 @@ func TestGroupToSingleEvent(t *testing.T) {
 	assert.Equal(t, val, mapstr.M{"processor_count": float64(2)})
 }
 
+func TestMatchByParentInstance(t *testing.T) {
+	_true := true
+	_false := false
+	reader := Reader{
+		query: pdh.Query{},
+		log:   nil,
+		config: Config{
+			MatchByParentInstance: &_true,
+		},
+		counters: []PerfCounter{
+			{
+				QueryField:    "%_processor_time",
+				QueryName:     `\Processor Information(*)\% Processor Time`,
+				Format:        "float",
+				ObjectName:    "Processor Information",
+				ObjectField:   "object",
+				InstanceName:  "*",
+				InstanceField: "instance",
+				ChildQueries:  []string{`\Processor Information(processor)\% Processor Time`, `\Processor Information(processor#1)\% Processor Time`},
+			},
+		},
+	}
+
+	counters := map[string][]pdh.CounterValue{
+		`\Processor Information(processor)\% Processor Time`: {
+			{
+				Instance:    "processor",
+				Measurement: 23,
+			},
+		},
+		`\Processor Information(processor#1)\% Processor Time`: {
+			{
+				Instance:    "processor#1",
+				Measurement: 21,
+			},
+		},
+	}
+
+	{
+		events := reader.groupToEvents(counters)
+		assert.NotNil(t, events)
+		assert.Equal(t, 2, len(events))
+		ok, err := events[0].MetricSetFields.HasKey("instance")
+		assert.NoError(t, err)
+		assert.True(t, ok)
+		ok, err = events[1].MetricSetFields.HasKey("instance")
+		assert.NoError(t, err)
+		assert.True(t, ok)
+		val1, err := events[0].MetricSetFields.GetValue("instance")
+		assert.NoError(t, err)
+		assert.Equal(t, val1, "processor")
+		val2, err := events[1].MetricSetFields.GetValue("instance")
+		assert.NoError(t, err)
+		assert.Equal(t, val2, "processor")
+	}
+
+	reader.config.MatchByParentInstance = &_false
+	{
+		events := reader.groupToEvents(counters)
+		assert.NotNil(t, events)
+		assert.Equal(t, 2, len(events))
+		ok, err := events[0].MetricSetFields.HasKey("instance")
+		assert.NoError(t, err)
+		assert.True(t, ok)
+		ok, err = events[1].MetricSetFields.HasKey("instance")
+		assert.NoError(t, err)
+		assert.True(t, ok)
+		val1, err := events[0].MetricSetFields.GetValue("instance")
+		assert.NoError(t, err)
+		assert.Equal(t, val1, "processor")
+		val2, err := events[1].MetricSetFields.GetValue("instance")
+		assert.NoError(t, err)
+		assert.Equal(t, val2, "processor#1")
+	}
+}
+
 func TestMatchesParentProcess(t *testing.T) {
 	ok, val := matchesParentProcess("svchost")
 	assert.True(t, ok)