Merge pull request #399 from askuy/feature/gracefulstop

graceful stop

ego.go

@@ -17,6 +17,7 @@ import (
 	"github.com/gotomicro/ego/server"
 	"github.com/gotomicro/ego/task/ecron"
 	"github.com/gotomicro/ego/task/ejob"
+	"go.uber.org/zap"
 )
 
 // Ego 分为三大部分
@@ -43,6 +44,13 @@ type Ego struct {
 
 	// 第三部分 可选方法
 	opts opts
+
+	// stopStartTime
+	stopInfo stopInfo
+}
+type stopInfo struct {
+	stopStartTime  time.Time
+	isGracefulStop bool
 }
 
 type opts struct {
@@ -232,11 +240,11 @@ func (e *Ego) Run() error {
 
 	// 阻塞，等待信号量
 	if err := <-e.cycle.Wait(e.opts.hang); err != nil {
-		e.logger.Error("Ego shutdown with error", elog.FieldComponent("app"), elog.FieldErr(err))
+		e.logger.Error("Ego shutdown with error", elog.FieldComponent("app"), elog.FieldErr(err), elog.FieldCost(time.Since(e.stopInfo.stopStartTime)), zap.Bool("grace", e.stopInfo.isGracefulStop), zap.String("stopTimeout", e.opts.stopTimeout.String()))
 		runSerialFuncLogError(e.opts.afterStopClean)
 		return err
 	}
-	e.logger.Info("stop Ego, bye!", elog.FieldComponent("app"))
+	e.logger.Info("stop ego, bye!", elog.FieldComponent("app"), elog.FieldCost(time.Since(e.stopInfo.stopStartTime)), zap.Bool("grace", e.stopInfo.isGracefulStop), zap.String("stopTimeout", e.opts.stopTimeout.String()))
 	// 运行停止后清理
 	runSerialFuncLogError(e.opts.afterStopClean)
 	return nil

ego_function.go

@@ -8,11 +8,11 @@ import (
 	"os/signal"
 	"runtime"
 	"syscall"
+	"time"
 
 	sentinelmetrics "github.com/alibaba/sentinel-golang/metrics"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/automaxprocs/maxprocs"
-	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/gotomicro/ego/core/constant"
@@ -42,23 +42,26 @@ func (e *Ego) waitSignals() {
 		grace := s != syscall.SIGQUIT
 		go func() {
 			// todo 父节点传context待考虑
-			stopCtx, cancel := context.WithTimeout(context.Background(), e.opts.stopTimeout)
+			e.stopInfo = stopInfo{
+				stopStartTime:  time.Now(),
+				isGracefulStop: grace,
+			}
+			stopCtx, cancel := context.WithTimeoutCause(context.Background(), e.opts.stopTimeout, fmt.Errorf("stop timeout %v", e.opts.stopTimeout))
+
 			defer func() {
 				signal.Stop(sig)
 				cancel()
 			}()
 
-			elog.Info("server stop", zap.Bool("graceful", grace))
-
 			_ = e.Stop(stopCtx, grace)
 			<-stopCtx.Done()
 			// 记录服务器关闭时候，由于关闭过慢，无法正常关闭，被强制cancel
 			if errors.Is(stopCtx.Err(), context.DeadlineExceeded) {
-				elog.Error("waitSignals stop context err", elog.FieldErr(stopCtx.Err()))
+				e.logger.Error("waitSignals stop context err", elog.FieldErr(stopCtx.Err()))
 			}
 		}()
 		<-sig
-		elog.Error("waitSignals quit")
+		e.logger.Error("waitSignals quit")
 		// 因为os.Signal长度为2，那么这里会阻塞住，如果发送两次信号量，强制退出
 		os.Exit(128 + int(s.(syscall.Signal))) // second signal. Exit directly.
 	}()

server/egin/component.go

@@ -181,7 +181,10 @@ func (c *Component) Stop() error {
 	c.mu.Lock()
 	err := c.Server.Close()
 	c.mu.Unlock()
-	return err
+	if err != nil {
+		return fmt.Errorf("egin Stop, err: %w", err)
+	}
+	return nil
 }
 
 // GracefulStop implements server.Component interface
@@ -190,7 +193,10 @@ func (c *Component) GracefulStop(ctx context.Context) error {
 	c.mu.Lock()
 	err := c.Server.Shutdown(ctx)
 	c.mu.Unlock()
-	return err
+	if err != nil {
+		return fmt.Errorf("egin GracefulStop, err: %w", err)
+	}
+	return nil
 }
 
 // Info returns server info, used by governor and consumer balancer

server/egovernor/component.go

@@ -3,6 +3,7 @@ package egovernor
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net"
 	"net/http"
 	"net/http/pprof"
@@ -153,12 +154,20 @@ func (c *Component) Start() error {
 
 // Stop ..
 func (c *Component) Stop() error {
-	return c.Server.Close()
+	err := c.Server.Close()
+	if err != nil {
+		return fmt.Errorf("egovernor Stop, err: %w", err)
+	}
+	return nil
 }
 
 // GracefulStop ..
 func (c *Component) GracefulStop(ctx context.Context) error {
-	return c.Server.Shutdown(ctx)
+	err := c.Server.Shutdown(ctx)
+	if err != nil {
+		return fmt.Errorf("egovernor GracefulStop, err: %w", err)
+	}
+	return nil
 }
 
 // Info ..

server/egrpc/component.go

@@ -2,6 +2,7 @@ package egrpc
 
 import (
 	"context"
+	"fmt"
 	"net"
 
 	"github.com/gotomicro/ego/core/constant"
@@ -157,7 +158,11 @@ func (c *Component) GracefulStop(ctx context.Context) error {
 	for {
 		select {
 		case <-ctx.Done():
-			return ctx.Err()
+			err := ctx.Err()
+			if err != nil {
+				return fmt.Errorf("egrpc GracefulStop, err: %w", err)
+			}
+			return nil
 		case <-c.quit:
 			return nil
 		}

test/gracefulstop/config.toml

@@ -0,0 +1,3 @@
+[server.http]
+    port = 9001
+    host = "0.0.0.0"

test/gracefulstop/gracefulstop.md

@@ -0,0 +1,5 @@
+构造一个接口需要响应5s
+发送信号量 kill -s SIGTERM PID
+
+curl http://127.0.0.1:9001/hello
+kill -s SIGTERM 53065

test/gracefulstop/main.go

@@ -0,0 +1,25 @@
+package main
+
+import (
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/gotomicro/ego"
+	"github.com/gotomicro/ego/core/elog"
+	"github.com/gotomicro/ego/server/egin"
+)
+
+// export EGO_DEBUG=true && go run main.go --config=config.toml
+func main() {
+	if err := ego.New().Serve(func() *egin.Component {
+		server := egin.Load("server.http").Build()
+		server.GET("/hello", func(ctx *gin.Context) {
+			time.Sleep(time.Second * 4)
+			ctx.JSON(200, "Hello EGO")
+			return
+		})
+		return server
+	}()).Run(); err != nil {
+		elog.Panic("startup", elog.FieldErr(err))
+	}
+}