Create new grpc storage configuration to align with OTEL

cmd/remote-storage/app/server.go

@@ -22,6 +22,7 @@ import (
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
+	_ "google.golang.org/grpc/encoding"
 	"google.golang.org/grpc/reflection"
 
 	"github.com/jaegertracing/jaeger/cmd/query/app/querysvc"

go.mod

@@ -151,7 +151,7 @@ require (
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
-	github.com/mostynb/go-grpc-compression v1.2.2 // indirect
+	github.com/mostynb/go-grpc-compression v1.2.2
 	github.com/oklog/run v1.1.0 // indirect
 	github.com/onsi/ginkgo v1.16.5 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opencensusexporter v0.97.0 // indirect
@@ -205,9 +205,9 @@ require (
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/collector v0.97.0 // indirect
 	go.opentelemetry.io/collector/config/configauth v0.97.0 // indirect
-	go.opentelemetry.io/collector/config/configcompression v1.4.0 // indirect
+	go.opentelemetry.io/collector/config/configcompression v1.4.0
 	go.opentelemetry.io/collector/config/confignet v0.97.0 // indirect
-	go.opentelemetry.io/collector/config/configopaque v1.4.0 // indirect
+	go.opentelemetry.io/collector/config/configopaque v1.4.0
 	go.opentelemetry.io/collector/config/configtelemetry v0.97.0 // indirect
 	go.opentelemetry.io/collector/config/internal v0.97.0 // indirect
 	go.opentelemetry.io/collector/confmap/converter/expandconverter v0.97.0 // indirect

plugin/storage/grpc/config/config.go

@@ -23,12 +23,16 @@
 
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-plugin"
+	"github.com/mostynb/go-grpc-compression/nonclobbering/snappy"
+	"github.com/mostynb/go-grpc-compression/nonclobbering/zstd"
+	"go.opentelemetry.io/collector/config/configcompression"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/encoding/gzip"
 
 	"github.com/jaegertracing/jaeger/pkg/config/tlscfg"
 	"github.com/jaegertracing/jaeger/pkg/tenancy"
@@ -51,6 +55,12 @@
 	pluginHealthCheckDone chan bool
 	pluginRPCClient       plugin.ClientProtocol
 	remoteConn            *grpc.ClientConn
+
+	Compression     configcompression.Type `yaml:"compression" mapstructure:"compression"`
+	ReadBufferSize  int                    `yaml:"read_buffer_size" mapstructure:"read_buffer_size"`
+	WriteBufferSize int                    `yaml:"write_buffer_size" mapstructure:"write_buffer_size"`
+	Authority       string                 `yaml:"authority" mapstructure:"authority"`
+	WaitForReady    bool                   `yaml:"wait_for_ready" mapstructure:"wait_for_ready"`
 }
 
 // ClientPluginServices defines services plugin can expose and its capabilities
@@ -90,7 +100,28 @@
 	opts := []grpc.DialOption{
 		grpc.WithStatsHandler(otelgrpc.NewClientHandler(otelgrpc.WithTracerProvider(tracerProvider))),
 		grpc.WithBlock(),
+		grpc.WithDefaultCallOptions(grpc.WaitForReady(c.WaitForReady)),
+	}
+
+	if c.Compression.IsCompressed() {
+		compressionType, err := getGRPCCompressionName(c.Compression)
+		if err != nil {
+			return nil, err
+		}
+		opts = append(opts, grpc.WithDefaultCallOptions(grpc.UseCompressor(compressionType)))
+	}
+
+	if c.ReadBufferSize > 0 {
+		opts = append(opts, grpc.WithReadBufferSize(c.ReadBufferSize))
 	}
+	if c.WriteBufferSize > 0 {
+		opts = append(opts, grpc.WithWriteBufferSize(c.WriteBufferSize))
+	}
+
+	if c.Authority != "" {
+		opts = append(opts, grpc.WithAuthority(c.Authority))
+	}
+
 	if c.RemoteTLS.Enabled {
 		tlsCfg, err := c.RemoteTLS.Config(logger)
 		if err != nil {
@@ -223,3 +254,16 @@
 
 	return c.pluginRPCClient.Ping()
 }
+
+func getGRPCCompressionName(compressionType configcompression.Type) (string, error) {
+	switch compressionType {
+	case configcompression.TypeGzip:
+		return gzip.Name, nil
+	case configcompression.TypeSnappy:
+		return snappy.Name, nil
+	case configcompression.TypeZstd:
+		return zstd.Name, nil
+	default:
+		return "", fmt.Errorf("unsupported compression type %q", compressionType)
+	}
+}