backport of commit f884c3c4cd (#22922)

Co-authored-by: Tom Proctor <tomhjp@users.noreply.github.com>

changelog/22914.txt

@@ -0,0 +1,6 @@
+```release-note:bug
+plugins: Fix instance where broken/unresponsive plugins could cause Vault to hang.
+```
+```release-note:bug
+plugins: Fix instance where Vault could fail to kill broken/unresponsive plugins.
+```

helper/testhelpers/pluginhelpers/pluginhelpers.go

@@ -154,6 +154,8 @@ func BuildPluginContainerImage(t testing.T, plugin TestPlugin, pluginDir string)
 	ref := plugin.Name
 	if plugin.Version != "" {
 		ref += ":" + strings.TrimPrefix(plugin.Version, "v")
+	} else {
+		ref += ":latest"
 	}
 	args := []string{"build", "--tag=" + ref, "--build-arg=plugin=" + plugin.FileName, "--file=vault/testdata/Dockerfile", pluginDir}
 	cmd := exec.Command("docker", args...)

sdk/plugin/grpc_backend_client.go

@@ -25,7 +25,7 @@ var (
 )
 
 // Validate backendGRPCPluginClient satisfies the logical.Backend interface
-var _ logical.Backend = &backendGRPCPluginClient{}
+var _ logical.Backend = (*backendGRPCPluginClient)(nil)
 
 // backendPluginClient implements logical.Backend and is the
 // go-plugin client.
@@ -183,17 +183,21 @@ func (b *backendGRPCPluginClient) Cleanup(ctx context.Context) {
 	defer close(quitCh)
 	defer cancel()
 
-	b.client.Cleanup(ctx, &pb.Empty{})
+	// Only wait on graceful cleanup if we can establish communication with the
-
+	// plugin, otherwise b.cleanupCh may never get closed.
+	if _, err := b.client.Cleanup(ctx, &pb.Empty{}); status.Code(err) != codes.Unavailable {
 		// This will block until Setup has run the function to create a new server
 		// in b.server. If we stop here before it has a chance to actually start
 		// listening, when it starts listening it will immediately error out and
 		// exit, which is fine. Overall this ensures that we do not miss stopping
 		// the server if it ends up being created after Cleanup is called.
-	<-b.cleanupCh
+		select {
+		case <-b.cleanupCh:
+		}
+	}
 	server := b.server.Load()
-	if server != nil {
+	if grpcServer, ok := server.(*grpc.Server); ok && grpcServer != nil {
-		server.(*grpc.Server).GracefulStop()
+		grpcServer.GracefulStop()
 	}
 }
 

vault/plugin_catalog.go

@@ -383,6 +383,7 @@ func (c *PluginCatalog) newPluginClient(ctx context.Context, pluginRunner *plugi
 
 	// Multiplexing support will always be false initially, but will be
 	// adjusted once we query from the plugin whether it can multiplex or not
+	var spawnedPlugin bool
 	if !extPlugin.multiplexingSupport || len(extPlugin.connections) == 0 {
 		c.logger.Debug("spawning a new plugin process", "plugin_name", pluginRunner.Name, "id", id)
 		client, err := pluginRunner.RunConfig(ctx,
@@ -398,6 +399,7 @@ func (c *PluginCatalog) newPluginClient(ctx context.Context, pluginRunner *plugi
 			return nil, err
 		}
 
+		spawnedPlugin = true
 		pc.client = client
 	} else {
 		c.logger.Debug("returning existing plugin client for multiplexed plugin", "id", id)
@@ -417,6 +419,11 @@ func (c *PluginCatalog) newPluginClient(ctx context.Context, pluginRunner *plugi
 	// Subsequent calls to this will return the same client.
 	rpcClient, err := pc.client.Client()
 	if err != nil {
+		// Make sure we kill any spawned plugins that didn't make it into our
+		// map of connections.
+		if spawnedPlugin {
+			pc.client.Kill()
+		}
 		return nil, err
 	}
 
@@ -427,6 +434,11 @@ func (c *PluginCatalog) newPluginClient(ctx context.Context, pluginRunner *plugi
 
 	muxed, err := pluginutil.MultiplexingSupported(ctx, clientConn, config.Name)
 	if err != nil {
+		// Make sure we kill any spawned plugins that didn't make it into our
+		// map of connections.
+		if spawnedPlugin {
+			pc.client.Kill()
+		}
 		return nil, err
 	}