[Metricbeat] Add region to googlecloud module config (elastic#16203)

* Add region to googlecloud module config

* Add changelog

* add warning when zone and region are both provided in config

* add more unit test for getFilterForMetric

* check if instance is nil before checking labels/machinetype

CHANGELOG.next.asciidoc

@@ -161,6 +161,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add collecting AuroraDB metrics in rds metricset. {issue}14142[14142] {pull}16004[16004]
 - Reuse connections in SQL module. {pull}16001[16001]
 - Improve the `logstash` module (when `xpack.enabled` is set to `true`) to use the override `cluster_uuid` returned by Logstash APIs. {issue}15772[15772] {pull}15795[15795]
+- Add region parameter in googlecloud module. {issue}15780[15780] {pull}16203[16203]
 - Add kubernetes storage class support via kube-state-metrics. {pull}16145[16145]
 - Add support for NATS 2.1. {pull}16317[16317]
 - Add Load Balancing metricset to GCP {pull}15559[15559]

metricbeat/docs/modules/googlecloud.asciidoc

@@ -17,6 +17,7 @@ Note: extra GCP charges on Stackdriver Monitoring API requests will be generated
 This is a list of the possible module parameters you can tune:
 
 * *zone*: A single string with the zone you want to monitor like "us-central1-a". If you need to fetch from multiple regions, you have to setup a different configuration for each (but you don't need a new instance of Metricbeat running)
+* *region*: A single string with the region you want to monitor like "us-central1". This will enable monitoring for all zones under this region.
 * *project_id*: A single string with your GCP Project ID
 * *credentials_file_path*: A single string pointing to the JSON file path reachable by Metricbeat that you have created using IAM.
 * *exclude_labels*: (`true`/`false` default `false`) Do not extract extra labels and metadata information from Metricsets and fetch metrics onlly. At the moment, *labels and metadata extraction is only supported* in Compute Metricset.

x-pack/metricbeat/module/googlecloud/_meta/docs.asciidoc

@@ -7,6 +7,7 @@ Note: extra GCP charges on Stackdriver Monitoring API requests will be generated
 This is a list of the possible module parameters you can tune:
 
 * *zone*: A single string with the zone you want to monitor like "us-central1-a". If you need to fetch from multiple regions, you have to setup a different configuration for each (but you don't need a new instance of Metricbeat running)
+* *region*: A single string with the region you want to monitor like "us-central1". This will enable monitoring for all zones under this region.
 * *project_id*: A single string with your GCP Project ID
 * *credentials_file_path*: A single string pointing to the JSON file path reachable by Metricbeat that you have created using IAM.
 * *exclude_labels*: (`true`/`false` default `false`) Do not extract extra labels and metadata information from Metricsets and fetch metrics onlly. At the moment, *labels and metadata extraction is only supported* in Compute Metricset.

x-pack/metricbeat/module/googlecloud/metadata.go

@@ -54,6 +54,7 @@ type MetadataCollectorInputData struct {
 	TimeSeries *monitoringpb.TimeSeries
 	ProjectID  string
 	Zone       string
+	Region     string
 	Point      *monitoringpb.Point
 	Timestamp  *time.Time
 }

x-pack/metricbeat/module/googlecloud/stackdriver/compute/metadata.go

@@ -20,10 +20,11 @@ import (
 )
 
 // NewMetadataService returns the specific Metadata service for a GCP Compute resource
-func NewMetadataService(projectID, zone string, opt ...option.ClientOption) (googlecloud.MetadataService, error) {
+func NewMetadataService(projectID, zone string, region string, opt ...option.ClientOption) (googlecloud.MetadataService, error) {
 	return &metadataCollector{
 		projectID:     projectID,
 		zone:          zone,
+		region:        region,
 		opt:           opt,
 		instanceCache: common.NewCache(30*time.Second, 13),
 	}, nil
@@ -48,6 +49,7 @@ type computeMetadata struct {
 type metadataCollector struct {
 	projectID string
 	zone      string
+	region    string
 	opt       []option.ClientOption
 
 	computeMetadata *computeMetadata
@@ -58,7 +60,7 @@ type metadataCollector struct {
 // Metadata implements googlecloud.MetadataCollector to the known set of labels from a Compute TimeSeries single point of data.
 func (s *metadataCollector) Metadata(ctx context.Context, resp *monitoringpb.TimeSeries) (googlecloud.MetadataCollectorData, error) {
 	if s.computeMetadata == nil {
-		_, err := s.instanceMetadata(ctx, s.instanceID(resp), s.zone)
+		_, err := s.instanceMetadata(ctx, s.instanceID(resp), s.zone, s.region)
 		if err != nil {
 			return googlecloud.MetadataCollectorData{}, err
 		}
@@ -102,8 +104,8 @@ func (s *metadataCollector) Metadata(ctx context.Context, resp *monitoringpb.Tim
 }
 
 // instanceMetadata returns the labels of an instance
-func (s *metadataCollector) instanceMetadata(ctx context.Context, instanceID, zone string) (*computeMetadata, error) {
-	i, err := s.instance(ctx, instanceID, zone)
+func (s *metadataCollector) instanceMetadata(ctx context.Context, instanceID, zone string, region string) (*computeMetadata, error) {
+	i, err := s.instance(ctx, instanceID, zone, region)
 	if err != nil {
 		return nil, errors.Wrapf(err, "error trying to get data from instance '%s' in zone '%s'", instanceID, zone)
 	}
@@ -113,6 +115,10 @@ func (s *metadataCollector) instanceMetadata(ctx context.Context, instanceID, zo
 		zone:       zone,
 	}
 
+	if i == nil {
+		return s.computeMetadata, nil
+	}
+
 	if i.Labels != nil {
 		s.computeMetadata.User = i.Labels
 	}
@@ -133,7 +139,7 @@ func (s *metadataCollector) instanceMetadata(ctx context.Context, instanceID, zo
 }
 
 // instance returns data from an instance ID using the cache or making a request
-func (s *metadataCollector) instance(ctx context.Context, instanceID, zone string) (i *compute.Instance, err error) {
+func (s *metadataCollector) instance(ctx context.Context, instanceID, zone string, region string) (*compute.Instance, error) {
 	service, err := compute.NewService(ctx, s.opt...)
 	if err != nil {
 		return nil, errors.Wrapf(err, "error getting client from Compute service")
@@ -146,13 +152,34 @@ func (s *metadataCollector) instance(ctx context.Context, instanceID, zone strin
 		}
 	}
 
-	instanceData, err := service.Instances.Get(s.projectID, zone, instanceID).Do()
-	if err != nil {
-		return nil, errors.Wrapf(err, "error getting instance information for instance with ID '%s'", instanceID)
+	if region != "" {
+		regionData, err := service.Regions.Get(s.projectID, region).Do()
+		if err != nil {
+			return nil, errors.Wrapf(err, "error getting region information for '%s'", region)
+		}
+
+		zones := regionData.Zones
+		for _, zone := range zones {
+			zString := strings.Split(zone, "/")
+			zName := zString[len(zString)-1]
+			instanceData, err := service.Instances.Get(s.projectID, zName, instanceID).Do()
+			if err != nil {
+				continue
+			}
+			s.instanceCache.Put(instanceID, instanceData)
+			return instanceData, nil
+		}
 	}
-	s.instanceCache.Put(instanceID, instanceData)
 
-	return instanceData, nil
+	if zone != "" {
+		instanceData, err := service.Instances.Get(s.projectID, zone, instanceID).Do()
+		if err != nil {
+			return nil, errors.Wrapf(err, "error getting instance information for instance with ID '%s'", instanceID)
+		}
+		s.instanceCache.Put(instanceID, instanceData)
+		return instanceData, nil
+	}
+	return nil, nil
 }
 
 func (s *metadataCollector) instanceID(ts *monitoringpb.TimeSeries) string {

x-pack/metricbeat/module/googlecloud/stackdriver/metadata_services.go

@@ -16,7 +16,7 @@ import (
 func NewMetadataServiceForConfig(c config) (googlecloud.MetadataService, error) {
 	switch c.ServiceName {
 	case googlecloud.ServiceCompute:
-		return compute.NewMetadataService(c.ProjectID, c.Zone, c.opt...)
+		return compute.NewMetadataService(c.ProjectID, c.Zone, c.Region, c.opt...)
 	case googlecloud.ServicePubsub, googlecloud.ServiceLoadBalancing:
 		return nil, nil
 	default:

x-pack/metricbeat/module/googlecloud/stackdriver/metrics_requester.go

@@ -52,13 +52,11 @@ type stackdriverMetricsRequester struct {
 func (r *stackdriverMetricsRequester) Metric(ctx context.Context, m string) (out []*monitoringpb.TimeSeries) {
 	out = make([]*monitoringpb.TimeSeries, 0)
 
-	filter := r.getFilterForMetric(m)
-
 	req := &monitoringpb.ListTimeSeriesRequest{
 		Name:     "projects/" + r.config.ProjectID,
 		Interval: r.interval,
 		View:     monitoringpb.ListTimeSeriesRequest_FULL,
-		Filter:   filter,
+		Filter:   r.getFilterForMetric(m),
 	}
 
 	it := r.client.ListTimeSeries(ctx, req)
@@ -79,6 +77,18 @@ func (r *stackdriverMetricsRequester) Metric(ctx context.Context, m string) (out
 	return
 }
 
+func constructFilter(m string, region string, zone string) string {
+	filter := fmt.Sprintf(`metric.type="%s" AND resource.labels.zone = `, m)
+	// If region is specified, use region as filter resource label.
+	// If region is empty but zone is given, use zone instead.
+	if region != "" {
+		filter += fmt.Sprintf(`starts_with("%s")`, region)
+	} else if zone != "" {
+		filter += fmt.Sprintf(`"%s"`, zone)
+	}
+	return filter
+}
+
 func (r *stackdriverMetricsRequester) Metrics(ctx context.Context, ms []string) ([]*monitoringpb.TimeSeries, error) {
 	var lock sync.Mutex
 	var wg sync.WaitGroup
@@ -120,9 +130,16 @@ func (r *stackdriverMetricsRequester) getFilterForMetric(m string) (f string) {
 	case googlecloud.ServicePubsub, googlecloud.ServiceLoadBalancing:
 		return
 	default:
-		f = fmt.Sprintf(`%s AND resource.labels.zone = "%s"`, f, r.config.Zone)
+		if r.config.Region != "" && r.config.Zone != "" {
+			r.logger.Warnf("when region %s and zone %s config parameter "+
+				"both are provided, only use region", r.config.Region, r.config.Zone)
+		}
+		if r.config.Region != "" {
+			f = fmt.Sprintf(`%s AND resource.labels.zone = starts_with("%s")`, f, r.config.Region)
+		} else if r.config.Zone != "" {
+			f = fmt.Sprintf(`%s AND resource.labels.zone = "%s"`, f, r.config.Zone)
+		}
 	}
-
 	return
 }
 

x-pack/metricbeat/module/googlecloud/stackdriver/metrics_requester_test.go

@@ -8,17 +8,98 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/libbeat/logp"
 )
 
-func TestGetFilterForMetric(t *testing.T) {
-	r := stackdriverMetricsRequester{config: config{Zone: "us-central1-a"}}
+func TestStringInSlice(t *testing.T) {
+	cases := []struct {
+		title          string
+		m              string
+		region         string
+		zone           string
+		expectedFilter string
+	}{
+		{
+			"construct filter with zone",
+			"compute.googleapis.com/instance/cpu/utilization",
+			"",
+			"us-east1-b",
+			"metric.type=\"compute.googleapis.com/instance/cpu/utilization\" AND resource.labels.zone = \"us-east1-b\"",
+		},
+		{
+			"construct filter with region",
+			"compute.googleapis.com/instance/cpu/utilization",
+			"us-east1",
+			"",
+			"metric.type=\"compute.googleapis.com/instance/cpu/utilization\" AND resource.labels.zone = starts_with(\"us-east1\")",
+		},
+	}
 
-	s := r.getFilterForMetric("compute.googleapis.com/firewall/dropped_bytes_count")
-	assert.Equal(t, `metric.type="compute.googleapis.com/firewall/dropped_bytes_count" AND resource.labels.zone = "us-central1-a"`, s)
+	for _, c := range cases {
+		t.Run(c.title, func(t *testing.T) {
+			filter := constructFilter(c.m, c.region, c.zone)
+			assert.Equal(t, c.expectedFilter, filter)
+		})
+	}
+}
 
-	s = r.getFilterForMetric("pubsub.googleapis.com/subscription/ack_message_count")
-	assert.Equal(t, `metric.type="pubsub.googleapis.com/subscription/ack_message_count"`, s)
+func TestGetFilterForMetric(t *testing.T) {
+	var logger = logp.NewLogger("test")
+	cases := []struct {
+		title          string
+		m              string
+		r              stackdriverMetricsRequester
+		expectedFilter string
+	}{
+		{
+			"compute service with zone in config",
+			"compute.googleapis.com/firewall/dropped_bytes_count",
+			stackdriverMetricsRequester{config: config{Zone: "us-central1-a"}},
+			"metric.type=\"compute.googleapis.com/firewall/dropped_bytes_count\" AND resource.labels.zone = \"us-central1-a\"",
+		},
+		{
+			"pubsub service with zone in config",
+			"pubsub.googleapis.com/subscription/ack_message_count",
+			stackdriverMetricsRequester{config: config{Zone: "us-central1-a"}},
+			"metric.type=\"pubsub.googleapis.com/subscription/ack_message_count\"",
+		},
+		{
+			"loadbalancing service with zone in config",
+			"loadbalancing.googleapis.com/https/backend_latencies",
+			stackdriverMetricsRequester{config: config{Zone: "us-central1-a"}},
+			"metric.type=\"loadbalancing.googleapis.com/https/backend_latencies\"",
+		},
+		{
+			"compute service with region in config",
+			"compute.googleapis.com/firewall/dropped_bytes_count",
+			stackdriverMetricsRequester{config: config{Region: "us-east1"}},
+			"metric.type=\"compute.googleapis.com/firewall/dropped_bytes_count\" AND resource.labels.zone = starts_with(\"us-east1\")",
+		},
+		{
+			"pubsub service with region in config",
+			"pubsub.googleapis.com/subscription/ack_message_count",
+			stackdriverMetricsRequester{config: config{Region: "us-east1"}},
+			"metric.type=\"pubsub.googleapis.com/subscription/ack_message_count\"",
+		},
+		{
+			"loadbalancing service with region in config",
+			"loadbalancing.googleapis.com/https/backend_latencies",
+			stackdriverMetricsRequester{config: config{Region: "us-east1"}},
+			"metric.type=\"loadbalancing.googleapis.com/https/backend_latencies\"",
+		},
+		{
+			"compute service with both region and zone in config",
+			"compute.googleapis.com/firewall/dropped_bytes_count",
+			stackdriverMetricsRequester{config: config{Region: "us-central1", Zone: "us-central1-a"}, logger: logger},
+			"metric.type=\"compute.googleapis.com/firewall/dropped_bytes_count\" AND resource.labels.zone = starts_with(\"us-central1\")",
+		},
+	}
 
-	s = r.getFilterForMetric("loadbalancing.googleapis.com/https/backend_latencies")
-	assert.Equal(t, `metric.type="loadbalancing.googleapis.com/https/backend_latencies"`, s)
+	for _, c := range cases {
+		t.Run(c.title, func(t *testing.T) {
+			filter := c.r.getFilterForMetric(c.m)
+			assert.Equal(t, c.expectedFilter, filter)
+		})
+	}
 }

x-pack/metricbeat/module/googlecloud/stackdriver/metricset.go

@@ -43,7 +43,8 @@ type MetricSet struct {
 
 type config struct {
 	Metrics             []string `config:"stackdriver.metrics" validate:"required"`
-	Zone                string   `config:"zone" validate:"required"`
+	Zone                string   `config:"zone"`
+	Region              string   `config:"region"`
 	ProjectID           string   `config:"project_id" validate:"required"`
 	ExcludeLabels       bool     `config:"exclude_labels"`
 	ServiceName         string   `config:"stackdriver.service"  validate:"required"`
@@ -78,12 +79,13 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 func (m *MetricSet) Fetch(ctx context.Context, reporter mb.ReporterV2) (err error) {
 	reqs, err := newStackdriverMetricsRequester(ctx, m.config, m.Module().Config().Period, m.Logger())
 	if err != nil {
-		return errors.Wrapf(err, "error trying to do create a request client to GCP project '%s' in zone '%s'", m.config.ProjectID, m.config.Zone)
+		return errors.Wrapf(err, "error trying to do create a request client to GCP project '%s' in zone '%s' or region '%s'", m.config.ProjectID, m.config.Zone, m.config.Region)
 	}
 
 	responses, err := reqs.Metrics(ctx, m.config.Metrics)
 	if err != nil {
-		return errors.Wrapf(err, "error trying to get metrics for project '%s' and zone '%s'", m.config.ProjectID, m.config.Zone)
+
+		return errors.Wrapf(err, "error trying to get metrics for project '%s' and zone '%s' or region '%s'", m.config.ProjectID, m.config.Zone, m.config.Region)
 	}
 
 	events, err := m.eventMapping(ctx, responses)
@@ -145,3 +147,10 @@ func validatePeriodForGCP(d time.Duration) (err error) {
 
 	return nil
 }
+
+func (c *config) Validate() error {
+	if c.Region == "" && c.Zone == "" {
+		return errors.New("region and zone in Google Cloud config file cannot both be empty")
+	}
+	return nil
+}

x-pack/metricbeat/module/googlecloud/stackdriver/timeseries.go

@@ -27,7 +27,7 @@ func (m *MetricSet) timeSeriesGrouped(ctx context.Context, gcpService googleclou
 			return nil, err
 		}
 
-		sdCollectorInputData := googlecloud.NewStackdriverCollectorInputData(ts, m.config.ProjectID, m.config.Zone)
+		sdCollectorInputData := googlecloud.NewStackdriverCollectorInputData(ts, m.config.ProjectID, m.config.Zone, m.config.Region)
 
 		for i := range keyValues {
 			sdCollectorInputData.Timestamp = &keyValues[i].Timestamp

x-pack/metricbeat/module/googlecloud/timeseries_metadata_collector.go

@@ -16,11 +16,12 @@ import (
 )
 
 // NewStackdriverCollectorInputData returns a ready to use MetadataCollectorInputData to be sent to Metadata collectors
-func NewStackdriverCollectorInputData(ts *monitoringpb.TimeSeries, projectID, zone string) *MetadataCollectorInputData {
+func NewStackdriverCollectorInputData(ts *monitoringpb.TimeSeries, projectID, zone string, region string) *MetadataCollectorInputData {
 	return &MetadataCollectorInputData{
 		TimeSeries: ts,
 		ProjectID:  projectID,
 		Zone:       zone,
+		Region:     region,
 	}
 }