[metricbeat] [gcp] add redis metadata

CHANGELOG.next.asciidoc

@@ -287,6 +287,7 @@ automatic splitting at root level, if root level element is an array. {pull}3415
 - Update README file on how to run Metricbeat on Kubernetes. {pull}33308[33308]
 - Add per-thread metrics to system_summary {pull}33614[33614]
 - Add GCP CloudSQL metadata {pull}33066[33066]
+- Add GCP Redis metadata {pull}33701[33701]
 - Remove GCP Compute metadata cache {pull}33655[33655]
 - Add support for multiple regions in GCP {pull}32964[32964]
 - Add GCP Redis regions support {pull}33728[33728]

go.mod

@@ -164,7 +164,7 @@ require (
 	golang.org/x/text v0.7.0
 	golang.org/x/time v0.3.0
 	golang.org/x/tools v0.1.12
-	google.golang.org/api v0.100.0
+	google.golang.org/api v0.102.0
 	google.golang.org/genproto v0.0.0-20221027153422-115e99e71e1c
 	google.golang.org/grpc v1.50.1
 	google.golang.org/protobuf v1.28.1
@@ -185,6 +185,7 @@ require (
 
 require (
 	cloud.google.com/go v0.105.0
+	cloud.google.com/go/redis v1.10.0
 	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v0.4.1
 	github.com/Azure/go-autorest/autorest/adal v0.9.14
 	github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.11.17
@@ -215,8 +216,10 @@ require (
 )
 
 require (
-	cloud.google.com/go/compute v1.10.0 // indirect
+	cloud.google.com/go/compute v1.12.1 // indirect
+	cloud.google.com/go/compute/metadata v0.2.1 // indirect
 	cloud.google.com/go/iam v0.6.0 // indirect
+	cloud.google.com/go/longrunning v0.1.1 // indirect
 	code.cloudfoundry.org/gofileutils v0.0.0-20170111115228-4d0c80011a0f // indirect
 	github.com/Azure/azure-amqp-common-go/v3 v3.2.1 // indirect
 	github.com/Azure/azure-pipeline-go v0.2.1 // indirect

go.sum

@@ -42,8 +42,10 @@ cloud.google.com/go/bigquery v1.42.0 h1:JuTk8po4bCKRwObdT0zLb1K0BGkGHJdtgs2GK3j2
 cloud.google.com/go/bigquery v1.42.0/go.mod h1:8dRTJxhtG+vwBKzE5OseQn/hiydoQN3EedCaOdYmxRA=
 cloud.google.com/go/bigtable v1.2.0/go.mod h1:JcVAOl45lrTmQfLj7T6TxyMzIN/3FGGcFm+2xVAli2o=
 cloud.google.com/go/bigtable v1.3.0/go.mod h1:z5EyKrPE8OQmeg4h5MNdKvuSnI9CCT49Ki3f23aBzio=
-cloud.google.com/go/compute v1.10.0 h1:aoLIYaA1fX3ywihqpBk2APQKOo20nXsp1GEZQbx5Jk4=
-cloud.google.com/go/compute v1.10.0/go.mod h1:ER5CLbMxl90o2jtNbGSbtfOpQKR0t15FOtRsugnLrlU=
+cloud.google.com/go/compute v1.12.1 h1:gKVJMEyqV5c/UnpzjjQbo3Rjvvqpr9B1DFSbJC4OXr0=
+cloud.google.com/go/compute v1.12.1/go.mod h1:e8yNOBcBONZU1vJKCvCoDw/4JQsA0dpM4x/6PIIOocU=
+cloud.google.com/go/compute/metadata v0.2.1 h1:efOwf5ymceDhK6PKMnnrTHP4pppY5L22mle96M1yP48=
+cloud.google.com/go/compute/metadata v0.2.1/go.mod h1:jgHgmJd2RKBGzXqF5LR2EZMGxBkeanZ9wwa75XHJgOM=
 cloud.google.com/go/datacatalog v1.7.0 h1:vYBwR8Sy0jVv6AIWCz37ylpDU7IQm2KgexqzOZePIEc=
 cloud.google.com/go/datastore v1.0.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7TKLgZqpHYE=
 cloud.google.com/go/datastore v1.1.0/go.mod h1:umbIZjpQpHh4hmRpGhH4tLFup+FVzqBi1b3c64qFpCk=
@@ -52,6 +54,7 @@ cloud.google.com/go/iam v0.6.0 h1:nsqQC88kT5Iwlm4MeNGTpfMWddp6NB/UOLFTH6m1QfQ=
 cloud.google.com/go/iam v0.6.0/go.mod h1:+1AH33ueBne5MzYccyMHtEKqLE4/kJOibtffMHDMFMc=
 cloud.google.com/go/kms v1.5.0 h1:uc58n3b/n/F2yDMJzHMbXORkJSh3fzO4/+jju6eR7Zg=
 cloud.google.com/go/longrunning v0.1.1 h1:y50CXG4j0+qvEukslYFBCrzaXX0qpFbBzc3PchSu/LE=
+cloud.google.com/go/longrunning v0.1.1/go.mod h1:UUFxuDWkv22EuY93jjmDMFT5GPQKeFVJBIF6QlTqdsE=
 cloud.google.com/go/monitoring v1.7.0 h1:zK12aN/jzLRzrRXopEOUaG6kvuF6WJmlv1mXn1L7a6E=
 cloud.google.com/go/monitoring v1.7.0/go.mod h1:HpYse6kkGo//7p6sT0wsIC6IBDET0RhIsnmlA53dvEk=
 cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
@@ -60,6 +63,8 @@ cloud.google.com/go/pubsub v1.2.0/go.mod h1:jhfEVHT8odbXTkndysNHCcx0awwzvfOlguIA
 cloud.google.com/go/pubsub v1.3.1/go.mod h1:i+ucay31+CNRpDW4Lu78I4xXG+O1r/MAHgjpRVR+TSU=
 cloud.google.com/go/pubsub v1.25.1 h1:l0wCNZKuEp2Q54wAy8283EV9O57+7biWOXnnU2/Tq/A=
 cloud.google.com/go/pubsub v1.25.1/go.mod h1:bY6l7rF8kCcwz6V3RaQ6kK4p5g7qc7PqjRoE9wDOqOU=
+cloud.google.com/go/redis v1.10.0 h1:/zTwwBKIAD2DEWTrXZp8WD9yD/gntReF/HkPssVYd0U=
+cloud.google.com/go/redis v1.10.0/go.mod h1:ThJf3mMBQtW18JzGgh41/Wld6vnDDc/F/F35UolRZPM=
 cloud.google.com/go/storage v1.0.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiyrjsg+URw=
 cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0ZeosJ0Rtdos=
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
@@ -2391,8 +2396,8 @@ google.golang.org/api v0.57.0/go.mod h1:dVPlbZyBo2/OjBpmvNdpn2GRm6rPy75jyU7bmhdr
 google.golang.org/api v0.59.0/go.mod h1:sT2boj7M9YJxZzgeZqXogmhfmRWDtPzT31xkieUbuZU=
 google.golang.org/api v0.61.0/go.mod h1:xQRti5UdCmoCEqFxcz93fTl338AVqDgyaDRuOZ3hg9I=
 google.golang.org/api v0.62.0/go.mod h1:dKmwPCydfsad4qCH08MSdgWjfHOyfpd4VtDGgRFdavw=
-google.golang.org/api v0.100.0 h1:LGUYIrbW9pzYQQ8NWXlaIVkgnfubVBZbMFb9P8TK374=
-google.golang.org/api v0.100.0/go.mod h1:ZE3Z2+ZOr87Rx7dqFsdRQkRBk36kDtp/h+QpHbB7a70=
+google.golang.org/api v0.102.0 h1:JxJl2qQ85fRMPNvlZY/enexbxpCjLwGhZUtgfGeQ51I=
+google.golang.org/api v0.102.0/go.mod h1:3VFl6/fzoA+qNuS1N1/VfXY4LjoXN/wzeIp7TweWwGo=
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.2.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=

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

@@ -8,6 +8,7 @@ import (
 	"github.com/elastic/beats/v7/x-pack/metricbeat/module/gcp"
 	"github.com/elastic/beats/v7/x-pack/metricbeat/module/gcp/metrics/cloudsql"
 	"github.com/elastic/beats/v7/x-pack/metricbeat/module/gcp/metrics/compute"
+	"github.com/elastic/beats/v7/x-pack/metricbeat/module/gcp/metrics/redis"
 )
 
 // NewMetadataServiceForConfig returns a service to fetch metadata from a config struct. It must return the Compute
@@ -18,6 +19,8 @@ func NewMetadataServiceForConfig(c config, serviceName string) (gcp.MetadataServ
 		return compute.NewMetadataService(c.ProjectID, c.Zone, c.Region, c.Regions, c.opt...)
 	case gcp.ServiceCloudSQL:
 		return cloudsql.NewMetadataService(c.ProjectID, c.Zone, c.Region, c.Regions, c.opt...)
+	case gcp.ServiceRedis:
+		return redis.NewMetadataService(c.ProjectID, c.Zone, c.Region, c.Regions, c.opt...)
 	default:
 		return nil, nil
 	}

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

@@ -0,0 +1,211 @@
+// 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 redis
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	monitoringpb "cloud.google.com/go/monitoring/apiv3/v2/monitoringpb"
+	redis "cloud.google.com/go/redis/apiv1"
+	"cloud.google.com/go/redis/apiv1/redispb"
+	"google.golang.org/api/iterator"
+	"google.golang.org/api/option"
+
+	"github.com/elastic/beats/v7/x-pack/metricbeat/module/gcp"
+	"github.com/elastic/elastic-agent-libs/logp"
+)
+
+// NewMetadataService returns the specific Metadata service for a GCP Redis resource
+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,
+		instances: make(map[string]*redispb.Instance),
+		logger:    logp.NewLogger("metrics-redis"),
+	}, nil
+}
+
+// redisMetadata 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 redisMetadata struct {
+	region       string
+	instanceID   string
+	instanceName string
+	machineType  string
+
+	User     map[string]string
+	Metadata map[string]string
+	Metrics  interface{}
+	System   interface{}
+}
+
+type metadataCollector struct {
+	projectID string
+	zone      string
+	region    string
+	regions   []string
+	opt       []option.ClientOption
+	// NOTE: instances holds data used for all metrics collected in a given period
+	// this avoids calling the remote endpoint for each metric, which would take a long time overall
+	instances map[string]*redispb.Instance
+	logger    *logp.Logger
+}
+
+func (s *metadataCollector) ID(ctx context.Context, in *gcp.MetadataCollectorInputData) (string, error) {
+	metadata, err := s.Metadata(ctx, in.TimeSeries)
+	if err != nil {
+		return "", err
+	}
+
+	metadata.ECS.Update(metadata.Labels)
+	if in.Timestamp != nil {
+		_, _ = metadata.ECS.Put("timestamp", in.Timestamp)
+	} else if in.Point != nil {
+		_, _ = metadata.ECS.Put("timestamp", in.Point.Interval.EndTime)
+	} else {
+		return "", fmt.Errorf("no timestamp information found")
+	}
+
+	return metadata.ECS.String(), nil
+}
+
+// Metadata implements googlecloud.MetadataCollector to the known set of labels from a Redis TimeSeries single point of data.
+func (s *metadataCollector) Metadata(ctx context.Context, resp *monitoringpb.TimeSeries) (gcp.MetadataCollectorData, error) {
+	metadata, err := s.instanceMetadata(ctx, s.instanceID(resp), s.instanceRegion(resp))
+	if err != nil {
+		return gcp.MetadataCollectorData{}, err
+	}
+
+	stackdriverLabels := gcp.NewStackdriverMetadataServiceForTimeSeries(resp)
+
+	metadataCollectorData, err := stackdriverLabels.Metadata(ctx, resp)
+	if err != nil {
+		return gcp.MetadataCollectorData{}, err
+	}
+
+	if resp.Resource != nil && resp.Resource.Labels != nil {
+		_, _ = metadataCollectorData.ECS.Put(gcp.ECSCloudInstanceIDKey, resp.Resource.Labels[gcp.TimeSeriesResponsePathForECSInstanceID])
+	}
+
+	_, _ = metadataCollectorData.ECS.Put(gcp.ECSCloudInstanceNameKey, metadata.instanceName)
+
+	if metadata.machineType != "" {
+		lastIndex := strings.LastIndex(metadata.machineType, "/")
+		_, _ = metadataCollectorData.ECS.Put(gcp.ECSCloudMachineTypeKey, metadata.machineType[lastIndex+1:])
+	}
+
+	metadata.Metrics = metadataCollectorData.Labels[gcp.LabelMetrics]
+	metadata.System = metadataCollectorData.Labels[gcp.LabelSystem]
+
+	if metadata.User != nil {
+		metadataCollectorData.Labels[gcp.LabelUser] = metadata.User
+	}
+
+	return metadataCollectorData, nil
+}
+
+// instanceMetadata returns the labels of an instance
+func (s *metadataCollector) instanceMetadata(ctx context.Context, instanceID, region string) (*redisMetadata, error) {
+	instance, err := s.instance(ctx, instanceID)
+	if err != nil {
+		return nil, fmt.Errorf("error trying to get data from instance '%s' %w", instanceID, err)
+	}
+
+	metadata := &redisMetadata{
+		instanceID: instanceID,
+		region:     region,
+	}
+
+	if instance == nil {
+		s.logger.Debugf("couldn't get instance '%s' call ListInstances API", instanceID)
+		return metadata, nil
+	}
+
+	if instance.Name != "" {
+		parts := strings.Split(instance.Name, "/")
+		metadata.instanceName = parts[len(parts)-1]
+	}
+
+	if instance.Labels != nil {
+		metadata.User = instance.Labels
+	}
+
+	if instance.Tier.String() != "" {
+		metadata.machineType = instance.Tier.String()
+	}
+
+	return metadata, nil
+}
+
+// instance returns data from an instance ID using the cache or making a request
+func (s *metadataCollector) instance(ctx context.Context, instanceID string) (*redispb.Instance, error) {
+	s.getInstances(ctx)
+
+	instance, ok := s.instances[instanceID]
+	if ok {
+		return instance, nil
+	}
+
+	s.instances = make(map[string]*redispb.Instance)
+
+	return nil, nil
+}
+
+func (s *metadataCollector) instanceID(ts *monitoringpb.TimeSeries) string {
+	if ts.Resource != nil && ts.Resource.Labels != nil {
+		return ts.Resource.Labels[gcp.TimeSeriesResponsePathForECSInstanceID]
+	}
+
+	return ""
+}
+
+func (s *metadataCollector) instanceRegion(ts *monitoringpb.TimeSeries) string {
+	if ts.Resource != nil && ts.Resource.Labels != nil {
+		return ts.Resource.Labels["region"]
+	}
+
+	return ""
+}
+
+func (s *metadataCollector) getInstances(ctx context.Context) {
+	if len(s.instances) > 0 {
+		return
+	}
+
+	s.logger.Debug("get redis instances with ListInstances API")
+
+	client, err := redis.NewCloudRedisClient(ctx, s.opt...)
+	if err != nil {
+		s.logger.Errorf("error getting client from redis service: %v", err)
+		return
+	}
+
+	defer client.Close()
+
+	// Use locations - (wildcard) to fetch all instances.
+	// https://pkg.go.dev/cloud.google.com/go/redis@v1.10.0/apiv1#CloudRedisClient.ListInstances
+	it := client.ListInstances(ctx, &redispb.ListInstancesRequest{
+		Parent: fmt.Sprintf("projects/%s/locations/-", s.projectID),
+	})
+	for {
+		instance, err := it.Next()
+		if errors.Is(err, iterator.Done) {
+			break
+		}
+
+		if err != nil {
+			s.logger.Errorf("redis ListInstances error: %v", err)
+			break
+		}
+
+		s.instances[instance.Name] = instance
+	}
+}

