Commit ⋅ yarpc/yarpc-go

package tchannel

import (
	"errors"
	"fmt"
	"time"

	"go.uber.org/yarpc/api/peer"
	"go.uber.org/yarpc/api/transport"
	yarpctls "go.uber.org/yarpc/api/transport/tls"
	"go.uber.org/yarpc/peer/hostport"

@@ -75,8 +77,53 @@

// 	  myservice:
// 	    tchannel:
// 	      peer: 127.0.0.1:4040
//
type OutboundConfig struct {
	yarpcconfig.PeerChooser
	// TLS config enables TLS outbound.
	//
	//  tchannel:
	//    peer: 127.0.0.1:4040
	//    tls:
	//      mode: enforced
	//      spiffe-ids:
	//        - destination-id
	TLS OutboundTLSConfig `config:"tls"`
}

// OutboundTLSConfig configures TLS for a TChannel outbound.
type OutboundTLSConfig struct {
	// Mode when set to Enforced enables TLS outbound and
	// outbound TLS configuration provider will be used for fetching tls.Config.
	Mode yarpctls.Mode `config:"mode,interpolate"`
	// SpiffeIDs is a list of the accepted server spiffe IDs.
	SpiffeIDs []string `config:"spiffe-ids"`
}

// getPeerTransport returns peer transport to be used in peer chooser creation.
func (c OutboundConfig) getPeerTransport(transport *Transport, destName string) (peer.Transport, error) {
	if c.TLS.Mode == yarpctls.Disabled {
		return transport, nil
	}

	if c.TLS.Mode == yarpctls.Permissive {
		return nil, errors.New("outbound does not support permissive TLS mode")
	}

	if transport.outboundTLSConfigProvider == nil {
		return nil, errors.New("outbound TLS enforced but outbound TLS config provider is nil")
	}

	if len(c.TLS.SpiffeIDs) == 0 {
		return nil, errors.New("outbound TLS enforced but no spiffe id is provided")
	}

	config, err := transport.outboundTLSConfigProvider.ClientTLSConfig(c.TLS.SpiffeIDs)
	if err != nil {
		return nil, err
	}

	return transport.CreateTLSOutboundChannel(config, destName)
}

// TransportSpec returns a TransportSpec for the TChannel unary transport.

@@ -166,7 +213,11 @@


func (ts *transportSpec) buildUnaryOutbound(oc *OutboundConfig, t transport.Transport, k *yarpcconfig.Kit) (transport.UnaryOutbound, error) {
	x := t.(*Transport)
	chooser, err := oc.BuildPeerChooser(x, hostport.Identify, k)
	pt, err := oc.getPeerTransport(x, k.OutboundServiceName())
	if err != nil {
		return nil, err
	}
	chooser, err := oc.BuildPeerChooser(pt, hostport.Identify, k)
	if err != nil {
		return nil, err
	}

@@ -104,9 +104,7 @@


// Call sends an RPC to this specific peer.
func (p *tchannelPeer) Call(ctx context.Context, req *transport.Request) (*transport.Response, error) {
	root := p.transport.ch.RootPeers()
	tp := root.GetOrAdd(p.HostPort())
	return callWithPeer(ctx, req, tp, p.transport.headerCase)
	return callWithPeer(ctx, req, p.getPeer(), p.transport.headerCase)
}

// callWithPeer sends a request with the chosen peer.

@@ -0,0 +1,77 @@

1	+	// Copyright (c) 2022 Uber Technologies, Inc.
2	+	//
3	+	// Permission is hereby granted, free of charge, to any person obtaining a copy
4	+	// of this software and associated documentation files (the "Software"), to deal
5	+	// in the Software without restriction, including without limitation the rights
6	+	// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
7	+	// copies of the Software, and to permit persons to whom the Software is
8	+	// furnished to do so, subject to the following conditions:
9	+	//
10	+	// The above copyright notice and this permission notice shall be included in
11	+	// all copies or substantial portions of the Software.
12	+	//
13	+	// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
14	+	// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15	+	// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16	+	// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17	+	// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
18	+	// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
19	+	// THE SOFTWARE.
20	+
21	+	package tchannel
22	+
23	+	import (
24	+	"net"
25	+
26	+	"github.com/uber/tchannel-go"
27	+	"go.uber.org/yarpc/api/peer"
28	+	"golang.org/x/net/context"
29	+	)
30	+
31	+	var _ peer.Transport = (*outboundChannel)(nil)
32	+
33	+	type dialerFunc = func(ctx context.Context, network, hostPort string) (net.Conn, error)
34	+
35	+	// outboundChannel holds a TChannel channel for outbound use only with custom
36	+	// dialer. This is used for creating TLS connections with different TLS config
37	+	// per outbound.
38	+	type outboundChannel struct {
39	+	t *Transport
40	+	dialer dialerFunc
41	+	ch *tchannel.Channel
42	+	}
43	+
44	+	// newOutboundChannel returns outbound channel with given transport and dialer.
45	+	func newOutboundChannel(t Transport, dialer dialerFunc) outboundChannel {
46	+	return &outboundChannel{
47	+	t: t,
48	+	dialer: dialer,
49	+	}
50	+	}
51	+
52	+	// RetainPeer delegates to the transport RetainPeer along with its channel.
53	+	func (o *outboundChannel) RetainPeer(pid peer.Identifier, sub peer.Subscriber) (peer.Peer, error) {
54	+	return o.t.retainPeer(pid, sub, o.ch)
55	+	}
56	+
57	+	// ReleasePeer delegates to the transport ReleasePeer.
58	+	func (o *outboundChannel) ReleasePeer(pid peer.Identifier, sub peer.Subscriber) error {
59	+	return o.t.ReleasePeer(pid, sub)
60	+	}
61	+
62	+	// start creates channel used for managing outbound peers.
63	+	// This is invoked by the transport when it is started.
64	+	func (o *outboundChannel) start() (err error) {
65	+	o.ch, err = tchannel.NewChannel(o.t.name, &tchannel.ChannelOptions{
66	+	Dialer: o.dialer,
67	+	OnPeerStatusChanged: o.t.onPeerStatusChanged,
68	+	Tracer: o.t.tracer,
69	+	})
70	+	return err
71	+	}
72	+
73	+	// stop closes the outbound channel stopping outbound connections.
74	+	// This is invoked by the transport when it is stopping.
75	+	func (o *outboundChannel) stop() {
76	+	o.ch.Close()
77	+	}

@@ -289,7 +289,7 @@

	}

	b.needTransport(spec)
	cv, err := spec.OnewayOutbound.Decode(attrs, config.InterpolateWith(b.kit.resolver))
	cv, err := spec.OnewayOutbound.Decode(attrs, config.InterpolateWith(b.kit.resolver), mapdecode.DecodeHook(tlsModeDecodeHook))
	if err != nil {
		return fmt.Errorf("failed to decode oneway outbound configuration: %v", err)
	}

@@ -68,6 +68,7 @@

	excludeServiceHeaderInResponse bool
	inboundTLSConfig               *tls.Config
	inboundTLSMode                 *yarpctls.Mode
	outboundTLSConfigProvider      yarpctls.OutboundTLSConfigProvider
}

