Repository: GoogleCloudPlatform/opentelemetry-operations-go
Pull request #576 (2023-01-26 19:41:17)

Add gzip compression option to metrics and logs
Fixes https://github.com/GoogleCloudPlatform/opentelemetry-operations-go/issues/544

This copies GMP's [compression setting](https://github.com/GoogleCloudPlatform/prometheus-engine/blob/30ce1d152c6ecfcde8e57b4553dadc895b233708/pkg/operator/apis/monitoring/v1/types.go#L115) by exposing a string option for users to specify the type of compression to enable for gRPC calls. Currently the only supported option is `gzip`.

This only works for logs and metrics, trying to wire it up for Cloud Trace gave this error:
```
2023/01/26 19:31:25 failed to export to Google Cloud Trace: rpc error: code = InvalidArgument desc = Error skipping unknown field of a message with tag:31, type:google.devtools.cloudtrace.v2.BatchWriteSpansRequest
error details: name = DebugInfo detail = Was unable to extract resource containers for Chemist from the incoming request. Error message during extraction: Error skipping unknown field of a message with tag:31, type:google.devtools.cloudtrace.v2.BatchWriteSpansRequest stack =
2023-01-26T19:31:25.980Z	info	exporterhelper/queued_retry.go:426	Exporting failed. Will retry the request after interval.	{"kind": "exporter", "data_type": "traces", "name": "googlecloud", "error": "rpc error: code = InvalidArgument desc = Error skipping unknown field of a message with tag:31, type:google.devtools.cloudtrace.v2.BatchWriteSpansRequest\nerror details: name = DebugInfo detail = Was unable to extract resource containers for Chemist from the incoming request. Error message during extraction: Error skipping unknown field of a message with tag:31, type:google.devtools.cloudtrace.v2.BatchWriteSpansRequest stack =", "interval": "6.722764325s"}
2023/01/26 19:31:32 failed to export to Google Cloud Trace: rpc error: code = InvalidArgument desc = Error skipping unknown field of a message with tag:31, type:google.devtools.cloudtrace.v2.BatchWriteSpansRequest
error details: name = DebugInfo detail = Was unable to extract resource containers for Chemist from the incoming request. Error message during extraction: Error skipping unknown field of a message with tag:31, type:google.devtools.cloudtrace.v2.BatchWriteSpansRequest stack =
```

