[Metricbeat] Add support for multiple regions in GCP (#32964)

* add regions config setting and pass it as argument

* add services region resource label constants

* add buildRegionsFilter and update getFilterForMetric logic

* add getFilterForMetric and buildRegionsFilter tests

* add changelog entry

* minor changes for golangci-lint

* add warn logs and remove redundant return

* add missing argument in warnf log

(cherry picked from commit 3bcefab)

# Conflicts:
#	x-pack/metricbeat/module/gcp/metrics/compute/metadata.go
#	x-pack/metricbeat/module/gcp/metrics/metrics_requester.go
#	x-pack/metricbeat/module/gcp/metrics/metrics_requester_test.go
#	x-pack/metricbeat/module/gcp/metrics/metricset.go

CHANGELOG-developer.next.asciidoc

@@ -118,6 +118,15 @@ The list below covers the major changes between 7.0.0-rc2 and master only.
 - Introduce `libbeat/beat.Beat.OutputConfigReloader` {pull}28048[28048]
 - Update Go version to 1.17.1. {pull}27543[27543]
 - Whitelist `GCP_*` environment variables in dev tools {pull}28364[28364]
+- Add support for `credentials_json` in `gcp` module, all metricsets {pull}29584[29584]
+- Add gcp firestore metricset. {pull}29918[29918]
+- Added TESTING_FILEBEAT_FILEPATTERN option for filebeat module pytests {pull}30103[30103]
+- Add gcp dataproc metricset. {pull}30008[30008]
+- Add Github action for linting
+- Add regex support for drop_fields processor.
+- Improve compatibility and reduce flakyness of Python tests {pull}31588[31588]
+- Added `.python-version` file {pull}32323[32323]
+- Add support for multiple regions in GCP {pull}32964[32964]
 
 ==== Deprecated
 

x-pack/metricbeat/module/gcp/constants.go

@@ -74,7 +74,14 @@ const (
 	LabelMetadata = "metadata"
 )
 
-// Available perSeriesAligner map
+const (
+	DefaultResourceLabelZone     = "resource.label.zone"
+	ComputeResourceLabelZone     = "resource.labels.zone"
+	GKEResourceLabelLocation     = "resource.label.location"
+	StorageResourceLabelLocation = "resource.label.location"
+)
+
+// AlignersMapToGCP map contains available perSeriesAligner
 // https://cloud.google.com/monitoring/api/ref_v3/rest/v3/projects.alertPolicies#Aligner
 var AlignersMapToGCP = map[string]monitoringpb.Aggregation_Aligner{
 	"ALIGN_NONE":           monitoringpb.Aggregation_ALIGN_NONE,

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

@@ -55,6 +55,7 @@ type MetadataCollectorInputData struct {
 	ProjectID  string
 	Zone       string
 	Region     string
+	Regions    []string
 	Point      *monitoringpb.Point
 	Timestamp  *time.Time
 }

x-pack/metricbeat/module/gcp/metrics/compute/metadata.go

@@ -6,10 +6,14 @@ package compute
 
 import (
 	"context"
+<<<<<<< HEAD
+=======
+	"fmt"
+	"strconv"
+>>>>>>> 3bcefabb28 ([Metricbeat] Add support for multiple regions in GCP (#32964))
 	"strings"
 	"time"
 
-	"github.com/pkg/errors"
 	"google.golang.org/api/compute/v1"
 	"google.golang.org/api/option"
 	monitoringpb "google.golang.org/genproto/googleapis/monitoring/v3"
@@ -20,11 +24,12 @@ import (
 )
 
 // NewMetadataService returns the specific Metadata service for a GCP Compute resource
-func NewMetadataService(projectID, zone string, region string, opt ...option.ClientOption) (gcp.MetadataService, error) {
+func NewMetadataService(projectID, zone string, region string, regions []string, opt ...option.ClientOption) (gcp.MetadataService, error) {
 	return &metadataCollector{
 		projectID:     projectID,
 		zone:          zone,
 		region:        region,
+		regions:       regions,
 		opt:           opt,
 		instanceCache: common.NewCache(30*time.Second, 13),
 		logger:        logp.NewLogger("metrics-compute"),
@@ -34,12 +39,12 @@ func NewMetadataService(projectID, zone string, region string, opt ...option.Cli
 // computeMetadata is an object to store data in between the extraction and the writing in the destination (to uncouple
 // reading and writing in the same method)
 type computeMetadata struct {
-	projectID   string
+	// projectID   string
 	zone        string
 	instanceID  string
 	machineType string
 
-	ts *monitoringpb.TimeSeries
+	// ts *monitoringpb.TimeSeries
 
 	User     map[string]string
 	Metadata map[string]string
@@ -48,13 +53,21 @@ type computeMetadata struct {
 }
 
 type metadataCollector struct {
+<<<<<<< HEAD
 	projectID string
 	zone      string
 	region    string
 	opt       []option.ClientOption
 
 	computeMetadata *computeMetadata
 
+=======
+	projectID     string
+	zone          string
+	region        string
+	regions       []string
+	opt           []option.ClientOption
+>>>>>>> 3bcefabb28 ([Metricbeat] Add support for multiple regions in GCP (#32964))
 	instanceCache *common.Cache
 	logger        *logp.Logger
 }
@@ -75,16 +88,22 @@ func (s *metadataCollector) Metadata(ctx context.Context, resp *monitoringpb.Tim
 	}
 
 	if resp.Resource != nil && resp.Resource.Labels != nil {
-		metadataCollectorData.ECS.Put(gcp.ECSCloudInstanceIDKey, resp.Resource.Labels[gcp.TimeSeriesResponsePathForECSInstanceID])
+		_, _ = metadataCollectorData.ECS.Put(gcp.ECSCloudInstanceIDKey, resp.Resource.Labels[gcp.TimeSeriesResponsePathForECSInstanceID])
 	}
 
 	if resp.Metric.Labels != nil {
-		metadataCollectorData.ECS.Put(gcp.ECSCloudInstanceNameKey, resp.Metric.Labels[gcp.TimeSeriesResponsePathForECSInstanceName])
+		_, _ = metadataCollectorData.ECS.Put(gcp.ECSCloudInstanceNameKey, resp.Metric.Labels[gcp.TimeSeriesResponsePathForECSInstanceName])
 	}
 
+<<<<<<< HEAD
 	if s.computeMetadata.machineType != "" {
 		lastIndex := strings.LastIndex(s.computeMetadata.machineType, "/")
 		metadataCollectorData.ECS.Put(gcp.ECSCloudMachineTypeKey, s.computeMetadata.machineType[lastIndex+1:])
+=======
+	if computeMetadata.machineType != "" {
+		lastIndex := strings.LastIndex(computeMetadata.machineType, "/")
+		_, _ = metadataCollectorData.ECS.Put(gcp.ECSCloudMachineTypeKey, computeMetadata.machineType[lastIndex+1:])
+>>>>>>> 3bcefabb28 ([Metricbeat] Add support for multiple regions in GCP (#32964))
 	}
 
 	s.computeMetadata.Metrics = metadataCollectorData.Labels[gcp.LabelMetrics]
@@ -110,7 +129,11 @@ func (s *metadataCollector) instanceMetadata(ctx context.Context, instanceID, zo
 	// FIXME: remove side effect on metadataCollector instance and use return value instead
 	i, err := s.instance(ctx, instanceID, zone)
 	if err != nil {
+<<<<<<< HEAD
 		return nil, errors.Wrapf(err, "error trying to get data from instance '%s' in zone '%s'", instanceID, zone)
+=======
+		return nil, fmt.Errorf("error trying to get data from instance '%s' %w", instanceID, err)
+>>>>>>> 3bcefabb28 ([Metricbeat] Add support for multiple regions in GCP (#32964))
 	}
 
 	s.computeMetadata = &computeMetadata{

x-pack/metricbeat/module/gcp/metrics/metadata_services.go

@@ -14,7 +14,7 @@ import (
 func NewMetadataServiceForConfig(c config, serviceName string) (gcp.MetadataService, error) {
 	switch serviceName {
 	case gcp.ServiceCompute:
-		return compute.NewMetadataService(c.ProjectID, c.Zone, c.Region, c.opt...)
+		return compute.NewMetadataService(c.ProjectID, c.Zone, c.Region, c.Regions, c.opt...)
 	default:
 		return nil, nil
 	}

x-pack/metricbeat/module/gcp/metrics/metrics_requester.go

@@ -94,15 +94,88 @@ func (r *metricsRequester) Metrics(ctx context.Context, serviceName string, alig
 	return results, nil
 }
 
+func (r *metricsRequester) buildRegionsFilter(regions []string, label string) string {
+	if len(regions) == 0 {
+		return ""
+	}
+
+	var filter strings.Builder
+
+	// No. of regions added to the filter string.
+	var regionsCount uint
+
+	for _, region := range regions {
+		// If 1 region has been added and the iteration continues, add the OR operator.
+		if regionsCount > 0 {
+			filter.WriteString("OR")
+			filter.WriteString(" ")
+		}
+
+		filter.WriteString(fmt.Sprintf("%s = starts_with(\"%s\")", label, strings.TrimSuffix(region, "*")))
+		filter.WriteString(" ")
+
+		regionsCount++
+	}
+
+	switch {
+	// If the filter string has more than 1 region, parentheses are added for better filter readability.
+	case regionsCount > 1:
+		return fmt.Sprintf("(%s)", strings.TrimSpace(filter.String()))
+	default:
+		return strings.TrimSpace(filter.String())
+	}
+}
+
 // getFilterForMetric returns the filter associated with the corresponding filter. Some services like Pub/Sub fails
 // if they have a region specified.
+<<<<<<< HEAD
 func (r *metricsRequester) getFilterForMetric(serviceName, m string) (f string) {
 	f = fmt.Sprintf(`metric.type="%s"`, m)
 	if r.config.Zone == "" && r.config.Region == "" {
 		return
+=======
+func (r *metricsRequester) getFilterForMetric(serviceName, m string) string {
+	f := fmt.Sprintf(`metric.type="%s"`, m)
+	if r.config.Zone == "" && r.config.Region == "" && len(r.config.Regions) == 0 {
+		return f
+>>>>>>> 3bcefabb28 ([Metricbeat] Add support for multiple regions in GCP (#32964))
 	}
 
 	switch serviceName {
+	case gcp.ServiceCompute:
+		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.Regions, r.config.Zone)
+		}
+
+		if r.config.Region != "" && len(r.config.Regions) != 0 {
+			r.logger.Warnf("when region %s and regions config parameters are both provided, use region", r.config.Region)
+		}
+
+		if r.config.Region != "" {
+			f = fmt.Sprintf(
+				"%s AND %s = starts_with(\"%s\")",
+				f,
+				gcp.ComputeResourceLabelZone,
+				strings.TrimSuffix(r.config.Region, "*"),
+			)
+			break
+		}
+
+		if r.config.Zone != "" {
+			f = fmt.Sprintf(
+				"%s AND %s = starts_with(\"%s\")",
+				f,
+				gcp.ComputeResourceLabelZone,
+				strings.TrimSuffix(r.config.Zone, "*"),
+			)
+			break
+		}
+
+		if len(r.config.Regions) != 0 {
+			regionsFilter := r.buildRegionsFilter(r.config.Regions, gcp.ComputeResourceLabelZone)
+			f = fmt.Sprintf("%s AND %s", f, regionsFilter)
+		}
 	case gcp.ServiceGKE:
 		if r.config.Region != "" && r.config.Zone != "" {
 			r.logger.Warnf("when region %s and zone %s config parameter "+
@@ -123,15 +196,32 @@ func (r *metricsRequester) getFilterForMetric(serviceName, m string) (f string)
 				zone = strings.TrimSuffix(zone, "*")
 			}
 			f = fmt.Sprintf("%s AND resource.label.location=starts_with(\"%s\")", f, zone)
+			break
+		}
+
+		if len(r.config.Regions) != 0 {
+			regionsFilter := r.buildRegionsFilter(r.config.Regions, gcp.GKEResourceLabelLocation)
+			f = fmt.Sprintf("%s AND %s", f, regionsFilter)
 		}
 	case gcp.ServicePubsub, gcp.ServiceLoadBalancing, gcp.ServiceCloudFunctions:
 		return
 	case gcp.ServiceStorage:
+<<<<<<< HEAD
 		if r.config.Region == "" {
 			return
+=======
+		if r.config.Region != "" && len(r.config.Regions) != 0 {
+			r.logger.Warnf("when region %s and regions config parameters are both provided, use region", r.config.Region)
+>>>>>>> 3bcefabb28 ([Metricbeat] Add support for multiple regions in GCP (#32964))
 		}
 
-		f = fmt.Sprintf(`%s AND resource.labels.location = "%s"`, f, r.config.Region)
+		switch {
+		case r.config.Region != "":
+			f = fmt.Sprintf(`%s AND resource.labels.location = "%s"`, f, r.config.Region)
+		case len(r.config.Regions) != 0:
+			regionsFilter := r.buildRegionsFilter(r.config.Regions, gcp.StorageResourceLabelLocation)
+			f = fmt.Sprintf("%s AND %s", f, regionsFilter)
+		}
 	default:
 		if r.config.Region != "" && r.config.Zone != "" {
 			r.logger.Warnf("when region %s and zone %s config parameter "+