x-pack/metricbeat/module/gcp/metrics/redis/metadata_test.go

@@ -0,0 +1,78 @@
+// 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 redis
+
+import (
+	"testing"
+
+	monitoring "cloud.google.com/go/monitoring/apiv3/v2/monitoringpb"
+	"github.com/golang/protobuf/ptypes/timestamp"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/genproto/googleapis/api/metric"
+	"google.golang.org/genproto/googleapis/api/monitoredres"
+)
+
+var fake = &monitoring.TimeSeries{
+	Resource: &monitoredres.MonitoredResource{
+		Type: "gce_instance",
+		Labels: map[string]string{
+			"instance_id": "4624337448093162893",
+			"project_id":  "elastic-metricbeat",
+			"region":      "us-central1",
+		},
+	},
+	Metadata: &monitoredres.MonitoredResourceMetadata{
+		UserLabels: map[string]string{
+			"user": "label",
+		},
+	},
+	Metric: &metric.Metric{
+		Labels: map[string]string{
+			"instance_name": "instance-1",
+		},
+		Type: "redis.googleapis.com/instance/cpu/usage_time",
+	},
+	MetricKind: metric.MetricDescriptor_GAUGE,
+	ValueType:  metric.MetricDescriptor_DOUBLE,
+	Points: []*monitoring.Point{{
+		Value: &monitoring.TypedValue{
+			Value: &monitoring.TypedValue_DoubleValue{DoubleValue: 0.0041224284852319215},
+		},
+		Interval: &monitoring.TimeInterval{
+			StartTime: &timestamp.Timestamp{
+				Seconds: 1569932700,
+			},
+			EndTime: &timestamp.Timestamp{
+				Seconds: 1569932700,
+			},
+		},
+	}, {
+		Value: &monitoring.TypedValue{
+			Value: &monitoring.TypedValue_DoubleValue{DoubleValue: 0.004205757571772513},
+		},
+		Interval: &monitoring.TimeInterval{
+			StartTime: &timestamp.Timestamp{
+				Seconds: 1569932640,
+			},
+			EndTime: &timestamp.Timestamp{
+				Seconds: 1569932640,
+			},
+		},
+	}},
+}
+
+var m = &metadataCollector{
+	projectID: "projectID",
+}
+
+func TestInstanceID(t *testing.T) {
+	instanceID := m.instanceID(fake)
+	assert.Equal(t, "4624337448093162893", instanceID)
+}
+
+func TestInstanceRegion(t *testing.T) {
+	zone := m.instanceRegion(fake)
+	assert.Equal(t, "us-central1", zone)
+}