aboutsummaryrefslogtreecommitdiffhomepage
path: root/vendor/github.com/hashicorp/go-plugin
diff options
context:
space:
mode:
Diffstat (limited to 'vendor/github.com/hashicorp/go-plugin')
-rw-r--r--vendor/github.com/hashicorp/go-plugin/README.md49
-rw-r--r--vendor/github.com/hashicorp/go-plugin/client.go293
-rw-r--r--vendor/github.com/hashicorp/go-plugin/grpc_broker.go455
-rw-r--r--vendor/github.com/hashicorp/go-plugin/grpc_broker.pb.go190
-rw-r--r--vendor/github.com/hashicorp/go-plugin/grpc_broker.proto14
-rw-r--r--vendor/github.com/hashicorp/go-plugin/grpc_client.go107
-rw-r--r--vendor/github.com/hashicorp/go-plugin/grpc_server.go132
-rw-r--r--vendor/github.com/hashicorp/go-plugin/log_entry.go73
-rw-r--r--vendor/github.com/hashicorp/go-plugin/plugin.go33
-rw-r--r--vendor/github.com/hashicorp/go-plugin/protocol.go45
-rw-r--r--vendor/github.com/hashicorp/go-plugin/rpc_client.go47
-rw-r--r--vendor/github.com/hashicorp/go-plugin/rpc_server.go20
-rw-r--r--vendor/github.com/hashicorp/go-plugin/server.go135
-rw-r--r--vendor/github.com/hashicorp/go-plugin/testing.go86
14 files changed, 1590 insertions, 89 deletions
diff --git a/vendor/github.com/hashicorp/go-plugin/README.md b/vendor/github.com/hashicorp/go-plugin/README.md
index 2058cfb..e4558db 100644
--- a/vendor/github.com/hashicorp/go-plugin/README.md
+++ b/vendor/github.com/hashicorp/go-plugin/README.md
@@ -1,10 +1,9 @@
1# Go Plugin System over RPC 1# Go Plugin System over RPC
2 2
3`go-plugin` is a Go (golang) plugin system over RPC. It is the plugin system 3`go-plugin` is a Go (golang) plugin system over RPC. It is the plugin system
4that has been in use by HashiCorp tooling for over 3 years. While initially 4that has been in use by HashiCorp tooling for over 4 years. While initially
5created for [Packer](https://www.packer.io), it has since been used by 5created for [Packer](https://www.packer.io), it is additionally in use by
6[Terraform](https://www.terraform.io) and [Otto](https://www.ottoproject.io), 6[Terraform](https://www.terraform.io), [Nomad](https://www.nomadproject.io), and
7with plans to also use it for [Nomad](https://www.nomadproject.io) and
8[Vault](https://www.vaultproject.io). 7[Vault](https://www.vaultproject.io).
9 8
10While the plugin system is over RPC, it is currently only designed to work 9While the plugin system is over RPC, it is currently only designed to work
@@ -24,6 +23,11 @@ interface as if it were going to run in the same process. For a plugin user:
24you just use and call functions on an interface as if it were in the same 23you just use and call functions on an interface as if it were in the same
25process. This plugin system handles the communication in between. 24process. This plugin system handles the communication in between.
26 25
26**Cross-language support.** Plugins can be written (and consumed) by
27almost every major language. This library supports serving plugins via
28[gRPC](http://www.grpc.io). gRPC-based plugins enable plugins to be written
29in any language.
30
27**Complex arguments and return values are supported.** This library 31**Complex arguments and return values are supported.** This library
28provides APIs for handling complex arguments and return values such 32provides APIs for handling complex arguments and return values such
29as interfaces, `io.Reader/Writer`, etc. We do this by giving you a library 33as interfaces, `io.Reader/Writer`, etc. We do this by giving you a library
@@ -37,7 +41,10 @@ and the plugin can call back into the host process.
37**Built-in Logging.** Any plugins that use the `log` standard library 41**Built-in Logging.** Any plugins that use the `log` standard library
38will have log data automatically sent to the host process. The host 42will have log data automatically sent to the host process. The host
39process will mirror this output prefixed with the path to the plugin 43process will mirror this output prefixed with the path to the plugin
40binary. This makes debugging with plugins simple. 44binary. This makes debugging with plugins simple. If the host system
45uses [hclog](https://github.com/hashicorp/go-hclog) then the log data
46will be structured. If the plugin also uses hclog, logs from the plugin
47will be sent to the host hclog and be structured.
41 48
42**Protocol Versioning.** A very basic "protocol version" is supported that 49**Protocol Versioning.** A very basic "protocol version" is supported that
43can be incremented to invalidate any previous plugins. This is useful when 50can be incremented to invalidate any previous plugins. This is useful when
@@ -62,13 +69,18 @@ This requires the host/plugin to know this is possible and daemonize
62properly. `NewClient` takes a `ReattachConfig` to determine if and how to 69properly. `NewClient` takes a `ReattachConfig` to determine if and how to
63reattach. 70reattach.
64 71
72**Cryptographically Secure Plugins.** Plugins can be verified with an expected
73checksum and RPC communications can be configured to use TLS. The host process
74must be properly secured to protect this configuration.
75
65## Architecture 76## Architecture
66 77
67The HashiCorp plugin system works by launching subprocesses and communicating 78The HashiCorp plugin system works by launching subprocesses and communicating
68over RPC (using standard `net/rpc`). A single connection is made between 79over RPC (using standard `net/rpc` or [gRPC](http://www.grpc.io)). A single
69any plugin and the host process, and we use a 80connection is made between any plugin and the host process. For net/rpc-based
70[connection multiplexing](https://github.com/hashicorp/yamux) 81plugins, we use a [connection multiplexing](https://github.com/hashicorp/yamux)
71library to multiplex any other connections on top. 82library to multiplex any other connections on top. For gRPC-based plugins,
83the HTTP2 protocol handles multiplexing.
72 84
73This architecture has a number of benefits: 85This architecture has a number of benefits:
74 86
@@ -76,8 +88,8 @@ This architecture has a number of benefits:
76 panic the plugin user. 88 panic the plugin user.
77 89
78 * Plugins are very easy to write: just write a Go application and `go build`. 90 * Plugins are very easy to write: just write a Go application and `go build`.
79 Theoretically you could also use another language as long as it can 91 Or use any other language to write a gRPC server with a tiny amount of
80 communicate the Go `net/rpc` protocol but this hasn't yet been tried. 92 boilerplate to support go-plugin.
81 93
82 * Plugins are very easy to install: just put the binary in a location where 94 * Plugins are very easy to install: just put the binary in a location where
83 the host will find it (depends on the host but this library also provides 95 the host will find it (depends on the host but this library also provides
@@ -85,8 +97,8 @@ This architecture has a number of benefits:
85 97
86 * Plugins can be relatively secure: The plugin only has access to the 98 * Plugins can be relatively secure: The plugin only has access to the
87 interfaces and args given to it, not to the entire memory space of the 99 interfaces and args given to it, not to the entire memory space of the
88 process. More security features are planned (see the coming soon section 100 process. Additionally, go-plugin can communicate with the plugin over
89 below). 101 TLS.
90 102
91## Usage 103## Usage
92 104
@@ -97,10 +109,9 @@ high-level steps that must be done. Examples are available in the
97 1. Choose the interface(s) you want to expose for plugins. 109 1. Choose the interface(s) you want to expose for plugins.
98 110
99 2. For each interface, implement an implementation of that interface 111 2. For each interface, implement an implementation of that interface
100 that communicates over an `*rpc.Client` (from the standard `net/rpc` 112 that communicates over a `net/rpc` connection or other a
101 package) for every function call. Likewise, implement the RPC server 113 [gRPC](http://www.grpc.io) connection or both. You'll have to implement
102 struct this communicates to which is then communicating to a real, 114 both a client and server implementation.
103 concrete implementation.
104 115
105 3. Create a `Plugin` implementation that knows how to create the RPC 116 3. Create a `Plugin` implementation that knows how to create the RPC
106 client/server for a given plugin type. 117 client/server for a given plugin type.
@@ -125,10 +136,6 @@ improvements we can make.
125 136
126At this point in time, the roadmap for the plugin system is: 137At this point in time, the roadmap for the plugin system is:
127 138
128**Cryptographically Secure Plugins.** We'll implement signing plugins
129and loading signed plugins in order to allow Vault to make use of multi-process
130in a secure way.
131
132**Semantic Versioning.** Plugins will be able to implement a semantic version. 139**Semantic Versioning.** Plugins will be able to implement a semantic version.
133This plugin system will give host processes a system for constraining 140This plugin system will give host processes a system for constraining
134versions. This is in addition to the protocol versioning already present 141versions. This is in addition to the protocol versioning already present
diff --git a/vendor/github.com/hashicorp/go-plugin/client.go b/vendor/github.com/hashicorp/go-plugin/client.go
index 9f8a0f2..b3e3b78 100644
--- a/vendor/github.com/hashicorp/go-plugin/client.go
+++ b/vendor/github.com/hashicorp/go-plugin/client.go
@@ -2,8 +2,12 @@ package plugin
2 2
3import ( 3import (
4 "bufio" 4 "bufio"
5 "context"
6 "crypto/subtle"
7 "crypto/tls"
5 "errors" 8 "errors"
6 "fmt" 9 "fmt"
10 "hash"
7 "io" 11 "io"
8 "io/ioutil" 12 "io/ioutil"
9 "log" 13 "log"
@@ -17,6 +21,8 @@ import (
17 "sync/atomic" 21 "sync/atomic"
18 "time" 22 "time"
19 "unicode" 23 "unicode"
24
25 hclog "github.com/hashicorp/go-hclog"
20) 26)
21 27
22// If this is 1, then we've called CleanupClients. This can be used 28// If this is 1, then we've called CleanupClients. This can be used
@@ -35,6 +41,22 @@ var (
35 // ErrProcessNotFound is returned when a client is instantiated to 41 // ErrProcessNotFound is returned when a client is instantiated to
36 // reattach to an existing process and it isn't found. 42 // reattach to an existing process and it isn't found.
37 ErrProcessNotFound = errors.New("Reattachment process not found") 43 ErrProcessNotFound = errors.New("Reattachment process not found")
44
45 // ErrChecksumsDoNotMatch is returned when binary's checksum doesn't match
46 // the one provided in the SecureConfig.
47 ErrChecksumsDoNotMatch = errors.New("checksums did not match")
48
49 // ErrSecureNoChecksum is returned when an empty checksum is provided to the
50 // SecureConfig.
51 ErrSecureConfigNoChecksum = errors.New("no checksum provided")
52
53 // ErrSecureNoHash is returned when a nil Hash object is provided to the
54 // SecureConfig.
55 ErrSecureConfigNoHash = errors.New("no hash implementation provided")
56
57 // ErrSecureConfigAndReattach is returned when both Reattach and
58 // SecureConfig are set.
59 ErrSecureConfigAndReattach = errors.New("only one of Reattach or SecureConfig can be set")
38) 60)
39 61
40// Client handles the lifecycle of a plugin application. It launches 62// Client handles the lifecycle of a plugin application. It launches
@@ -55,7 +77,10 @@ type Client struct {
55 l sync.Mutex 77 l sync.Mutex
56 address net.Addr 78 address net.Addr
57 process *os.Process 79 process *os.Process
58 client *RPCClient 80 client ClientProtocol
81 protocol Protocol
82 logger hclog.Logger
83 doneCtx context.Context
59} 84}
60 85
61// ClientConfig is the configuration used to initialize a new 86// ClientConfig is the configuration used to initialize a new
@@ -79,6 +104,13 @@ type ClientConfig struct {
79 Cmd *exec.Cmd 104 Cmd *exec.Cmd
80 Reattach *ReattachConfig 105 Reattach *ReattachConfig
81 106
107 // SecureConfig is configuration for verifying the integrity of the
108 // executable. It can not be used with Reattach.
109 SecureConfig *SecureConfig
110
111 // TLSConfig is used to enable TLS on the RPC client.
112 TLSConfig *tls.Config
113
82 // Managed represents if the client should be managed by the 114 // Managed represents if the client should be managed by the
83 // plugin package or not. If true, then by calling CleanupClients, 115 // plugin package or not. If true, then by calling CleanupClients,
84 // it will automatically be cleaned up. Otherwise, the client 116 // it will automatically be cleaned up. Otherwise, the client
@@ -109,14 +141,74 @@ type ClientConfig struct {
109 // sync any of these streams. 141 // sync any of these streams.
110 SyncStdout io.Writer 142 SyncStdout io.Writer
111 SyncStderr io.Writer 143 SyncStderr io.Writer
144
145 // AllowedProtocols is a list of allowed protocols. If this isn't set,
146 // then only netrpc is allowed. This is so that older go-plugin systems
147 // can show friendly errors if they see a plugin with an unknown
148 // protocol.
149 //
150 // By setting this, you can cause an error immediately on plugin start
151 // if an unsupported protocol is used with a good error message.
152 //
153 // If this isn't set at all (nil value), then only net/rpc is accepted.
154 // This is done for legacy reasons. You must explicitly opt-in to
155 // new protocols.
156 AllowedProtocols []Protocol
157
158 // Logger is the logger that the client will used. If none is provided,
159 // it will default to hclog's default logger.
160 Logger hclog.Logger
112} 161}
113 162
114// ReattachConfig is used to configure a client to reattach to an 163// ReattachConfig is used to configure a client to reattach to an
115// already-running plugin process. You can retrieve this information by 164// already-running plugin process. You can retrieve this information by
116// calling ReattachConfig on Client. 165// calling ReattachConfig on Client.
117type ReattachConfig struct { 166type ReattachConfig struct {
118 Addr net.Addr 167 Protocol Protocol
119 Pid int 168 Addr net.Addr
169 Pid int
170}
171
172// SecureConfig is used to configure a client to verify the integrity of an
173// executable before running. It does this by verifying the checksum is
174// expected. Hash is used to specify the hashing method to use when checksumming
175// the file. The configuration is verified by the client by calling the
176// SecureConfig.Check() function.
177//
178// The host process should ensure the checksum was provided by a trusted and
179// authoritative source. The binary should be installed in such a way that it
180// can not be modified by an unauthorized user between the time of this check
181// and the time of execution.
182type SecureConfig struct {
183 Checksum []byte
184 Hash hash.Hash
185}
186
187// Check takes the filepath to an executable and returns true if the checksum of
188// the file matches the checksum provided in the SecureConfig.
189func (s *SecureConfig) Check(filePath string) (bool, error) {
190 if len(s.Checksum) == 0 {
191 return false, ErrSecureConfigNoChecksum
192 }
193
194 if s.Hash == nil {
195 return false, ErrSecureConfigNoHash
196 }
197
198 file, err := os.Open(filePath)
199 if err != nil {
200 return false, err
201 }
202 defer file.Close()
203
204 _, err = io.Copy(s.Hash, file)
205 if err != nil {
206 return false, err
207 }
208
209 sum := s.Hash.Sum(nil)
210
211 return subtle.ConstantTimeCompare(sum, s.Checksum) == 1, nil
120} 212}
121 213
122// This makes sure all the managed subprocesses are killed and properly 214// This makes sure all the managed subprocesses are killed and properly
@@ -174,7 +266,22 @@ func NewClient(config *ClientConfig) (c *Client) {
174 config.SyncStderr = ioutil.Discard 266 config.SyncStderr = ioutil.Discard
175 } 267 }
176 268
177 c = &Client{config: config} 269 if config.AllowedProtocols == nil {
270 config.AllowedProtocols = []Protocol{ProtocolNetRPC}
271 }
272
273 if config.Logger == nil {
274 config.Logger = hclog.New(&hclog.LoggerOptions{
275 Output: hclog.DefaultOutput,
276 Level: hclog.Trace,
277 Name: "plugin",
278 })
279 }
280
281 c = &Client{
282 config: config,
283 logger: config.Logger,
284 }
178 if config.Managed { 285 if config.Managed {
179 managedClientsLock.Lock() 286 managedClientsLock.Lock()
180 managedClients = append(managedClients, c) 287 managedClients = append(managedClients, c)
@@ -184,11 +291,11 @@ func NewClient(config *ClientConfig) (c *Client) {
184 return 291 return
185} 292}
186 293
187// Client returns an RPC client for the plugin. 294// Client returns the protocol client for this connection.
188// 295//
189// Subsequent calls to this will return the same RPC client. 296// Subsequent calls to this will return the same client.
190func (c *Client) Client() (*RPCClient, error) { 297func (c *Client) Client() (ClientProtocol, error) {
191 addr, err := c.Start() 298 _, err := c.Start()
192 if err != nil { 299 if err != nil {
193 return nil, err 300 return nil, err
194 } 301 }
@@ -200,29 +307,18 @@ func (c *Client) Client() (*RPCClient, error) {
200 return c.client, nil 307 return c.client, nil
201 } 308 }
202 309
203 // Connect to the client 310 switch c.protocol {
204 conn, err := net.Dial(addr.Network(), addr.String()) 311 case ProtocolNetRPC:
205 if err != nil { 312 c.client, err = newRPCClient(c)
206 return nil, err
207 }
208 if tcpConn, ok := conn.(*net.TCPConn); ok {
209 // Make sure to set keep alive so that the connection doesn't die
210 tcpConn.SetKeepAlive(true)
211 }
212 313
213 // Create the actual RPC client 314 case ProtocolGRPC:
214 c.client, err = NewRPCClient(conn, c.config.Plugins) 315 c.client, err = newGRPCClient(c.doneCtx, c)
215 if err != nil { 316
216 conn.Close() 317 default:
217 return nil, err 318 return nil, fmt.Errorf("unknown server protocol: %s", c.protocol)
218 } 319 }
219 320
220 // Begin the stream syncing so that stdin, out, err work properly
221 err = c.client.SyncStreams(
222 c.config.SyncStdout,
223 c.config.SyncStderr)
224 if err != nil { 321 if err != nil {
225 c.client.Close()
226 c.client = nil 322 c.client = nil
227 return nil, err 323 return nil, err
228 } 324 }
@@ -274,8 +370,7 @@ func (c *Client) Kill() {
274 if err != nil { 370 if err != nil {
275 // If there was an error just log it. We're going to force 371 // If there was an error just log it. We're going to force
276 // kill in a moment anyways. 372 // kill in a moment anyways.
277 log.Printf( 373 c.logger.Warn("error closing client during Kill", "err", err)
278 "[WARN] plugin: error closing client during Kill: %s", err)
279 } 374 }
280 } 375 }
281 } 376 }
@@ -318,13 +413,21 @@ func (c *Client) Start() (addr net.Addr, err error) {
318 { 413 {
319 cmdSet := c.config.Cmd != nil 414 cmdSet := c.config.Cmd != nil
320 attachSet := c.config.Reattach != nil 415 attachSet := c.config.Reattach != nil
416 secureSet := c.config.SecureConfig != nil
321 if cmdSet == attachSet { 417 if cmdSet == attachSet {
322 return nil, fmt.Errorf("Only one of Cmd or Reattach must be set") 418 return nil, fmt.Errorf("Only one of Cmd or Reattach must be set")
323 } 419 }
420
421 if secureSet && attachSet {
422 return nil, ErrSecureConfigAndReattach
423 }
324 } 424 }
325 425
326 // Create the logging channel for when we kill 426 // Create the logging channel for when we kill
327 c.doneLogging = make(chan struct{}) 427 c.doneLogging = make(chan struct{})
428 // Create a context for when we kill
429 var ctxCancel context.CancelFunc
430 c.doneCtx, ctxCancel = context.WithCancel(context.Background())
328 431
329 if c.config.Reattach != nil { 432 if c.config.Reattach != nil {
330 // Verify the process still exists. If not, then it is an error 433 // Verify the process still exists. If not, then it is an error
@@ -350,7 +453,7 @@ func (c *Client) Start() (addr net.Addr, err error) {
350 pidWait(pid) 453 pidWait(pid)
351 454
352 // Log so we can see it 455 // Log so we can see it
353 log.Printf("[DEBUG] plugin: reattached plugin process exited\n") 456 c.logger.Debug("reattached plugin process exited")
354 457
355 // Mark it 458 // Mark it
356 c.l.Lock() 459 c.l.Lock()
@@ -359,11 +462,19 @@ func (c *Client) Start() (addr net.Addr, err error) {
359 462
360 // Close the logging channel since that doesn't work on reattach 463 // Close the logging channel since that doesn't work on reattach
361 close(c.doneLogging) 464 close(c.doneLogging)
465
466 // Cancel the context
467 ctxCancel()
362 }(p.Pid) 468 }(p.Pid)
363 469
364 // Set the address and process 470 // Set the address and process
365 c.address = c.config.Reattach.Addr 471 c.address = c.config.Reattach.Addr
366 c.process = p 472 c.process = p
473 c.protocol = c.config.Reattach.Protocol
474 if c.protocol == "" {
475 // Default the protocol to net/rpc for backwards compatibility
476 c.protocol = ProtocolNetRPC
477 }
367 478
368 return c.address, nil 479 return c.address, nil
369 } 480 }
@@ -384,7 +495,15 @@ func (c *Client) Start() (addr net.Addr, err error) {
384 cmd.Stderr = stderr_w 495 cmd.Stderr = stderr_w
385 cmd.Stdout = stdout_w 496 cmd.Stdout = stdout_w
386 497
387 log.Printf("[DEBUG] plugin: starting plugin: %s %#v", cmd.Path, cmd.Args) 498 if c.config.SecureConfig != nil {
499 if ok, err := c.config.SecureConfig.Check(cmd.Path); err != nil {
500 return nil, fmt.Errorf("error verifying checksum: %s", err)
501 } else if !ok {
502 return nil, ErrChecksumsDoNotMatch
503 }
504 }
505
506 c.logger.Debug("starting plugin", "path", cmd.Path, "args", cmd.Args)
388 err = cmd.Start() 507 err = cmd.Start()
389 if err != nil { 508 if err != nil {
390 return 509 return
@@ -418,12 +537,15 @@ func (c *Client) Start() (addr net.Addr, err error) {
418 cmd.Wait() 537 cmd.Wait()
419 538
420 // Log and make sure to flush the logs write away 539 // Log and make sure to flush the logs write away
421 log.Printf("[DEBUG] plugin: %s: plugin process exited\n", cmd.Path) 540 c.logger.Debug("plugin process exited", "path", cmd.Path)
422 os.Stderr.Sync() 541 os.Stderr.Sync()
423 542
424 // Mark that we exited 543 // Mark that we exited
425 close(exitCh) 544 close(exitCh)
426 545
546 // Cancel the context, marking that we exited
547 ctxCancel()
548
427 // Set that we exited, which takes a lock 549 // Set that we exited, which takes a lock
428 c.l.Lock() 550 c.l.Lock()
429 defer c.l.Unlock() 551 defer c.l.Unlock()
@@ -465,7 +587,7 @@ func (c *Client) Start() (addr net.Addr, err error) {
465 timeout := time.After(c.config.StartTimeout) 587 timeout := time.After(c.config.StartTimeout)
466 588
467 // Start looking for the address 589 // Start looking for the address
468 log.Printf("[DEBUG] plugin: waiting for RPC address for: %s", cmd.Path) 590 c.logger.Debug("waiting for RPC address", "path", cmd.Path)
469 select { 591 select {
470 case <-timeout: 592 case <-timeout:
471 err = errors.New("timeout while waiting for plugin to start") 593 err = errors.New("timeout while waiting for plugin to start")
@@ -475,7 +597,7 @@ func (c *Client) Start() (addr net.Addr, err error) {
475 // Trim the line and split by "|" in order to get the parts of 597 // Trim the line and split by "|" in order to get the parts of
476 // the output. 598 // the output.
477 line := strings.TrimSpace(string(lineBytes)) 599 line := strings.TrimSpace(string(lineBytes))
478 parts := strings.SplitN(line, "|", 4) 600 parts := strings.SplitN(line, "|", 6)
479 if len(parts) < 4 { 601 if len(parts) < 4 {
480 err = fmt.Errorf( 602 err = fmt.Errorf(
481 "Unrecognized remote plugin message: %s\n\n"+ 603 "Unrecognized remote plugin message: %s\n\n"+
@@ -495,7 +617,7 @@ func (c *Client) Start() (addr net.Addr, err error) {
495 617
496 if int(coreProtocol) != CoreProtocolVersion { 618 if int(coreProtocol) != CoreProtocolVersion {
497 err = fmt.Errorf("Incompatible core API version with plugin. "+ 619 err = fmt.Errorf("Incompatible core API version with plugin. "+
498 "Plugin version: %s, Ours: %d\n\n"+ 620 "Plugin version: %s, Core version: %d\n\n"+
499 "To fix this, the plugin usually only needs to be recompiled.\n"+ 621 "To fix this, the plugin usually only needs to be recompiled.\n"+
500 "Please report this to the plugin author.", parts[0], CoreProtocolVersion) 622 "Please report this to the plugin author.", parts[0], CoreProtocolVersion)
501 return 623 return
@@ -513,7 +635,7 @@ func (c *Client) Start() (addr net.Addr, err error) {
513 // Test the API version 635 // Test the API version
514 if uint(protocol) != c.config.ProtocolVersion { 636 if uint(protocol) != c.config.ProtocolVersion {
515 err = fmt.Errorf("Incompatible API version with plugin. "+ 637 err = fmt.Errorf("Incompatible API version with plugin. "+
516 "Plugin version: %s, Ours: %d", parts[1], c.config.ProtocolVersion) 638 "Plugin version: %s, Core version: %d", parts[1], c.config.ProtocolVersion)
517 return 639 return
518 } 640 }
519 641
@@ -525,6 +647,27 @@ func (c *Client) Start() (addr net.Addr, err error) {
525 default: 647 default:
526 err = fmt.Errorf("Unknown address type: %s", parts[3]) 648 err = fmt.Errorf("Unknown address type: %s", parts[3])
527 } 649 }
650
651 // If we have a server type, then record that. We default to net/rpc
652 // for backwards compatibility.
653 c.protocol = ProtocolNetRPC
654 if len(parts) >= 5 {
655 c.protocol = Protocol(parts[4])
656 }
657
658 found := false
659 for _, p := range c.config.AllowedProtocols {
660 if p == c.protocol {
661 found = true
662 break
663 }
664 }
665 if !found {
666 err = fmt.Errorf("Unsupported plugin protocol %q. Supported: %v",
667 c.protocol, c.config.AllowedProtocols)
668 return
669 }
670
528 } 671 }
529 672
530 c.address = addr 673 c.address = addr
@@ -555,9 +698,57 @@ func (c *Client) ReattachConfig() *ReattachConfig {
555 } 698 }
556 699
557 return &ReattachConfig{ 700 return &ReattachConfig{
558 Addr: c.address, 701 Protocol: c.protocol,
559 Pid: c.config.Cmd.Process.Pid, 702 Addr: c.address,
703 Pid: c.config.Cmd.Process.Pid,
704 }
705}
706
707// Protocol returns the protocol of server on the remote end. This will
708// start the plugin process if it isn't already started. Errors from
709// starting the plugin are surpressed and ProtocolInvalid is returned. It
710// is recommended you call Start explicitly before calling Protocol to ensure
711// no errors occur.
712func (c *Client) Protocol() Protocol {
713 _, err := c.Start()
714 if err != nil {
715 return ProtocolInvalid
716 }
717
718 return c.protocol
719}
720
721func netAddrDialer(addr net.Addr) func(string, time.Duration) (net.Conn, error) {
722 return func(_ string, _ time.Duration) (net.Conn, error) {
723 // Connect to the client
724 conn, err := net.Dial(addr.Network(), addr.String())
725 if err != nil {
726 return nil, err
727 }
728 if tcpConn, ok := conn.(*net.TCPConn); ok {
729 // Make sure to set keep alive so that the connection doesn't die
730 tcpConn.SetKeepAlive(true)
731 }
732
733 return conn, nil
734 }
735}
736
737// dialer is compatible with grpc.WithDialer and creates the connection
738// to the plugin.
739func (c *Client) dialer(_ string, timeout time.Duration) (net.Conn, error) {
740 conn, err := netAddrDialer(c.address)("", timeout)
741 if err != nil {
742 return nil, err
560 } 743 }
744
745 // If we have a TLS config we wrap our connection. We only do this
746 // for net/rpc since gRPC uses its own mechanism for TLS.
747 if c.protocol == ProtocolNetRPC && c.config.TLSConfig != nil {
748 conn = tls.Client(conn, c.config.TLSConfig)
749 }
750
751 return conn, nil
561} 752}
562 753
563func (c *Client) logStderr(r io.Reader) { 754func (c *Client) logStderr(r io.Reader) {
@@ -566,9 +757,31 @@ func (c *Client) logStderr(r io.Reader) {
566 line, err := bufR.ReadString('\n') 757 line, err := bufR.ReadString('\n')
567 if line != "" { 758 if line != "" {
568 c.config.Stderr.Write([]byte(line)) 759 c.config.Stderr.Write([]byte(line))
569
570 line = strings.TrimRightFunc(line, unicode.IsSpace) 760 line = strings.TrimRightFunc(line, unicode.IsSpace)
571 log.Printf("[DEBUG] plugin: %s: %s", filepath.Base(c.config.Cmd.Path), line) 761
762 l := c.logger.Named(filepath.Base(c.config.Cmd.Path))
763
764 entry, err := parseJSON(line)
765 // If output is not JSON format, print directly to Debug
766 if err != nil {
767 l.Debug(line)
768 } else {
769 out := flattenKVPairs(entry.KVPairs)
770
771 l = l.With("timestamp", entry.Timestamp.Format(hclog.TimeFormat))
772 switch hclog.LevelFromString(entry.Level) {
773 case hclog.Trace:
774 l.Trace(entry.Message, out...)
775 case hclog.Debug:
776 l.Debug(entry.Message, out...)
777 case hclog.Info:
778 l.Info(entry.Message, out...)
779 case hclog.Warn:
780 l.Warn(entry.Message, out...)
781 case hclog.Error:
782 l.Error(entry.Message, out...)
783 }
784 }
572 } 785 }
573 786
574 if err == io.EOF { 787 if err == io.EOF {
diff --git a/vendor/github.com/hashicorp/go-plugin/grpc_broker.go b/vendor/github.com/hashicorp/go-plugin/grpc_broker.go
new file mode 100644
index 0000000..49fd21c
--- /dev/null
+++ b/vendor/github.com/hashicorp/go-plugin/grpc_broker.go
@@ -0,0 +1,455 @@
1package plugin
2
3import (
4 "context"
5 "crypto/tls"
6 "errors"
7 "fmt"
8 "log"
9 "net"
10 "sync"
11 "sync/atomic"
12 "time"
13
14 "github.com/oklog/run"
15 "google.golang.org/grpc"
16 "google.golang.org/grpc/credentials"
17)
18
19// streamer interface is used in the broker to send/receive connection
20// information.
21type streamer interface {
22 Send(*ConnInfo) error
23 Recv() (*ConnInfo, error)
24 Close()
25}
26
27// sendErr is used to pass errors back during a send.
28type sendErr struct {
29 i *ConnInfo
30 ch chan error
31}
32
33// gRPCBrokerServer is used by the plugin to start a stream and to send
34// connection information to/from the plugin. Implements GRPCBrokerServer and
35// streamer interfaces.
36type gRPCBrokerServer struct {
37 // send is used to send connection info to the gRPC stream.
38 send chan *sendErr
39
40 // recv is used to receive connection info from the gRPC stream.
41 recv chan *ConnInfo
42
43 // quit closes down the stream.
44 quit chan struct{}
45
46 // o is used to ensure we close the quit channel only once.
47 o sync.Once
48}
49
50func newGRPCBrokerServer() *gRPCBrokerServer {
51 return &gRPCBrokerServer{
52 send: make(chan *sendErr),
53 recv: make(chan *ConnInfo),
54 quit: make(chan struct{}),
55 }
56}
57
58// StartStream implements the GRPCBrokerServer interface and will block until
59// the quit channel is closed or the context reports Done. The stream will pass
60// connection information to/from the client.
61func (s *gRPCBrokerServer) StartStream(stream GRPCBroker_StartStreamServer) error {
62 doneCh := stream.Context().Done()
63 defer s.Close()
64
65 // Proccess send stream
66 go func() {
67 for {
68 select {
69 case <-doneCh:
70 return
71 case <-s.quit:
72 return
73 case se := <-s.send:
74 err := stream.Send(se.i)
75 se.ch <- err
76 }
77 }
78 }()
79
80 // Process receive stream
81 for {
82 i, err := stream.Recv()
83 if err != nil {
84 return err
85 }
86 select {
87 case <-doneCh:
88 return nil
89 case <-s.quit:
90 return nil
91 case s.recv <- i:
92 }
93 }
94
95 return nil
96}
97
98// Send is used by the GRPCBroker to pass connection information into the stream
99// to the client.
100func (s *gRPCBrokerServer) Send(i *ConnInfo) error {
101 ch := make(chan error)
102 defer close(ch)
103
104 select {
105 case <-s.quit:
106 return errors.New("broker closed")
107 case s.send <- &sendErr{
108 i: i,
109 ch: ch,
110 }:
111 }
112
113 return <-ch
114}
115
116// Recv is used by the GRPCBroker to pass connection information that has been
117// sent from the client from the stream to the broker.
118func (s *gRPCBrokerServer) Recv() (*ConnInfo, error) {
119 select {
120 case <-s.quit:
121 return nil, errors.New("broker closed")
122 case i := <-s.recv:
123 return i, nil
124 }
125}
126
127// Close closes the quit channel, shutting down the stream.
128func (s *gRPCBrokerServer) Close() {
129 s.o.Do(func() {
130 close(s.quit)
131 })
132}
133
134// gRPCBrokerClientImpl is used by the client to start a stream and to send
135// connection information to/from the client. Implements GRPCBrokerClient and
136// streamer interfaces.
137type gRPCBrokerClientImpl struct {
138 // client is the underlying GRPC client used to make calls to the server.
139 client GRPCBrokerClient
140
141 // send is used to send connection info to the gRPC stream.
142 send chan *sendErr
143
144 // recv is used to receive connection info from the gRPC stream.
145 recv chan *ConnInfo
146
147 // quit closes down the stream.
148 quit chan struct{}
149
150 // o is used to ensure we close the quit channel only once.
151 o sync.Once
152}
153
154func newGRPCBrokerClient(conn *grpc.ClientConn) *gRPCBrokerClientImpl {
155 return &gRPCBrokerClientImpl{
156 client: NewGRPCBrokerClient(conn),
157 send: make(chan *sendErr),
158 recv: make(chan *ConnInfo),
159 quit: make(chan struct{}),
160 }
161}
162
163// StartStream implements the GRPCBrokerClient interface and will block until
164// the quit channel is closed or the context reports Done. The stream will pass
165// connection information to/from the plugin.
166func (s *gRPCBrokerClientImpl) StartStream() error {
167 ctx, cancelFunc := context.WithCancel(context.Background())
168 defer cancelFunc()
169 defer s.Close()
170
171 stream, err := s.client.StartStream(ctx)
172 if err != nil {
173 return err
174 }
175 doneCh := stream.Context().Done()
176
177 go func() {
178 for {
179 select {
180 case <-doneCh:
181 return
182 case <-s.quit:
183 return
184 case se := <-s.send:
185 err := stream.Send(se.i)
186 se.ch <- err
187 }
188 }
189 }()
190
191 for {
192 i, err := stream.Recv()
193 if err != nil {
194 return err
195 }
196 select {
197 case <-doneCh:
198 return nil
199 case <-s.quit:
200 return nil
201 case s.recv <- i:
202 }
203 }
204
205 return nil
206}
207
208// Send is used by the GRPCBroker to pass connection information into the stream
209// to the plugin.
210func (s *gRPCBrokerClientImpl) Send(i *ConnInfo) error {
211 ch := make(chan error)
212 defer close(ch)
213
214 select {
215 case <-s.quit:
216 return errors.New("broker closed")
217 case s.send <- &sendErr{
218 i: i,
219 ch: ch,
220 }:
221 }
222
223 return <-ch
224}
225
226// Recv is used by the GRPCBroker to pass connection information that has been
227// sent from the plugin to the broker.
228func (s *gRPCBrokerClientImpl) Recv() (*ConnInfo, error) {
229 select {
230 case <-s.quit:
231 return nil, errors.New("broker closed")
232 case i := <-s.recv:
233 return i, nil
234 }
235}
236
237// Close closes the quit channel, shutting down the stream.
238func (s *gRPCBrokerClientImpl) Close() {
239 s.o.Do(func() {
240 close(s.quit)
241 })
242}
243
244// GRPCBroker is responsible for brokering connections by unique ID.
245//
246// It is used by plugins to create multiple gRPC connections and data
247// streams between the plugin process and the host process.
248//
249// This allows a plugin to request a channel with a specific ID to connect to
250// or accept a connection from, and the broker handles the details of
251// holding these channels open while they're being negotiated.
252//
253// The Plugin interface has access to these for both Server and Client.
254// The broker can be used by either (optionally) to reserve and connect to
255// new streams. This is useful for complex args and return values,
256// or anything else you might need a data stream for.
257type GRPCBroker struct {
258 nextId uint32
259 streamer streamer
260 streams map[uint32]*gRPCBrokerPending
261 tls *tls.Config
262 doneCh chan struct{}
263 o sync.Once
264
265 sync.Mutex
266}
267
268type gRPCBrokerPending struct {
269 ch chan *ConnInfo
270 doneCh chan struct{}
271}
272
273func newGRPCBroker(s streamer, tls *tls.Config) *GRPCBroker {
274 return &GRPCBroker{
275 streamer: s,
276 streams: make(map[uint32]*gRPCBrokerPending),
277 tls: tls,
278 doneCh: make(chan struct{}),
279 }
280}
281
282// Accept accepts a connection by ID.
283//
284// This should not be called multiple times with the same ID at one time.
285func (b *GRPCBroker) Accept(id uint32) (net.Listener, error) {
286 listener, err := serverListener()
287 if err != nil {
288 return nil, err
289 }
290
291 err = b.streamer.Send(&ConnInfo{
292 ServiceId: id,
293 Network: listener.Addr().Network(),
294 Address: listener.Addr().String(),
295 })
296 if err != nil {
297 return nil, err
298 }
299
300 return listener, nil
301}
302
303// AcceptAndServe is used to accept a specific stream ID and immediately
304// serve a gRPC server on that stream ID. This is used to easily serve
305// complex arguments. Each AcceptAndServe call opens a new listener socket and
306// sends the connection info down the stream to the dialer. Since a new
307// connection is opened every call, these calls should be used sparingly.
308// Multiple gRPC server implementations can be registered to a single
309// AcceptAndServe call.
310func (b *GRPCBroker) AcceptAndServe(id uint32, s func([]grpc.ServerOption) *grpc.Server) {
311 listener, err := b.Accept(id)
312 if err != nil {
313 log.Printf("[ERR] plugin: plugin acceptAndServe error: %s", err)
314 return
315 }
316 defer listener.Close()
317
318 var opts []grpc.ServerOption
319 if b.tls != nil {
320 opts = []grpc.ServerOption{grpc.Creds(credentials.NewTLS(b.tls))}
321 }
322
323 server := s(opts)
324
325 // Here we use a run group to close this goroutine if the server is shutdown
326 // or the broker is shutdown.
327 var g run.Group
328 {
329 // Serve on the listener, if shutting down call GracefulStop.
330 g.Add(func() error {
331 return server.Serve(listener)
332 }, func(err error) {
333 server.GracefulStop()
334 })
335 }
336 {
337 // block on the closeCh or the doneCh. If we are shutting down close the
338 // closeCh.
339 closeCh := make(chan struct{})
340 g.Add(func() error {
341 select {
342 case <-b.doneCh:
343 case <-closeCh:
344 }
345 return nil
346 }, func(err error) {
347 close(closeCh)
348 })
349 }
350
351 // Block until we are done
352 g.Run()
353}
354
355// Close closes the stream and all servers.
356func (b *GRPCBroker) Close() error {
357 b.streamer.Close()
358 b.o.Do(func() {
359 close(b.doneCh)
360 })
361 return nil
362}
363
364// Dial opens a connection by ID.
365func (b *GRPCBroker) Dial(id uint32) (conn *grpc.ClientConn, err error) {
366 var c *ConnInfo
367
368 // Open the stream
369 p := b.getStream(id)
370 select {
371 case c = <-p.ch:
372 close(p.doneCh)
373 case <-time.After(5 * time.Second):
374 return nil, fmt.Errorf("timeout waiting for connection info")
375 }
376
377 var addr net.Addr
378 switch c.Network {
379 case "tcp":
380 addr, err = net.ResolveTCPAddr("tcp", c.Address)
381 case "unix":
382 addr, err = net.ResolveUnixAddr("unix", c.Address)
383 default:
384 err = fmt.Errorf("Unknown address type: %s", c.Address)
385 }
386 if err != nil {
387 return nil, err
388 }
389
390 return dialGRPCConn(b.tls, netAddrDialer(addr))
391}
392
393// NextId returns a unique ID to use next.
394//
395// It is possible for very long-running plugin hosts to wrap this value,
396// though it would require a very large amount of calls. In practice
397// we've never seen it happen.
398func (m *GRPCBroker) NextId() uint32 {
399 return atomic.AddUint32(&m.nextId, 1)
400}
401
402// Run starts the brokering and should be executed in a goroutine, since it
403// blocks forever, or until the session closes.
404//
405// Uses of GRPCBroker never need to call this. It is called internally by
406// the plugin host/client.
407func (m *GRPCBroker) Run() {
408 for {
409 stream, err := m.streamer.Recv()
410 if err != nil {
411 // Once we receive an error, just exit
412 break
413 }
414
415 // Initialize the waiter
416 p := m.getStream(stream.ServiceId)
417 select {
418 case p.ch <- stream:
419 default:
420 }
421
422 go m.timeoutWait(stream.ServiceId, p)
423 }
424}
425
426func (m *GRPCBroker) getStream(id uint32) *gRPCBrokerPending {
427 m.Lock()
428 defer m.Unlock()
429
430 p, ok := m.streams[id]
431 if ok {
432 return p
433 }
434
435 m.streams[id] = &gRPCBrokerPending{
436 ch: make(chan *ConnInfo, 1),
437 doneCh: make(chan struct{}),
438 }
439 return m.streams[id]
440}
441
442func (m *GRPCBroker) timeoutWait(id uint32, p *gRPCBrokerPending) {
443 // Wait for the stream to either be picked up and connected, or
444 // for a timeout.
445 select {
446 case <-p.doneCh:
447 case <-time.After(5 * time.Second):
448 }
449
450 m.Lock()
451 defer m.Unlock()
452
453 // Delete the stream so no one else can grab it
454 delete(m.streams, id)
455}
diff --git a/vendor/github.com/hashicorp/go-plugin/grpc_broker.pb.go b/vendor/github.com/hashicorp/go-plugin/grpc_broker.pb.go
new file mode 100644
index 0000000..d490daf
--- /dev/null
+++ b/vendor/github.com/hashicorp/go-plugin/grpc_broker.pb.go
@@ -0,0 +1,190 @@
1// Code generated by protoc-gen-go. DO NOT EDIT.
2// source: grpc_broker.proto
3
4/*
5Package plugin is a generated protocol buffer package.
6
7It is generated from these files:
8 grpc_broker.proto
9
10It has these top-level messages:
11 ConnInfo
12*/
13package plugin
14
15import proto "github.com/golang/protobuf/proto"
16import fmt "fmt"
17import math "math"
18
19import (
20 context "golang.org/x/net/context"
21 grpc "google.golang.org/grpc"
22)
23
24// Reference imports to suppress errors if they are not otherwise used.
25var _ = proto.Marshal
26var _ = fmt.Errorf
27var _ = math.Inf
28
29// This is a compile-time assertion to ensure that this generated file
30// is compatible with the proto package it is being compiled against.
31// A compilation error at this line likely means your copy of the
32// proto package needs to be updated.
33const _ = proto.ProtoPackageIsVersion2 // please upgrade the proto package
34
35type ConnInfo struct {
36 ServiceId uint32 `protobuf:"varint,1,opt,name=service_id,json=serviceId" json:"service_id,omitempty"`
37 Network string `protobuf:"bytes,2,opt,name=network" json:"network,omitempty"`
38 Address string `protobuf:"bytes,3,opt,name=address" json:"address,omitempty"`
39}
40
41func (m *ConnInfo) Reset() { *m = ConnInfo{} }
42func (m *ConnInfo) String() string { return proto.CompactTextString(m) }
43func (*ConnInfo) ProtoMessage() {}
44func (*ConnInfo) Descriptor() ([]byte, []int) { return fileDescriptor0, []int{0} }
45
46func (m *ConnInfo) GetServiceId() uint32 {
47 if m != nil {
48 return m.ServiceId
49 }
50 return 0
51}
52
53func (m *ConnInfo) GetNetwork() string {
54 if m != nil {
55 return m.Network
56 }
57 return ""
58}
59
60func (m *ConnInfo) GetAddress() string {
61 if m != nil {
62 return m.Address
63 }
64 return ""
65}
66
67func init() {
68 proto.RegisterType((*ConnInfo)(nil), "plugin.ConnInfo")
69}
70
71// Reference imports to suppress errors if they are not otherwise used.
72var _ context.Context
73var _ grpc.ClientConn
74
75// This is a compile-time assertion to ensure that this generated file
76// is compatible with the grpc package it is being compiled against.
77const _ = grpc.SupportPackageIsVersion4
78
79// Client API for GRPCBroker service
80
81type GRPCBrokerClient interface {
82 StartStream(ctx context.Context, opts ...grpc.CallOption) (GRPCBroker_StartStreamClient, error)
83}
84
85type gRPCBrokerClient struct {
86 cc *grpc.ClientConn
87}
88
89func NewGRPCBrokerClient(cc *grpc.ClientConn) GRPCBrokerClient {
90 return &gRPCBrokerClient{cc}
91}
92
93func (c *gRPCBrokerClient) StartStream(ctx context.Context, opts ...grpc.CallOption) (GRPCBroker_StartStreamClient, error) {
94 stream, err := grpc.NewClientStream(ctx, &_GRPCBroker_serviceDesc.Streams[0], c.cc, "/plugin.GRPCBroker/StartStream", opts...)
95 if err != nil {
96 return nil, err
97 }
98 x := &gRPCBrokerStartStreamClient{stream}
99 return x, nil
100}
101
102type GRPCBroker_StartStreamClient interface {
103 Send(*ConnInfo) error
104 Recv() (*ConnInfo, error)
105 grpc.ClientStream
106}
107
108type gRPCBrokerStartStreamClient struct {
109 grpc.ClientStream
110}
111
112func (x *gRPCBrokerStartStreamClient) Send(m *ConnInfo) error {
113 return x.ClientStream.SendMsg(m)
114}
115
116func (x *gRPCBrokerStartStreamClient) Recv() (*ConnInfo, error) {
117 m := new(ConnInfo)
118 if err := x.ClientStream.RecvMsg(m); err != nil {
119 return nil, err
120 }
121 return m, nil
122}
123
124// Server API for GRPCBroker service
125
126type GRPCBrokerServer interface {
127 StartStream(GRPCBroker_StartStreamServer) error
128}
129
130func RegisterGRPCBrokerServer(s *grpc.Server, srv GRPCBrokerServer) {
131 s.RegisterService(&_GRPCBroker_serviceDesc, srv)
132}
133
134func _GRPCBroker_StartStream_Handler(srv interface{}, stream grpc.ServerStream) error {
135 return srv.(GRPCBrokerServer).StartStream(&gRPCBrokerStartStreamServer{stream})
136}
137
138type GRPCBroker_StartStreamServer interface {
139 Send(*ConnInfo) error
140 Recv() (*ConnInfo, error)
141 grpc.ServerStream
142}
143
144type gRPCBrokerStartStreamServer struct {
145 grpc.ServerStream
146}
147
148func (x *gRPCBrokerStartStreamServer) Send(m *ConnInfo) error {
149 return x.ServerStream.SendMsg(m)
150}
151
152func (x *gRPCBrokerStartStreamServer) Recv() (*ConnInfo, error) {
153 m := new(ConnInfo)
154 if err := x.ServerStream.RecvMsg(m); err != nil {
155 return nil, err
156 }
157 return m, nil
158}
159
160var _GRPCBroker_serviceDesc = grpc.ServiceDesc{
161 ServiceName: "plugin.GRPCBroker",
162 HandlerType: (*GRPCBrokerServer)(nil),
163 Methods: []grpc.MethodDesc{},
164 Streams: []grpc.StreamDesc{
165 {
166 StreamName: "StartStream",
167 Handler: _GRPCBroker_StartStream_Handler,
168 ServerStreams: true,
169 ClientStreams: true,
170 },
171 },
172 Metadata: "grpc_broker.proto",
173}
174
175func init() { proto.RegisterFile("grpc_broker.proto", fileDescriptor0) }
176
177var fileDescriptor0 = []byte{
178 // 170 bytes of a gzipped FileDescriptorProto
179 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0x12, 0x4c, 0x2f, 0x2a, 0x48,
180 0x8e, 0x4f, 0x2a, 0xca, 0xcf, 0x4e, 0x2d, 0xd2, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0x62, 0x2b,
181 0xc8, 0x29, 0x4d, 0xcf, 0xcc, 0x53, 0x8a, 0xe5, 0xe2, 0x70, 0xce, 0xcf, 0xcb, 0xf3, 0xcc, 0x4b,
182 0xcb, 0x17, 0x92, 0xe5, 0xe2, 0x2a, 0x4e, 0x2d, 0x2a, 0xcb, 0x4c, 0x4e, 0x8d, 0xcf, 0x4c, 0x91,
183 0x60, 0x54, 0x60, 0xd4, 0xe0, 0x0d, 0xe2, 0x84, 0x8a, 0x78, 0xa6, 0x08, 0x49, 0x70, 0xb1, 0xe7,
184 0xa5, 0x96, 0x94, 0xe7, 0x17, 0x65, 0x4b, 0x30, 0x29, 0x30, 0x6a, 0x70, 0x06, 0xc1, 0xb8, 0x20,
185 0x99, 0xc4, 0x94, 0x94, 0xa2, 0xd4, 0xe2, 0x62, 0x09, 0x66, 0x88, 0x0c, 0x94, 0x6b, 0xe4, 0xcc,
186 0xc5, 0xe5, 0x1e, 0x14, 0xe0, 0xec, 0x04, 0xb6, 0x5a, 0xc8, 0x94, 0x8b, 0x3b, 0xb8, 0x24, 0xb1,
187 0xa8, 0x24, 0xb8, 0xa4, 0x28, 0x35, 0x31, 0x57, 0x48, 0x40, 0x0f, 0xe2, 0x08, 0x3d, 0x98, 0x0b,
188 0xa4, 0x30, 0x44, 0x34, 0x18, 0x0d, 0x18, 0x93, 0xd8, 0xc0, 0x4e, 0x36, 0x06, 0x04, 0x00, 0x00,
189 0xff, 0xff, 0x7b, 0x5d, 0xfb, 0xe1, 0xc7, 0x00, 0x00, 0x00,
190}
diff --git a/vendor/github.com/hashicorp/go-plugin/grpc_broker.proto b/vendor/github.com/hashicorp/go-plugin/grpc_broker.proto
new file mode 100644
index 0000000..f578348
--- /dev/null
+++ b/vendor/github.com/hashicorp/go-plugin/grpc_broker.proto
@@ -0,0 +1,14 @@
1syntax = "proto3";
2package plugin;
3
4message ConnInfo {
5 uint32 service_id = 1;
6 string network = 2;
7 string address = 3;
8}
9
10service GRPCBroker {
11 rpc StartStream(stream ConnInfo) returns (stream ConnInfo);
12}
13
14
diff --git a/vendor/github.com/hashicorp/go-plugin/grpc_client.go b/vendor/github.com/hashicorp/go-plugin/grpc_client.go
new file mode 100644
index 0000000..44294d0
--- /dev/null
+++ b/vendor/github.com/hashicorp/go-plugin/grpc_client.go
@@ -0,0 +1,107 @@
1package plugin
2
3import (
4 "crypto/tls"
5 "fmt"
6 "net"
7 "time"
8
9 "golang.org/x/net/context"
10 "google.golang.org/grpc"
11 "google.golang.org/grpc/credentials"
12 "google.golang.org/grpc/health/grpc_health_v1"
13)
14
15func dialGRPCConn(tls *tls.Config, dialer func(string, time.Duration) (net.Conn, error)) (*grpc.ClientConn, error) {
16 // Build dialing options.
17 opts := make([]grpc.DialOption, 0, 5)
18
19 // We use a custom dialer so that we can connect over unix domain sockets
20 opts = append(opts, grpc.WithDialer(dialer))
21
22 // go-plugin expects to block the connection
23 opts = append(opts, grpc.WithBlock())
24
25 // Fail right away
26 opts = append(opts, grpc.FailOnNonTempDialError(true))
27
28 // If we have no TLS configuration set, we need to explicitly tell grpc
29 // that we're connecting with an insecure connection.
30 if tls == nil {
31 opts = append(opts, grpc.WithInsecure())
32 } else {
33 opts = append(opts, grpc.WithTransportCredentials(
34 credentials.NewTLS(tls)))
35 }
36
37 // Connect. Note the first parameter is unused because we use a custom
38 // dialer that has the state to see the address.
39 conn, err := grpc.Dial("unused", opts...)
40 if err != nil {
41 return nil, err
42 }
43
44 return conn, nil
45}
46
47// newGRPCClient creates a new GRPCClient. The Client argument is expected
48// to be successfully started already with a lock held.
49func newGRPCClient(doneCtx context.Context, c *Client) (*GRPCClient, error) {
50 conn, err := dialGRPCConn(c.config.TLSConfig, c.dialer)
51 if err != nil {
52 return nil, err
53 }
54
55 // Start the broker.
56 brokerGRPCClient := newGRPCBrokerClient(conn)
57 broker := newGRPCBroker(brokerGRPCClient, c.config.TLSConfig)
58 go broker.Run()
59 go brokerGRPCClient.StartStream()
60
61 return &GRPCClient{
62 Conn: conn,
63 Plugins: c.config.Plugins,
64 doneCtx: doneCtx,
65 broker: broker,
66 }, nil
67}
68
69// GRPCClient connects to a GRPCServer over gRPC to dispense plugin types.
70type GRPCClient struct {
71 Conn *grpc.ClientConn
72 Plugins map[string]Plugin
73
74 doneCtx context.Context
75 broker *GRPCBroker
76}
77
78// ClientProtocol impl.
79func (c *GRPCClient) Close() error {
80 c.broker.Close()
81 return c.Conn.Close()
82}
83
84// ClientProtocol impl.
85func (c *GRPCClient) Dispense(name string) (interface{}, error) {
86 raw, ok := c.Plugins[name]
87 if !ok {
88 return nil, fmt.Errorf("unknown plugin type: %s", name)
89 }
90
91 p, ok := raw.(GRPCPlugin)
92 if !ok {
93 return nil, fmt.Errorf("plugin %q doesn't support gRPC", name)
94 }
95
96 return p.GRPCClient(c.doneCtx, c.broker, c.Conn)
97}
98
99// ClientProtocol impl.
100func (c *GRPCClient) Ping() error {
101 client := grpc_health_v1.NewHealthClient(c.Conn)
102 _, err := client.Check(context.Background(), &grpc_health_v1.HealthCheckRequest{
103 Service: GRPCServiceName,
104 })
105
106 return err
107}
diff --git a/vendor/github.com/hashicorp/go-plugin/grpc_server.go b/vendor/github.com/hashicorp/go-plugin/grpc_server.go
new file mode 100644
index 0000000..3a72739
--- /dev/null
+++ b/vendor/github.com/hashicorp/go-plugin/grpc_server.go
@@ -0,0 +1,132 @@
1package plugin
2
3import (
4 "bytes"
5 "crypto/tls"
6 "encoding/json"
7 "fmt"
8 "io"
9 "net"
10
11 "google.golang.org/grpc"
12 "google.golang.org/grpc/credentials"
13 "google.golang.org/grpc/health"
14 "google.golang.org/grpc/health/grpc_health_v1"
15)
16
17// GRPCServiceName is the name of the service that the health check should
18// return as passing.
19const GRPCServiceName = "plugin"
20
21// DefaultGRPCServer can be used with the "GRPCServer" field for Server
22// as a default factory method to create a gRPC server with no extra options.
23func DefaultGRPCServer(opts []grpc.ServerOption) *grpc.Server {
24 return grpc.NewServer(opts...)
25}
26
27// GRPCServer is a ServerType implementation that serves plugins over
28// gRPC. This allows plugins to easily be written for other languages.
29//
30// The GRPCServer outputs a custom configuration as a base64-encoded
31// JSON structure represented by the GRPCServerConfig config structure.
32type GRPCServer struct {
33 // Plugins are the list of plugins to serve.
34 Plugins map[string]Plugin
35
36 // Server is the actual server that will accept connections. This
37 // will be used for plugin registration as well.
38 Server func([]grpc.ServerOption) *grpc.Server
39
40 // TLS should be the TLS configuration if available. If this is nil,
41 // the connection will not have transport security.
42 TLS *tls.Config
43
44 // DoneCh is the channel that is closed when this server has exited.
45 DoneCh chan struct{}
46
47 // Stdout/StderrLis are the readers for stdout/stderr that will be copied
48 // to the stdout/stderr connection that is output.
49 Stdout io.Reader
50 Stderr io.Reader
51
52 config GRPCServerConfig
53 server *grpc.Server
54 broker *GRPCBroker
55}
56
57// ServerProtocol impl.
58func (s *GRPCServer) Init() error {
59 // Create our server
60 var opts []grpc.ServerOption
61 if s.TLS != nil {
62 opts = append(opts, grpc.Creds(credentials.NewTLS(s.TLS)))
63 }
64 s.server = s.Server(opts)
65
66 // Register the health service
67 healthCheck := health.NewServer()
68 healthCheck.SetServingStatus(
69 GRPCServiceName, grpc_health_v1.HealthCheckResponse_SERVING)
70 grpc_health_v1.RegisterHealthServer(s.server, healthCheck)
71
72 // Register the broker service
73 brokerServer := newGRPCBrokerServer()
74 RegisterGRPCBrokerServer(s.server, brokerServer)
75 s.broker = newGRPCBroker(brokerServer, s.TLS)
76 go s.broker.Run()
77
78 // Register all our plugins onto the gRPC server.
79 for k, raw := range s.Plugins {
80 p, ok := raw.(GRPCPlugin)
81 if !ok {
82 return fmt.Errorf("%q is not a GRPC-compatible plugin", k)
83 }
84
85 if err := p.GRPCServer(s.broker, s.server); err != nil {
86 return fmt.Errorf("error registring %q: %s", k, err)
87 }
88 }
89
90 return nil
91}
92
93// Stop calls Stop on the underlying grpc.Server
94func (s *GRPCServer) Stop() {
95 s.server.Stop()
96}
97
98// GracefulStop calls GracefulStop on the underlying grpc.Server
99func (s *GRPCServer) GracefulStop() {
100 s.server.GracefulStop()
101}
102
103// Config is the GRPCServerConfig encoded as JSON then base64.
104func (s *GRPCServer) Config() string {
105 // Create a buffer that will contain our final contents
106 var buf bytes.Buffer
107
108 // Wrap the base64 encoding with JSON encoding.
109 if err := json.NewEncoder(&buf).Encode(s.config); err != nil {
110 // We panic since ths shouldn't happen under any scenario. We
111 // carefully control the structure being encoded here and it should
112 // always be successful.
113 panic(err)
114 }
115
116 return buf.String()
117}
118
119func (s *GRPCServer) Serve(lis net.Listener) {
120 // Start serving in a goroutine
121 go s.server.Serve(lis)
122
123 // Wait until graceful completion
124 <-s.DoneCh
125}
126
127// GRPCServerConfig is the extra configuration passed along for consumers
128// to facilitate using GRPC plugins.
129type GRPCServerConfig struct {
130 StdoutAddr string `json:"stdout_addr"`
131 StderrAddr string `json:"stderr_addr"`
132}
diff --git a/vendor/github.com/hashicorp/go-plugin/log_entry.go b/vendor/github.com/hashicorp/go-plugin/log_entry.go
new file mode 100644
index 0000000..2996c14
--- /dev/null
+++ b/vendor/github.com/hashicorp/go-plugin/log_entry.go
@@ -0,0 +1,73 @@
1package plugin
2
3import (
4 "encoding/json"
5 "time"
6)
7
8// logEntry is the JSON payload that gets sent to Stderr from the plugin to the host
9type logEntry struct {
10 Message string `json:"@message"`
11 Level string `json:"@level"`
12 Timestamp time.Time `json:"timestamp"`
13 KVPairs []*logEntryKV `json:"kv_pairs"`
14}
15
16// logEntryKV is a key value pair within the Output payload
17type logEntryKV struct {
18 Key string `json:"key"`
19 Value interface{} `json:"value"`
20}
21
22// flattenKVPairs is used to flatten KVPair slice into []interface{}
23// for hclog consumption.
24func flattenKVPairs(kvs []*logEntryKV) []interface{} {
25 var result []interface{}
26 for _, kv := range kvs {
27 result = append(result, kv.Key)
28 result = append(result, kv.Value)
29 }
30
31 return result
32}
33
34// parseJSON handles parsing JSON output
35func parseJSON(input string) (*logEntry, error) {
36 var raw map[string]interface{}
37 entry := &logEntry{}
38
39 err := json.Unmarshal([]byte(input), &raw)
40 if err != nil {
41 return nil, err
42 }
43
44 // Parse hclog-specific objects
45 if v, ok := raw["@message"]; ok {
46 entry.Message = v.(string)
47 delete(raw, "@message")
48 }
49
50 if v, ok := raw["@level"]; ok {
51 entry.Level = v.(string)
52 delete(raw, "@level")
53 }
54
55 if v, ok := raw["@timestamp"]; ok {
56 t, err := time.Parse("2006-01-02T15:04:05.000000Z07:00", v.(string))
57 if err != nil {
58 return nil, err
59 }
60 entry.Timestamp = t
61 delete(raw, "@timestamp")
62 }
63
64 // Parse dynamic KV args from the hclog payload.
65 for k, v := range raw {
66 entry.KVPairs = append(entry.KVPairs, &logEntryKV{
67 Key: k,
68 Value: v,
69 })
70 }
71
72 return entry, nil
73}
diff --git a/vendor/github.com/hashicorp/go-plugin/plugin.go b/vendor/github.com/hashicorp/go-plugin/plugin.go
index 37c8fd6..79d9674 100644
--- a/vendor/github.com/hashicorp/go-plugin/plugin.go
+++ b/vendor/github.com/hashicorp/go-plugin/plugin.go
@@ -9,7 +9,11 @@
9package plugin 9package plugin
10 10
11import ( 11import (
12 "context"
13 "errors"
12 "net/rpc" 14 "net/rpc"
15
16 "google.golang.org/grpc"
13) 17)
14 18
15// Plugin is the interface that is implemented to serve/connect to an 19// Plugin is the interface that is implemented to serve/connect to an
@@ -23,3 +27,32 @@ type Plugin interface {
23 // serving that communicates to the server end of the plugin. 27 // serving that communicates to the server end of the plugin.
24 Client(*MuxBroker, *rpc.Client) (interface{}, error) 28 Client(*MuxBroker, *rpc.Client) (interface{}, error)
25} 29}
30
31// GRPCPlugin is the interface that is implemented to serve/connect to
32// a plugin over gRPC.
33type GRPCPlugin interface {
34 // GRPCServer should register this plugin for serving with the
35 // given GRPCServer. Unlike Plugin.Server, this is only called once
36 // since gRPC plugins serve singletons.
37 GRPCServer(*GRPCBroker, *grpc.Server) error
38
39 // GRPCClient should return the interface implementation for the plugin
40 // you're serving via gRPC. The provided context will be canceled by
41 // go-plugin in the event of the plugin process exiting.
42 GRPCClient(context.Context, *GRPCBroker, *grpc.ClientConn) (interface{}, error)
43}
44
45// NetRPCUnsupportedPlugin implements Plugin but returns errors for the
46// Server and Client functions. This will effectively disable support for
47// net/rpc based plugins.
48//
49// This struct can be embedded in your struct.
50type NetRPCUnsupportedPlugin struct{}
51
52func (p NetRPCUnsupportedPlugin) Server(*MuxBroker) (interface{}, error) {
53 return nil, errors.New("net/rpc plugin protocol not supported")
54}
55
56func (p NetRPCUnsupportedPlugin) Client(*MuxBroker, *rpc.Client) (interface{}, error) {
57 return nil, errors.New("net/rpc plugin protocol not supported")
58}
diff --git a/vendor/github.com/hashicorp/go-plugin/protocol.go b/vendor/github.com/hashicorp/go-plugin/protocol.go
new file mode 100644
index 0000000..0cfc19e
--- /dev/null
+++ b/vendor/github.com/hashicorp/go-plugin/protocol.go
@@ -0,0 +1,45 @@
1package plugin
2
3import (
4 "io"
5 "net"
6)
7
8// Protocol is an enum representing the types of protocols.
9type Protocol string
10
11const (
12 ProtocolInvalid Protocol = ""
13 ProtocolNetRPC Protocol = "netrpc"
14 ProtocolGRPC Protocol = "grpc"
15)
16
17// ServerProtocol is an interface that must be implemented for new plugin
18// protocols to be servers.
19type ServerProtocol interface {
20 // Init is called once to configure and initialize the protocol, but
21 // not start listening. This is the point at which all validation should
22 // be done and errors returned.
23 Init() error
24
25 // Config is extra configuration to be outputted to stdout. This will
26 // be automatically base64 encoded to ensure it can be parsed properly.
27 // This can be an empty string if additional configuration is not needed.
28 Config() string
29
30 // Serve is called to serve connections on the given listener. This should
31 // continue until the listener is closed.
32 Serve(net.Listener)
33}
34
35// ClientProtocol is an interface that must be implemented for new plugin
36// protocols to be clients.
37type ClientProtocol interface {
38 io.Closer
39
40 // Dispense dispenses a new instance of the plugin with the given name.
41 Dispense(string) (interface{}, error)
42
43 // Ping checks that the client connection is still healthy.
44 Ping() error
45}
diff --git a/vendor/github.com/hashicorp/go-plugin/rpc_client.go b/vendor/github.com/hashicorp/go-plugin/rpc_client.go
index 29f9bf0..f30a4b1 100644
--- a/vendor/github.com/hashicorp/go-plugin/rpc_client.go
+++ b/vendor/github.com/hashicorp/go-plugin/rpc_client.go
@@ -1,6 +1,7 @@
1package plugin 1package plugin
2 2
3import ( 3import (
4 "crypto/tls"
4 "fmt" 5 "fmt"
5 "io" 6 "io"
6 "net" 7 "net"
@@ -19,6 +20,42 @@ type RPCClient struct {
19 stdout, stderr net.Conn 20 stdout, stderr net.Conn
20} 21}
21 22
23// newRPCClient creates a new RPCClient. The Client argument is expected
24// to be successfully started already with a lock held.
25func newRPCClient(c *Client) (*RPCClient, error) {
26 // Connect to the client
27 conn, err := net.Dial(c.address.Network(), c.address.String())
28 if err != nil {
29 return nil, err
30 }
31 if tcpConn, ok := conn.(*net.TCPConn); ok {
32 // Make sure to set keep alive so that the connection doesn't die
33 tcpConn.SetKeepAlive(true)
34 }
35
36 if c.config.TLSConfig != nil {
37 conn = tls.Client(conn, c.config.TLSConfig)
38 }
39
40 // Create the actual RPC client
41 result, err := NewRPCClient(conn, c.config.Plugins)
42 if err != nil {
43 conn.Close()
44 return nil, err
45 }
46
47 // Begin the stream syncing so that stdin, out, err work properly
48 err = result.SyncStreams(
49 c.config.SyncStdout,
50 c.config.SyncStderr)
51 if err != nil {
52 result.Close()
53 return nil, err
54 }
55
56 return result, nil
57}
58
22// NewRPCClient creates a client from an already-open connection-like value. 59// NewRPCClient creates a client from an already-open connection-like value.
23// Dial is typically used instead. 60// Dial is typically used instead.
24func NewRPCClient(conn io.ReadWriteCloser, plugins map[string]Plugin) (*RPCClient, error) { 61func NewRPCClient(conn io.ReadWriteCloser, plugins map[string]Plugin) (*RPCClient, error) {
@@ -121,3 +158,13 @@ func (c *RPCClient) Dispense(name string) (interface{}, error) {
121 158
122 return p.Client(c.broker, rpc.NewClient(conn)) 159 return p.Client(c.broker, rpc.NewClient(conn))
123} 160}
161
162// Ping pings the connection to ensure it is still alive.
163//
164// The error from the RPC call is returned exactly if you want to inspect
165// it for further error analysis. Any error returned from here would indicate
166// that the connection to the plugin is not healthy.
167func (c *RPCClient) Ping() error {
168 var empty struct{}
169 return c.control.Call("Control.Ping", true, &empty)
170}
diff --git a/vendor/github.com/hashicorp/go-plugin/rpc_server.go b/vendor/github.com/hashicorp/go-plugin/rpc_server.go
index 3984dc8..5bb18dd 100644
--- a/vendor/github.com/hashicorp/go-plugin/rpc_server.go
+++ b/vendor/github.com/hashicorp/go-plugin/rpc_server.go
@@ -34,10 +34,14 @@ type RPCServer struct {
34 lock sync.Mutex 34 lock sync.Mutex
35} 35}
36 36
37// Accept accepts connections on a listener and serves requests for 37// ServerProtocol impl.
38// each incoming connection. Accept blocks; the caller typically invokes 38func (s *RPCServer) Init() error { return nil }
39// it in a go statement. 39
40func (s *RPCServer) Accept(lis net.Listener) { 40// ServerProtocol impl.
41func (s *RPCServer) Config() string { return "" }
42
43// ServerProtocol impl.
44func (s *RPCServer) Serve(lis net.Listener) {
41 for { 45 for {
42 conn, err := lis.Accept() 46 conn, err := lis.Accept()
43 if err != nil { 47 if err != nil {
@@ -122,6 +126,14 @@ type controlServer struct {
122 server *RPCServer 126 server *RPCServer
123} 127}
124 128
129// Ping can be called to verify the connection (and likely the binary)
130// is still alive to a plugin.
131func (c *controlServer) Ping(
132 null bool, response *struct{}) error {
133 *response = struct{}{}
134 return nil
135}
136
125func (c *controlServer) Quit( 137func (c *controlServer) Quit(
126 null bool, response *struct{}) error { 138 null bool, response *struct{}) error {
127 // End the server 139 // End the server
diff --git a/vendor/github.com/hashicorp/go-plugin/server.go b/vendor/github.com/hashicorp/go-plugin/server.go
index b5c5270..1e808b9 100644
--- a/vendor/github.com/hashicorp/go-plugin/server.go
+++ b/vendor/github.com/hashicorp/go-plugin/server.go
@@ -1,6 +1,8 @@
1package plugin 1package plugin
2 2
3import ( 3import (
4 "crypto/tls"
5 "encoding/base64"
4 "errors" 6 "errors"
5 "fmt" 7 "fmt"
6 "io/ioutil" 8 "io/ioutil"
@@ -11,6 +13,10 @@ import (
11 "runtime" 13 "runtime"
12 "strconv" 14 "strconv"
13 "sync/atomic" 15 "sync/atomic"
16
17 "github.com/hashicorp/go-hclog"
18
19 "google.golang.org/grpc"
14) 20)
15 21
16// CoreProtocolVersion is the ProtocolVersion of the plugin system itself. 22// CoreProtocolVersion is the ProtocolVersion of the plugin system itself.
@@ -45,14 +51,41 @@ type ServeConfig struct {
45 // HandshakeConfig is the configuration that must match clients. 51 // HandshakeConfig is the configuration that must match clients.
46 HandshakeConfig 52 HandshakeConfig
47 53
54 // TLSProvider is a function that returns a configured tls.Config.
55 TLSProvider func() (*tls.Config, error)
56
48 // Plugins are the plugins that are served. 57 // Plugins are the plugins that are served.
49 Plugins map[string]Plugin 58 Plugins map[string]Plugin
59
60 // GRPCServer should be non-nil to enable serving the plugins over
61 // gRPC. This is a function to create the server when needed with the
62 // given server options. The server options populated by go-plugin will
63 // be for TLS if set. You may modify the input slice.
64 //
65 // Note that the grpc.Server will automatically be registered with
66 // the gRPC health checking service. This is not optional since go-plugin
67 // relies on this to implement Ping().
68 GRPCServer func([]grpc.ServerOption) *grpc.Server
69
70 // Logger is used to pass a logger into the server. If none is provided the
71 // server will create a default logger.
72 Logger hclog.Logger
73}
74
75// Protocol returns the protocol that this server should speak.
76func (c *ServeConfig) Protocol() Protocol {
77 result := ProtocolNetRPC
78 if c.GRPCServer != nil {
79 result = ProtocolGRPC
80 }
81
82 return result
50} 83}
51 84
52// Serve serves the plugins given by ServeConfig. 85// Serve serves the plugins given by ServeConfig.
53// 86//
54// Serve doesn't return until the plugin is done being executed. Any 87// Serve doesn't return until the plugin is done being executed. Any
55// errors will be outputted to the log. 88// errors will be outputted to os.Stderr.
56// 89//
57// This is the method that plugins should call in their main() functions. 90// This is the method that plugins should call in their main() functions.
58func Serve(opts *ServeConfig) { 91func Serve(opts *ServeConfig) {
@@ -77,6 +110,16 @@ func Serve(opts *ServeConfig) {
77 // Logging goes to the original stderr 110 // Logging goes to the original stderr
78 log.SetOutput(os.Stderr) 111 log.SetOutput(os.Stderr)
79 112
113 logger := opts.Logger
114 if logger == nil {
115 // internal logger to os.Stderr
116 logger = hclog.New(&hclog.LoggerOptions{
117 Level: hclog.Trace,
118 Output: os.Stderr,
119 JSONFormat: true,
120 })
121 }
122
80 // Create our new stdout, stderr files. These will override our built-in 123 // Create our new stdout, stderr files. These will override our built-in
81 // stdout/stderr so that it works across the stream boundary. 124 // stdout/stderr so that it works across the stream boundary.
82 stdout_r, stdout_w, err := os.Pipe() 125 stdout_r, stdout_w, err := os.Pipe()
@@ -93,30 +136,86 @@ func Serve(opts *ServeConfig) {
93 // Register a listener so we can accept a connection 136 // Register a listener so we can accept a connection
94 listener, err := serverListener() 137 listener, err := serverListener()
95 if err != nil { 138 if err != nil {
96 log.Printf("[ERR] plugin: plugin init: %s", err) 139 logger.Error("plugin init error", "error", err)
97 return 140 return
98 } 141 }
99 defer listener.Close() 142
143 // Close the listener on return. We wrap this in a func() on purpose
144 // because the "listener" reference may change to TLS.
145 defer func() {
146 listener.Close()
147 }()
148
149 var tlsConfig *tls.Config
150 if opts.TLSProvider != nil {
151 tlsConfig, err = opts.TLSProvider()
152 if err != nil {
153 logger.Error("plugin tls init", "error", err)
154 return
155 }
156 }
100 157
101 // Create the channel to tell us when we're done 158 // Create the channel to tell us when we're done
102 doneCh := make(chan struct{}) 159 doneCh := make(chan struct{})
103 160
104 // Create the RPC server to dispense 161 // Build the server type
105 server := &RPCServer{ 162 var server ServerProtocol
106 Plugins: opts.Plugins, 163 switch opts.Protocol() {
107 Stdout: stdout_r, 164 case ProtocolNetRPC:
108 Stderr: stderr_r, 165 // If we have a TLS configuration then we wrap the listener
109 DoneCh: doneCh, 166 // ourselves and do it at that level.
167 if tlsConfig != nil {
168 listener = tls.NewListener(listener, tlsConfig)
169 }
170
171 // Create the RPC server to dispense
172 server = &RPCServer{
173 Plugins: opts.Plugins,
174 Stdout: stdout_r,
175 Stderr: stderr_r,
176 DoneCh: doneCh,
177 }
178
179 case ProtocolGRPC:
180 // Create the gRPC server
181 server = &GRPCServer{
182 Plugins: opts.Plugins,
183 Server: opts.GRPCServer,
184 TLS: tlsConfig,
185 Stdout: stdout_r,
186 Stderr: stderr_r,
187 DoneCh: doneCh,
188 }
189
190 default:
191 panic("unknown server protocol: " + opts.Protocol())
110 } 192 }
111 193
194 // Initialize the servers
195 if err := server.Init(); err != nil {
196 logger.Error("protocol init", "error", err)
197 return
198 }
199
200 // Build the extra configuration
201 extra := ""
202 if v := server.Config(); v != "" {
203 extra = base64.StdEncoding.EncodeToString([]byte(v))
204 }
205 if extra != "" {
206 extra = "|" + extra
207 }
208
209 logger.Debug("plugin address", "network", listener.Addr().Network(), "address", listener.Addr().String())
210
112 // Output the address and service name to stdout so that core can bring it up. 211 // Output the address and service name to stdout so that core can bring it up.
113 log.Printf("[DEBUG] plugin: plugin address: %s %s\n", 212 fmt.Printf("%d|%d|%s|%s|%s%s\n",
114 listener.Addr().Network(), listener.Addr().String())
115 fmt.Printf("%d|%d|%s|%s\n",
116 CoreProtocolVersion, 213 CoreProtocolVersion,
117 opts.ProtocolVersion, 214 opts.ProtocolVersion,
118 listener.Addr().Network(), 215 listener.Addr().Network(),
119 listener.Addr().String()) 216 listener.Addr().String(),
217 opts.Protocol(),
218 extra)
120 os.Stdout.Sync() 219 os.Stdout.Sync()
121 220
122 // Eat the interrupts 221 // Eat the interrupts
@@ -127,9 +226,7 @@ func Serve(opts *ServeConfig) {
127 for { 226 for {
128 <-ch 227 <-ch
129 newCount := atomic.AddInt32(&count, 1) 228 newCount := atomic.AddInt32(&count, 1)
130 log.Printf( 229 logger.Debug("plugin received interrupt signal, ignoring", "count", newCount)
131 "[DEBUG] plugin: received interrupt signal (count: %d). Ignoring.",
132 newCount)
133 } 230 }
134 }() 231 }()
135 232
@@ -137,10 +234,8 @@ func Serve(opts *ServeConfig) {
137 os.Stdout = stdout_w 234 os.Stdout = stdout_w
138 os.Stderr = stderr_w 235 os.Stderr = stderr_w
139 236
140 // Serve 237 // Accept connections and wait for completion
141 go server.Accept(listener) 238 go server.Serve(listener)
142
143 // Wait for the graceful exit
144 <-doneCh 239 <-doneCh
145} 240}
146 241
diff --git a/vendor/github.com/hashicorp/go-plugin/testing.go b/vendor/github.com/hashicorp/go-plugin/testing.go
index 9086a1b..df29593 100644
--- a/vendor/github.com/hashicorp/go-plugin/testing.go
+++ b/vendor/github.com/hashicorp/go-plugin/testing.go
@@ -2,9 +2,12 @@ package plugin
2 2
3import ( 3import (
4 "bytes" 4 "bytes"
5 "context"
5 "net" 6 "net"
6 "net/rpc" 7 "net/rpc"
7 "testing" 8
9 "github.com/mitchellh/go-testing-interface"
10 "google.golang.org/grpc"
8) 11)
9 12
10// The testing file contains test helpers that you can use outside of 13// The testing file contains test helpers that you can use outside of
@@ -12,7 +15,7 @@ import (
12 15
13// TestConn is a helper function for returning a client and server 16// TestConn is a helper function for returning a client and server
14// net.Conn connected to each other. 17// net.Conn connected to each other.
15func TestConn(t *testing.T) (net.Conn, net.Conn) { 18func TestConn(t testing.T) (net.Conn, net.Conn) {
16 // Listen to any local port. This listener will be closed 19 // Listen to any local port. This listener will be closed
17 // after a single connection is established. 20 // after a single connection is established.
18 l, err := net.Listen("tcp", "127.0.0.1:0") 21 l, err := net.Listen("tcp", "127.0.0.1:0")
@@ -46,7 +49,7 @@ func TestConn(t *testing.T) (net.Conn, net.Conn) {
46} 49}
47 50
48// TestRPCConn returns a rpc client and server connected to each other. 51// TestRPCConn returns a rpc client and server connected to each other.
49func TestRPCConn(t *testing.T) (*rpc.Client, *rpc.Server) { 52func TestRPCConn(t testing.T) (*rpc.Client, *rpc.Server) {
50 clientConn, serverConn := TestConn(t) 53 clientConn, serverConn := TestConn(t)
51 54
52 server := rpc.NewServer() 55 server := rpc.NewServer()
@@ -58,7 +61,7 @@ func TestRPCConn(t *testing.T) (*rpc.Client, *rpc.Server) {
58 61
59// TestPluginRPCConn returns a plugin RPC client and server that are connected 62// TestPluginRPCConn returns a plugin RPC client and server that are connected
60// together and configured. 63// together and configured.
61func TestPluginRPCConn(t *testing.T, ps map[string]Plugin) (*RPCClient, *RPCServer) { 64func TestPluginRPCConn(t testing.T, ps map[string]Plugin) (*RPCClient, *RPCServer) {
62 // Create two net.Conns we can use to shuttle our control connection 65 // Create two net.Conns we can use to shuttle our control connection
63 clientConn, serverConn := TestConn(t) 66 clientConn, serverConn := TestConn(t)
64 67
@@ -74,3 +77,78 @@ func TestPluginRPCConn(t *testing.T, ps map[string]Plugin) (*RPCClient, *RPCServ
74 77
75 return client, server 78 return client, server
76} 79}
80
81// TestGRPCConn returns a gRPC client conn and grpc server that are connected
82// together and configured. The register function is used to register services
83// prior to the Serve call. This is used to test gRPC connections.
84func TestGRPCConn(t testing.T, register func(*grpc.Server)) (*grpc.ClientConn, *grpc.Server) {
85 // Create a listener
86 l, err := net.Listen("tcp", "127.0.0.1:0")
87 if err != nil {
88 t.Fatalf("err: %s", err)
89 }
90
91 server := grpc.NewServer()
92 register(server)
93 go server.Serve(l)
94
95 // Connect to the server
96 conn, err := grpc.Dial(
97 l.Addr().String(),
98 grpc.WithBlock(),
99 grpc.WithInsecure())
100 if err != nil {
101 t.Fatalf("err: %s", err)
102 }
103
104 // Connection successful, close the listener
105 l.Close()
106
107 return conn, server
108}
109
110// TestPluginGRPCConn returns a plugin gRPC client and server that are connected
111// together and configured. This is used to test gRPC connections.
112func TestPluginGRPCConn(t testing.T, ps map[string]Plugin) (*GRPCClient, *GRPCServer) {
113 // Create a listener
114 l, err := net.Listen("tcp", "127.0.0.1:0")
115 if err != nil {
116 t.Fatalf("err: %s", err)
117 }
118
119 // Start up the server
120 server := &GRPCServer{
121 Plugins: ps,
122 Server: DefaultGRPCServer,
123 Stdout: new(bytes.Buffer),
124 Stderr: new(bytes.Buffer),
125 }
126 if err := server.Init(); err != nil {
127 t.Fatalf("err: %s", err)
128 }
129 go server.Serve(l)
130
131 // Connect to the server
132 conn, err := grpc.Dial(
133 l.Addr().String(),
134 grpc.WithBlock(),
135 grpc.WithInsecure())
136 if err != nil {
137 t.Fatalf("err: %s", err)
138 }
139
140 brokerGRPCClient := newGRPCBrokerClient(conn)
141 broker := newGRPCBroker(brokerGRPCClient, nil)
142 go broker.Run()
143 go brokerGRPCClient.StartStream()
144
145 // Create the client
146 client := &GRPCClient{
147 Conn: conn,
148 Plugins: ps,
149 broker: broker,
150 doneCtx: context.Background(),
151 }
152
153 return client, server
154}
generated by cgit v1.2.3 (git 2.25.1) at 2024-07-05 23:26:56 +0000