// newTransportOptions constructs the default transport options struct

@@ -112,6 +113,14 @@

	}
}

// OutboundTLSConfigProvider returns a TransportOption that provides the
// outbound TLS config provider.
func OutboundTLSConfigProvider(provider yarpctls.OutboundTLSConfigProvider) TransportOption {
	return func(transportOptions *transportOptions) {
		transportOptions.outboundTLSConfigProvider = provider
	}
}

// WithChannel specifies the TChannel Channel to use to send and receive YARPC
// requests. The instance may already have handlers registered against it;
// these will be left unchanged.

@@ -37,6 +37,7 @@

	"go.uber.org/yarpc/api/transport"
	yarpctls "go.uber.org/yarpc/api/transport/tls"
	"go.uber.org/yarpc/pkg/lifecycle"
	"go.uber.org/yarpc/transport/internal/tls/dialer"
	"go.uber.org/yarpc/transport/internal/tls/muxlistener"
	"go.uber.org/zap"
)

@@ -81,6 +82,9 @@


	inboundTLSConfig *tls.Config
	inboundTLSMode   *yarpctls.Mode

	outboundTLSConfigProvider yarpctls.OutboundTLSConfigProvider
	outboundChannels          []*outboundChannel
}

// NewTransport is a YARPC transport that facilitates sending and receiving

@@ -131,6 +135,7 @@

		excludeServiceHeaderInResponse: o.excludeServiceHeaderInResponse,
		inboundTLSConfig:               o.inboundTLSConfig,
		inboundTLSMode:                 o.inboundTLSMode,
		outboundTLSConfigProvider:      o.outboundTLSConfigProvider,
	}
}


@@ -142,22 +147,26 @@

// RetainPeer adds a peer subscriber (typically a peer chooser) and causes the
// transport to maintain persistent connections with that peer.
func (t *Transport) RetainPeer(pid peer.Identifier, sub peer.Subscriber) (peer.Peer, error) {
	return t.retainPeer(pid, sub, t.ch)
}

func (t *Transport) retainPeer(pid peer.Identifier, sub peer.Subscriber, ch *tchannel.Channel) (peer.Peer, error) {
	t.lock.Lock()
	defer t.lock.Unlock()

	p := t.getOrCreatePeer(pid)
	p := t.getOrCreatePeer(pid, ch)
	p.Subscribe(sub)
	return p, nil
}

