sink-server: update Shutdown interface method, update server shutdown…

… sequence
streamingfast · Nov 29, 2023 · 9bf67a0 · 9bf67a0
1 parent 20bb46f
commit 9bf67a0
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 4 deletions.
diff --git a/sink-server/docker/docker.go b/sink-server/docker/docker.go
@@ -499,9 +499,10 @@ func toDuration(in time.Duration) *types.Duration {
 	return deref(types.Duration(in))
 }
 
-func (e *DockerEngine) Shutdown(ctx context.Context, zlog *zap.Logger) (err error) {
+func (e *DockerEngine) Shutdown(ctx context.Context, _ error, zlog *zap.Logger) error {
 	e.mutex.Lock()
 	defer e.mutex.Unlock()
+
 	deps, err := e.list(ctx, zlog)
 	if err != nil {
 		return fmt.Errorf("cannot list deployments: %w", err)
@@ -515,6 +516,7 @@ func (e *DockerEngine) Shutdown(ctx context.Context, zlog *zap.Logger) (err erro
 			err = errors.Join(err, e)
 		}
 	}
+
 	return err
 }
 

diff --git a/sink-server/interface.go b/sink-server/interface.go
@@ -23,7 +23,7 @@ type Engine interface {
 	Info(ctx context.Context, deploymentID string, zlog *zap.Logger) (*pbsinksvc.InfoResponse, error)
 	List(ctx context.Context, zlog *zap.Logger) ([]*pbsinksvc.DeploymentWithStatus, error)
 
-	Shutdown(ctx context.Context, zlog *zap.Logger) error
+	Shutdown(ctx context.Context, err error, zlog *zap.Logger) error
 }
 
 var _ Engine = &docker.DockerEngine{}
diff --git a/sink-server/server.go b/sink-server/server.go
@@ -83,10 +83,15 @@ func (s *server) Run(ctx context.Context) {
 	addr := strings.ReplaceAll(s.httpListenAddr, "*", "")
 
 	s.OnTerminating(func(err error) {
-		time.Sleep(time.Second)
+		shutdownErr := s.engine.Shutdown(ctx, err, s.logger)
+		if shutdownErr != nil {
+			s.logger.Warn("failed to shutdown engine", zap.Error(shutdownErr))
+		}
+
+		time.Sleep(1 * time.Second)
+
 		s.logger.Info("shutting down connect web server")
 		srv.Shutdown(nil)
-		s.engine.Shutdown(ctx, s.logger)
 	})
 
 	srv.Launch(addr)