vendor/github.com/docker/swarmkit/connectionbroker/broker.go

   1 // Package connectionbroker is a layer on top of remotes that returns
   2 // a gRPC connection to a manager. The connection may be a local connection
   3 // using a local socket such as a UNIX socket.
   4 package connectionbroker
   5
   6 import (
   7         "sync"
   8
   9         "github.com/docker/swarmkit/api"
  10         "github.com/docker/swarmkit/remotes"
  11         grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
  12         "google.golang.org/grpc"
  13 )
  14
  15 // Broker is a simple connection broker. It can either return a fresh
  16 // connection to a remote manager selected with weighted randomization, or a
  17 // local gRPC connection to the local manager.
  18 type Broker struct {
  19         mu        sync.Mutex
  20         remotes   remotes.Remotes
  21         localConn *grpc.ClientConn
  22 }
  23
  24 // New creates a new connection broker.
  25 func New(remotes remotes.Remotes) *Broker {
  26         return &Broker{
  27                 remotes: remotes,
  28         }
  29 }
  30
  31 // SetLocalConn changes the local gRPC connection used by the connection broker.
  32 func (b *Broker) SetLocalConn(localConn *grpc.ClientConn) {
  33         b.mu.Lock()
  34         defer b.mu.Unlock()
  35
  36         b.localConn = localConn
  37 }
  38
  39 // Select a manager from the set of available managers, and return a connection.
  40 func (b *Broker) Select(dialOpts ...grpc.DialOption) (*Conn, error) {
  41         b.mu.Lock()
  42         localConn := b.localConn
  43         b.mu.Unlock()
  44
  45         if localConn != nil {
  46                 return &Conn{
  47                         ClientConn: localConn,
  48                         isLocal:    true,
  49                 }, nil
  50         }
  51
  52         return b.SelectRemote(dialOpts...)
  53 }
  54
  55 // SelectRemote chooses a manager from the remotes, and returns a TCP
  56 // connection.
  57 func (b *Broker) SelectRemote(dialOpts ...grpc.DialOption) (*Conn, error) {
  58         peer, err := b.remotes.Select()
  59         if err != nil {
  60                 return nil, err
  61         }
  62
  63         dialOpts = append(dialOpts,
  64                 grpc.WithUnaryInterceptor(grpc_prometheus.UnaryClientInterceptor),
  65                 grpc.WithStreamInterceptor(grpc_prometheus.StreamClientInterceptor))
  66
  67         cc, err := grpc.Dial(peer.Addr, dialOpts...)
  68         if err != nil {
  69                 b.remotes.ObserveIfExists(peer, -remotes.DefaultObservationWeight)
  70                 return nil, err
  71         }
  72
  73         return &Conn{
  74                 ClientConn: cc,
  75                 remotes:    b.remotes,
  76                 peer:       peer,
  77         }, nil
  78 }
  79
  80 // Remotes returns the remotes interface used by the broker, so the caller
  81 // can make observations or see weights directly.
  82 func (b *Broker) Remotes() remotes.Remotes {
  83         return b.remotes
  84 }
  85
  86 // Conn is a wrapper around a gRPC client connection.
  87 type Conn struct {
  88         *grpc.ClientConn
  89         isLocal bool
  90         remotes remotes.Remotes
  91         peer    api.Peer
  92 }
  93
  94 // Close closes the client connection if it is a remote connection. It also
  95 // records a positive experience with the remote peer if success is true,
  96 // otherwise it records a negative experience. If a local connection is in use,
  97 // Close is a noop.
  98 func (c *Conn) Close(success bool) error {
  99         if c.isLocal {
 100                 return nil
 101         }
 102
 103         if success {
 104                 c.remotes.ObserveIfExists(c.peer, remotes.DefaultObservationWeight)
 105         } else {
 106                 c.remotes.ObserveIfExists(c.peer, -remotes.DefaultObservationWeight)
 107         }
 108
 109         return c.ClientConn.Close()
 110 }