// **NOTE** should only be called while the lock write mutex is acquired
func (t *Transport) getOrCreatePeer(pid peer.Identifier) *tchannelPeer {
func (t *Transport) getOrCreatePeer(pid peer.Identifier, ch *tchannel.Channel) *tchannelPeer {
	addr := pid.Identifier()
	if p, ok := t.peers[addr]; ok {
		return p
	}

	p := newPeer(addr, t)
	p := newPeer(addr, t, ch)
	t.peers[addr] = p
	// Start a peer connection loop
	t.connectorsGroup.Add(1)

@@ -193,17 +202,6 @@

	return nil
}

func (t *Transport) peerList() *tchannel.RootPeerList {
	t.lock.Lock()
	defer t.lock.Unlock()

	if t.ch == nil {
		return nil
	}

	return t.ch.RootPeers()
}

// Start starts the TChannel transport. This starts making connections and
// accepting inbound requests. All inbounds must have been assigned a router
// to accept inbound requests before this is called.

@@ -289,6 +287,11 @@

	}
	t.addr = t.ch.PeerInfo().HostPort

	for _, outboundChannel := range t.outboundChannels {
		if err := outboundChannel.start(); err != nil {
			return err
		}
	}
	return nil
}


@@ -302,6 +305,9 @@


func (t *Transport) stop() error {
	t.ch.Close()
	for _, outboundChannel := range t.outboundChannels {
		outboundChannel.stop()
	}
	t.connectorsGroup.Wait()
	return nil
}

@@ -323,3 +329,34 @@

323	329		}
324	330		p.notifyConnectionStatusChanged()
325	331		}
	332	+
	333	+	// CreateTLSOutboundChannel creates a outbound channel for managing tls
	334	+	// connections with the given tls config and destination name.
	335	+	// Usage:
	336	+	// tr, _ := tchannel.NewTransport(...)
	337	+	// outboundCh, _ := tr.CreateTLSOutboundChannel(tls-config, "dest-name")
	338	+	// outbound := tr.NewOutbound(peer.NewSingle(id, outboundCh))
	339	+	func (t Transport) CreateTLSOutboundChannel(tlsConfig tls.Config, destinationName string) (peer.Transport, error) {
	340	+	params := dialer.Params{
	341	+	Config: tlsConfig,
	342	+	Meter: t.meter,
	343	+	Logger: t.logger,
	344	+	ServiceName: t.name,
	345	+	TransportName: TransportName,
	346	+	Dest: destinationName,
	347	+	Dialer: t.dialer,
	348	+	}
	349	+	return t.createOutboundChannel(dialer.NewTLSDialer(params).DialContext)
	350	+	}
	351	+
	352	+	func (t *Transport) createOutboundChannel(dialerFunc dialerFunc) (peer.Transport, error) {
	353	+	t.lock.Lock()
	354	+	defer t.lock.Unlock()
	355	+
	356	+	if t.once.State() != lifecycle.Idle {
	357	+	return nil, errors.New("tchannel outbound channel cannot be created after starting transport")
	358	+	}
	359	+	outboundChannel := newOutboundChannel(t, dialerFunc)
	360	+	t.outboundChannels = append(t.outboundChannels, outboundChannel)
	361	+	return outboundChannel, nil
	362	+	}

@@ -24,6 +24,7 @@

	"context"
	"time"

	"github.com/uber/tchannel-go"
	"go.uber.org/yarpc/api/peer"
	"go.uber.org/yarpc/peer/abstractpeer"
	"go.uber.org/zap"

@@ -33,13 +34,14 @@

	*abstractpeer.Peer

	transport *Transport
	ch        *tchannel.Channel
	addr      string
	changed   chan struct{}
	released  chan struct{}
	timer     *time.Timer
}

