Update go-plugin

Signed-off-by: Carlisia <carlisiac@vmware.com>
2026-09-21 07:24:19 +00:00 · 2019-03-20 13:02:51 -07:00
parent 4c73e23ce8
commit 01f2ae76e2
404 changed files with 118785 additions and 24794 deletions
@@ -2,14 +2,16 @@ package plugin

 import (
 	"bufio"
+	"context"
 	"crypto/subtle"
 	"crypto/tls"
+	"crypto/x509"
+	"encoding/base64"
 	"errors"
 	"fmt"
 	"hash"
 	"io"
 	"io/ioutil"
-	"log"
 	"net"
 	"os"
 	"os/exec"
@@ -19,7 +21,6 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
-	"unicode"

 	hclog "github.com/hashicorp/go-hclog"
 )
@@ -70,15 +71,31 @@ var (
 //
 // See NewClient and ClientConfig for using a Client.
 type Client struct {
-	config      *ClientConfig
-	exited      bool
-	doneLogging chan struct{}
-	l           sync.Mutex
-	address     net.Addr
-	process     *os.Process
-	client      ClientProtocol
-	protocol    Protocol
-	logger      hclog.Logger
+	config            *ClientConfig
+	exited            bool
+	l                 sync.Mutex
+	address           net.Addr
+	process           *os.Process
+	client            ClientProtocol
+	protocol          Protocol
+	logger            hclog.Logger
+	doneCtx           context.Context
+	ctxCancel         context.CancelFunc
+	negotiatedVersion int
+
+	// clientWaitGroup is used to manage the lifecycle of the plugin management
+	// goroutines.
+	clientWaitGroup sync.WaitGroup
+
+	// processKilled is used for testing only, to flag when the process was
+	// forcefully killed.
+	processKilled bool
+}
+
+// NegotiatedVersion returns the protocol version negotiated with the server.
+// This is only valid after Start() is called.
+func (c *Client) NegotiatedVersion() int {
+	return c.negotiatedVersion
 }

 // ClientConfig is the configuration used to initialize a new
@@ -89,7 +106,13 @@ type ClientConfig struct {
 	HandshakeConfig

 	// Plugins are the plugins that can be consumed.
-	Plugins map[string]Plugin
+	// The implied version of this PluginSet is the Handshake.ProtocolVersion.
+	Plugins PluginSet
+
+	// VersionedPlugins is a map of PluginSets for specific protocol versions.
+	// These can be used to negotiate a compatible version between client and
+	// server. If this is set, Handshake.ProtocolVersion is not required.
+	VersionedPlugins map[int]PluginSet

 	// One of the following must be set, but not both.
 	//
@@ -156,6 +179,29 @@ type ClientConfig struct {
 	// Logger is the logger that the client will used. If none is provided,
 	// it will default to hclog's default logger.
 	Logger hclog.Logger
+
+	// AutoMTLS has the client and server automatically negotiate mTLS for
+	// transport authentication. This ensures that only the original client will
+	// be allowed to connect to the server, and all other connections will be
+	// rejected. The client will also refuse to connect to any server that isn't
+	// the original instance started by the client.
+	//
+	// In this mode of operation, the client generates a one-time use tls
+	// certificate, sends the public x.509 certificate to the new server, and
+	// the server generates a one-time use tls certificate, and sends the public
+	// x.509 certificate back to the client. These are used to authenticate all
+	// rpc connections between the client and server.
+	//
+	// Setting AutoMTLS to true implies that the server must support the
+	// protocol, and correctly negotiate the tls certificates, or a connection
+	// failure will result.
+	//
+	// The client should not set TLSConfig, nor should the server set a
+	// TLSProvider, because AutoMTLS implies that a new certificate and tls
+	// configuration will be generated at startup.
+	//
+	// You cannot Reattach to a server with this option enabled.
+	AutoMTLS bool
 }

 // ReattachConfig is used to configure a client to reattach to an
@@ -232,7 +278,6 @@ func CleanupClients() {
 	}
 	managedClientsLock.Unlock()

-	log.Println("[DEBUG] plugin: waiting for all plugin processes to complete...")
 	wg.Wait()
 }

@@ -310,7 +355,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)
@@ -331,6 +376,14 @@ func (c *Client) Exited() bool {
 	return c.exited
 }

+// killed is used in tests to check if a process failed to exit gracefully, and
+// needed to be killed.
+func (c *Client) killed() bool {
+	c.l.Lock()
+	defer c.l.Unlock()
+	return c.processKilled
+}
+
 // End the executing subprocess (if it is running) and perform any cleanup
 // tasks necessary such as capturing any remaining logs and so on.
 //
@@ -342,14 +395,24 @@ func (c *Client) Kill() {
 	c.l.Lock()
 	process := c.process
 	addr := c.address
-	doneCh := c.doneLogging
 	c.l.Unlock()

-	// If there is no process, we never started anything. Nothing to kill.
+	// If there is no process, there is nothing to kill.
 	if process == nil {
 		return
 	}

+	defer func() {
+		// Wait for the all client goroutines to finish.
+		c.clientWaitGroup.Wait()
+
+		// Make sure there is no reference to the old process after it has been
+		// killed.
+		c.l.Lock()
+		c.process = nil
+		c.l.Unlock()
+	}()
+
 	// We need to check for address here. It is possible that the plugin
 	// started (process != nil) but has no address (addr == nil) if the
 	// plugin failed at startup. If we do have an address, we need to close
@@ -370,6 +433,8 @@ func (c *Client) Kill() {
 				// kill in a moment anyways.
 				c.logger.Warn("error closing client during Kill", "err", err)
 			}
+		} else {
+			c.logger.Error("client", "error", err)
 		}
 	}

@@ -378,17 +443,20 @@ func (c *Client) Kill() {
 	// doneCh which would be closed if the process exits.
 	if graceful {
 		select {
-		case <-doneCh:
+		case <-c.doneCtx.Done():
+			c.logger.Debug("plugin exited")
 			return
-		case <-time.After(250 * time.Millisecond):
+		case <-time.After(2 * time.Second):
 		}
 	}

 	// If graceful exiting failed, just kill it
+	c.logger.Warn("plugin failed to exit gracefully")
 	process.Kill()

-	// Wait for the client to finish logging so we have a complete log
-	<-doneCh
+	c.l.Lock()
+	c.processKilled = true
+	c.l.Unlock()
 }

 // Starts the underlying subprocess, communicating with it to negotiate
