Start,stop,exit the grpc server gracefully

Author: jshahriddhi

Dockerfile

@@ -25,4 +25,4 @@ WORKDIR /root/
 RUN apk add gcompat
 COPY --from=consul-dataplane-binary /cdp/consul-dataplane /usr/local/bin/consul-dataplane
 COPY --from=envoy-binary /usr/local/bin/envoy /usr/local/bin/envoy
-ENTRYPOINT [ "./consul-dataplane" ]
+ENTRYPOINT [ "consul-dataplane" ]

pkg/consuldp/bootstrap.go

@@ -25,9 +25,9 @@ const (
 
 // bootstrapConfig generates the Envoy bootstrap config in JSON format.
 func (cdp *ConsulDataplane) bootstrapConfig(ctx context.Context) ([]byte, error) {
-	cdpFullAddr := strings.Split(cdp.gRPCListener.Addr().String(), ":")
-	cdpAddr := cdpFullAddr[0]
-	cdpPort := cdpFullAddr[1]
+	cdpGRPCFullAddr := strings.Split(cdp.gRPCServer.listener.Addr().String(), ":")
+	cdpGRPCAddr := cdpGRPCFullAddr[0]
+	cdpGRPCPort := cdpGRPCFullAddr[1]
 	svc := cdp.cfg.Service
 	envoy := cdp.cfg.Envoy
 
@@ -54,9 +54,8 @@ func (cdp *ConsulDataplane) bootstrapConfig(ctx context.Context) ([]byte, error)
 
 	args := &bootstrap.BootstrapTplArgs{
 		GRPC: bootstrap.GRPC{
-			// For now we just give the server address directly.
-			AgentAddress: cdpAddr,
-			AgentPort:    cdpPort,
+			AgentAddress: cdpGRPCAddr,
+			AgentPort:    cdpGRPCPort,
 			AgentTLS:     false,
 		},
 		ProxyCluster:          rsp.Service,

pkg/consuldp/consul_dataplane.go

@@ -25,18 +25,23 @@ type consulServer struct {
 	// supportedFeatures is a map of the dataplane features supported by the Consul server
 	supportedFeatures map[pbdataplane.DataplaneFeatures]bool
 
+	// grpcClientConn is the gRPC connection to the Consul server
 	grpcClientConn *grpc.ClientConn
 }
 
+type gRPCServer struct {
+	listener net.Listener
+	server   *grpc.Server
+	exitedCh chan struct{}
+}
+
 // ConsulDataplane represents the consul-dataplane process
 type ConsulDataplane struct {
 	logger          hclog.Logger
 	cfg             *Config
 	consulServer    *consulServer
 	dpServiceClient pbdataplane.DataplaneServiceClient
-
-	gRPCListener net.Listener
-	gRPCServer   *grpc.Server
+	gRPCServer      *gRPCServer
 }
 
 // NewConsulDP creates a new instance of ConsulDataplane
@@ -137,6 +142,7 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error {
 		return err
 	}
 	defer grpcClientConn.Close()
+	// TODO (NET-148): Ensure the server connection here is the one acquired via the server discovery library
 	cdp.consulServer.grpcClientConn = grpcClientConn
 	cdp.logger.Info("connected to consul server over grpc", "grpc-target", gRPCTarget)
 
@@ -184,9 +190,16 @@ func (cdp *ConsulDataplane) Run(ctx context.Context) error {
 			if err := proxy.Stop(); err != nil {
 				cdp.logger.Error("failed to stop proxy", "error", err)
 			}
+			cdp.stopGRPCServer()
 			doneCh <- nil
 		case <-proxy.Exited():
+			cdp.stopGRPCServer()
 			doneCh <- errors.New("envoy proxy exited unexpectedly")
+		case <-cdp.gRPCServerExited():
+			if err := proxy.Stop(); err != nil {
+				cdp.logger.Error("failed to stop proxy", "error", err)
+			}
+			doneCh <- errors.New("gRPC server exited unexpectedly")
 		}
 	}()
 	return <-doneCh

pkg/consuldp/grpc.go

@@ -23,36 +23,44 @@ func (cdp *ConsulDataplane) setupGRPCServer() error {
 		cdp.logger.Error("failed to create gRPC/TCP listener: %v", err)
 		return err
 	}
-	cdp.gRPCListener = lis
 
 	// create gRPC server
 	// one main role of this gRPC server in consul-dataplane is to proxy envoy ADS requests
 	// to the connected Consul server.
 	director := func(ctx context.Context, fullMethodName string) (context.Context, *grpc.ClientConn, error) {
+		if !strings.Contains(fullMethodName, "envoy.service.discovery.v3.AggregatedDiscoveryService/DeltaAggregatedResources") {
+			return ctx, nil, status.Errorf(codes.Unimplemented, fmt.Sprintf("Unknown method %s", fullMethodName))
+		}
+
 		md, _ := metadata.FromIncomingContext(ctx)
 		mdCopy := md.Copy()
 		// TODO (NET-148): Inject the ACL token acquired from the server discovery library
 		mdCopy[metadataKeyToken] = []string{cdp.cfg.Consul.Credentials.Static.Token}
 		outCtx := metadata.NewOutgoingContext(ctx, mdCopy)
-		if !strings.Contains(fullMethodName, "envoy.service.discovery.v3.AggregatedDiscoveryService/DeltaAggregatedResources") {
-			return outCtx, nil, status.Errorf(codes.Unimplemented, fmt.Sprintf("Unknown method %s", fullMethodName))
-		}
-		// TODO (NET-148): Ensure the server connection here is the one acquired via the server discovery library
 		return outCtx, cdp.consulServer.grpcClientConn, nil
 	}
-	gRPCServer := grpc.NewServer(grpc.UnknownServiceHandler(proxy.TransparentHandler(director)))
-	cdp.gRPCServer = gRPCServer
+	newGRPCServer := grpc.NewServer(grpc.UnknownServiceHandler(proxy.TransparentHandler(director)))
 
-	cdp.logger.Info("created gRPC server", "address", lis.Addr().String())
+	cdp.gRPCServer = &gRPCServer{listener: lis, server: newGRPCServer, exitedCh: make(chan struct{})}
+	cdp.logger.Trace("created gRPC server", "address", lis.Addr().String())
 	return nil
 }
 
 func (cdp *ConsulDataplane) startGRPCServer() {
-	cdp.logger.Trace("starting gRPC server")
+	cdp.logger.Info("starting gRPC server", "address", cdp.gRPCServer.listener.Addr().String())
 
-	if err := cdp.gRPCServer.Serve(cdp.gRPCListener); err != nil {
-		cdp.logger.Error("failed to serve gRPC requests: %v", err)
-		cdp.gRPCListener.Close()
-		// TODO: gracefully exit
+	if err := cdp.gRPCServer.server.Serve(cdp.gRPCServer.listener); err != nil {
+		cdp.logger.Error("failed to serve gRPC requests", "error", err)
+		cdp.gRPCServer.listener.Close()
+		close(cdp.gRPCServer.exitedCh)
 	}
 }
+
+func (cdp *ConsulDataplane) stopGRPCServer() {
+	if cdp.gRPCServer != nil {
+		cdp.logger.Debug("stopping gRPC server")
+		cdp.gRPCServer.server.Stop()
+	}
+}
+
+func (cdp *ConsulDataplane) gRPCServerExited() chan struct{} { return cdp.gRPCServer.exitedCh }