func newPeer(addr string, t *Transport) *tchannelPeer {
func newPeer(addr string, t *Transport, ch *tchannel.Channel) *tchannelPeer {
	// Create a defused timer for later use.
	timer := time.NewTimer(0)
	if !timer.Stop() {

@@ -48,6 +50,7 @@


	return &tchannelPeer{
		addr:      addr,
		ch:        ch,
		Peer:      abstractpeer.NewPeer(abstractpeer.PeerIdentifier(addr), t),
		transport: t,
		changed:   make(chan struct{}, 1),

@@ -64,7 +67,7 @@


	// Wait for start (so we can be certain that we have a channel).
	<-p.transport.once.Started()
	pl := p.transport.peerList()
	pl := p.getRootPeers()
	if pl == nil {
		return
	}

@@ -165,6 +168,22 @@

	return false
}

func (p *tchannelPeer) getPeer() *tchannel.Peer {
	return p.getRootPeers().GetOrAdd(p.HostPort())
}

func (p *tchannelPeer) getRootPeers() *tchannel.RootPeerList {
	ch := p.ch
	if ch == nil {
		if p.transport.ch == nil {
			return nil
		}
		ch = p.transport.ch
	}

	return ch.RootPeers()
}

// StartRequest and EndRequest are no-ops now.
// They previously aggregated pending request count from all subscibed peer
// lists and distributed change notifications.

Files		•	•	•	Coverage
api	642	624	3	15	97.20%
compressor	98	91	3	4	92.86%
encoding	1,911	1,476	79	356	77.24%
internal	5,042	3,999	117	926	79.31%
peer	2,383	2,081	53	249	87.33%
pkg	169	161	2	6	95.27%
transport	3,457	3,120	124	213	90.25%
x/debug	39	36	1	2	92.31%
yarpcconfig	862	800	17	45	92.81%
yarpcerrors	192	190	1	1	98.96%
call.go	38	38	0	0	100.00%
config.go	36	29	2	5	80.56%
dispatcher.go	212	204	3	5	96.23%
dispatcher_introspection.go	49	42	3	4	85.71%
dispatcher_startup.go	112	102	5	5	91.07%
errors.go	6	6	0	0	100.00%
header.go	2	2	0	0	100.00%
inject.go	58	56	1	1	96.55%
middleware.go	12	12	0	0	100.00%
router.go	75	72	1	2	96.00%
serialize/serialize.go	68	52	8	8	76.47%
Project Totals (270 files)	15,463	13,193	423	1,847	85.32%

View CI Build Download

codecov:
  branch: dev
coverage:
  range: 50...90
  status:
    project:
      default:
        enabled: yes
        target: 80%
    patch:
      default:
        enabled: yes
        target: 80%
# This list is generated using scripts/generate-cover-ignore.sh
# All packages with .nocover files in them MUST be listed here
ignore:
 - /api/transport/tls/
 - /encoding/protobuf/internal/testpb/
 - /encoding/protobuf/internal/testpb/v2/
 - /encoding/thrift/internal/
 - /encoding/thrift/internal/observabilitytest/test/
 - /encoding/thrift/internal/observabilitytest/test/testserviceclient/
 - /encoding/thrift/internal/observabilitytest/test/testservicefx/
 - /encoding/thrift/internal/observabilitytest/test/testserviceserver/
 - /encoding/thrift/internal/observabilitytest/test/testservicetest/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/NOSERVICES/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/readonlystoreclient/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/readonlystorefx/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/readonlystoreserver/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/readonlystoretest/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/storeclient/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/storefx/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/storeserver/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/storetest/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/baseserviceclient/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/baseservicefx/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/baseserviceserver/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/baseservicetest/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/emptyserviceclient/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/emptyservicefx/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/emptyserviceserver/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/emptyservicetest/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendemptyclient/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendemptyfx/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendemptyserver/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendemptytest/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendonlyclient/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendonlyfx/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendonlyserver/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendonlytest/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/barclient/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/barfx/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/barserver/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/bartest/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/fooclient/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/foofx/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/fooserver/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/footest/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/nameclient/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/namefx/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/nameserver/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/nametest/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/weather/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/weather/weatherclient/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/weather/weatherfx/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/weather/weatherserver/
 - /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/weather/weathertest/
 - /internal/cover/
 - /internal/crossdock/crossdockpb/
 - /internal/crossdock/thrift/echo/
 - /internal/crossdock/thrift/echo/echoclient/
 - /internal/crossdock/thrift/echo/echofx/
 - /internal/crossdock/thrift/echo/echoserver/
 - /internal/crossdock/thrift/echo/echotest/
 - /internal/crossdock/thrift/gauntlet/
 - /internal/crossdock/thrift/gauntlet/secondserviceclient/
 - /internal/crossdock/thrift/gauntlet/secondservicefx/
 - /internal/crossdock/thrift/gauntlet/secondserviceserver/
 - /internal/crossdock/thrift/gauntlet/secondservicetest/
 - /internal/crossdock/thrift/gauntlet/thrifttestclient/
 - /internal/crossdock/thrift/gauntlet/thrifttestfx/
 - /internal/crossdock/thrift/gauntlet/thrifttestserver/
 - /internal/crossdock/thrift/gauntlet/thrifttesttest/
 - /internal/crossdock/thrift/gen-go/echo/
 - /internal/crossdock/thrift/gen-go/gauntlet_apache/
 - /internal/crossdock/thrift/gen-go/gauntlet_tchannel/
 - /internal/crossdock/thrift/oneway/
 - /internal/crossdock/thrift/oneway/onewayclient/
 - /internal/crossdock/thrift/oneway/onewayfx/
 - /internal/crossdock/thrift/oneway/onewayserver/
 - /internal/crossdock/thrift/oneway/onewaytest/
 - /internal/examples/json-keyvalue/client/
 - /internal/examples/json-keyvalue/server/
 - /internal/examples/protobuf/
 - /internal/examples/protobuf/example/
 - /internal/examples/protobuf/examplepb/
 - /internal/examples/protobuf/exampleutil/
 - /internal/examples/streaming/
 - /internal/examples/streaming/client/
 - /internal/examples/streaming/server/
 - /internal/examples/thrift-hello/hello/
 - /internal/examples/thrift-hello/hello/echo/
 - /internal/examples/thrift-hello/hello/echo/helloclient/
 - /internal/examples/thrift-hello/hello/echo/hellofx/
 - /internal/examples/thrift-hello/hello/echo/helloserver/
 - /internal/examples/thrift-hello/hello/echo/hellotest/
 - /internal/examples/thrift-keyvalue/keyvalue/client/
 - /internal/examples/thrift-keyvalue/keyvalue/kv/
 - /internal/examples/thrift-keyvalue/keyvalue/kv/keyvalueclient/
 - /internal/examples/thrift-keyvalue/keyvalue/kv/keyvaluefx/
 - /internal/examples/thrift-keyvalue/keyvalue/kv/keyvalueserver/
 - /internal/examples/thrift-keyvalue/keyvalue/kv/keyvaluetest/
 - /internal/examples/thrift-keyvalue/keyvalue/server/
 - /internal/examples/thrift-oneway/
 - /internal/examples/thrift-oneway/sink/
 - /internal/examples/thrift-oneway/sink/helloclient/
 - /internal/examples/thrift-oneway/sink/hellofx/
 - /internal/examples/thrift-oneway/sink/helloserver/
 - /internal/examples/thrift-oneway/sink/hellotest/
 - /internal/prototest/example/
 - /internal/prototest/examplepb/
 - /internal/prototest/exampleutil/
 - /internal/service-test/
 - /internal/testutils/
 - /serialize/internal/

Read the docs

Interact now

The inner-most circle is the entire project, moving away from the center are folders then, finally, a single file. The size and color of each slice is representing the number of statements and the coverage, respectively.

SVG

The top section represents the entire project. Proceeding with folders and finally individual files. The size and color of each slice is representing the number of statements and the coverage, respectively.

SVG

Each block represents a single file in the project. The size and color of each block is represented by the number of statements and the coverage, respectively.

SVG

Learn more about the coverage sunburst graph

166	213
167	214		func (ts transportSpec) buildUnaryOutbound(oc OutboundConfig, t transport.Transport, k *yarpcconfig.Kit) (transport.UnaryOutbound, error) {
168	215		x := t.(*Transport)
169		-	chooser, err := oc.BuildPeerChooser(x, hostport.Identify, k)
	216	+	pt, err := oc.getPeerTransport(x, k.OutboundServiceName())
	217	+	if err != nil {
	218	+	return nil, err
	219	+	}
	220	+	chooser, err := oc.BuildPeerChooser(pt, hostport.Identify, k)
170	221		if err != nil {
171	222		return nil, err
172	223		}

104	104
105	105		// Call sends an RPC to this specific peer.
106	106		func (p tchannelPeer) Call(ctx context.Context, req transport.Request) (*transport.Response, error) {
107		-	root := p.transport.ch.RootPeers()
108		-	tp := root.GetOrAdd(p.HostPort())
109		-	return callWithPeer(ctx, req, tp, p.transport.headerCase)
	107	+	return callWithPeer(ctx, req, p.getPeer(), p.transport.headerCase)
110	108		}
111	109
112	110		// callWithPeer sends a request with the chosen peer.

289	289		}
290	290
291	291		b.needTransport(spec)
292		-	cv, err := spec.OnewayOutbound.Decode(attrs, config.InterpolateWith(b.kit.resolver))
	292	+	cv, err := spec.OnewayOutbound.Decode(attrs, config.InterpolateWith(b.kit.resolver), mapdecode.DecodeHook(tlsModeDecodeHook))
293	293		if err != nil {
294	294		return fmt.Errorf("failed to decode oneway outbound configuration: %v", err)
295	295		}

131	135		excludeServiceHeaderInResponse: o.excludeServiceHeaderInResponse,
132	136		inboundTLSConfig: o.inboundTLSConfig,
133	137		inboundTLSMode: o.inboundTLSMode,
	138	+	outboundTLSConfigProvider: o.outboundTLSConfigProvider,
134	139		}
135	140		}
136	141

142	147		// RetainPeer adds a peer subscriber (typically a peer chooser) and causes the
143	148		// transport to maintain persistent connections with that peer.
144	149		func (t *Transport) RetainPeer(pid peer.Identifier, sub peer.Subscriber) (peer.Peer, error) {
	150	+	return t.retainPeer(pid, sub, t.ch)
	151	+	}
	152	+
	153	+	func (t Transport) retainPeer(pid peer.Identifier, sub peer.Subscriber, ch tchannel.Channel) (peer.Peer, error) {
145	154		t.lock.Lock()
146	155		defer t.lock.Unlock()
147	156
148		-	p := t.getOrCreatePeer(pid)
	157	+	p := t.getOrCreatePeer(pid, ch)
149	158		p.Subscribe(sub)
150	159		return p, nil
151	160		}
152	161
153	162		// NOTE should only be called while the lock write mutex is acquired
154		-	func (t Transport) getOrCreatePeer(pid peer.Identifier) tchannelPeer {
	163	+	func (t Transport) getOrCreatePeer(pid peer.Identifier, ch tchannel.Channel) *tchannelPeer {
155	164		addr := pid.Identifier()
156	165		if p, ok := t.peers[addr]; ok {
157	166		return p
158	167		}
159	168
160		-	p := newPeer(addr, t)
	169	+	p := newPeer(addr, t, ch)
161	170		t.peers[addr] = p
162	171		// Start a peer connection loop
163	172		t.connectorsGroup.Add(1)

21	21		package tchannel
22	22
23	23		import (
	24	+	"errors"
24	25		"fmt"
25	26		"time"
26	27
	28	+	"go.uber.org/yarpc/api/peer"
27	29		"go.uber.org/yarpc/api/transport"
28	30		yarpctls "go.uber.org/yarpc/api/transport/tls"
29	31		"go.uber.org/yarpc/peer/hostport"

75	77		// myservice:
76	78		// tchannel:
77	79		// peer: 127.0.0.1:4040
	80	+	//
78	81		type OutboundConfig struct {
79	82		yarpcconfig.PeerChooser
	83	+	// TLS config enables TLS outbound.
	84	+	//
	85	+	// tchannel:
	86	+	// peer: 127.0.0.1:4040
	87	+	// tls:
	88	+	// mode: enforced
	89	+	// spiffe-ids:
	90	+	// - destination-id
	91	+	TLS OutboundTLSConfig `config:"tls"`
	92	+	}
	93	+
	94	+	// OutboundTLSConfig configures TLS for a TChannel outbound.
	95	+	type OutboundTLSConfig struct {
	96	+	// Mode when set to Enforced enables TLS outbound and
	97	+	// outbound TLS configuration provider will be used for fetching tls.Config.
	98	+	Mode yarpctls.Mode `config:"mode,interpolate"`
	99	+	// SpiffeIDs is a list of the accepted server spiffe IDs.
	100	+	SpiffeIDs []string `config:"spiffe-ids"`
	101	+	}
	102	+
	103	+	// getPeerTransport returns peer transport to be used in peer chooser creation.
	104	+	func (c OutboundConfig) getPeerTransport(transport *Transport, destName string) (peer.Transport, error) {
	105	+	if c.TLS.Mode == yarpctls.Disabled {
	106	+	return transport, nil
	107	+	}
	108	+
	109	+	if c.TLS.Mode == yarpctls.Permissive {
	110	+	return nil, errors.New("outbound does not support permissive TLS mode")
	111	+	}
	112	+
	113	+	if transport.outboundTLSConfigProvider == nil {
	114	+	return nil, errors.New("outbound TLS enforced but outbound TLS config provider is nil")
	115	+	}
	116	+
	117	+	if len(c.TLS.SpiffeIDs) == 0 {
	118	+	return nil, errors.New("outbound TLS enforced but no spiffe id is provided")
	119	+	}
	120	+
	121	+	config, err := transport.outboundTLSConfigProvider.ClientTLSConfig(c.TLS.SpiffeIDs)
	122	+	if err != nil {
	123	+	return nil, err
	124	+	}
	125	+
	126	+	return transport.CreateTLSOutboundChannel(config, destName)
80	127		}
81	128
82	129		// TransportSpec returns a TransportSpec for the TChannel unary transport.

68	68		excludeServiceHeaderInResponse bool
69	69		inboundTLSConfig *tls.Config
70	70		inboundTLSMode *yarpctls.Mode
	71	+	outboundTLSConfigProvider yarpctls.OutboundTLSConfigProvider
71	72		}
72	73
73	74		// newTransportOptions constructs the default transport options struct

112	113		}
113	114		}
114	115
	116	+	// OutboundTLSConfigProvider returns a TransportOption that provides the
	117	+	// outbound TLS config provider.
	118	+	func OutboundTLSConfigProvider(provider yarpctls.OutboundTLSConfigProvider) TransportOption {
	119	+	return func(transportOptions *transportOptions) {
	120	+	transportOptions.outboundTLSConfigProvider = provider
	121	+	}
	122	+	}
	123	+
115	124		// WithChannel specifies the TChannel Channel to use to send and receive YARPC
116	125		// requests. The instance may already have handlers registered against it;
117	126		// these will be left unchanged.

37	37		"go.uber.org/yarpc/api/transport"
38	38		yarpctls "go.uber.org/yarpc/api/transport/tls"
39	39		"go.uber.org/yarpc/pkg/lifecycle"
	40	+	"go.uber.org/yarpc/transport/internal/tls/dialer"
40	41		"go.uber.org/yarpc/transport/internal/tls/muxlistener"
41	42		"go.uber.org/zap"
42	43		)

