Merge pull request #2202 from yarpc/ronakj/release

Preparing release v1.70.0

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [1.70.0] - 2023-02-16
+### Fixed
+- grpc: apply compression only on the outbounds with compression enabled.
+
 ## [1.69.1] - 2023-1-24
 ### Changed
 -- yarpcerrors: export logic to get server and client fault type
@@ -1462,6 +1466,7 @@ This release requires regeneration of ThriftRW code.
 ## 0.1.0 - 2016-08-31
 
 - Initial release.
+[1.70.0]: https://github.com/yarpc/yarpc-go/compare/v1.69.1...v1.70.0
 [1.69.1]: https://github.com/yarpc/yarpc-go/compare/v1.69.0...v1.69.1
 [1.69.0]: https://github.com/yarpc/yarpc-go/compare/v1.68.0...v1.69.0
 [1.68.0]: https://github.com/yarpc/yarpc-go/compare/v1.67.0...v1.68.0

transport/grpc/config.go

@@ -192,7 +192,6 @@ func (c OutboundConfig) dialOptions(kit *yarpcconfig.Kit, tlsConfigProvider yarp
 	if err != nil {
 		return nil, err
 	}
-	opts = append(opts, Compressor(kit.Compressor(c.Compressor)))
 
 	keepaliveOpts, err := c.Keepalive.dialOptions()
 	if err != nil {
@@ -375,8 +374,7 @@ func (t *transportSpec) buildOutbound(outboundConfig *OutboundConfig, tr transpo
 		return nil, newTransportCastError(tr)
 	}
 
-	outboundOpts := newOutboundOptions(t.OutboundOptions)
-	dialOpts, err := outboundConfig.dialOptions(kit, outboundOpts.tlsConfigProvider)
+	dialOpts, err := outboundConfig.dialOptions(kit, newOutboundOptions(t.OutboundOptions).tlsConfigProvider)
 	if err != nil {
 		return nil, err
 	}
@@ -397,7 +395,9 @@ func (t *transportSpec) buildOutbound(outboundConfig *OutboundConfig, tr transpo
 		}
 	}
 
-	return trans.NewOutbound(chooser, t.OutboundOptions...), nil
+	outboundOpts := []OutboundOption{OutboundCompressor(kit.Compressor(outboundConfig.Compressor))}
+	outboundOpts = append(outboundOpts, t.OutboundOptions...)
+	return trans.NewOutbound(chooser, outboundOpts...), nil
 }
 
 func newTransportCastError(tr transport.Transport) error {

transport/grpc/integration_test.go

@@ -445,6 +445,56 @@ func TestTLSWithYARPCAndGRPC(t *testing.T) {
 	}
 }
 
+// TestCompressionWithMultipleOutbounds creates multiple outbound for the
+// same hostport where one outbound has compression enabled.
+// Validates compression is applied for the outbound with compression enabled
+// and rest of the outbounds are still uncompressed.
+func TestCompressionWithMultipleOutbounds(t *testing.T) {
+	env, err := newTestEnv(t, nil, nil, nil, nil)
+	require.NoError(t, err)
+	defer func() { assert.NoError(t, env.Close()) }()
+
+	chooser := peer.NewSingle(hostport.Identify(env.Inbound.Addr().String()), env.Transport.NewDialer())
+	compressedOutbound := env.Transport.NewOutbound(chooser, OutboundCompressor(_goodCompressor))
+	require.NoError(t, compressedOutbound.Start())
+	defer compressedOutbound.Stop()
+
+	caller := "example-client"
+	service := "example"
+	clientConfig := clientconfig.MultiOutbound(
+		caller,
+		service,
+		transport.Outbounds{
+			ServiceName: caller,
+			Unary:       compressedOutbound,
+		},
+	)
+	compressedClient := examplepb.NewKeyValueYARPCClient(clientConfig)
+
+	ctx, cancel := context.WithTimeout(context.Background(), testtime.Second*5)
+	defer cancel()
+
+	// Send request over uncompressed outbound and assert compression metric
+	// is empty.
+	_metrics.reset()
+	require.NoError(t, env.SetValueYARPC(ctx, "foo", strings.Repeat("a", 32*1024)))
+	assert.Equal(t, &metricCollection{metrics: []metric{}}, _metrics)
+
+	// Send request over compressed outbound and assert compression metric
+	// is seen.
+	_metrics.reset()
+	_, err = compressedClient.SetValue(ctx, &examplepb.SetValueRequest{Key: "foo", Value: strings.Repeat("a", 32*1024)})
+	require.NoError(t, err)
+	wantMetric := []metric{
+		{32777, map[string]string{"stage": "compress"}},
+		{32777, map[string]string{"stage": "decompress"}},
+		{0, map[string]string{"stage": "compress"}},
+	}
+	assert.Equal(t, newMetrics(wantMetric, map[string]string{
+		"compressor": _goodCompressor.name,
+	}), _metrics)
+}
+
 func TestGRPCHeaderListSize(t *testing.T) {
 	tests := []struct {
 		desc       string

transport/grpc/options.go

@@ -212,6 +212,16 @@ func OutboundTLSConfigProvider(provider yarpctls.OutboundTLSConfigProvider) Outb
 	}
 }
 
