Backend image rendering as plugin (#11966)

* rendering: headless chrome progress

* renderer: minor change

* grpc: version hell

* updated grpc libs

* wip: minor progess

* rendering: new image rendering plugin is starting to work

* feat: now phantomjs works as well and updated alerting to use new rendering service

* refactor: renamed renderer package and service to rendering to make renderer name less confusing (rendering is internal service that handles the renderer plugin now)

* rendering: now render key is passed and render auth is working in plugin mode

* removed unneeded lines from gitignore

* rendering: now plugin mode supports waiting for all panels to complete rendering

* fix: LastSeenAt fix for render calls, was not set which causes a lot of updates to Last Seen at during rendering, this should fix sqlite db locked issues in seen in previous releases

* change: changed render tz url parameter to use proper timezone name as chrome does not handle UTC offset TZ values

* fix: another update to tz param generation

* renderer: added http mode to renderer service, new ini setting [rendering] server_url

vendor/github.com/hashicorp/go-plugin/client.go

@@ -2,6 +2,7 @@ package plugin
 
 import (
 	"bufio"
+	"context"
 	"crypto/subtle"
 	"crypto/tls"
 	"errors"
@@ -9,7 +10,6 @@ import (
 	"hash"
 	"io"
 	"io/ioutil"
-	"log"
 	"net"
 	"os"
 	"os/exec"
@@ -79,6 +79,7 @@ type Client struct {
 	client      ClientProtocol
 	protocol    Protocol
 	logger      hclog.Logger
+	doneCtx     context.Context
 }
 
 // ClientConfig is the configuration used to initialize a new
@@ -232,7 +233,6 @@ func CleanupClients() {
 	}
 	managedClientsLock.Unlock()
 
-	log.Println("[DEBUG] plugin: waiting for all plugin processes to complete...")
 	wg.Wait()
 }
 
@@ -310,7 +310,7 @@ func (c *Client) Client() (ClientProtocol, error) {
 		c.client, err = newRPCClient(c)
 
 	case ProtocolGRPC:
-		c.client, err = newGRPCClient(c)
+		c.client, err = newGRPCClient(c.doneCtx, c)
 
 	default:
 		return nil, fmt.Errorf("unknown server protocol: %s", c.protocol)
@@ -423,6 +423,9 @@ func (c *Client) Start() (addr net.Addr, err error) {
 
 	// Create the logging channel for when we kill
 	c.doneLogging = make(chan struct{})
+	// Create a context for when we kill
+	var ctxCancel context.CancelFunc
+	c.doneCtx, ctxCancel = context.WithCancel(context.Background())
 
 	if c.config.Reattach != nil {
 		// Verify the process still exists. If not, then it is an error
@@ -457,6 +460,9 @@ func (c *Client) Start() (addr net.Addr, err error) {
 
 			// Close the logging channel since that doesn't work on reattach
 			close(c.doneLogging)
+
+			// Cancel the context
+			ctxCancel()
 		}(p.Pid)
 
 		// Set the address and process
@@ -535,6 +541,9 @@ func (c *Client) Start() (addr net.Addr, err error) {
 		// Mark that we exited
 		close(exitCh)
 
+		// Cancel the context, marking that we exited
+		ctxCancel()
+
 		// Set that we exited, which takes a lock
 		c.l.Lock()
 		defer c.l.Unlock()
@@ -606,7 +615,7 @@ func (c *Client) Start() (addr net.Addr, err error) {
 
 			if int(coreProtocol) != CoreProtocolVersion {
 				err = fmt.Errorf("Incompatible core API version with plugin. "+
-					"Plugin version: %s, Ours: %d\n\n"+
+					"Plugin version: %s, Core version: %d\n\n"+
 					"To fix this, the plugin usually only needs to be recompiled.\n"+
 					"Please report this to the plugin author.", parts[0], CoreProtocolVersion)
 				return
@@ -624,7 +633,7 @@ func (c *Client) Start() (addr net.Addr, err error) {
 		// Test the API version
 		if uint(protocol) != c.config.ProtocolVersion {
 			err = fmt.Errorf("Incompatible API version with plugin. "+
-				"Plugin version: %s, Ours: %d", parts[1], c.config.ProtocolVersion)
+				"Plugin version: %s, Core version: %d", parts[1], c.config.ProtocolVersion)
 			return
 		}
 
@@ -707,18 +716,29 @@ func (c *Client) Protocol() Protocol {
 	return c.protocol
 }
 
+func netAddrDialer(addr net.Addr) func(string, time.Duration) (net.Conn, error) {
+	return func(_ string, _ time.Duration) (net.Conn, error) {
+		// Connect to the client
+		conn, err := net.Dial(addr.Network(), addr.String())
+		if err != nil {
+			return nil, err
+		}
+		if tcpConn, ok := conn.(*net.TCPConn); ok {
+			// Make sure to set keep alive so that the connection doesn't die
+			tcpConn.SetKeepAlive(true)
+		}
+
+		return conn, nil
+	}
+}
+
 // dialer is compatible with grpc.WithDialer and creates the connection
 // to the plugin.
 func (c *Client) dialer(_ string, timeout time.Duration) (net.Conn, error) {
-	// Connect to the client
-	conn, err := net.Dial(c.address.Network(), c.address.String())
+	conn, err := netAddrDialer(c.address)("", timeout)
 	if err != nil {
 		return nil, err
 	}
-	if tcpConn, ok := conn.(*net.TCPConn); ok {
-		// Make sure to set keep alive so that the connection doesn't die
-		tcpConn.SetKeepAlive(true)
-	}
 
 	// If we have a TLS config we wrap our connection. We only do this
 	// for net/rpc since gRPC uses its own mechanism for TLS.
@@ -731,14 +751,14 @@ func (c *Client) dialer(_ string, timeout time.Duration) (net.Conn, error) {
 
 func (c *Client) logStderr(r io.Reader) {
 	bufR := bufio.NewReader(r)
+	l := c.logger.Named(filepath.Base(c.config.Cmd.Path))
+
 	for {
 		line, err := bufR.ReadString('\n')
 		if line != "" {
 			c.config.Stderr.Write([]byte(line))
 			line = strings.TrimRightFunc(line, unicode.IsSpace)
 
-			l := c.logger.Named(filepath.Base(c.config.Cmd.Path))
-
 			entry, err := parseJSON(line)
 			// If output is not JSON format, print directly to Debug
 			if err != nil {
@@ -746,7 +766,7 @@ func (c *Client) logStderr(r io.Reader) {
 			} else {
 				out := flattenKVPairs(entry.KVPairs)
 
-				l = l.With("timestamp", entry.Timestamp.Format(hclog.TimeFormat))
+				out = append(out, "timestamp", entry.Timestamp.Format(hclog.TimeFormat))
 				switch hclog.LevelFromString(entry.Level) {
 				case hclog.Trace:
 					l.Trace(entry.Message, out...)

vendor/github.com/hashicorp/go-plugin/grpc_broker.go

@@ -0,0 +1,455 @@
+package plugin
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"log"
+	"net"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/oklog/run"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+)
+
+// streamer interface is used in the broker to send/receive connection
+// information.
+type streamer interface {
+	Send(*ConnInfo) error
+	Recv() (*ConnInfo, error)
+	Close()
+}
+
+// sendErr is used to pass errors back during a send.
+type sendErr struct {
+	i  *ConnInfo
+	ch chan error
+}
+
+// gRPCBrokerServer is used by the plugin to start a stream and to send
+// connection information to/from the plugin. Implements GRPCBrokerServer and
+// streamer interfaces.
+type gRPCBrokerServer struct {
+	// send is used to send connection info to the gRPC stream.
+	send chan *sendErr
+
+	// recv is used to receive connection info from the gRPC stream.
+	recv chan *ConnInfo
+
+	// quit closes down the stream.
+	quit chan struct{}
+
+	// o is used to ensure we close the quit channel only once.
+	o sync.Once
+}
+
+func newGRPCBrokerServer() *gRPCBrokerServer {
+	return &gRPCBrokerServer{
+		send: make(chan *sendErr),
+		recv: make(chan *ConnInfo),
+		quit: make(chan struct{}),
+	}
+}
+
+// StartStream implements the GRPCBrokerServer interface and will block until
+// the quit channel is closed or the context reports Done. The stream will pass
+// connection information to/from the client.
+func (s *gRPCBrokerServer) StartStream(stream GRPCBroker_StartStreamServer) error {
+	doneCh := stream.Context().Done()
+	defer s.Close()
+
+	// Proccess send stream
+	go func() {
+		for {
+			select {
+			case <-doneCh:
+				return
+			case <-s.quit:
+				return
+			case se := <-s.send:
+				err := stream.Send(se.i)
+				se.ch <- err
+			}
+		}
+	}()
+
+	// Process receive stream
+	for {
+		i, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+		select {
+		case <-doneCh:
+			return nil
+		case <-s.quit:
+			return nil
+		case s.recv <- i:
+		}
+	}
+
+	return nil
+}
+
+// Send is used by the GRPCBroker to pass connection information into the stream
+// to the client.
+func (s *gRPCBrokerServer) Send(i *ConnInfo) error {
+	ch := make(chan error)
+	defer close(ch)
+
+	select {
+	case <-s.quit:
+		return errors.New("broker closed")
+	case s.send <- &sendErr{
+		i:  i,
+		ch: ch,
+	}:
+	}
+
+	return <-ch
+}
+
+// Recv is used by the GRPCBroker to pass connection information that has been
+// sent from the client from the stream to the broker.
+func (s *gRPCBrokerServer) Recv() (*ConnInfo, error) {
+	select {
+	case <-s.quit:
+		return nil, errors.New("broker closed")
+	case i := <-s.recv:
+		return i, nil
+	}
+}
+
+// Close closes the quit channel, shutting down the stream.
+func (s *gRPCBrokerServer) Close() {
+	s.o.Do(func() {
+		close(s.quit)
+	})
+}
+
+// gRPCBrokerClientImpl is used by the client to start a stream and to send
+// connection information to/from the client. Implements GRPCBrokerClient and
+// streamer interfaces.
+type gRPCBrokerClientImpl struct {
+	// client is the underlying GRPC client used to make calls to the server.
+	client GRPCBrokerClient
+
+	// send is used to send connection info to the gRPC stream.
+	send chan *sendErr
+
+	// recv is used to receive connection info from the gRPC stream.
+	recv chan *ConnInfo
+
+	// quit closes down the stream.
+	quit chan struct{}
+
+	// o is used to ensure we close the quit channel only once.
+	o sync.Once
+}
+
+func newGRPCBrokerClient(conn *grpc.ClientConn) *gRPCBrokerClientImpl {
+	return &gRPCBrokerClientImpl{
+		client: NewGRPCBrokerClient(conn),
+		send:   make(chan *sendErr),
+		recv:   make(chan *ConnInfo),
+		quit:   make(chan struct{}),
+	}
+}
+
+// StartStream implements the GRPCBrokerClient interface and will block until
+// the quit channel is closed or the context reports Done. The stream will pass
+// connection information to/from the plugin.
+func (s *gRPCBrokerClientImpl) StartStream() error {
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	defer cancelFunc()
+	defer s.Close()
+
+	stream, err := s.client.StartStream(ctx)
+	if err != nil {
+		return err
+	}
+	doneCh := stream.Context().Done()
+
+	go func() {
+		for {
+			select {
+			case <-doneCh:
+				return
+			case <-s.quit:
+				return
+			case se := <-s.send:
+				err := stream.Send(se.i)
+				se.ch <- err
+			}
+		}
+	}()
+
+	for {
+		i, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+		select {
+		case <-doneCh:
+			return nil
+		case <-s.quit:
+			return nil
+		case s.recv <- i:
+		}
+	}
+
+	return nil
+}
+
+// Send is used by the GRPCBroker to pass connection information into the stream
+// to the plugin.
+func (s *gRPCBrokerClientImpl) Send(i *ConnInfo) error {
+	ch := make(chan error)
+	defer close(ch)
+
+	select {
+	case <-s.quit:
+		return errors.New("broker closed")
+	case s.send <- &sendErr{
+		i:  i,
+		ch: ch,
+	}:
+	}
+
+	return <-ch
+}
+
+// Recv is used by the GRPCBroker to pass connection information that has been
+// sent from the plugin to the broker.
+func (s *gRPCBrokerClientImpl) Recv() (*ConnInfo, error) {
+	select {
+	case <-s.quit:
+		return nil, errors.New("broker closed")
+	case i := <-s.recv:
+		return i, nil
+	}
+}
+
+// Close closes the quit channel, shutting down the stream.
+func (s *gRPCBrokerClientImpl) Close() {
+	s.o.Do(func() {
+		close(s.quit)
+	})
+}
+
+// GRPCBroker is responsible for brokering connections by unique ID.
+//
+// It is used by plugins to create multiple gRPC connections and data
+// streams between the plugin process and the host process.
+//
+// This allows a plugin to request a channel with a specific ID to connect to
+// or accept a connection from, and the broker handles the details of
+// holding these channels open while they're being negotiated.
+//
+// The Plugin interface has access to these for both Server and Client.
+// The broker can be used by either (optionally) to reserve and connect to
+// new streams. This is useful for complex args and return values,
+// or anything else you might need a data stream for.
+type GRPCBroker struct {
+	nextId   uint32
+	streamer streamer
+	streams  map[uint32]*gRPCBrokerPending
+	tls      *tls.Config
+	doneCh   chan struct{}
+	o        sync.Once
+
+	sync.Mutex
+}
+
+type gRPCBrokerPending struct {
+	ch     chan *ConnInfo
+	doneCh chan struct{}
+}
+
+func newGRPCBroker(s streamer, tls *tls.Config) *GRPCBroker {
+	return &GRPCBroker{
+		streamer: s,
+		streams:  make(map[uint32]*gRPCBrokerPending),
+		tls:      tls,
+		doneCh:   make(chan struct{}),
+	}
+}
+
+// Accept accepts a connection by ID.
+//
+// This should not be called multiple times with the same ID at one time.
+func (b *GRPCBroker) Accept(id uint32) (net.Listener, error) {
+	listener, err := serverListener()
+	if err != nil {
+		return nil, err
+	}
+
+	err = b.streamer.Send(&ConnInfo{
+		ServiceId: id,
+		Network:   listener.Addr().Network(),
+		Address:   listener.Addr().String(),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	return listener, nil
+}
+
+// AcceptAndServe is used to accept a specific stream ID and immediately
+// serve a gRPC server on that stream ID. This is used to easily serve
+// complex arguments. Each AcceptAndServe call opens a new listener socket and
+// sends the connection info down the stream to the dialer. Since a new
+// connection is opened every call, these calls should be used sparingly.
+// Multiple gRPC server implementations can be registered to a single
+// AcceptAndServe call.
+func (b *GRPCBroker) AcceptAndServe(id uint32, s func([]grpc.ServerOption) *grpc.Server) {
+	listener, err := b.Accept(id)
+	if err != nil {
+		log.Printf("[ERR] plugin: plugin acceptAndServe error: %s", err)
+		return
+	}
+	defer listener.Close()
+
+	var opts []grpc.ServerOption
+	if b.tls != nil {
+		opts = []grpc.ServerOption{grpc.Creds(credentials.NewTLS(b.tls))}
+	}
+
+	server := s(opts)
+
+	// Here we use a run group to close this goroutine if the server is shutdown
+	// or the broker is shutdown.
+	var g run.Group
+	{
+		// Serve on the listener, if shutting down call GracefulStop.
+		g.Add(func() error {
+			return server.Serve(listener)
+		}, func(err error) {
+			server.GracefulStop()
+		})
+	}
+	{
+		// block on the closeCh or the doneCh. If we are shutting down close the
+		// closeCh.
+		closeCh := make(chan struct{})
+		g.Add(func() error {
+			select {
+			case <-b.doneCh:
+			case <-closeCh:
+			}
+			return nil
+		}, func(err error) {
+			close(closeCh)
+		})
+	}
+
+	// Block until we are done
+	g.Run()
+}
+
+// Close closes the stream and all servers.
+func (b *GRPCBroker) Close() error {
+	b.streamer.Close()
+	b.o.Do(func() {
+		close(b.doneCh)
+	})
+	return nil
+}
+
+// Dial opens a connection by ID.
+func (b *GRPCBroker) Dial(id uint32) (conn *grpc.ClientConn, err error) {
+	var c *ConnInfo
+
+	// Open the stream
+	p := b.getStream(id)
+	select {
+	case c = <-p.ch:
+		close(p.doneCh)
+	case <-time.After(5 * time.Second):
+		return nil, fmt.Errorf("timeout waiting for connection info")
+	}
+
+	var addr net.Addr
+	switch c.Network {
+	case "tcp":
+		addr, err = net.ResolveTCPAddr("tcp", c.Address)
+	case "unix":
+		addr, err = net.ResolveUnixAddr("unix", c.Address)
+	default:
+		err = fmt.Errorf("Unknown address type: %s", c.Address)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	return dialGRPCConn(b.tls, netAddrDialer(addr))
+}
+
+// NextId returns a unique ID to use next.
+//
+// It is possible for very long-running plugin hosts to wrap this value,
+// though it would require a very large amount of calls. In practice
+// we've never seen it happen.
+func (m *GRPCBroker) NextId() uint32 {
+	return atomic.AddUint32(&m.nextId, 1)
+}
+
+// Run starts the brokering and should be executed in a goroutine, since it
+// blocks forever, or until the session closes.
+//
+// Uses of GRPCBroker never need to call this. It is called internally by
+// the plugin host/client.
+func (m *GRPCBroker) Run() {
+	for {
+		stream, err := m.streamer.Recv()
+		if err != nil {
+			// Once we receive an error, just exit
+			break
+		}
+
+		// Initialize the waiter
+		p := m.getStream(stream.ServiceId)
+		select {
+		case p.ch <- stream:
+		default:
+		}
+
+		go m.timeoutWait(stream.ServiceId, p)
+	}
+}
+
+func (m *GRPCBroker) getStream(id uint32) *gRPCBrokerPending {
+	m.Lock()
+	defer m.Unlock()
+
+	p, ok := m.streams[id]
+	if ok {
+		return p
+	}
+
+	m.streams[id] = &gRPCBrokerPending{
+		ch:     make(chan *ConnInfo, 1),
+		doneCh: make(chan struct{}),
+	}
+	return m.streams[id]
+}
+
+func (m *GRPCBroker) timeoutWait(id uint32, p *gRPCBrokerPending) {
+	// Wait for the stream to either be picked up and connected, or
+	// for a timeout.
+	select {
+	case <-p.doneCh:
+	case <-time.After(5 * time.Second):
+	}
+
+	m.Lock()
+	defer m.Unlock()
+
+	// Delete the stream so no one else can grab it
+	delete(m.streams, id)
+}

vendor/github.com/hashicorp/go-plugin/grpc_broker.pb.go

@@ -0,0 +1,190 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// source: grpc_broker.proto
+
+/*
+Package plugin is a generated protocol buffer package.
+
+It is generated from these files:
+	grpc_broker.proto
+
+It has these top-level messages:
+	ConnInfo
+*/
+package plugin
+
+import proto "github.com/golang/protobuf/proto"
+import fmt "fmt"
+import math "math"
+
+import (
+	context "golang.org/x/net/context"
+	grpc "google.golang.org/grpc"
+)
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
+
+type ConnInfo struct {
+	ServiceId uint32 `protobuf:"varint,1,opt,name=service_id,json=serviceId" json:"service_id,omitempty"`
+	Network   string `protobuf:"bytes,2,opt,name=network" json:"network,omitempty"`
+	Address   string `protobuf:"bytes,3,opt,name=address" json:"address,omitempty"`
+}
+
+func (m *ConnInfo) Reset()                    { *m = ConnInfo{} }
+func (m *ConnInfo) String() string            { return proto.CompactTextString(m) }
+func (*ConnInfo) ProtoMessage()               {}
+func (*ConnInfo) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{0} }
+
+func (m *ConnInfo) GetServiceId() uint32 {
+	if m != nil {
+		return m.ServiceId
+	}
+	return 0
+}
+
+func (m *ConnInfo) GetNetwork() string {
+	if m != nil {
+		return m.Network
+	}
+	return ""
+}
+
+func (m *ConnInfo) GetAddress() string {
+	if m != nil {
+		return m.Address
+	}
+	return ""
+}
+
+func init() {
+	proto.RegisterType((*ConnInfo)(nil), "plugin.ConnInfo")
+}
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ context.Context
+var _ grpc.ClientConn
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+const _ = grpc.SupportPackageIsVersion4
+
+// Client API for GRPCBroker service
+
+type GRPCBrokerClient interface {
+	StartStream(ctx context.Context, opts ...grpc.CallOption) (GRPCBroker_StartStreamClient, error)
+}
+
+type gRPCBrokerClient struct {
+	cc *grpc.ClientConn
+}
+
+func NewGRPCBrokerClient(cc *grpc.ClientConn) GRPCBrokerClient {
+	return &gRPCBrokerClient{cc}
+}
+
+func (c *gRPCBrokerClient) StartStream(ctx context.Context, opts ...grpc.CallOption) (GRPCBroker_StartStreamClient, error) {
+	stream, err := grpc.NewClientStream(ctx, &_GRPCBroker_serviceDesc.Streams[0], c.cc, "/plugin.GRPCBroker/StartStream", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &gRPCBrokerStartStreamClient{stream}
+	return x, nil
+}
+
+type GRPCBroker_StartStreamClient interface {
+	Send(*ConnInfo) error
+	Recv() (*ConnInfo, error)
+	grpc.ClientStream
+}
+
+type gRPCBrokerStartStreamClient struct {
+	grpc.ClientStream
+}
+
+func (x *gRPCBrokerStartStreamClient) Send(m *ConnInfo) error {
+	return x.ClientStream.SendMsg(m)
+}
+
+func (x *gRPCBrokerStartStreamClient) Recv() (*ConnInfo, error) {
+	m := new(ConnInfo)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+// Server API for GRPCBroker service
+
+type GRPCBrokerServer interface {
+	StartStream(GRPCBroker_StartStreamServer) error
+}
+
+func RegisterGRPCBrokerServer(s *grpc.Server, srv GRPCBrokerServer) {
+	s.RegisterService(&_GRPCBroker_serviceDesc, srv)
+}
+
+func _GRPCBroker_StartStream_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(GRPCBrokerServer).StartStream(&gRPCBrokerStartStreamServer{stream})
+}
+
+type GRPCBroker_StartStreamServer interface {
+	Send(*ConnInfo) error
+	Recv() (*ConnInfo, error)
+	grpc.ServerStream
+}
+
+type gRPCBrokerStartStreamServer struct {
+	grpc.ServerStream
+}
+
+func (x *gRPCBrokerStartStreamServer) Send(m *ConnInfo) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func (x *gRPCBrokerStartStreamServer) Recv() (*ConnInfo, error) {
+	m := new(ConnInfo)
+	if err := x.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+var _GRPCBroker_serviceDesc = grpc.ServiceDesc{
+	ServiceName: "plugin.GRPCBroker",
+	HandlerType: (*GRPCBrokerServer)(nil),
+	Methods:     []grpc.MethodDesc{},
+	Streams: []grpc.StreamDesc{
+		{
+			StreamName:    "StartStream",
+			Handler:       _GRPCBroker_StartStream_Handler,
+			ServerStreams: true,
+			ClientStreams: true,
+		},
+	},
+	Metadata: "grpc_broker.proto",
+}
+
+func init() { proto.RegisterFile("grpc_broker.proto", fileDescriptor0) }
+
+var fileDescriptor0 = []byte{
+	// 170 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0x12, 0x4c, 0x2f, 0x2a, 0x48,
+	0x8e, 0x4f, 0x2a, 0xca, 0xcf, 0x4e, 0x2d, 0xd2, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x62, 0x2b,
+	0xc8, 0x29, 0x4d, 0xcf, 0xcc, 0x53, 0x8a, 0xe5, 0xe2, 0x70, 0xce, 0xcf, 0xcb, 0xf3, 0xcc, 0x4b,
+	0xcb, 0x17, 0x92, 0xe5, 0xe2, 0x2a, 0x4e, 0x2d, 0x2a, 0xcb, 0x4c, 0x4e, 0x8d, 0xcf, 0x4c, 0x91,
+	0x60, 0x54, 0x60, 0xd4, 0xe0, 0x0d, 0xe2, 0x84, 0x8a, 0x78, 0xa6, 0x08, 0x49, 0x70, 0xb1, 0xe7,
+	0xa5, 0x96, 0x94, 0xe7, 0x17, 0x65, 0x4b, 0x30, 0x29, 0x30, 0x6a, 0x70, 0x06, 0xc1, 0xb8, 0x20,
+	0x99, 0xc4, 0x94, 0x94, 0xa2, 0xd4, 0xe2, 0x62, 0x09, 0x66, 0x88, 0x0c, 0x94, 0x6b, 0xe4, 0xcc,
+	0xc5, 0xe5, 0x1e, 0x14, 0xe0, 0xec, 0x04, 0xb6, 0x5a, 0xc8, 0x94, 0x8b, 0x3b, 0xb8, 0x24, 0xb1,
+	0xa8, 0x24, 0xb8, 0xa4, 0x28, 0x35, 0x31, 0x57, 0x48, 0x40, 0x0f, 0xe2, 0x08, 0x3d, 0x98, 0x0b,
+	0xa4, 0x30, 0x44, 0x34, 0x18, 0x0d, 0x18, 0x93, 0xd8, 0xc0, 0x4e, 0x36, 0x06, 0x04, 0x00, 0x00,
+	0xff, 0xff, 0x7b, 0x5d, 0xfb, 0xe1, 0xc7, 0x00, 0x00, 0x00,
+}

vendor/github.com/hashicorp/go-plugin/grpc_client.go

@@ -1,7 +1,10 @@
 package plugin
 
 import (
+	"crypto/tls"
 	"fmt"
+	"net"
+	"time"
 
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
@@ -9,14 +12,12 @@ import (
 	"google.golang.org/grpc/health/grpc_health_v1"
 )
 
-// newGRPCClient creates a new GRPCClient. The Client argument is expected
-// to be successfully started already with a lock held.
-func newGRPCClient(c *Client) (*GRPCClient, error) {
+func dialGRPCConn(tls *tls.Config, dialer func(string, time.Duration) (net.Conn, error)) (*grpc.ClientConn, error) {
 	// Build dialing options.
 	opts := make([]grpc.DialOption, 0, 5)
 
 	// We use a custom dialer so that we can connect over unix domain sockets
-	opts = append(opts, grpc.WithDialer(c.dialer))
+	opts = append(opts, grpc.WithDialer(dialer))
 
 	// go-plugin expects to block the connection
 	opts = append(opts, grpc.WithBlock())
@@ -26,11 +27,11 @@ func newGRPCClient(c *Client) (*GRPCClient, error) {
 
 	// If we have no TLS configuration set, we need to explicitly tell grpc
 	// that we're connecting with an insecure connection.
-	if c.config.TLSConfig == nil {
+	if tls == nil {
 		opts = append(opts, grpc.WithInsecure())
 	} else {
 		opts = append(opts, grpc.WithTransportCredentials(
-			credentials.NewTLS(c.config.TLSConfig)))
+			credentials.NewTLS(tls)))
 	}
 
 	// Connect. Note the first parameter is unused because we use a custom
@@ -40,9 +41,28 @@ func newGRPCClient(c *Client) (*GRPCClient, error) {
 		return nil, err
 	}
 
+	return conn, nil
+}
+
+// newGRPCClient creates a new GRPCClient. The Client argument is expected
+// to be successfully started already with a lock held.
+func newGRPCClient(doneCtx context.Context, c *Client) (*GRPCClient, error) {
+	conn, err := dialGRPCConn(c.config.TLSConfig, c.dialer)
+	if err != nil {
+		return nil, err
+	}
+
+	// Start the broker.
+	brokerGRPCClient := newGRPCBrokerClient(conn)
+	broker := newGRPCBroker(brokerGRPCClient, c.config.TLSConfig)
+	go broker.Run()
+	go brokerGRPCClient.StartStream()
+
 	return &GRPCClient{
 		Conn:    conn,
 		Plugins: c.config.Plugins,
+		doneCtx: doneCtx,
+		broker:  broker,
 	}, nil
 }
 
@@ -50,10 +70,14 @@ func newGRPCClient(c *Client) (*GRPCClient, error) {
 type GRPCClient struct {
 	Conn    *grpc.ClientConn
 	Plugins map[string]Plugin
+
+	doneCtx context.Context
+	broker  *GRPCBroker
 }
 
 // ClientProtocol impl.
 func (c *GRPCClient) Close() error {
+	c.broker.Close()
 	return c.Conn.Close()
 }
 
@@ -69,7 +93,7 @@ func (c *GRPCClient) Dispense(name string) (interface{}, error) {
 		return nil, fmt.Errorf("plugin %q doesn't support gRPC", name)
 	}
 
-	return p.GRPCClient(c.Conn)
+	return p.GRPCClient(c.doneCtx, c.broker, c.Conn)
 }
 
 // ClientProtocol impl.

vendor/github.com/hashicorp/go-plugin/grpc_server.go

@@ -51,6 +51,7 @@ type GRPCServer struct {
 
 	config GRPCServerConfig
 	server *grpc.Server
+	broker *GRPCBroker
 }
 
 // ServerProtocol impl.
@@ -68,14 +69,20 @@ func (s *GRPCServer) Init() error {
 		GRPCServiceName, grpc_health_v1.HealthCheckResponse_SERVING)
 	grpc_health_v1.RegisterHealthServer(s.server, healthCheck)
 
+	// Register the broker service
+	brokerServer := newGRPCBrokerServer()
+	RegisterGRPCBrokerServer(s.server, brokerServer)
+	s.broker = newGRPCBroker(brokerServer, s.TLS)
+	go s.broker.Run()
+
 	// Register all our plugins onto the gRPC server.
 	for k, raw := range s.Plugins {
 		p, ok := raw.(GRPCPlugin)
 		if !ok {
-			return fmt.Errorf("%q is not a GRPC-compatibile plugin", k)
+			return fmt.Errorf("%q is not a GRPC-compatible plugin", k)
 		}
 
-		if err := p.GRPCServer(s.server); err != nil {
+		if err := p.GRPCServer(s.broker, s.server); err != nil {
 			return fmt.Errorf("error registring %q: %s", k, err)
 		}
 	}
@@ -83,6 +90,16 @@ func (s *GRPCServer) Init() error {
 	return nil
 }
 
+// Stop calls Stop on the underlying grpc.Server
+func (s *GRPCServer) Stop() {
+	s.server.Stop()
+}
+
+// GracefulStop calls GracefulStop on the underlying grpc.Server
+func (s *GRPCServer) GracefulStop() {
+	s.server.GracefulStop()
+}
+
 // Config is the GRPCServerConfig encoded as JSON then base64.
 func (s *GRPCServer) Config() string {
 	// Create a buffer that will contain our final contents

vendor/github.com/hashicorp/go-plugin/plugin.go

@@ -9,6 +9,7 @@
 package plugin
 
 import (
+	"context"
 	"errors"
 	"net/rpc"
 
@@ -33,11 +34,12 @@ type GRPCPlugin interface {
 	// GRPCServer should register this plugin for serving with the
 	// given GRPCServer. Unlike Plugin.Server, this is only called once
 	// since gRPC plugins serve singletons.
-	GRPCServer(*grpc.Server) error
+	GRPCServer(*GRPCBroker, *grpc.Server) error
 
 	// GRPCClient should return the interface implementation for the plugin
-	// you're serving via gRPC.
-	GRPCClient(*grpc.ClientConn) (interface{}, error)
+	// you're serving via gRPC. The provided context will be canceled by
+	// go-plugin in the event of the plugin process exiting.
+	GRPCClient(context.Context, *GRPCBroker, *grpc.ClientConn) (interface{}, error)
 }
 
 // NetRPCUnsupportedPlugin implements Plugin but returns errors for the

vendor/github.com/hashicorp/go-plugin/server.go

@@ -66,6 +66,10 @@ type ServeConfig struct {
 	// the gRPC health checking service. This is not optional since go-plugin
 	// relies on this to implement Ping().
 	GRPCServer func([]grpc.ServerOption) *grpc.Server
+
+	// Logger is used to pass a logger into the server. If none is provided the
+	// server will create a default logger.
+	Logger hclog.Logger
 }
 
 // Protocol returns the protocol that this server should speak.
@@ -106,12 +110,15 @@ func Serve(opts *ServeConfig) {
 	// Logging goes to the original stderr
 	log.SetOutput(os.Stderr)
 
-	// internal logger to os.Stderr
-	logger := hclog.New(&hclog.LoggerOptions{
-		Level:      hclog.Trace,
-		Output:     os.Stderr,
-		JSONFormat: true,
-	})
+	logger := opts.Logger
+	if logger == nil {
+		// internal logger to os.Stderr
+		logger = hclog.New(&hclog.LoggerOptions{
+			Level:      hclog.Trace,
+			Output:     os.Stderr,
+			JSONFormat: true,
+		})
+	}
 
 	// Create our new stdout, stderr files. These will override our built-in
 	// stdout/stderr so that it works across the stream boundary.

vendor/github.com/hashicorp/go-plugin/testing.go

@@ -2,6 +2,8 @@ package plugin
 
 import (
 	"bytes"
+	"context"
+	"io"
 	"net"
 	"net/rpc"
 
@@ -9,6 +11,18 @@ import (
 	"google.golang.org/grpc"
 )
 
+// TestOptions allows specifying options that can affect the behavior of the
+// test functions
+type TestOptions struct {
+	//ServerStdout causes the given value to be used in place of a blank buffer
+	//for RPCServer's Stdout
+	ServerStdout io.ReadCloser
+
+	//ServerStderr causes the given value to be used in place of a blank buffer
+	//for RPCServer's Stderr
+	ServerStderr io.ReadCloser
+}
+
 // The testing file contains test helpers that you can use outside of
 // this package for making it easier to test plugins themselves.
 
@@ -60,12 +74,20 @@ func TestRPCConn(t testing.T) (*rpc.Client, *rpc.Server) {
 
 // TestPluginRPCConn returns a plugin RPC client and server that are connected
 // together and configured.
-func TestPluginRPCConn(t testing.T, ps map[string]Plugin) (*RPCClient, *RPCServer) {
+func TestPluginRPCConn(t testing.T, ps map[string]Plugin, opts *TestOptions) (*RPCClient, *RPCServer) {
 	// Create two net.Conns we can use to shuttle our control connection
 	clientConn, serverConn := TestConn(t)
 
 	// Start up the server
 	server := &RPCServer{Plugins: ps, Stdout: new(bytes.Buffer), Stderr: new(bytes.Buffer)}
+	if opts != nil {
+		if opts.ServerStdout != nil {
+			server.Stdout = opts.ServerStdout
+		}
+		if opts.ServerStderr != nil {
+			server.Stderr = opts.ServerStderr
+		}
+	}
 	go server.ServeConn(serverConn)
 
 	// Connect the client to the server
@@ -77,6 +99,35 @@ func TestPluginRPCConn(t testing.T, ps map[string]Plugin) (*RPCClient, *RPCServe
 	return client, server
 }
 
+// TestGRPCConn returns a gRPC client conn and grpc server that are connected
+// together and configured. The register function is used to register services
+// prior to the Serve call. This is used to test gRPC connections.
+func TestGRPCConn(t testing.T, register func(*grpc.Server)) (*grpc.ClientConn, *grpc.Server) {
+	// Create a listener
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	server := grpc.NewServer()
+	register(server)
+	go server.Serve(l)
+
+	// Connect to the server
+	conn, err := grpc.Dial(
+		l.Addr().String(),
+		grpc.WithBlock(),
+		grpc.WithInsecure())
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	// Connection successful, close the listener
+	l.Close()
+
+	return conn, server
+}
+
 // TestPluginGRPCConn returns a plugin gRPC client and server that are connected
 // together and configured. This is used to test gRPC connections.
 func TestPluginGRPCConn(t testing.T, ps map[string]Plugin) (*GRPCClient, *GRPCServer) {
@@ -107,13 +158,17 @@ func TestPluginGRPCConn(t testing.T, ps map[string]Plugin) (*GRPCClient, *GRPCSe
 		t.Fatalf("err: %s", err)
 	}
 
-	// Connection successful, close the listener
-	l.Close()
+	brokerGRPCClient := newGRPCBrokerClient(conn)
+	broker := newGRPCBroker(brokerGRPCClient, nil)
+	go broker.Run()
+	go brokerGRPCClient.StartStream()
 
 	// Create the client
 	client := &GRPCClient{
 		Conn:    conn,
 		Plugins: ps,
+		broker:  broker,
+		doneCtx: context.Background(),
 	}
 
 	return client, server