Ftr: Gracefully shutdown

before_ut.bat

@@ -1,3 +1,19 @@
+::
+::  Licensed to the Apache Software Foundation (ASF) under one or more
+::  contributor license agreements.  See the NOTICE file distributed with
+::  this work for additional information regarding copyright ownership.
+::  The ASF licenses this file to You under the Apache License, Version 2.0
+::  (the "License"); you may not use this file except in compliance with
+::  the License.  You may obtain a copy of the License at
+::
+::      http://www.apache.org/licenses/LICENSE-2.0
+::
+::  Unless required by applicable law or agreed to in writing, software
+::  distributed under the License is distributed on an "AS IS" BASIS,
+::  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+::  See the License for the specific language governing permissions and
+::  limitations under the License.
+
 set zkJar=zookeeper-3.4.9-fatjar.jar
 md remoting\zookeeper\zookeeper-4unittest\contrib\fatjar config_center\zookeeper\zookeeper-4unittest\contrib\fatjar registry\zookeeper\zookeeper-4unittest\contrib\fatjar
 curl -L https://github.com/dubbogo/resources/raw/master/zookeeper-4unitest/contrib/fatjar/%zkJar% -o remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/%zkJar%

before_ut.sh

@@ -1,3 +1,20 @@
+#
+#  Licensed to the Apache Software Foundation (ASF) under one or more
+#  contributor license agreements.  See the NOTICE file distributed with
+#  this work for additional information regarding copyright ownership.
+#  The ASF licenses this file to You under the Apache License, Version 2.0
+#  (the "License"); you may not use this file except in compliance with
+#  the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+
 mkdir -p remoting/zookeeper/zookeeper-4unittest/contrib/fatjar config_center/zookeeper/zookeeper-4unittest/contrib/fatjar registry/zookeeper/zookeeper-4unittest/contrib/fatjar
 wget -P "remoting/zookeeper/zookeeper-4unittest/contrib/fatjar" https://github.com/dubbogo/resources/raw/master/zookeeper-4unitest/contrib/fatjar/zookeeper-3.4.9-fatjar.jar
 cp remoting/zookeeper/zookeeper-4unittest/contrib/fatjar/zookeeper-3.4.9-fatjar.jar config_center/zookeeper/zookeeper-4unittest/contrib/fatjar/

common/constant/default.go

@@ -46,8 +46,8 @@ const (
 const (
 	DEFAULT_KEY               = "default"
 	PREFIX_DEFAULT_KEY        = "default."
-	DEFAULT_SERVICE_FILTERS   = "echo,token,accesslog,tps,execute"
-	DEFAULT_REFERENCE_FILTERS = ""
+	DEFAULT_SERVICE_FILTERS   = "echo,token,accesslog,tps,execute,pshutdown"
+	DEFAULT_REFERENCE_FILTERS = "cshutdown"
 	GENERIC_REFERENCE_FILTERS = "generic"
 	GENERIC                   = "$invoke"
 	ECHO                      = "$echo"

common/constant/key.go

@@ -71,6 +71,8 @@ const (
 	EXECUTE_LIMIT_KEY                      = "execute.limit"
 	DEFAULT_EXECUTE_LIMIT                  = "-1"
 	EXECUTE_REJECTED_EXECUTION_HANDLER_KEY = "execute.limit.rejected.handler"
+	PROVIDER_SHUTDOWN_FILTER               = "pshutdown"
+	CONSUMER_SHUTDOWN_FILTER               = "cshutdown"
 )
 
 const (
@@ -112,6 +114,7 @@ const (
 	ProtocolConfigPrefix       = "dubbo.protocols."
 	ProviderConfigPrefix       = "dubbo.provider."
 	ConsumerConfigPrefix       = "dubbo.consumer."
+	ShutdownConfigPrefix       = "dubbo.shutdown."
 )
 
 const (

common/extension/graceful_shutdown.go

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package extension
+
+import (
+	"container/list"
+)
+
+var (
+	customShutdownCallbacks = list.New()
+)
+
+/**
+ * you should not make any assumption about the order.
+ * For example, if you have more than one callbacks, and you wish the order is:
+ * callback1()
+ * callback2()
+ * ...
+ * callbackN()
+ * Then you should put then together:
+ * func callback() {
+ *     callback1()
+ *     callback2()
+ *     ...
+ *     callbackN()
+ * }
+ * I think the order of custom callbacks should be decided by the users.
+ * Even though I can design a mechanism to support the ordered custom callbacks,
+ * the benefit of that mechanism is low.
+ * And it may introduce much complication for another users.
+ */
+func AddCustomShutdownCallback(callback func()) {
+	customShutdownCallbacks.PushBack(callback)
+}
+
+func GetAllCustomShutdownCallbacks() *list.List {
+	return customShutdownCallbacks
+}

config/base_config_test.go

@@ -39,6 +39,7 @@ func Test_refresh(t *testing.T) {
 	mockMap["dubbo.com.MockService.MockService.GetUser.retries"] = "10"
 	mockMap["dubbo.consumer.check"] = "false"
 	mockMap["dubbo.application.name"] = "dubbo"
+	mockMap["dubbo.shutdown.timeout"] = "12s"
 
 	config.GetEnvInstance().UpdateExternalConfigMap(mockMap)
 
@@ -113,6 +114,13 @@ func Test_refresh(t *testing.T) {
 				},
 			},
 		},
+		ShutdownConfig: &ShutdownConfig{
+			Timeout:              "12s",
+			StepTimeout:          "2s",
+			RejectRequestHandler: "mock",
+			RejectRequest:        false,
+			RequestsFinished:     false,
+		},
 	}
 
 	c.SetFatherConfig(father)

config/config_loader.go

@@ -149,6 +149,8 @@ func Load() {
 			}
 		}
 	}
