agent: Properly stop the gRPC server

This commit attempts to close cleanly the gRPC server so that tracing
will be ended properly.

Fixes #445

Signed-off-by: Sebastien Boeuf <sebastien.boeuf@intel.com>

agent.go

@@ -120,6 +120,7 @@ type sandbox struct {
 	enableGrpcTrace   bool
 	sandboxPidNs      bool
 	storages          map[string]*sandboxStorage
+	stopServer        chan struct{}
 }
 
 var agentFields = logrus.Fields{
@@ -524,6 +525,36 @@ func (s *sandbox) teardownSharedPidNs() error {
 	return nil
 }
 
+func (s *sandbox) waitForStopServer() {
+	fieldLogger := agentLog.WithField("subsystem", "stopserverwatcher")
+
+	fieldLogger.Info("Waiting for stopServer signal...")
+
+	// Wait for DestroySandbox() to signal this thread about the need to
+	// stop the server.
+	<-s.stopServer
+
+	defer fieldLogger.Info("gRPC server stopped")
+
+	// Try to gracefully stop the server for a minute
+	timeout := time.Minute
+	done := make(chan struct{})
+	go func() {
+		s.gracefulStopGRPC()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		return
+	case <-time.After(timeout):
+		fieldLogger.Warnf("Could not gracefully stop the server after %v", timeout)
+	}
+
+	fieldLogger.Info("Force stopping the server now")
+	s.stopGRPC()
+}
+
 func (s *sandbox) listenToUdevEvents() {
 	fieldLogger := agentLog.WithField("subsystem", "udevlistener")
 
@@ -810,6 +841,7 @@ func (s *sandbox) startGRPC() {
 		defer s.wg.Done()
 
 		var err error
+		var servErr error
 		for {
 			agentLog.Info("agent grpc server starts")
 
@@ -833,19 +865,37 @@ func (s *sandbox) startGRPC() {
 			}
 
 			// l is closed when Serve() returns
-			err = grpcServer.Serve(l)
-			if err != nil {
+			servErr = grpcServer.Serve(l)
+			if servErr != nil {
 				agentLog.WithError(err).Warn("agent grpc server quits")
 			}
 
 			err = s.channel.teardown()
 			if err != nil {
 				agentLog.WithError(err).Warn("agent grpc channel teardown failed")
 			}
+
+			// Based on the definition of grpc.Serve(), the function
+			// returns nil in case of a proper stop triggered by either
+			// grpc.GracefulStop() or grpc.Stop(). Those calls can only
+			// be issued by the chain of events coming from DestroySandbox
+			// and explicitely means the server should not try to listen
+			// again, as the sandbox is being completely removed.
+			if servErr == nil {
+				agentLog.Info("ggent grpc server has been explicitely stopped")
+				return
+			}
 		}
 	}()
 }
 
+func (s *sandbox) gracefulStopGRPC() {
+	if s.server != nil {
+		s.server.GracefulStop()
+		s.server = nil
+	}
+}
+
 func (s *sandbox) stopGRPC() {
 	if s.server != nil {
 		s.server.Stop()
@@ -1019,6 +1069,7 @@ func realMain() {
 		pciDeviceMap:   make(map[string]string),
 		deviceWatchers: make(map[string](chan string)),
 		storages:       make(map[string]*sandboxStorage),
+		stopServer:     make(chan struct{}),
 	}
 
 	if err = s.initLogger(); err != nil {
@@ -1046,6 +1097,8 @@ func realMain() {
 	// Start gRPC server.
 	s.startGRPC()
 
+	go s.waitForStopServer()
+
 	go s.listenToUdevEvents()
 
 	s.wg.Wait()

grpc.go

@@ -1341,6 +1341,10 @@ func (a *agentGRPC) DestroySandbox(ctx context.Context, req *pb.DestroySandboxRe
 		return emptyResp, err
 	}
 
+	// Close stopServer channel to signal the main agent code to stop
+	// the server when all gRPC calls will be completed.
+	close(a.sandbox.stopServer)
+
 	a.sandbox.hostname = ""
 	a.sandbox.id = ""
 	a.sandbox.containers = make(map[string]*container)