+// OutboundCompressor returns an OutboundOption that applies compressorion
+// for requests in the outbound.
+func OutboundCompressor(compressor transport.Compressor) OutboundOption {
+	return func(outboundOptions *outboundOptions) {
+		if compressor != nil {
+			outboundOptions.compressor = compressor.Name()
+		}
+	}
+}
+
 // DialOption is an option that influences grpc.Dial.
 type DialOption func(*dialOptions)
 
@@ -323,6 +333,7 @@ func newInboundOptions(options []InboundOption) *inboundOptions {
 }
 
 type outboundOptions struct {
+	compressor        string
 	tlsConfigProvider yarpctls.OutboundTLSConfigProvider
 }
 

transport/grpc/outbound.go

@@ -178,6 +178,9 @@ func (o *Outbound) invoke(
 	if responseMD != nil {
 		callOptions = []grpc.CallOption{grpc.Trailer(responseMD)}
 	}
+	if o.options.compressor != "" {
+		callOptions = append(callOptions, grpc.UseCompressor(o.options.compressor))
+	}
 	apiPeer, onFinish, err := o.peerChooser.Choose(ctx, request)
 	if err != nil {
 		return err

transport/http/outbound.go

@@ -75,7 +75,7 @@ func URLTemplate(template string) OutboundOption {
 // AddHeader specifies that an HTTP outbound should always include the given
 // header in outgoung requests.
 //
-// 	httpTransport.NewOutbound(chooser, http.AddHeader("X-Token", "TOKEN"))
+//	httpTransport.NewOutbound(chooser, http.AddHeader("X-Token", "TOKEN"))
 //
 // Note that headers starting with "Rpc-" are reserved by YARPC. This function
 // will panic if the header starts with "Rpc-".
@@ -135,7 +135,12 @@ func (t *Transport) NewOutbound(chooser peer.Chooser, opts ...OutboundOption) *O
 	client := t.client
 	if o.tlsConfig != nil {
 		client = createTLSClient(o)
-		o.urlTemplate.Scheme = "https"
+		// Create a copy of the url template to avoid scheme changes impacting
+		// other outbounds as the base url template is shared across http
+		// outbounds.
+		ut := *o.urlTemplate
+		ut.Scheme = "https"
+		o.urlTemplate = &ut
 	}
 	o.client = client
 	o.sender = &transportSender{Client: client}
@@ -537,15 +542,15 @@ func checkServiceMatch(reqSvcName string, resHeaders http.Header) (bool, string)
 //
 // Sample usage:
 //
-//  client := http.Client{Transport: outbound}
+//	client := http.Client{Transport: outbound}
 //
 // Thereafter use the Golang standard library HTTP to send requests with this client.
 //
-//  ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-//  defer cancel()
-//  req, err := http.NewRequest("GET", "http://example.com/", nil /* body */)
-//  req = req.WithContext(ctx)
-//  res, err := client.Do(req)
+//	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+//	defer cancel()
+//	req, err := http.NewRequest("GET", "http://example.com/", nil /* body */)
+//	req = req.WithContext(ctx)
+//	res, err := client.Do(req)
 //
 // All requests must have a deadline on the context.
 // The peer chooser for raw HTTP requests will receive a YARPC transport.Request with no body.

transport/http/outbound_test.go

@@ -23,6 +23,7 @@ package http
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
 	"errors"
 	"io/ioutil"
 	"net/http"
@@ -837,3 +838,12 @@ func TestCallOneWayResponseCloseError(t *testing.T) {
 	})
 	require.Errorf(t, err, "Received unexpected error:code:internal message:test error")
 }
+
+func TestIsolatedSchemaChange(t *testing.T) {
+	tr := &Transport{client: &http.Client{Transport: http.DefaultTransport}}
+	plainOutbound := tr.NewOutbound(nil)
+	tlsOutbound := tr.NewOutbound(nil, OutboundTLSConfiguration(&tls.Config{}))
+	assert.NotEqual(t, plainOutbound.urlTemplate, tlsOutbound.urlTemplate)
+	assert.Equal(t, "http", plainOutbound.urlTemplate.Scheme)
+	assert.Equal(t, "https", tlsOutbound.urlTemplate.Scheme)
+}

version.go

@@ -21,4 +21,4 @@
 package yarpc // import "go.uber.org/yarpc"
 
 // Version is the current version of YARPC.
-const Version = "1.69.1"
+const Version = "1.70.0"