Make compute_vm light metricset and add update cloud instance id

CHANGELOG.next.asciidoc

@@ -667,6 +667,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add state_daemonset metricset for Kubernetes Metricbeat module {pull}20649[20649]
 - Add host inventory metrics to azure compute_vm metricset. {pull}20641[20641]
 - Add host inventory metrics to googlecloud compute metricset. {pull}20391[20391]
+- Migrate `compute_vm` metricset to a light one, map `cloud.instance.id` field. {pull}20889[20889]
 - Request prometheus endpoints to be gzipped by default {pull}20766[20766]
 - Add billing metricset into googlecloud module. {pull}20812[20812] {issue}20738[20738]
 

metricbeat/docs/modules/azure/compute_vm.asciidoc

@@ -8,6 +8,7 @@ This file is generated! See scripts/mage/docs_collector.go
 
 include::../../../../x-pack/metricbeat/module/azure/compute_vm/_meta/docs.asciidoc[]
 
+This is a default metricset. If the host module is unconfigured, this metricset is enabled by default.
 
 ==== Fields
 

x-pack/metricbeat/module/azure/add_metadata.go

@@ -0,0 +1,50 @@
+// 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.
+
+package azure
+
+import (
+	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/metricbeat/mb"
+)
+
+func addHostMetadata(event *mb.Event, metricList common.MapStr) {
+	hostFieldTable := map[string]string{
+		"percentage_cpu.avg":      "host.cpu.pct",
+		"network_in_total.total":  "host.network.in.bytes",
+		"network_in.total":        "host.network.in.packets",
+		"network_out_total.total": "host.network.out.bytes",
+		"network_out.total":       "host.network.out.packets",
+		"disk_read_bytes.total":   "host.disk.read.bytes",
+		"disk_write_bytes.total":  "host.disk.write.bytes",
+	}
+
+	for metricName, hostName := range hostFieldTable {
+		metricValue, err := metricList.GetValue(metricName)
+		if err != nil {
+			continue
+		}
+
+		if value, ok := metricValue.(float64); ok {
+			if metricName == "percentage_cpu.avg" {
+				value = value / 100
+			}
+			event.RootFields.Put(hostName, value)
+		}
+	}
+}
+
+func addCloudVMMetadata(event *mb.Event, resource Resource) {
+	event.RootFields.Put("cloud.instance.name", resource.Name)
+	event.RootFields.Put("host.name", resource.Name)
+	if resource.Vm != (VmResource{}) {
+		if resource.Vm.Id != "" {
+			event.RootFields.Put("cloud.instance.id", resource.Vm.Id)
+			event.RootFields.Put("host.id", resource.Vm.Id)
+		}
+		if resource.Vm.Size != "" {
+			event.RootFields.Put("cloud.machine.type", resource.Vm.Size)
+		}
+	}
+}

x-pack/metricbeat/module/azure/azure.go

@@ -6,50 +6,12 @@ package azure
 
 import (
 	"fmt"
-	"time"
 
 	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/v7/metricbeat/mb"
 )
 