193	202		return nil
194	203		}
195	204
196		-	func (t Transport) peerList() tchannel.RootPeerList {
197		-	t.lock.Lock()
198		-	defer t.lock.Unlock()
199		-
200		-	if t.ch == nil {
201		-	return nil
202		-	}
203		-
204		-	return t.ch.RootPeers()
205		-	}
206		-
207	205		// Start starts the TChannel transport. This starts making connections and
208	206		// accepting inbound requests. All inbounds must have been assigned a router
209	207		// to accept inbound requests before this is called.

289	287		}
290	288		t.addr = t.ch.PeerInfo().HostPort
291	289
	290	+	for _, outboundChannel := range t.outboundChannels {
	291	+	if err := outboundChannel.start(); err != nil {
	292	+	return err
	293	+	}
	294	+	}
292	295		return nil
293	296		}
294	297

302	305
303	306		func (t *Transport) stop() error {
304	307		t.ch.Close()
	308	+	for _, outboundChannel := range t.outboundChannels {
	309	+	outboundChannel.stop()
	310	+	}
305	311		t.connectorsGroup.Wait()
306	312		return nil
307	313		}

24	24		"context"
25	25		"time"
26	26
	27	+	"github.com/uber/tchannel-go"
27	28		"go.uber.org/yarpc/api/peer"
28	29		"go.uber.org/yarpc/peer/abstractpeer"
29	30		"go.uber.org/zap"