codecov[bot]: # [Codecov](https://codecov.io/gh/GoogleCloudPlatform/opentelemetry-operations-go/pull/576?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) Report
> Merging [#576](https://codecov.io/gh/GoogleCloudPlatform/opentelemetry-operations-go/pull/576?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (040e366) into [main](https://codecov.io/gh/GoogleCloudPlatform/opentelemetry-operations-go/commit/9c0f81b80da79dce32aaff35814e21e8d9711bac?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (9c0f81b) will **decrease** coverage by `0.59%`.
> The diff coverage is `0.00%`.

```diff
@@            Coverage Diff             @@
##             main     #576      +/-   ##
==========================================
- Coverage   64.68%   64.09%   -0.59%     
==========================================
  Files          38       38              
  Lines        4281     4314      +33     
==========================================
- Hits         2769     2765       -4     
- Misses       1399     1432      +33     
- Partials      113      117       +4     
```


| [Impacted Files](https://codecov.io/gh/GoogleCloudPlatform/opentelemetry-operations-go/pull/576?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) | Coverage Δ | |
|---|---|---|
| [exporter/collector/config.go](https://codecov.io/gh/GoogleCloudPlatform/opentelemetry-operations-go/pull/576?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-ZXhwb3J0ZXIvY29sbGVjdG9yL2NvbmZpZy5nbw==) | `43.47% <0.00%> (-3.70%)` | :arrow_down: |
| [exporter/collector/logs.go](https://codecov.io/gh/GoogleCloudPlatform/opentelemetry-operations-go/pull/576?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-ZXhwb3J0ZXIvY29sbGVjdG9yL2xvZ3MuZ28=) | `40.70% <0.00%> (-0.53%)` | :arrow_down: |
| [exporter/collector/metrics.go](https://codecov.io/gh/GoogleCloudPlatform/opentelemetry-operations-go/pull/576?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-ZXhwb3J0ZXIvY29sbGVjdG9yL21ldHJpY3MuZ28=) | `67.07% <0.00%> (-0.61%)` | :arrow_down: |
| [exporter/metric/metric.go](https://codecov.io/gh/GoogleCloudPlatform/opentelemetry-operations-go/pull/576?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-ZXhwb3J0ZXIvbWV0cmljL21ldHJpYy5nbw==) | `71.07% <0.00%> (-1.32%)` | :arrow_down: |
| [exporter/metric/option.go](https://codecov.io/gh/GoogleCloudPlatform/opentelemetry-operations-go/pull/576?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-ZXhwb3J0ZXIvbWV0cmljL29wdGlvbi5nbw==) | `64.70% <0.00%> (-8.63%)` | :arrow_down: |
| [...lector/internal/datapointstorage/datapointcache.go](https://codecov.io/gh/GoogleCloudPlatform/opentelemetry-operations-go/pull/576?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-ZXhwb3J0ZXIvY29sbGVjdG9yL2ludGVybmFsL2RhdGFwb2ludHN0b3JhZ2UvZGF0YXBvaW50Y2FjaGUuZ28=) | `80.30% <0.00%> (-3.04%)` | :arrow_down: |

:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform)

damemi: Trying to add tests to this, my goal is to send gzip-compressed metrics/logs to our mock servers and have the mock return an error (because the mock doesn't have the gzip compressor registered). Then, ideally a 2nd test with a mock server that has gzip decoding enabled would pass.

However it looks like the gzip compressor is registered automatically by importing the `gzip` package https://stackoverflow.com/a/49507459. Calling [`encoding.GetCompressor("gzip")`](https://pkg.go.dev/google.golang.org/grpc/encoding#GetCompressor) in the mock confirms this.

So, I haven't found a way to get the error to trigger. I did add a test that sets compression and expects no error, so that is something at least.



Patch:
diff --git a/exporter/collector/README.md b/exporter/collector/README.md
index c025d67..99bcb6c 100644
--- a/exporter/collector/README.md
+++ b/exporter/collector/README.md
@@ -123,6 +123,7 @@ The following configuration options are supported:
 - `user_agent` (optional): Override the user agent string sent on requests to Cloud Monitoring (currently only applies to metrics). Specify `{{version}}` to include the application version number. Defaults to `opentelemetry-collector-contrib {{version}}`.
 - `use_insecure` (optional): If true. use gRPC as their communication transport. Only has effect if Endpoint is not "".
 - `timeout` (optional): Timeout for all API calls. If not set, defaults to 12 seconds.
+- `compression` (optional): Enable gzip compression on gRPC calls for Metrics or Logs. Valid values: `gzip`.
 - `resource_mappings` (optional): ResourceMapping defines mapping of resources from source (OpenCensus) to target (Google Cloud).
   - `label_mappings` (optional): Optional flag signals whether we can proceed with transformation if a label is missing in the resource.
 - `retry_on_failure` (optional): Configuration for how to handle retries when sending data to Google Cloud fails.
@@ -185,6 +186,7 @@ exporters:
     metric:
       prefix: prefix
       skip_create_descriptor: true
+      compression: gzip
 
     log:
       default_log_name: my-app
diff --git a/exporter/collector/config.go b/exporter/collector/config.go
index fbda5ed..9b94232 100644
--- a/exporter/collector/config.go
+++ b/exporter/collector/config.go
@@ -31,6 +31,7 @@ import (
 	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/encoding/gzip"
 )
 
 const (
@@ -62,6 +63,9 @@ type ClientConfig struct {
 	GetClientOptions func() []option.ClientOption
 
 	Endpoint string `mapstructure:"endpoint"`
+	// Compression specifies the compression format for Metrics and Logging gRPC requests.
+	// Supported values: gzip.
+	Compression string `mapstructure:"compression"`
 	// Only has effect if Endpoint is not ""
 	UseInsecure bool `mapstructure:"use_insecure"`
 	// GRPCPoolSize sets the size of the connection pool in the GCP client
@@ -211,6 +215,17 @@ func ValidateConfig(cfg Config) error {
 			return fmt.Errorf("unable to parse resource filter regex: %s", err.Error())
 		}
 	}
+
+	if len(cfg.LogConfig.ClientConfig.Compression) > 0 && cfg.LogConfig.ClientConfig.Compression != gzip.Name {
+		return fmt.Errorf("unknown compression option '%s', allowed values: '', 'gzip'", cfg.LogConfig.ClientConfig.Compression)
+	}
+	if len(cfg.MetricConfig.ClientConfig.Compression) > 0 && cfg.MetricConfig.ClientConfig.Compression != gzip.Name {
+		return fmt.Errorf("unknown compression option '%s', allowed values: '', 'gzip'", cfg.MetricConfig.ClientConfig.Compression)
+	}
+	if len(cfg.TraceConfig.ClientConfig.Compression) > 0 {
+		return fmt.Errorf("traces.compression invalid: compression is only available for logs and metrics")
+	}
+
 	return nil
 }
 
diff --git a/exporter/collector/go.mod b/exporter/collector/go.mod
index edd5028..642ad51 100644
--- a/exporter/collector/go.mod
+++ b/exporter/collector/go.mod
@@ -10,6 +10,7 @@ require (
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.34.2
 	github.com/census-instrumentation/opencensus-proto v0.4.1
 	github.com/google/go-cmp v0.5.9
+	github.com/googleapis/gax-go/v2 v2.7.0
 	github.com/stretchr/testify v1.8.1
 	go.opencensus.io v0.24.0
 	go.opentelemetry.io/collector/pdata v1.0.0-rc2
@@ -40,7 +41,6 @@ require (
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.2.1 // indirect
-	github.com/googleapis/gax-go/v2 v2.7.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/kr/pretty v0.3.0 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
diff --git a/exporter/collector/logs.go b/exporter/collector/logs.go
index 156b4ea..d98b7b3 100644
--- a/exporter/collector/logs.go
+++ b/exporter/collector/logs.go
@@ -29,8 +29,12 @@ import (
 	loggingv2 "cloud.google.com/go/logging/apiv2"
 	logpb "cloud.google.com/go/logging/apiv2/loggingpb"
 	"google.golang.org/genproto/googleapis/api/monitoredres"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/encoding/gzip"
 	"google.golang.org/protobuf/proto"
 
+	"github.com/googleapis/gax-go/v2"
+
 	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping"
 
 	"go.opentelemetry.io/collector/pdata/pcommon"
@@ -140,6 +144,11 @@ func NewGoogleCloudLogsExporter(
 		return nil, err
 	}
 
+	if cfg.LogConfig.ClientConfig.Compression == gzip.Name {
+		loggingClient.CallOptions.WriteLogEntries = append(loggingClient.CallOptions.WriteLogEntries,
+			gax.WithGRPCOptions(grpc.UseCompressor(gzip.Name)))
+	}
+
 	obs := selfObservability{
 		log: log,
 	}
diff --git a/exporter/collector/metrics.go b/exporter/collector/metrics.go
index dcae3c4..ea2c39c 100644
--- a/exporter/collector/metrics.go
+++ b/exporter/collector/metrics.go
@@ -42,11 +42,15 @@ import (
 	"google.golang.org/genproto/googleapis/api/label"
 	metricpb "google.golang.org/genproto/googleapis/api/metric"
 	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/encoding/gzip"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
+	"github.com/googleapis/gax-go/v2"
+
 	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector/internal/datapointstorage"
 	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/collector/internal/normalization"
 	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping"
@@ -145,6 +149,16 @@ func NewGoogleCloudMetricsExporter(
 	if err != nil {
 		return nil, err
 	}
+
+	if cfg.MetricConfig.ClientConfig.Compression == gzip.Name {
+		client.CallOptions.CreateMetricDescriptor = append(client.CallOptions.CreateMetricDescriptor,
+			gax.WithGRPCOptions(grpc.UseCompressor(gzip.Name)))
+		client.CallOptions.CreateTimeSeries = append(client.CallOptions.CreateTimeSeries,
+			gax.WithGRPCOptions(grpc.UseCompressor(gzip.Name)))
+		client.CallOptions.CreateServiceTimeSeries = append(client.CallOptions.CreateServiceTimeSeries,
+			gax.WithGRPCOptions(grpc.UseCompressor(gzip.Name)))
+	}
+
 	obs := selfObservability{log: log}
 	shutdown := make(chan struct{})
 	normalizer := normalization.NewDisabledNormalizer()
diff --git a/exporter/metric/go.mod b/exporter/metric/go.mod
index 7e88ee6..eea75eb 100644
--- a/exporter/metric/go.mod
+++ b/exporter/metric/go.mod
@@ -21,6 +21,7 @@ require (
 require (
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.34.2
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.34.2
+	github.com/googleapis/gax-go/v2 v2.7.0
 	github.com/stretchr/testify v1.8.1
 	go.uber.org/multierr v1.8.0
 )
@@ -37,7 +38,6 @@ require (
 	github.com/golang/protobuf v1.5.2 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.2.1 // indirect
-	github.com/googleapis/gax-go/v2 v2.7.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/otel/trace v1.11.2 // indirect
diff --git a/exporter/metric/metric.go b/exporter/metric/metric.go
index 107171b..ef3aaaa 100644
--- a/exporter/metric/metric.go
+++ b/exporter/metric/metric.go
@@ -40,9 +40,13 @@ import (
 	"google.golang.org/genproto/googleapis/api/label"
 	googlemetricpb "google.golang.org/genproto/googleapis/api/metric"
 	monitoredrespb "google.golang.org/genproto/googleapis/api/monitoredres"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/encoding/gzip"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
+	"github.com/googleapis/gax-go/v2"
+
 	"github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping"
 )
 
@@ -104,6 +108,15 @@ func newMetricExporter(o *options) (*metricExporter, error) {
 		return nil, err
 	}
 
+	if o.compression == "gzip" {
+		client.CallOptions.GetMetricDescriptor = append(client.CallOptions.CreateMetricDescriptor,
+			gax.WithGRPCOptions(grpc.UseCompressor(gzip.Name)))
+		client.CallOptions.CreateMetricDescriptor = append(client.CallOptions.CreateMetricDescriptor,
+			gax.WithGRPCOptions(grpc.UseCompressor(gzip.Name)))
+		client.CallOptions.CreateTimeSeries = append(client.CallOptions.CreateTimeSeries,
+			gax.WithGRPCOptions(grpc.UseCompressor(gzip.Name)))
+	}
+
 	cache := map[key]*googlemetricpb.MetricDescriptor{}
 	e := &metricExporter{
 		o:        o,
diff --git a/exporter/metric/option.go b/exporter/metric/option.go
index a4c5311..fab442f 100644
--- a/exporter/metric/option.go
+++ b/exporter/metric/option.go
@@ -56,6 +56,8 @@ type options struct {
 	// resource if the resource does not inherently belong to a specific
 	// project, e.g. on-premise resource like k8s_container or generic_task.
 	projectID string
+	// compression enables gzip compression on gRPC calls.
+	compression string
 	// monitoringClientOptions are additional options to be passed
 	// to the underlying Stackdriver Monitoring API client.
 	// Optional.
@@ -138,3 +140,10 @@ func WithDisableCreateMetricDescriptors() func(o *options) {
 		o.disableCreateMetricDescriptors = true
 	}
 }
+
+// WithCompression sets the compression to use for gRPC requests.
+func WithCompression(c string) func(o *options) {
+	return func(o *options) {
+		o.compression = c
+	}
+}


Test patch:
diff --git a/exporter/collector/integrationtest/testcases/testcases_metrics.go b/exporter/collector/integrationtest/testcases/testcases_metrics.go
index 31f3abd..d4bab82 100644
--- a/exporter/collector/integrationtest/testcases/testcases_metrics.go
+++ b/exporter/collector/integrationtest/testcases/testcases_metrics.go
@@ -218,5 +218,13 @@ var MetricsTestCases = []TestCase{
 		ExpectFixturePath:    "testdata/fixtures/metrics/prometheus_empty_buckets_expected.json",
 		SkipForSDK:           true,
 	},
+	{
+		Name:                 "Basic Counter with gzip compression",
+		OTLPInputFixturePath: "testdata/fixtures/metrics/basic_counter_metrics.json",
+		ExpectFixturePath:    "testdata/fixtures/metrics/basic_counter_metrics_expect.json",
+		ConfigureCollector: func(cfg *collector.Config) {
+			cfg.MetricConfig.ClientConfig.Compression = "gzip"
+		},
+	},
 	// TODO: Add integration tests for workload.googleapis.com metrics from the ops agent
 }