-// Config options
-type Config struct {
-	ClientId            string           `config:"client_id"`
-	ClientSecret        string           `config:"client_secret"`
-	TenantId            string           `config:"tenant_id"`
-	SubscriptionId      string           `config:"subscription_id"`
-	Period              time.Duration    `config:"period" validate:"nonzero,required"`
-	Resources           []ResourceConfig `config:"resources"`
-	RefreshListInterval time.Duration    `config:"refresh_list_interval"`
-	DefaultResourceType string           `config:"default_resource_type"`
-}
-
-// ResourceConfig contains resource and metric list specific configuration.
-type ResourceConfig struct {
-	Id          []string       `config:"resource_id"`
-	Group       []string       `config:"resource_group"`
-	Metrics     []MetricConfig `config:"metrics"`
-	Type        string         `config:"resource_type"`
-	Query       string         `config:"resource_query"`
-	ServiceType []string       `config:"service_type"`
-}
-
-// MetricConfig contains metric specific configuration.
-type MetricConfig struct {
-	Name         []string          `config:"name"`
-	Namespace    string            `config:"namespace"`
-	Aggregations []string          `config:"aggregations"`
-	Dimensions   []DimensionConfig `config:"dimensions"`
-	Timegrain    string            `config:"timegrain"`
-}
-
-// DimensionConfig contains dimensions specific configuration.
-type DimensionConfig struct {
-	Name  string `config:"name"`
-	Value string `config:"value"`
-}
-
 func init() {
 	// Register the ModuleFactory function for the "azure" module.
 	if err := mb.Registry.AddModule("azure", newModule); err != nil {
@@ -127,20 +89,20 @@ func NewMetricSet(base mb.BaseMetricSet) (*MetricSet, error) {
 // It publishes the event which is then forwarded to the output. In case
 // of an error set the Error field of mb.Event or simply call report.Error().
 func (m *MetricSet) Fetch(report mb.ReporterV2) error {
-	err := m.Client.InitResources(m.MapMetrics, report)
+	err := m.Client.InitResources(m.MapMetrics)
 	if err != nil {
 		return err
 	}
-	if len(m.Client.Resources.Metrics) == 0 {
+	if len(m.Client.ResourceConfigurations.Metrics) == 0 {
 		// error message is previously logged in the InitResources, no error event should be created
 		return nil
 	}
 	// retrieve metrics
-	groupedMetrics := groupMetricsByResource(m.Client.Resources.Metrics)
+	groupedMetrics := groupMetricsByResource(m.Client.ResourceConfigurations.Metrics)
 
 	for _, metrics := range groupedMetrics {
 		results := m.Client.GetMetricValues(metrics, report)
-		err := EventsMapping(results, m.BaseMetricSet.Name(), report)
+		err := EventsMapping(results, m.Client, report)
 		if err != nil {
 			return errors.Wrap(err, "error running EventsMapping")
 		}
@@ -160,19 +122,3 @@ func hasConfigOptions(config []string) bool {
 	}
 	return true
 }
-
-func (conf *Config) Validate() error {
-	if conf.SubscriptionId == "" {
-		return errors.New("no subscription ID has been configured")
-	}
-	if conf.ClientSecret == "" {
-		return errors.New("no client secret has been configured")
-	}
-	if conf.ClientId == "" {
-		return errors.New("no client ID has been configured")
-	}
-	if conf.TenantId == "" {
-		return errors.New("no tenant ID has been configured")
-	}
-	return nil
-}

x-pack/metricbeat/module/azure/client.go

@@ -20,10 +20,11 @@ import (
 
 // Client represents the azure client which will make use of the azure sdk go metrics related clients
 type Client struct {
-	AzureMonitorService Service
-	Config              Config
-	Resources           ResourceConfiguration
-	Log                 *logp.Logger
+	AzureMonitorService    Service
+	Config                 Config
+	ResourceConfigurations ResourceConfiguration
+	Log                    *logp.Logger
+	Resources              []Resource
 }
 
 // mapResourceMetrics function type will map the configuration options to client metrics (depending on the metricset)
@@ -40,21 +41,23 @@ func NewClient(config Config) (*Client, error) {
 		Config:              config,
 		Log:                 logp.NewLogger("azure monitor client"),
 	}
-	client.Resources.RefreshInterval = config.RefreshListInterval
+	client.ResourceConfigurations.RefreshInterval = config.RefreshListInterval
 	return client, nil
 }
 
 // InitResources function will retrieve and validate the resources configured by the users and then map the information configured to client metrics.
 // the mapMetric function sent in this case will handle the mapping part as different metric and aggregation options work for different metricsets
-func (client *Client) InitResources(fn mapResourceMetrics, report mb.ReporterV2) error {
+func (client *Client) InitResources(fn mapResourceMetrics) error {
 	if len(client.Config.Resources) == 0 {
 		return errors.New("no resource options defined")
 	}
 	// check if refresh interval has been set and if it has expired
-	if !client.Resources.Expired() {
+	if !client.ResourceConfigurations.Expired() {
 		return nil
 	}
 	var metrics []Metric
+	//reset client resources
+	client.Resources = []Resource{}
 	for _, resource := range client.Config.Resources {
 		// retrieve azure resources information
 		resourceList, err := client.AzureMonitorService.GetResourceDefinitions(resource.Id, resource.Group, resource.Type, resource.Query)
@@ -68,6 +71,19 @@ func (client *Client) InitResources(fn mapResourceMetrics, report mb.ReporterV2)
 			client.Log.Error(err)
 			continue
 		}
+		//map resources to the client
+		for _, resource := range resourceList.Values() {
+			if !containsResource(*resource.ID, client.Resources) {
+				client.Resources = append(client.Resources, Resource{
+					Id:           *resource.ID,
+					Name:         *resource.Name,
+					Location:     *resource.Location,
+					Type:         *resource.Type,
+					Group:        getResourceGroupFromId(*resource.ID),
+					Tags:         mapTags(resource.Tags),
+					Subscription: client.Config.SubscriptionId})
+			}
+		}
 		resourceMetrics, err := fn(client, resourceList.Values(), resource)
 		if err != nil {
 			return err
@@ -79,7 +95,7 @@ func (client *Client) InitResources(fn mapResourceMetrics, report mb.ReporterV2)
 	if len(metrics) == 0 {
 		client.Log.Debug("no resources were found based on all the configurations options entered")
 	}
-	client.Resources.Metrics = metrics
+	client.ResourceConfigurations.Metrics = metrics
 	return nil
 }
 
@@ -107,21 +123,21 @@ func (client *Client) GetMetricValues(metrics []Metric, report mb.ReporterV2) []
 			}
 			filter = strings.Join(filterList, " AND ")
 		}
-		resp, timegrain, err := client.AzureMonitorService.GetMetricValues(metric.Resource.SubId, metric.Namespace, metric.TimeGrain, timespan, metric.Names,
+		resp, timegrain, err := client.AzureMonitorService.GetMetricValues(metric.ResourceSubId, metric.Namespace, metric.TimeGrain, timespan, metric.Names,
 			metric.Aggregations, filter)
 		if err != nil {
-			err = errors.Wrapf(err, "error while listing metric values by resource ID %s and namespace  %s", metric.Resource.SubId, metric.Namespace)
+			err = errors.Wrapf(err, "error while listing metric values by resource ID %s and namespace  %s", metric.ResourceSubId, metric.Namespace)
 			client.Log.Error(err)
 			report.Error(err)
 		} else {
-			for i, currentMetric := range client.Resources.Metrics {
+			for i, currentMetric := range client.ResourceConfigurations.Metrics {
 				if matchMetrics(currentMetric, metric) {
 					current := mapMetricValues(resp, currentMetric.Values, endTime.Truncate(time.Minute).Add(interval*(-1)), endTime.Truncate(time.Minute))
-					client.Resources.Metrics[i].Values = current
-					if client.Resources.Metrics[i].TimeGrain == "" {
-						client.Resources.Metrics[i].TimeGrain = timegrain
+					client.ResourceConfigurations.Metrics[i].Values = current
+					if client.ResourceConfigurations.Metrics[i].TimeGrain == "" {
+						client.ResourceConfigurations.Metrics[i].TimeGrain = timegrain
 					}
-					resultedMetrics = append(resultedMetrics, client.Resources.Metrics[i])
+					resultedMetrics = append(resultedMetrics, client.ResourceConfigurations.Metrics[i])
 				}
 			}
 		}
@@ -130,26 +146,20 @@ func (client *Client) GetMetricValues(metrics []Metric, report mb.ReporterV2) []
 }
 
 // CreateMetric function will create a client metric based on the resource and metrics configured
-func (client *Client) CreateMetric(selectedResourceID string, resource resources.GenericResource, resourceSize string, namespace string, metrics []string, aggregations string, dimensions []Dimension, timegrain string) Metric {
+func (client *Client) CreateMetric(resourceId string, subResourceId string, namespace string, metrics []string, aggregations string, dimensions []Dimension, timegrain string) Metric {
+	if subResourceId == "" {
+		subResourceId = resourceId
+	}
 	met := Metric{
-		Resource: Resource{
-			SubId:        selectedResourceID,
-			Id:           *resource.ID,
-			Name:         *resource.Name,
-			Location:     *resource.Location,
-			Type:         *resource.Type,
-			Group:        getResourceGroupFromId(*resource.ID),
-			Tags:         mapTags(resource.Tags),
-			Subscription: client.Config.SubscriptionId,
-			Size:         resourceSize,
-		},
-		Namespace:    namespace,
-		Names:        metrics,
-		Dimensions:   dimensions,
-		Aggregations: aggregations,
-		TimeGrain:    timegrain,
+		ResourceId:    resourceId,
+		ResourceSubId: subResourceId,
+		Namespace:     namespace,
+		Names:         metrics,
+		Dimensions:    dimensions,
+		Aggregations:  aggregations,
+		TimeGrain:     timegrain,
 	}
-	for _, prevMet := range client.Resources.Metrics {
+	for _, prevMet := range client.ResourceConfigurations.Metrics {
 		if len(prevMet.Values) != 0 && matchMetrics(prevMet, met) {
 			met.Values = prevMet.Values
 		}
@@ -158,7 +168,7 @@ func (client *Client) CreateMetric(selectedResourceID string, resource resources
 }
 
 // MapMetricByPrimaryAggregation will map the primary aggregation of the metric definition to the client metric
-func (client *Client) MapMetricByPrimaryAggregation(metrics []insights.MetricDefinition, resource resources.GenericResource, selectedResourceID string, resourceSize string, namespace string, dim []Dimension, timegrain string) []Metric {
+func (client *Client) MapMetricByPrimaryAggregation(metrics []insights.MetricDefinition, resourceId string, subResourceId string, namespace string, dim []Dimension, timegrain string) []Metric {
 	var clientMetrics []Metric
 	metricGroups := make(map[string][]insights.MetricDefinition)
 
@@ -170,10 +180,50 @@ func (client *Client) MapMetricByPrimaryAggregation(metrics []insights.MetricDef
 		for _, metricName := range metricGroup {
 			metricNames = append(metricNames, *metricName.Name.Value)
 		}
-		if selectedResourceID == "" {
-			selectedResourceID = *resource.ID
-		}
-		clientMetrics = append(clientMetrics, client.CreateMetric(selectedResourceID, resource, resourceSize, namespace, metricNames, key, dim, timegrain))
+		clientMetrics = append(clientMetrics, client.CreateMetric(resourceId, subResourceId, namespace, metricNames, key, dim, timegrain))
 	}
 	return clientMetrics
 }
+
+func (client *Client) GetResourceForData(resId string) Resource {
+	for i, res := range client.Resources {
+		if res.Id == resId {
+			var vmSize string
+			var vmId string
+			if client.Config.AddCloudMetadata && res.Vm == (VmResource{}) {
+				expandedResource, err := client.AzureMonitorService.GetResourceDefinitionById(res.Id)
+				if err != nil {
+					client.Log.Error(err, "could not retrieve the resource details by resource ID %s", res.Id)
+					return Resource{}
+				}
+				if expandedResource.Properties != nil {
+					if properties, ok := expandedResource.Properties.(map[string]interface{}); ok {
+						if hardware, ok := properties["hardwareProfile"]; ok {
+							if vmSz, ok := hardware.(map[string]interface{})["vmSize"]; ok {
+								vmSize = vmSz.(string)
+							}
+							if vmID, ok := properties["vmId"]; ok {
+								vmId = vmID.(string)
+							}
+						}
+					}
+				}
+				client.Resources[i].Vm = VmResource{Size: vmSize, Id: vmId}
+				return client.Resources[i]
+			}
+			return res
+		}
+	}
+	return Resource{}
+}
+
+// NewMockClient instantiates a new client with the mock azure service
+func NewMockClient() *Client {
+	azureMockService := new(MockService)
+	client := &Client{
+		AzureMonitorService: azureMockService,
+		Config:              Config{},
+		Log:                 logp.NewLogger("test azure monitor"),
+	}
+	return client
+}