33	34		*abstractpeer.Peer
34	35
35	36		transport *Transport
	37	+	ch *tchannel.Channel
36	38		addr string
37	39		changed chan struct{}
38	40		released chan struct{}
39	41		timer *time.Timer
40	42		}
41	43
42		-	func newPeer(addr string, t Transport) tchannelPeer {
	44	+	func newPeer(addr string, t Transport, ch tchannel.Channel) *tchannelPeer {
43	45		// Create a defused timer for later use.
44	46		timer := time.NewTimer(0)
45	47		if !timer.Stop() {

Navigation	Overlay
`t` Navigate files	`h` Toggle hits
`y` Change url to tip of branch	`m` Toggle misses
`b / v` Jump to prev/next hit line	`p` Toggle partial
`z / x` Jump to prev/next missed or partial line	`1..9` Toggle flags
`shift + o` Open current page in GitHub	`a` Toggle all on
`/ or ?` Show keyboard shortcuts dialog	`c` Toggle context lines or commits

48	50
49	51		return &tchannelPeer{
50	52		addr: addr,
	53	+	ch: ch,
51	54		Peer: abstractpeer.NewPeer(abstractpeer.PeerIdentifier(addr), t),
52	55		transport: t,
53	56		changed: make(chan struct{}, 1),

64	67
65	68		// Wait for start (so we can be certain that we have a channel).
66	69		<-p.transport.once.Started()
67		-	pl := p.transport.peerList()
	70	+	pl := p.getRootPeers()
68	71		if pl == nil {
69	72		return
70	73		}

165	168		return false
166	169		}
167	170
	171	+	func (p tchannelPeer) getPeer() tchannel.Peer {
	172	+	return p.getRootPeers().GetOrAdd(p.HostPort())
	173	+	}
	174	+
	175	+	func (p tchannelPeer) getRootPeers() tchannel.RootPeerList {
	176	+	ch := p.ch
	177	+	if ch == nil {
	178	+	if p.transport.ch == nil {
	179	+	return nil
	180	+	}
	181	+	ch = p.transport.ch
	182	+	}
	183	+
	184	+	return ch.RootPeers()
	185	+	}
	186	+
168	187		// StartRequest and EndRequest are no-ops now.
169	188		// They previously aggregated pending request count from all subscibed peer
170	189		// lists and distributed change notifications.

1	codecov:
2	branch: dev
3	coverage:
4	range: 50...90
5	status:
6	project:
7	default:
8	enabled: yes
9	target: 80%
10	patch:
11	default:
12	enabled: yes
13	target: 80%
14	# This list is generated using scripts/generate-cover-ignore.sh
15	# All packages with .nocover files in them MUST be listed here
16	ignore:
17	- /api/transport/tls/
18	- /encoding/protobuf/internal/testpb/
19	- /encoding/protobuf/internal/testpb/v2/
20	- /encoding/thrift/internal/
21	- /encoding/thrift/internal/observabilitytest/test/
22	- /encoding/thrift/internal/observabilitytest/test/testserviceclient/
23	- /encoding/thrift/internal/observabilitytest/test/testservicefx/
24	- /encoding/thrift/internal/observabilitytest/test/testserviceserver/
25	- /encoding/thrift/internal/observabilitytest/test/testservicetest/
26	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/NOSERVICES/
27	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/
28	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/readonlystoreclient/
29	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/readonlystorefx/
30	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/readonlystoreserver/
31	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/readonlystoretest/
32	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/storeclient/
33	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/storefx/
34	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/storeserver/
35	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/atomic/storetest/
36	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/
37	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/baseserviceclient/
38	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/baseservicefx/
39	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/baseserviceserver/
40	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/baseservicetest/
41	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/emptyserviceclient/
42	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/emptyservicefx/
43	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/emptyserviceserver/
44	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/emptyservicetest/
45	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendemptyclient/
46	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendemptyfx/
47	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendemptyserver/
48	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendemptytest/
49	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendonlyclient/
50	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendonlyfx/
51	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendonlyserver/
52	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/common/extendonlytest/
53	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/
54	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/barclient/
55	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/barfx/
56	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/barserver/
57	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/bartest/
58	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/fooclient/
59	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/foofx/
60	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/fooserver/
61	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/footest/
62	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/nameclient/
63	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/namefx/
64	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/nameserver/
65	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/extends/nametest/
66	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/weather/
67	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/weather/weatherclient/
68	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/weather/weatherfx/
69	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/weather/weatherserver/
70	- /encoding/thrift/thriftrw-plugin-yarpc/internal/tests/weather/weathertest/
71	- /internal/cover/
72	- /internal/crossdock/crossdockpb/
73	- /internal/crossdock/thrift/echo/
74	- /internal/crossdock/thrift/echo/echoclient/
75	- /internal/crossdock/thrift/echo/echofx/
76	- /internal/crossdock/thrift/echo/echoserver/
77	- /internal/crossdock/thrift/echo/echotest/
78	- /internal/crossdock/thrift/gauntlet/
79	- /internal/crossdock/thrift/gauntlet/secondserviceclient/
80	- /internal/crossdock/thrift/gauntlet/secondservicefx/
81	- /internal/crossdock/thrift/gauntlet/secondserviceserver/
82	- /internal/crossdock/thrift/gauntlet/secondservicetest/
83	- /internal/crossdock/thrift/gauntlet/thrifttestclient/
84	- /internal/crossdock/thrift/gauntlet/thrifttestfx/
85	- /internal/crossdock/thrift/gauntlet/thrifttestserver/
86	- /internal/crossdock/thrift/gauntlet/thrifttesttest/
87	- /internal/crossdock/thrift/gen-go/echo/
88	- /internal/crossdock/thrift/gen-go/gauntlet_apache/
89	- /internal/crossdock/thrift/gen-go/gauntlet_tchannel/
90	- /internal/crossdock/thrift/oneway/
91	- /internal/crossdock/thrift/oneway/onewayclient/
92	- /internal/crossdock/thrift/oneway/onewayfx/
93	- /internal/crossdock/thrift/oneway/onewayserver/
94	- /internal/crossdock/thrift/oneway/onewaytest/
95	- /internal/examples/json-keyvalue/client/
96	- /internal/examples/json-keyvalue/server/
97	- /internal/examples/protobuf/
98	- /internal/examples/protobuf/example/
99	- /internal/examples/protobuf/examplepb/
100	- /internal/examples/protobuf/exampleutil/
101	- /internal/examples/streaming/
102	- /internal/examples/streaming/client/
103	- /internal/examples/streaming/server/
104	- /internal/examples/thrift-hello/hello/
105	- /internal/examples/thrift-hello/hello/echo/
106	- /internal/examples/thrift-hello/hello/echo/helloclient/
107	- /internal/examples/thrift-hello/hello/echo/hellofx/
108	- /internal/examples/thrift-hello/hello/echo/helloserver/
109	- /internal/examples/thrift-hello/hello/echo/hellotest/
110	- /internal/examples/thrift-keyvalue/keyvalue/client/
111	- /internal/examples/thrift-keyvalue/keyvalue/kv/
112	- /internal/examples/thrift-keyvalue/keyvalue/kv/keyvalueclient/
113	- /internal/examples/thrift-keyvalue/keyvalue/kv/keyvaluefx/
114	- /internal/examples/thrift-keyvalue/keyvalue/kv/keyvalueserver/
115	- /internal/examples/thrift-keyvalue/keyvalue/kv/keyvaluetest/
116	- /internal/examples/thrift-keyvalue/keyvalue/server/
117	- /internal/examples/thrift-oneway/
118	- /internal/examples/thrift-oneway/sink/
119	- /internal/examples/thrift-oneway/sink/helloclient/
120	- /internal/examples/thrift-oneway/sink/hellofx/
121	- /internal/examples/thrift-oneway/sink/helloserver/
122	- /internal/examples/thrift-oneway/sink/hellotest/
123	- /internal/prototest/example/
124	- /internal/prototest/examplepb/
125	- /internal/prototest/exampleutil/
126	- /internal/service-test/
127	- /internal/testutils/
128	- /serialize/internal/

Updating our web app