@@ -407,7 +475,7 @@ func (c *Client) Start() (addr net.Addr, err error) {

 	// If one of cmd or reattach isn't set, then it is an error. We wrap
 	// this in a {} for scoping reasons, and hopeful that the escape
-	// analysis will pop the stock here.
+	// analysis will pop the stack here.
 	{
 		cmdSet := c.config.Cmd != nil
 		attachSet := c.config.Reattach != nil
@@ -421,71 +489,49 @@ func (c *Client) Start() (addr net.Addr, err error) {
 		}
 	}

-	// Create the logging channel for when we kill
-	c.doneLogging = make(chan struct{})
-
 	if c.config.Reattach != nil {
-		// Verify the process still exists. If not, then it is an error
-		p, err := os.FindProcess(c.config.Reattach.Pid)
-		if err != nil {
-			return nil, err
-		}
+		return c.reattach()
+	}

-		// Attempt to connect to the addr since on Unix systems FindProcess
-		// doesn't actually return an error if it can't find the process.
-		conn, err := net.Dial(
-			c.config.Reattach.Addr.Network(),
-			c.config.Reattach.Addr.String())
-		if err != nil {
-			p.Kill()
-			return nil, ErrProcessNotFound
-		}
-		conn.Close()
+	if c.config.VersionedPlugins == nil {
+		c.config.VersionedPlugins = make(map[int]PluginSet)
+	}

-		// Goroutine to mark exit status
-		go func(pid int) {
-			// Wait for the process to die
-			pidWait(pid)
+	// handle all plugins as versioned, using the handshake config as the default.
+	version := int(c.config.ProtocolVersion)

-			// Log so we can see it
-			c.logger.Debug("reattached plugin process exited")
+	// Make sure we're not overwriting a real version 0. If ProtocolVersion was
+	// non-zero, then we have to just assume the user made sure that
+	// VersionedPlugins doesn't conflict.
+	if _, ok := c.config.VersionedPlugins[version]; !ok && c.config.Plugins != nil {
+		c.config.VersionedPlugins[version] = c.config.Plugins
+	}

-			// Mark it
-			c.l.Lock()
-			defer c.l.Unlock()
-			c.exited = true
-
-			// Close the logging channel since that doesn't work on reattach
-			close(c.doneLogging)
-		}(p.Pid)
-
-		// Set the address and process
-		c.address = c.config.Reattach.Addr
-		c.process = p
-		c.protocol = c.config.Reattach.Protocol
-		if c.protocol == "" {
-			// Default the protocol to net/rpc for backwards compatibility
-			c.protocol = ProtocolNetRPC
-		}
-
-		return c.address, nil
+	var versionStrings []string
+	for v := range c.config.VersionedPlugins {
+		versionStrings = append(versionStrings, strconv.Itoa(v))
 	}

 	env := []string{
 		fmt.Sprintf("%s=%s", c.config.MagicCookieKey, c.config.MagicCookieValue),
 		fmt.Sprintf("PLUGIN_MIN_PORT=%d", c.config.MinPort),
 		fmt.Sprintf("PLUGIN_MAX_PORT=%d", c.config.MaxPort),
+		fmt.Sprintf("PLUGIN_PROTOCOL_VERSIONS=%s", strings.Join(versionStrings, ",")),
 	}

-	stdout_r, stdout_w := io.Pipe()
-	stderr_r, stderr_w := io.Pipe()
-
 	cmd := c.config.Cmd
 	cmd.Env = append(cmd.Env, os.Environ()...)
 	cmd.Env = append(cmd.Env, env...)
 	cmd.Stdin = os.Stdin
-	cmd.Stderr = stderr_w
-	cmd.Stdout = stdout_w
+
+	cmdStdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+	cmdStderr, err := cmd.StderrPipe()
+	if err != nil {
+		return nil, err
+	}

 	if c.config.SecureConfig != nil {
 		if ok, err := c.config.SecureConfig.Check(cmd.Path); err != nil {
@@ -495,6 +541,29 @@ func (c *Client) Start() (addr net.Addr, err error) {
 		}
 	}

+	// Setup a temporary certificate for client/server mtls, and send the public
+	// certificate to the plugin.
+	if c.config.AutoMTLS {
+		c.logger.Info("configuring client automatic mTLS")
+		certPEM, keyPEM, err := generateCert()
+		if err != nil {
+			c.logger.Error("failed to generate client certificate", "error", err)
+			return nil, err
+		}
+		cert, err := tls.X509KeyPair(certPEM, keyPEM)
+		if err != nil {
+			c.logger.Error("failed to parse client certificate", "error", err)
+			return nil, err
+		}
+
+		cmd.Env = append(cmd.Env, fmt.Sprintf("PLUGIN_CLIENT_CERT=%s", certPEM))
+
+		c.config.TLSConfig = &tls.Config{
+			Certificates: []tls.Certificate{cert},
+			ServerName:   "localhost",
+		}
+	}
+
 	c.logger.Debug("starting plugin", "path", cmd.Path, "args", cmd.Args)
 	err = cmd.Start()
 	if err != nil {
@@ -503,6 +572,7 @@ func (c *Client) Start() (addr net.Addr, err error) {

 	// Set the process
 	c.process = cmd.Process
+	c.logger.Debug("plugin started", "path", cmd.Path, "pid", c.process.Pid)

 	// Make sure the command is properly cleaned up if there is an error
 	defer func() {
@@ -517,24 +587,37 @@ func (c *Client) Start() (addr net.Addr, err error) {
 		}
 	}()

-	// Start goroutine to wait for process to exit
-	exitCh := make(chan struct{})
+	// Create a context for when we kill
+	c.doneCtx, c.ctxCancel = context.WithCancel(context.Background())
+
+	c.clientWaitGroup.Add(1)
 	go func() {
-		// Make sure we close the write end of our stderr/stdout so
-		// that the readers send EOF properly.
-		defer stderr_w.Close()
-		defer stdout_w.Close()
+		// ensure the context is cancelled when we're done
+		defer c.ctxCancel()
+
+		defer c.clientWaitGroup.Done()
+
+		// get the cmd info early, since the process information will be removed
+		// in Kill.
+		pid := c.process.Pid
+		path := cmd.Path

 		// Wait for the command to end.
-		cmd.Wait()
+		err := cmd.Wait()
+
+		debugMsgArgs := []interface{}{
+			"path", path,
+			"pid", pid,
+		}
+		if err != nil {
+			debugMsgArgs = append(debugMsgArgs,
+				[]interface{}{"error", err.Error()}...)
+		}

 		// Log and make sure to flush the logs write away
-		c.logger.Debug("plugin process exited", "path", cmd.Path)
+		c.logger.Debug("plugin process exited", debugMsgArgs...)
 		os.Stderr.Sync()

-		// Mark that we exited
-		close(exitCh)
-
 		// Set that we exited, which takes a lock
 		c.l.Lock()
 		defer c.l.Unlock()
@@ -542,32 +625,33 @@ func (c *Client) Start() (addr net.Addr, err error) {
 	}()

 	// Start goroutine that logs the stderr
-	go c.logStderr(stderr_r)
+	c.clientWaitGroup.Add(1)
+	// logStderr calls Done()
+	go c.logStderr(cmdStderr)

 	// Start a goroutine that is going to be reading the lines
 	// out of stdout
-	linesCh := make(chan []byte)
+	linesCh := make(chan string)
+	c.clientWaitGroup.Add(1)
 	go func() {
+		defer c.clientWaitGroup.Done()
 		defer close(linesCh)

-		buf := bufio.NewReader(stdout_r)
-		for {
-			line, err := buf.ReadBytes('\n')
-			if line != nil {
-				linesCh <- line
-			}
-
-			if err == io.EOF {
-				return
-			}
+		scanner := bufio.NewScanner(cmdStdout)
+		for scanner.Scan() {
+			linesCh <- scanner.Text()
 		}
 	}()

 	// Make sure after we exit we read the lines from stdout forever
-	// so they don't block since it is an io.Pipe
+	// so they don't block since it is a pipe.
+	// The scanner goroutine above will close this, but track it with a wait
+	// group for completeness.
+	c.clientWaitGroup.Add(1)
 	defer func() {
 		go func() {
-			for _ = range linesCh {
+			defer c.clientWaitGroup.Done()
+			for range linesCh {
 			}
 		}()
 	}()
@@ -580,12 +664,12 @@ func (c *Client) Start() (addr net.Addr, err error) {
 	select {
 	case <-timeout:
 		err = errors.New("timeout while waiting for plugin to start")
-	case <-exitCh:
+	case <-c.doneCtx.Done():
 		err = errors.New("plugin exited before we could connect")
-	case lineBytes := <-linesCh:
+	case line := <-linesCh:
 		// Trim the line and split by "|" in order to get the parts of
 		// the output.
-		line := strings.TrimSpace(string(lineBytes))
+		line = strings.TrimSpace(line)
 		parts := strings.SplitN(line, "|", 6)
 		if len(parts) < 4 {
 			err = fmt.Errorf(
@@ -613,20 +697,18 @@ func (c *Client) Start() (addr net.Addr, err error) {
 			}
 		}

-		// Parse the protocol version
-		var protocol int64
-		protocol, err = strconv.ParseInt(parts[1], 10, 0)
+		// Test the API version
+		version, pluginSet, err := c.checkProtoVersion(parts[1])
 		if err != nil {
-			err = fmt.Errorf("Error parsing protocol version: %s", err)
-			return
+			return addr, err
 		}

-		// Test the API version
-		if uint(protocol) != c.config.ProtocolVersion {
-			err = fmt.Errorf("Incompatible API version with plugin. "+
-				"Plugin version: %s, Core version: %d", parts[1], c.config.ProtocolVersion)
-			return
-		}
+		// set the Plugins value to the compatible set, so the version
+		// doesn't need to be passed through to the ClientProtocol
+		// implementation.
+		c.config.Plugins = pluginSet
+		c.negotiatedVersion = version
+		c.logger.Debug("using plugin", "version", version)

 		switch parts[2] {
 		case "tcp":
@@ -654,15 +736,125 @@ func (c *Client) Start() (addr net.Addr, err error) {
 		if !found {
 			err = fmt.Errorf("Unsupported plugin protocol %q. Supported: %v",
 				c.protocol, c.config.AllowedProtocols)
-			return
+			return addr, err
 		}

+		// See if we have a TLS certificate from the server.
+		// Checking if the length is > 50 rules out catching the unused "extra"
+		// data returned from some older implementations.
+		if len(parts) >= 6 && len(parts[5]) > 50 {
+			err := c.loadServerCert(parts[5])
+			if err != nil {
+				return nil, fmt.Errorf("error parsing server cert: %s", err)
+			}
+		}
 	}

 	c.address = addr
 	return
 }

+// loadServerCert is used by AutoMTLS to read an x.509 cert returned by the
+// server, and load it as the RootCA for the client TLSConfig.
+func (c *Client) loadServerCert(cert string) error {
+	certPool := x509.NewCertPool()
+
+	asn1, err := base64.RawStdEncoding.DecodeString(cert)
+	if err != nil {
+		return err
+	}
+
+	x509Cert, err := x509.ParseCertificate([]byte(asn1))
+	if err != nil {
+		return err
+	}
+
+	certPool.AddCert(x509Cert)
+
+	c.config.TLSConfig.RootCAs = certPool
+	return nil
+}
+
+func (c *Client) reattach() (net.Addr, error) {
+	// Verify the process still exists. If not, then it is an error
+	p, err := os.FindProcess(c.config.Reattach.Pid)
+	if err != nil {
+		return nil, err
+	}
+
+	// Attempt to connect to the addr since on Unix systems FindProcess
+	// doesn't actually return an error if it can't find the process.
+	conn, err := net.Dial(
+		c.config.Reattach.Addr.Network(),
+		c.config.Reattach.Addr.String())
+	if err != nil {
+		p.Kill()
+		return nil, ErrProcessNotFound
+	}
+	conn.Close()
+
+	// Create a context for when we kill
+	c.doneCtx, c.ctxCancel = context.WithCancel(context.Background())
+
+	c.clientWaitGroup.Add(1)
+	// Goroutine to mark exit status
+	go func(pid int) {
+		defer c.clientWaitGroup.Done()
+
+		// ensure the context is cancelled when we're done
+		defer c.ctxCancel()
+
+		// Wait for the process to die
+		pidWait(pid)
+
+		// Log so we can see it
+		c.logger.Debug("reattached plugin process exited")
+
+		// Mark it
+		c.l.Lock()
+		defer c.l.Unlock()
+		c.exited = true
+	}(p.Pid)
+
+	// Set the address and process
+	c.address = c.config.Reattach.Addr
+	c.process = p
+	c.protocol = c.config.Reattach.Protocol
+	if c.protocol == "" {
+		// Default the protocol to net/rpc for backwards compatibility
+		c.protocol = ProtocolNetRPC
+	}
+
+	return c.address, nil
+}
+
+// checkProtoVersion returns the negotiated version and PluginSet.
+// This returns an error if the server returned an incompatible protocol
+// version, or an invalid handshake response.
+func (c *Client) checkProtoVersion(protoVersion string) (int, PluginSet, error) {
+	serverVersion, err := strconv.Atoi(protoVersion)
+	if err != nil {
+		return 0, nil, fmt.Errorf("Error parsing protocol version %q: %s", protoVersion, err)
+	}
+
+	// record these for the error message
+	var clientVersions []int
+
+	// all versions, including the legacy ProtocolVersion have been added to
+	// the versions set
+	for version, plugins := range c.config.VersionedPlugins {
+		clientVersions = append(clientVersions, version)
+
+		if serverVersion != version {
+			continue
+		}
+		return version, plugins, nil
+	}
+
+	return 0, nil, fmt.Errorf("Incompatible API version with plugin. "+
+		"Plugin version: %d, Client versions: %d", serverVersion, clientVersions)
+}
+
 // ReattachConfig returns the information that must be provided to NewClient
 // to reattach to the plugin process that this client started. This is
 // useful for plugins that detach from their parent process.
@@ -707,18 +899,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.
@@ -729,44 +932,64 @@ func (c *Client) dialer(_ string, timeout time.Duration) (net.Conn, error) {
 	return conn, nil
 }

+var stdErrBufferSize = 64 * 1024
+
 func (c *Client) logStderr(r io.Reader) {
-	bufR := bufio.NewReader(r)
+	defer c.clientWaitGroup.Done()
+	l := c.logger.Named(filepath.Base(c.config.Cmd.Path))
+
+	reader := bufio.NewReaderSize(r, stdErrBufferSize)
+	// continuation indicates the previous line was a prefix
+	continuation := false
+
 	for {
-		line, err := bufR.ReadString('\n')
-		if line != "" {
-			c.config.Stderr.Write([]byte(line))
-			line = strings.TrimRightFunc(line, unicode.IsSpace)
+		line, isPrefix, err := reader.ReadLine()
+		switch {
+		case err == io.EOF:
+			return
+		case err != nil:
+			l.Error("reading plugin stderr", "error", err)
+			return
+		}

-			l := c.logger.Named(filepath.Base(c.config.Cmd.Path))
+		c.config.Stderr.Write(line)

-			entry, err := parseJSON(line)
-			// If output is not JSON format, print directly to Debug
-			if err != nil {
-				l.Debug(line)
-			} else {
-				out := flattenKVPairs(entry.KVPairs)
+		// The line was longer than our max token size, so it's likely
+		// incomplete and won't unmarshal.
+		if isPrefix || continuation {
+			l.Debug(string(line))

-				l = l.With("timestamp", entry.Timestamp.Format(hclog.TimeFormat))
-				switch hclog.LevelFromString(entry.Level) {
-				case hclog.Trace:
-					l.Trace(entry.Message, out...)
-				case hclog.Debug:
-					l.Debug(entry.Message, out...)
-				case hclog.Info:
-					l.Info(entry.Message, out...)
-				case hclog.Warn:
-					l.Warn(entry.Message, out...)
-				case hclog.Error:
-					l.Error(entry.Message, out...)
-				}
+			// if we're finishing a continued line, add the newline back in
+			if !isPrefix {
+				c.config.Stderr.Write([]byte{'\n'})
+			}
+
+			continuation = isPrefix
+			continue
+		}
+
+		c.config.Stderr.Write([]byte{'\n'})
+
+		entry, err := parseJSON(line)
+		// If output is not JSON format, print directly to Debug
+		if err != nil {
+			l.Debug(string(line))
+		} else {
+			out := flattenKVPairs(entry.KVPairs)
+
+			out = append(out, "timestamp", entry.Timestamp.Format(hclog.TimeFormat))
+			switch hclog.LevelFromString(entry.Level) {
+			case hclog.Trace:
+				l.Trace(entry.Message, out...)
+			case hclog.Debug:
+				l.Debug(entry.Message, out...)
+			case hclog.Info:
+				l.Info(entry.Message, out...)
+			case hclog.Warn:
+				l.Warn(entry.Message, out...)
+			case hclog.Error:
+				l.Error(entry.Message, out...)
 			}
 		}
-
-		if err == io.EOF {
-			break
-		}
 	}
-
-	// Flag that we've completed logging for others
-	close(c.doneLogging)
 }
@@ -0,0 +1,457 @@
+package plugin
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"log"
+	"net"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/hashicorp/go-plugin/internal/plugin"
+
+	"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(*plugin.ConnInfo) error
+	Recv() (*plugin.ConnInfo, error)
+	Close()
+}
+
+// sendErr is used to pass errors back during a send.
+type sendErr struct {
+	i  *plugin.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 *plugin.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 *plugin.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 plugin.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 *plugin.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() (*plugin.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 plugin.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 *plugin.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: plugin.NewGRPCBrokerClient(conn),
+		send:   make(chan *sendErr),
+		recv:   make(chan *plugin.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 *plugin.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() (*plugin.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 *plugin.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(&plugin.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 *plugin.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 *plugin.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)
+}
@@ -1,36 +1,35 @@
 package plugin

 import (
+	"crypto/tls"
 	"fmt"
+	"net"
+	"time"

+	"github.com/hashicorp/go-plugin/internal/plugin"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"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))
-
-	// go-plugin expects to block the connection
-	opts = append(opts, grpc.WithBlock())
+	// We use a custom dialer so that we can connect over unix domain sockets.
+	opts = append(opts, grpc.WithDialer(dialer))

 	// Fail right away
 	opts = append(opts, grpc.FailOnNonTempDialError(true))

 	// 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,20 +39,49 @@ func newGRPCClient(c *Client) (*GRPCClient, error) {
 		return nil, err
 	}

-	return &GRPCClient{
-		Conn:    conn,
-		Plugins: c.config.Plugins,
-	}, nil
+	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()
+
+	cl := &GRPCClient{
+		Conn:       conn,
+		Plugins:    c.config.Plugins,
+		doneCtx:    doneCtx,
+		broker:     broker,
+		controller: plugin.NewGRPCControllerClient(conn),
+	}
+
+	return cl, nil
 }

 // GRPCClient connects to a GRPCServer over gRPC to dispense plugin types.
 type GRPCClient struct {
 	Conn    *grpc.ClientConn
 	Plugins map[string]Plugin
+
+	doneCtx context.Context
+	broker  *GRPCBroker
+
+	controller plugin.GRPCControllerClient
 }

 // ClientProtocol impl.
 func (c *GRPCClient) Close() error {
+	c.broker.Close()
+	c.controller.Shutdown(c.doneCtx, &plugin.Empty{})
 	return c.Conn.Close()
 }

@@ -69,7 +97,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.
@@ -0,0 +1,23 @@
+package plugin
+
+import (
+	"context"
+
+	"github.com/hashicorp/go-plugin/internal/plugin"
+)
+
+// GRPCControllerServer handles shutdown calls to terminate the server when the
+// plugin client is closed.
+type grpcControllerServer struct {
+	server *GRPCServer
+}
+
+// Shutdown stops the grpc server. It first will attempt a graceful stop, then a
+// full stop on the server.
+func (s *grpcControllerServer) Shutdown(ctx context.Context, _ *plugin.Empty) (*plugin.Empty, error) {
+	resp := &plugin.Empty{}
+
+	// TODO: figure out why GracefullStop doesn't work.
+	s.server.Stop()
+	return resp, nil
+}
@@ -8,6 +8,8 @@ import (
 	"io"
 	"net"

+	hclog "github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-plugin/internal/plugin"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/health"
@@ -51,6 +53,9 @@ type GRPCServer struct {

 	config GRPCServerConfig
 	server *grpc.Server
+	broker *GRPCBroker
+
+	logger hclog.Logger
 }

 // ServerProtocol impl.
@@ -68,21 +73,43 @@ func (s *GRPCServer) Init() error {
 		GRPCServiceName, grpc_health_v1.HealthCheckResponse_SERVING)
 	grpc_health_v1.RegisterHealthServer(s.server, healthCheck)

+	// Register the broker service
+	brokerServer := newGRPCBrokerServer()
+	plugin.RegisterGRPCBrokerServer(s.server, brokerServer)
+	s.broker = newGRPCBroker(brokerServer, s.TLS)
+	go s.broker.Run()
+
+	// Register the controller
+	controllerServer := &grpcControllerServer{
+		server: s,
+	}
+	plugin.RegisterGRPCControllerServer(s.server, controllerServer)
+
 	// 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 {
-			return fmt.Errorf("error registring %q: %s", k, err)
+		if err := p.GRPCServer(s.broker, s.server); err != nil {
+			return fmt.Errorf("error registering %q: %s", k, err)
 		}
 	}

 	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
@@ -100,11 +127,11 @@ func (s *GRPCServer) Config() string {
 }

 func (s *GRPCServer) Serve(lis net.Listener) {
-	// Start serving in a goroutine
-	go s.server.Serve(lis)
-
-	// Wait until graceful completion
-	<-s.DoneCh
+	defer close(s.DoneCh)
+	err := s.server.Serve(lis)
+	if err != nil {
+		s.logger.Error("grpc server", "error", err)
+	}
 }

 // GRPCServerConfig is the extra configuration passed along for consumers
@@ -0,0 +1,3 @@
+//go:generate protoc -I ./ ./grpc_broker.proto ./grpc_controller.proto --go_out=plugins=grpc:.
+
+package plugin
@@ -0,0 +1,203 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// source: grpc_broker.proto
+
+package plugin
+
+import (
+	fmt "fmt"
+	proto "github.com/golang/protobuf/proto"
+	context "golang.org/x/net/context"
+	grpc "google.golang.org/grpc"
+	math "math"
+)
+
+// 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,proto3" json:"service_id,omitempty"`
+	Network              string   `protobuf:"bytes,2,opt,name=network,proto3" json:"network,omitempty"`
+	Address              string   `protobuf:"bytes,3,opt,name=address,proto3" json:"address,omitempty"`
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *ConnInfo) Reset()         { *m = ConnInfo{} }
+func (m *ConnInfo) String() string { return proto.CompactTextString(m) }
+func (*ConnInfo) ProtoMessage()    {}
+func (*ConnInfo) Descriptor() ([]byte, []int) {
+	return fileDescriptor_802e9beed3ec3b28, []int{0}
+}
+
+func (m *ConnInfo) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_ConnInfo.Unmarshal(m, b)
+}
+func (m *ConnInfo) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_ConnInfo.Marshal(b, m, deterministic)
+}
+func (m *ConnInfo) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_ConnInfo.Merge(m, src)
+}
+func (m *ConnInfo) XXX_Size() int {
+	return xxx_messageInfo_ConnInfo.Size(m)
+}
+func (m *ConnInfo) XXX_DiscardUnknown() {
+	xxx_messageInfo_ConnInfo.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_ConnInfo proto.InternalMessageInfo
+
+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")
+}
+
+func init() { proto.RegisterFile("grpc_broker.proto", fileDescriptor_802e9beed3ec3b28) }
+
+var fileDescriptor_802e9beed3ec3b28 = []byte{
+	// 175 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, 0x9d, 0x38, 0xa2, 0xa0, 0xae, 0x4d, 0x62, 0x03, 0x3b,
+	0xde, 0x18, 0x10, 0x00, 0x00, 0xff, 0xff, 0x10, 0x15, 0x39, 0x47, 0xd1, 0x00, 0x00, 0x00,
+}
+
+// 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
+
+// GRPCBrokerClient is the client API for GRPCBroker service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
+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 := c.cc.NewStream(ctx, &_GRPCBroker_serviceDesc.Streams[0], "/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
+}
+
+// GRPCBrokerServer is the 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",
+}
@@ -0,0 +1,143 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// source: grpc_controller.proto
+
+package plugin
+
+import (
+	fmt "fmt"
+	proto "github.com/golang/protobuf/proto"
+	context "golang.org/x/net/context"
+	grpc "google.golang.org/grpc"
+	math "math"
+)
+
+// 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 Empty struct {
+	XXX_NoUnkeyedLiteral struct{} `json:"-"`
+	XXX_unrecognized     []byte   `json:"-"`
+	XXX_sizecache        int32    `json:"-"`
+}
+
+func (m *Empty) Reset()         { *m = Empty{} }
+func (m *Empty) String() string { return proto.CompactTextString(m) }
+func (*Empty) ProtoMessage()    {}
+func (*Empty) Descriptor() ([]byte, []int) {
+	return fileDescriptor_23c2c7e42feab570, []int{0}
+}
+
+func (m *Empty) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_Empty.Unmarshal(m, b)
+}
+func (m *Empty) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_Empty.Marshal(b, m, deterministic)
+}
+func (m *Empty) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_Empty.Merge(m, src)
+}
+func (m *Empty) XXX_Size() int {
+	return xxx_messageInfo_Empty.Size(m)
+}
+func (m *Empty) XXX_DiscardUnknown() {
+	xxx_messageInfo_Empty.DiscardUnknown(m)
+}
+
+var xxx_messageInfo_Empty proto.InternalMessageInfo
+
+func init() {
+	proto.RegisterType((*Empty)(nil), "plugin.Empty")
+}
+
+func init() { proto.RegisterFile("grpc_controller.proto", fileDescriptor_23c2c7e42feab570) }
+
+var fileDescriptor_23c2c7e42feab570 = []byte{
+	// 108 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0x12, 0x4d, 0x2f, 0x2a, 0x48,
+	0x8e, 0x4f, 0xce, 0xcf, 0x2b, 0x29, 0xca, 0xcf, 0xc9, 0x49, 0x2d, 0xd2, 0x2b, 0x28, 0xca, 0x2f,
+	0xc9, 0x17, 0x62, 0x2b, 0xc8, 0x29, 0x4d, 0xcf, 0xcc, 0x53, 0x62, 0xe7, 0x62, 0x75, 0xcd, 0x2d,
+	0x28, 0xa9, 0x34, 0xb2, 0xe2, 0xe2, 0x73, 0x0f, 0x0a, 0x70, 0x76, 0x86, 0x2b, 0x14, 0xd2, 0xe0,
+	0xe2, 0x08, 0xce, 0x28, 0x2d, 0x49, 0xc9, 0x2f, 0xcf, 0x13, 0xe2, 0xd5, 0x83, 0xa8, 0xd7, 0x03,
+	0x2b, 0x96, 0x42, 0xe5, 0x3a, 0x71, 0x44, 0x41, 0x8d, 0x4b, 0x62, 0x03, 0x9b, 0x6e, 0x0c, 0x08,
+	0x00, 0x00, 0xff, 0xff, 0xab, 0x7c, 0x27, 0xe5, 0x76, 0x00, 0x00, 0x00,
+}
+
+// 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
+
+// GRPCControllerClient is the client API for GRPCController service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
+type GRPCControllerClient interface {
+	Shutdown(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*Empty, error)
+}
+
+type gRPCControllerClient struct {
+	cc *grpc.ClientConn
+}
+
+func NewGRPCControllerClient(cc *grpc.ClientConn) GRPCControllerClient {
+	return &gRPCControllerClient{cc}
+}
+
+func (c *gRPCControllerClient) Shutdown(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*Empty, error) {
+	out := new(Empty)
+	err := c.cc.Invoke(ctx, "/plugin.GRPCController/Shutdown", in, out, opts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// GRPCControllerServer is the server API for GRPCController service.
+type GRPCControllerServer interface {
+	Shutdown(context.Context, *Empty) (*Empty, error)
+}
+
+func RegisterGRPCControllerServer(s *grpc.Server, srv GRPCControllerServer) {
+	s.RegisterService(&_GRPCController_serviceDesc, srv)
+}
+
+func _GRPCController_Shutdown_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(Empty)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(GRPCControllerServer).Shutdown(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: "/plugin.GRPCController/Shutdown",
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(GRPCControllerServer).Shutdown(ctx, req.(*Empty))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+var _GRPCController_serviceDesc = grpc.ServiceDesc{
+	ServiceName: "plugin.GRPCController",
+	HandlerType: (*GRPCControllerServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "Shutdown",
+			Handler:    _GRPCController_Shutdown_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "grpc_controller.proto",
+}
@@ -32,11 +32,11 @@ func flattenKVPairs(kvs []*logEntryKV) []interface{} {
 }

 // parseJSON handles parsing JSON output
-func parseJSON(input string) (*logEntry, error) {
+func parseJSON(input []byte) (*logEntry, error) {
 	var raw map[string]interface{}
 	entry := &logEntry{}

-	err := json.Unmarshal([]byte(input), &raw)
+	err := json.Unmarshal(input, &raw)
 	if err != nil {
 		return nil, err
 	}
@@ -0,0 +1,73 @@
+package plugin
+
+import (
+	"bytes"
+	"crypto/ecdsa"
+	"crypto/elliptic"
+	"crypto/rand"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
+	"math/big"
+	"time"
+)
+
+// generateCert generates a temporary certificate for plugin authentication. The
+// certificate and private key are returns in PEM format.
+func generateCert() (cert []byte, privateKey []byte, err error) {
+	key, err := ecdsa.GenerateKey(elliptic.P521(), rand.Reader)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	serialNumberLimit := new(big.Int).Lsh(big.NewInt(1), 128)
+	sn, err := rand.Int(rand.Reader, serialNumberLimit)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	host := "localhost"
+
+	template := &x509.Certificate{
+		Subject: pkix.Name{
+			CommonName:   host,
+			Organization: []string{"HashiCorp"},
+		},
+		DNSNames: []string{host},
+		ExtKeyUsage: []x509.ExtKeyUsage{
+			x509.ExtKeyUsageClientAuth,
+			x509.ExtKeyUsageServerAuth,
+		},
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment | x509.KeyUsageKeyAgreement | x509.KeyUsageCertSign,
+		BasicConstraintsValid: true,
+		SerialNumber:          sn,
+		NotBefore:             time.Now().Add(-30 * time.Second),
+		NotAfter:              time.Now().Add(262980 * time.Hour),
+		IsCA:                  true,
+	}
+
+	der, err := x509.CreateCertificate(rand.Reader, template, template, key.Public(), key)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	var certOut bytes.Buffer
+	if err := pem.Encode(&certOut, &pem.Block{Type: "CERTIFICATE", Bytes: der}); err != nil {
+		return nil, nil, err
+	}
+
+	keyBytes, err := x509.MarshalECPrivateKey(key)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	var keyOut bytes.Buffer
+	if err := pem.Encode(&keyOut, &pem.Block{Type: "EC PRIVATE KEY", Bytes: keyBytes}); err != nil {
+		return nil, nil, err
+	}
+
+	cert = certOut.Bytes()
+	privateKey = keyOut.Bytes()
+
+	return cert, privateKey, nil
+}
@@ -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
@@ -2,6 +2,7 @@ package plugin

 import (
 	"crypto/tls"
+	"crypto/x509"
 	"encoding/base64"
 	"errors"
 	"fmt"
@@ -11,7 +12,9 @@ import (
 	"os"
 	"os/signal"
 	"runtime"
+	"sort"
 	"strconv"
+	"strings"
 	"sync/atomic"

 	"github.com/hashicorp/go-hclog"
@@ -36,6 +39,8 @@ type HandshakeConfig struct {
 	// ProtocolVersion is the version that clients must match on to
 	// agree they can communicate. This should match the ProtocolVersion
 	// set on ClientConfig when using a plugin.
+	// This field is not required if VersionedPlugins are being used in the
+	// Client or Server configurations.
 	ProtocolVersion uint

 	// MagicCookieKey and value are used as a very basic verification
@@ -46,6 +51,10 @@ type HandshakeConfig struct {
 	MagicCookieValue string
 }

+// PluginSet is a set of plugins provided to be registered in the plugin
+// server.
+type PluginSet map[string]Plugin
+
 // ServeConfig configures what sorts of plugins are served.
 type ServeConfig struct {
 	// HandshakeConfig is the configuration that must match clients.
@@ -55,7 +64,13 @@ type ServeConfig struct {
 	TLSProvider func() (*tls.Config, error)

 	// Plugins are the plugins that are served.
-	Plugins map[string]Plugin
+	// The implied version of this PluginSet is the Handshake.ProtocolVersion.
+	Plugins PluginSet
+
+	// VersionedPlugins is a map of PluginSets for specific protocol versions.
+	// These can be used to negotiate a compatible version between client and
+	// server. If this is set, Handshake.ProtocolVersion is not required.
+	VersionedPlugins map[int]PluginSet

 	// GRPCServer should be non-nil to enable serving the plugins over
 	// gRPC. This is a function to create the server when needed with the
@@ -66,16 +81,89 @@ 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.
-func (c *ServeConfig) Protocol() Protocol {
-	result := ProtocolNetRPC
-	if c.GRPCServer != nil {
-		result = ProtocolGRPC
+// protocolVersion determines the protocol version and plugin set to be used by
+// the server. In the event that there is no suitable version, the last version
+// in the config is returned leaving the client to report the incompatibility.
+func protocolVersion(opts *ServeConfig) (int, Protocol, PluginSet) {
+	protoVersion := int(opts.ProtocolVersion)
+	pluginSet := opts.Plugins
+	protoType := ProtocolNetRPC
+	// Check if the client sent a list of acceptable versions
+	var clientVersions []int
+	if vs := os.Getenv("PLUGIN_PROTOCOL_VERSIONS"); vs != "" {
+		for _, s := range strings.Split(vs, ",") {
+			v, err := strconv.Atoi(s)
+			if err != nil {
+				fmt.Fprintf(os.Stderr, "server sent invalid plugin version %q", s)
+				continue
+			}
+			clientVersions = append(clientVersions, v)
+		}
 	}

-	return result
+	// We want to iterate in reverse order, to ensure we match the newest
+	// compatible plugin version.
+	sort.Sort(sort.Reverse(sort.IntSlice(clientVersions)))
+
+	// set the old un-versioned fields as if they were versioned plugins
+	if opts.VersionedPlugins == nil {
+		opts.VersionedPlugins = make(map[int]PluginSet)
+	}
+
+	if pluginSet != nil {
+		opts.VersionedPlugins[protoVersion] = pluginSet
+	}
+
+	// Sort the version to make sure we match the latest first
+	var versions []int
+	for v := range opts.VersionedPlugins {
+		versions = append(versions, v)
+	}
+
+	sort.Sort(sort.Reverse(sort.IntSlice(versions)))
+
+	// See if we have multiple versions of Plugins to choose from
+	for _, version := range versions {
+		// Record each version, since we guarantee that this returns valid
+		// values even if they are not a protocol match.
+		protoVersion = version
+		pluginSet = opts.VersionedPlugins[version]
+
+		// If we have a configured gRPC server we should select a protocol
+		if opts.GRPCServer != nil {
+			// All plugins in a set must use the same transport, so check the first
+			// for the protocol type
+			for _, p := range pluginSet {
+				switch p.(type) {
+				case GRPCPlugin:
+					protoType = ProtocolGRPC
+				default:
+					protoType = ProtocolNetRPC
+				}
+				break
+			}
+		}
+
+		for _, clientVersion := range clientVersions {
+			if clientVersion == protoVersion {
+				return protoVersion, protoType, pluginSet
+			}
+		}
+	}
+
+	// Return the lowest version as the fallback.
+	// Since we iterated over all the versions in reverse order above, these
+	// values are from the lowest version number plugins (which may be from
+	// a combination of the Handshake.ProtocolVersion and ServeConfig.Plugins
+	// fields). This allows serving the oldest version of our plugins to a
+	// legacy client that did not send a PLUGIN_PROTOCOL_VERSIONS list.
+	return protoVersion, protoType, pluginSet
 }

 // Serve serves the plugins given by ServeConfig.
@@ -103,15 +191,22 @@ func Serve(opts *ServeConfig) {
 		os.Exit(1)
 	}

+	// negotiate the version and plugins
+	// start with default version in the handshake config
+	protoVersion, protoType, pluginSet := protocolVersion(opts)
+
 	// 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.
@@ -148,12 +243,47 @@ func Serve(opts *ServeConfig) {
 		}
 	}

+	var serverCert string
+	clientCert := os.Getenv("PLUGIN_CLIENT_CERT")
+	// If the client is configured using AutoMTLS, the certificate will be here,
+	// and we need to generate our own in response.
+	if tlsConfig == nil && clientCert != "" {
+		logger.Info("configuring server automatic mTLS")
+		clientCertPool := x509.NewCertPool()
+		if !clientCertPool.AppendCertsFromPEM([]byte(clientCert)) {
+			logger.Error("client cert provided but failed to parse", "cert", clientCert)
+		}
+
+		certPEM, keyPEM, err := generateCert()
+		if err != nil {
+			logger.Error("failed to generate client certificate", "error", err)
+			panic(err)
+		}
+
+		cert, err := tls.X509KeyPair(certPEM, keyPEM)
+		if err != nil {
+			logger.Error("failed to parse client certificate", "error", err)
+			panic(err)
+		}
+
+		tlsConfig = &tls.Config{
+			Certificates: []tls.Certificate{cert},
+			ClientAuth:   tls.RequireAndVerifyClientCert,
+			ClientCAs:    clientCertPool,
+			MinVersion:   tls.VersionTLS12,
+		}
+
+		// We send back the raw leaf cert data for the client rather than the
+		// PEM, since the protocol can't handle newlines.
+		serverCert = base64.RawStdEncoding.EncodeToString(cert.Certificate[0])
+	}
+
 	// Create the channel to tell us when we're done
 	doneCh := make(chan struct{})

 	// Build the server type
 	var server ServerProtocol
-	switch opts.Protocol() {
+	switch protoType {
 	case ProtocolNetRPC:
 		// If we have a TLS configuration then we wrap the listener
 		// ourselves and do it at that level.
@@ -163,7 +293,7 @@ func Serve(opts *ServeConfig) {

 		// Create the RPC server to dispense
 		server = &RPCServer{
-			Plugins: opts.Plugins,
+			Plugins: pluginSet,
 			Stdout:  stdout_r,
 			Stderr:  stderr_r,
 			DoneCh:  doneCh,
@@ -172,16 +302,17 @@ func Serve(opts *ServeConfig) {
 	case ProtocolGRPC:
 		// Create the gRPC server
 		server = &GRPCServer{
-			Plugins: opts.Plugins,
+			Plugins: pluginSet,
 			Server:  opts.GRPCServer,
 			TLS:     tlsConfig,
 			Stdout:  stdout_r,
 			Stderr:  stderr_r,
 			DoneCh:  doneCh,
+			logger:  logger,
 		}

 	default:
-		panic("unknown server protocol: " + opts.Protocol())
+		panic("unknown server protocol: " + protoType)
 	}

 	// Initialize the servers
@@ -190,25 +321,16 @@ func Serve(opts *ServeConfig) {
 		return
 	}

-	// Build the extra configuration
-	extra := ""
-	if v := server.Config(); v != "" {
-		extra = base64.StdEncoding.EncodeToString([]byte(v))
-	}
-	if extra != "" {
-		extra = "|" + extra
-	}
-
 	logger.Debug("plugin address", "network", listener.Addr().Network(), "address", listener.Addr().String())

-	// Output the address and service name to stdout so that core can bring it up.
-	fmt.Printf("%d|%d|%s|%s|%s%s\n",
+	// Output the address and service name to stdout so that the client can bring it up.
+	fmt.Printf("%d|%d|%s|%s|%s|%s\n",
 		CoreProtocolVersion,
-		opts.ProtocolVersion,
+		protoVersion,
 		listener.Addr().Network(),
 		listener.Addr().String(),
-		opts.Protocol(),
-		extra)
+		protoType,
+		serverCert)
 	os.Stdout.Sync()

 	// Eat the interrupts
@@ -2,13 +2,29 @@ package plugin

 import (
 	"bytes"
+	"context"
+	"io"
 	"net"
 	"net/rpc"

 	"github.com/mitchellh/go-testing-interface"
+	hclog "github.com/hashicorp/go-hclog"
+	"github.com/hashicorp/go-plugin/internal/plugin"
 	"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 +76,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 +101,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) {
@@ -89,9 +142,11 @@ func TestPluginGRPCConn(t testing.T, ps map[string]Plugin) (*GRPCClient, *GRPCSe
 	// Start up the server
 	server := &GRPCServer{
 		Plugins: ps,
+		DoneCh:  make(chan struct{}),
 		Server:  DefaultGRPCServer,
 		Stdout:  new(bytes.Buffer),
 		Stderr:  new(bytes.Buffer),
+		logger:  hclog.Default(),
 	}
 	if err := server.Init(); err != nil {
 		t.Fatalf("err: %s", err)
@@ -107,13 +162,18 @@ 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,
+		Conn:       conn,
+		Plugins:    ps,
+		broker:     broker,
+		doneCtx:    context.Background(),
+		controller: plugin.NewGRPCControllerClient(conn),
 	}

 	return client, server