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

   1 package plugin
   2
   3 import (
   4         "bytes"
   5         "crypto/tls"
   6         "encoding/json"
   7         "fmt"
   8         "io"
   9         "net"
  10
  11         hclog "github.com/hashicorp/go-hclog"
  12         "github.com/hashicorp/go-plugin/internal/plugin"
  13         "google.golang.org/grpc"
  14         "google.golang.org/grpc/credentials"
  15         "google.golang.org/grpc/health"
  16         "google.golang.org/grpc/health/grpc_health_v1"
  17 )
  18
  19 // GRPCServiceName is the name of the service that the health check should
  20 // return as passing.
  21 const GRPCServiceName = "plugin"
  22
  23 // DefaultGRPCServer can be used with the "GRPCServer" field for Server
  24 // as a default factory method to create a gRPC server with no extra options.
  25 func DefaultGRPCServer(opts []grpc.ServerOption) *grpc.Server {
  26         return grpc.NewServer(opts...)
  27 }
  28
  29 // GRPCServer is a ServerType implementation that serves plugins over
  30 // gRPC. This allows plugins to easily be written for other languages.
  31 //
  32 // The GRPCServer outputs a custom configuration as a base64-encoded
  33 // JSON structure represented by the GRPCServerConfig config structure.
  34 type GRPCServer struct {
  35         // Plugins are the list of plugins to serve.
  36         Plugins map[string]Plugin
  37
  38         // Server is the actual server that will accept connections. This
  39         // will be used for plugin registration as well.
  40         Server func([]grpc.ServerOption) *grpc.Server
  41
  42         // TLS should be the TLS configuration if available. If this is nil,
  43         // the connection will not have transport security.
  44         TLS *tls.Config
  45
  46         // DoneCh is the channel that is closed when this server has exited.
  47         DoneCh chan struct{}
  48
  49         // Stdout/StderrLis are the readers for stdout/stderr that will be copied
  50         // to the stdout/stderr connection that is output.
  51         Stdout io.Reader
  52         Stderr io.Reader
  53
  54         config GRPCServerConfig
  55         server *grpc.Server
  56         broker *GRPCBroker
  57
  58         logger hclog.Logger
  59 }
  60
  61 // ServerProtocol impl.
  62 func (s *GRPCServer) Init() error {
  63         // Create our server
  64         var opts []grpc.ServerOption
  65         if s.TLS != nil {
  66                 opts = append(opts, grpc.Creds(credentials.NewTLS(s.TLS)))
  67         }
  68         s.server = s.Server(opts)
  69
  70         // Register the health service
  71         healthCheck := health.NewServer()
  72         healthCheck.SetServingStatus(
  73                 GRPCServiceName, grpc_health_v1.HealthCheckResponse_SERVING)
  74         grpc_health_v1.RegisterHealthServer(s.server, healthCheck)
  75
  76         // Register the broker service
  77         brokerServer := newGRPCBrokerServer()
  78         plugin.RegisterGRPCBrokerServer(s.server, brokerServer)
  79         s.broker = newGRPCBroker(brokerServer, s.TLS)
  80         go s.broker.Run()
  81
  82         // Register the controller
  83         controllerServer := &grpcControllerServer{
  84                 server: s,
  85         }
  86         plugin.RegisterGRPCControllerServer(s.server, controllerServer)
  87
  88         // Register all our plugins onto the gRPC server.
  89         for k, raw := range s.Plugins {
  90                 p, ok := raw.(GRPCPlugin)
  91                 if !ok {
  92                         return fmt.Errorf("%q is not a GRPC-compatible plugin", k)
  93                 }
  94
  95                 if err := p.GRPCServer(s.broker, s.server); err != nil {
  96                         return fmt.Errorf("error registering %q: %s", k, err)
  97                 }
  98         }
  99
 100         return nil
 101 }
 102
 103 // Stop calls Stop on the underlying grpc.Server
 104 func (s *GRPCServer) Stop() {
 105         s.server.Stop()
 106 }
 107
 108 // GracefulStop calls GracefulStop on the underlying grpc.Server
 109 func (s *GRPCServer) GracefulStop() {
 110         s.server.GracefulStop()
 111 }
 112
 113 // Config is the GRPCServerConfig encoded as JSON then base64.
 114 func (s *GRPCServer) Config() string {
 115         // Create a buffer that will contain our final contents
 116         var buf bytes.Buffer
 117
 118         // Wrap the base64 encoding with JSON encoding.
 119         if err := json.NewEncoder(&buf).Encode(s.config); err != nil {
 120                 // We panic since ths shouldn't happen under any scenario. We
 121                 // carefully control the structure being encoded here and it should
 122                 // always be successful.
 123                 panic(err)
 124         }
 125
 126         return buf.String()
 127 }
 128
 129 func (s *GRPCServer) Serve(lis net.Listener) {
 130         defer close(s.DoneCh)
 131         err := s.server.Serve(lis)
 132         if err != nil {
 133                 s.logger.Error("grpc server", "error", err)
 134         }
 135 }
 136
 137 // GRPCServerConfig is the extra configuration passed along for consumers
 138 // to facilitate using GRPC plugins.
 139 type GRPCServerConfig struct {
 140         StdoutAddr string `json:"stdout_addr"`
 141         StderrAddr string `json:"stderr_addr"`
 142 }