+	// init the shutdown callback
+	GracefulShutdownInit()
 }
 
 // get rpc service for consumer

config/consumer_config.go

@@ -52,11 +52,12 @@ type ConsumerConfig struct {
 	ProxyFactory    string `yaml:"proxy_factory" default:"default" json:"proxy_factory,omitempty" property:"proxy_factory"`
 	Check           *bool  `yaml:"check"  json:"check,omitempty" property:"check"`
 
-	Registry     *RegistryConfig             `yaml:"registry" json:"registry,omitempty" property:"registry"`
-	Registries   map[string]*RegistryConfig  `yaml:"registries" json:"registries,omitempty" property:"registries"`
-	References   map[string]*ReferenceConfig `yaml:"references" json:"references,omitempty" property:"references"`
-	ProtocolConf interface{}                 `yaml:"protocol_conf" json:"protocol_conf,omitempty" property:"protocol_conf"`
-	FilterConf   interface{}                 `yaml:"filter_conf" json:"filter_conf,omitempty" property:"filter_conf" `
+	Registry       *RegistryConfig             `yaml:"registry" json:"registry,omitempty" property:"registry"`
+	Registries     map[string]*RegistryConfig  `yaml:"registries" json:"registries,omitempty" property:"registries"`
+	References     map[string]*ReferenceConfig `yaml:"references" json:"references,omitempty" property:"references"`
+	ProtocolConf   interface{}                 `yaml:"protocol_conf" json:"protocol_conf,omitempty" property:"protocol_conf"`
+	FilterConf     interface{}                 `yaml:"filter_conf" json:"filter_conf,omitempty" property:"filter_conf" `
+	ShutdownConfig *ShutdownConfig             `yaml:"shutdown_conf" json:"shutdown_conf,omitempty" property:"shutdown_conf" `
 }
 
 func (c *ConsumerConfig) UnmarshalYAML(unmarshal func(interface{}) error) error {

config/graceful_shutdown.go

@@ -0,0 +1,220 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package config
+
+import (
+	"os"
+	"os/signal"
+	"runtime/debug"
+	"syscall"
+	"time"
+)
+
+import (
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/common/logger"
+)
+
+/*
+ * The key point is that find out the signals to handle.
+ * The most important documentation is https://golang.org/pkg/os/signal/
+ * From this documentation, we can know that:
+ * 1. The signals SIGKILL and SIGSTOP may not be caught by signal package;
+ * 2. SIGHUP, SIGINT, or SIGTERM signal causes the program to exit
+ * 3. SIGQUIT, SIGILL, SIGTRAP, SIGABRT, SIGSTKFLT, SIGEMT, or SIGSYS signal causes the program to exit with a stack dump
+ * 4. The invocation of Notify(signal...) will disable the default behavior of those signals.
+ *
+ * So the signals SIGKILL, SIGSTOP, SIGHUP, SIGINT, SIGTERM, SIGQUIT, SIGILL, SIGTRAP, SIGABRT, SIGSTKFLT, SIGEMT, SIGSYS
+ * should be processed.
+ * syscall.SIGEMT cannot be found in CI
+ * It's seems that the Unix/Linux does not have the signal SIGSTKFLT. https://github.com/golang/go/issues/33381
+ * So this signal will be ignored.
+ *
+ */
+
+func GracefulShutdownInit() {
+
+	signals := make(chan os.Signal, 1)
+
+	signal.Notify(signals, os.Interrupt, os.Kill, syscall.SIGKILL, syscall.SIGSTOP,
+		syscall.SIGHUP, syscall.SIGINT, syscall.SIGQUIT, syscall.SIGILL, syscall.SIGTRAP,
+		syscall.SIGABRT, syscall.SIGSYS,
+	)
+
+	go func() {
+		select {
+		case sig := <-signals:
+			logger.Infof("get signal %s, application will shutdown.", sig.String())
+			// gracefulShutdownOnce.Do(func() {
+			BeforeShutdown()
+
+			switch sig {
+			// those signals' original behavior is exit with dump ths stack, so we try to keep the behavior
+			case syscall.SIGQUIT, syscall.SIGILL, syscall.SIGTRAP,
+				syscall.SIGABRT, syscall.SIGSYS:
+				debug.WriteHeapDump(os.Stdout.Fd())
+			default:
+				time.AfterFunc(totalTimeout(), func() {
+					logger.Warn("Shutdown gracefully timeout, application will shutdown immediately. ")
+					os.Exit(0)
+				})
+			}
+			os.Exit(0)
+		}
+	}()
+}
+
+func BeforeShutdown() {
+
+	destroyAllRegistries()
+	// waiting for a short time so that the clients have enough time to get the notification that server shutdowns
+	// The value of configuration depends on how long the clients will get notification.
+	waitAndAcceptNewRequests()
+
+	time.Sleep(1 * time.Minute)
+	// reject the new request, but keeping waiting for accepting requests
+	waitForReceivingRequests()
+
+	// If this application is not the provider, it will do nothing
+	destroyProviderProtocols()
+
+	// reject sending the new request, and waiting for response of sending requests
+	waitForSendingRequests()
+
+	// If this application is not the consumer, it will do nothing
+	destroyConsumerProtocols()
+
+	logger.Info("Graceful shutdown --- Execute the custom callbacks.")
+	customCallbacks := extension.GetAllCustomShutdownCallbacks()
+	for callback := customCallbacks.Front(); callback != nil; callback = callback.Next() {
+		callback.Value.(func())()
+	}
+}
+
+func destroyAllRegistries() {
+	logger.Info("Graceful shutdown --- Destroy all registries. ")
+	registryProtocol := extension.GetProtocol(constant.REGISTRY_KEY)
+	registryProtocol.Destroy()
+}
+
+func destroyConsumerProtocols() {
+	logger.Info("Graceful shutdown --- Destroy consumer's protocols. ")
+	if consumerConfig == nil || consumerConfig.ProtocolConf == nil {
+		return
+	}
+	protocols := consumerConfig.ProtocolConf.(map[interface{}]interface{})
+	for name, _ := range protocols {
+		extension.GetProtocol(name.(string)).Destroy()
+	}
+}
+
+/**
+ * destroy the provider's protocol.
+ * if the protocol is consumer's protocol too, we will keep it.
+ */
+func destroyProviderProtocols() {
+
+	logger.Info("Graceful shutdown --- Destroy provider's protocols. ")
+
+	if providerConfig == nil || providerConfig.ProtocolConf == nil {
+		return
+	}
+
+	consumerProtocol := make(map[interface{}]interface{}, 0)
+	if consumerConfig != nil && consumerConfig.ProtocolConf != nil {
+		consumerProtocol = consumerConfig.ProtocolConf.(map[interface{}]interface{})
+	}
+
+	protocols := providerConfig.ProtocolConf.(map[interface{}]interface{})
+	for name, _ := range protocols {
+		_, found := consumerProtocol[name]
+
+		// the protocol is the consumer's protocol, we can not destroy it.
+		if found {
+			continue
+		}
+		extension.GetProtocol(name.(string)).Destroy()
+	}
+}
+
+func waitAndAcceptNewRequests() {
+
+	logger.Info("Graceful shutdown --- Keep waiting and accept new requests for a short time. ")
+	if providerConfig == nil || providerConfig.ShutdownConfig == nil {
+		return
+	}
+
+	timeout := providerConfig.ShutdownConfig.GetStepTimeout()
+
+	// ignore this step
+	if timeout < 0 {
+		return
+	}
+	time.Sleep(timeout)
+}
+
+// for provider. It will wait for processing receiving requests
+func waitForReceivingRequests() {
+	logger.Info("Graceful shutdown --- Keep waiting until accepting requests finish or timeout. ")
+	if providerConfig == nil || providerConfig.ShutdownConfig == nil {
+		// ignore this step
+		return
+	}
+	waitingProcessedTimeout(providerConfig.ShutdownConfig)
+}
+
+// for consumer. It will wait for the response of sending requests
+func waitForSendingRequests() {
+	logger.Info("Graceful shutdown --- Keep waiting until sending requests getting response or timeout ")
+	if consumerConfig == nil || consumerConfig.ShutdownConfig == nil {
+		// ignore this step
+		return
+	}
+	waitingProcessedTimeout(consumerConfig.ShutdownConfig)
+}
+
+func waitingProcessedTimeout(shutdownConfig *ShutdownConfig) {
+	timeout := shutdownConfig.GetStepTimeout()
+	if timeout <= 0 {
+		return
+	}
+	start := time.Now().UnixNano()
+	for time.Now().UnixNano()-start < timeout.Nanoseconds() && !shutdownConfig.RequestsFinished {
+		// sleep 10 ms and then we check it again
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+func totalTimeout() time.Duration {
+	var providerShutdown time.Duration = 0
+	if providerConfig != nil && providerConfig.ShutdownConfig != nil {
+		providerShutdown = providerConfig.ShutdownConfig.GetTimeout()
+	}
+
+	var consumerShutdown time.Duration = 0
+	if consumerConfig != nil && consumerConfig.ShutdownConfig != nil {
+		consumerShutdown = consumerConfig.ShutdownConfig.GetTimeout()
+	}
+
+	var timeout = providerShutdown
+	if consumerShutdown > providerShutdown {
+		timeout = consumerShutdown
+	}
+	return timeout
+}