fix: panic when type assertion fails at runtime

client/circuit_breaker.go

@@ -21,7 +21,6 @@ import (
 	"github.com/cloudwego/kitex/pkg/circuitbreak"
 	"github.com/cloudwego/kitex/pkg/rpcinfo"
 	"github.com/kitex-contrib/config-file/monitor"
-	"github.com/kitex-contrib/config-file/parser"
 	"github.com/kitex-contrib/config-file/utils"
 )
 
@@ -44,9 +43,12 @@ func initCircuitBreaker(service string, watcher monitor.ConfigMonitor) (*circuit
 
 	onChangeCallback := func() {
 		set := utils.Set{}
-		configs := watcher.Config().(*parser.ClientFileConfig).Circuitbreaker
+		config := getFileConfig(watcher)
+		if config == nil {
+			return // config is nil, do nothing, log will be printed in getFileConfig
+		}
 
-		for method, config := range configs {
+		for method, config := range config.Circuitbreaker {
 			set[method] = true
 			key := genServiceCBKey(service, method)
 			cb.UpdateServiceCBConfig(key, *config)

client/client.go

@@ -0,0 +1,21 @@
+package client
+
+import (
+	"github.com/cloudwego/kitex/pkg/klog"
+	"github.com/kitex-contrib/config-file/monitor"
+	"github.com/kitex-contrib/config-file/parser"
+)
+
+// getFileConfig returns the config from the watcher.
+// if the config type is not *parser.ClientFileConfig, it will log an error and return nil.
+func getFileConfig(watcher monitor.ConfigMonitor) *parser.ClientFileConfig {
+	config, ok := watcher.Config().(*parser.ClientFileConfig)
+	if !ok {
+		// This should never happen.
+		// But if it does, we should log it and do nothing.
+		// Otherwise, the program will panic.
+		klog.Errorf("[local] Invalid config type: %T", watcher.Config())
+		return nil
+	}
+	return config
+}

client/retry.go

@@ -19,7 +19,6 @@ import (
 	"github.com/cloudwego/kitex/pkg/klog"
 	"github.com/cloudwego/kitex/pkg/retry"
 	"github.com/kitex-contrib/config-file/monitor"
-	"github.com/kitex-contrib/config-file/parser"
 	"github.com/kitex-contrib/config-file/utils"
 )
 
@@ -43,7 +42,11 @@ func initRetryContainer(watcher monitor.ConfigMonitor) (*retry.Container, int64)
 
 	onChangeCallback := func() {
 		// the key is method name, wildcard "*" can match anything.
-		rcs := watcher.Config().(*parser.ClientFileConfig).Retry
+		config := getFileConfig(watcher)
+		if config == nil {
+			return // config is nil, do nothing, log will be printed in getFileConfig
+		}
+		rcs := config.Retry
 		set := utils.Set{}
 
 		for method, policy := range rcs {

client/rpc_timeout.go

@@ -19,7 +19,6 @@ import (
 	"github.com/cloudwego/kitex/pkg/rpcinfo"
 	"github.com/cloudwego/kitex/pkg/rpctimeout"
 	"github.com/kitex-contrib/config-file/monitor"
-	"github.com/kitex-contrib/config-file/parser"
 )
 
 // WithRPCTimeout returns a server.Option that sets the timeout provider for the client.
@@ -40,8 +39,11 @@ func initRPCTimeout(watcher monitor.ConfigMonitor) (rpcinfo.TimeoutProvider, int
 
 	onChangeCallback := func() {
 		// the key is method name, wildcard "*" can match anything.
-		configs := watcher.Config().(*parser.ClientFileConfig).Timeout
-		rpcTimeoutContainer.NotifyPolicyChange(configs)
+		config := getFileConfig(watcher)
+		if config == nil {
+			return // config is nil, do nothing, log will be printed in getFileConfig
+		}
+		rpcTimeoutContainer.NotifyPolicyChange(config.Timeout)
 	}
 
 	keyRPCTimeout := watcher.RegisterCallback(onChangeCallback)

example/go.mod

@@ -10,6 +10,8 @@ require (
 
 replace github.com/kitex-contrib/config-file => ../.
 
+replace github.com/apache/thrift => github.com/apache/thrift v0.13.0
+
 require (
 	github.com/apache/thrift v0.16.0 // indirect
 	github.com/bytedance/gopkg v0.0.0-20230728082804-614d0af6619b // indirect
@@ -49,6 +51,7 @@ require (
 	golang.org/x/sys v0.13.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
 	google.golang.org/genproto v0.0.0-20231012201019-e917dd12ba7a // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b // indirect
 	google.golang.org/protobuf v1.31.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 	sigs.k8s.io/yaml v1.4.0 // indirect

filewatcher/filewatcher.go

@@ -92,6 +92,8 @@ func (fw *fileWatcher) RegisterCallback(callback func(data []byte)) int64 {
 		fw.callbacks = make(map[int64]func(data []byte), 0)
 	}
 
+	klog.Debugf("[local] filewatcher to %v registered callback\n", fw.filePath)
+
 	uniqueID := fw.counter.Add(1)
 	fw.callbacks[uniqueID] = callback
 	return uniqueID
@@ -122,7 +124,7 @@ func (fw *fileWatcher) StartWatching() error {
 	go func() {
 		defer func() {
 			if r := recover(); r != nil {
-				klog.Errorf("file watcher panic: %v\n", r)
+				klog.Errorf("[local] file watcher panic: %v\n", r)
 			}
 		}()
 		fw.start()
@@ -176,8 +178,12 @@ func (fw *fileWatcher) CallOnceAll() error {
 		return err
 	}
 
-	for _, v := range fw.callbacks {
-		v(data)
+	for key, callback := range fw.callbacks {
+		if callback == nil {
+			fw.DeregisterCallback(key) // When encountering Nil's callback function, directly cancel it here.
+			continue
+		}
+		callback(data)
 	}
 	return nil
 }
@@ -192,7 +198,7 @@ func (fw *fileWatcher) CallOnceSpecific(uniqueID int64) error {
 	if callback, ok := fw.callbacks[uniqueID]; ok {
 		callback(data)
 	} else {
-		return errors.New("not found callback for uniqueID: " + strconv.FormatInt(uniqueID, 10))
+		return errors.New("not found callback for id: " + strconv.FormatInt(uniqueID, 10))
 	}
 	return nil
 }

monitor/monitor.go

@@ -114,6 +114,8 @@ func (c *configMonitor) RegisterCallback(callback func()) int64 {
 
 	key := c.counter.Add(1)
 	c.callbacks[key] = callback
+
+	klog.Debugf("[local] config monitor registered callback, id: %v\n", key)
 	return key
 }
 
@@ -145,7 +147,11 @@ func (c *configMonitor) parseHandler(data []byte) {
 	}
 
 	if len(c.callbacks) > 0 {
-		for _, callback := range c.callbacks {
+		for key, callback := range c.callbacks {
+			if callback == nil {
+				c.DeregisterCallback(key) // When encountering Nil's callback function, directly cancel it here.
+				continue
+			}
 			callback()
 		}
 	}

server/limit.go

@@ -21,7 +21,6 @@ import (
 	"github.com/cloudwego/kitex/pkg/limit"
 	kitexserver "github.com/cloudwego/kitex/server"
 	"github.com/kitex-contrib/config-file/monitor"
-	"github.com/kitex-contrib/config-file/parser"
 )
 
 // WithLimiter returns a server.Option that sets the limiter for the server.
@@ -43,7 +42,11 @@ func initLimitOptions(watcher monitor.ConfigMonitor) (*limit.Option, int64) {
 	}
 
 	onChangeCallback := func() {
-		lc := watcher.Config().(*parser.ServerFileConfig).Limit
+		config := getFileConfig(watcher)
+		if config == nil {
+			return // config is nil, do nothing, log will be printed in getFileConfig
+		}
+		lc := config.Limit
 
 		opt.MaxConnections = int(lc.ConnectionLimit)
 		opt.MaxQPS = int(lc.QPSLimit)

server/server.go

@@ -0,0 +1,21 @@
+package server
+
+import (
+	"github.com/cloudwego/kitex/pkg/klog"
+	"github.com/kitex-contrib/config-file/monitor"
+	"github.com/kitex-contrib/config-file/parser"
+)
+
+// getFileConfig returns the config from the watcher.
+// if the config type is not *parser.ServerFileConfig, it will log an error and return nil.
+func getFileConfig(watcher monitor.ConfigMonitor) *parser.ServerFileConfig {
+	config, ok := watcher.Config().(*parser.ServerFileConfig)
+	if !ok {
+		// This should never happen.
+		// But if it does, we should log it and do nothing.
+		// Otherwise, the program will panic.
+		klog.Errorf("[local] Invalid config type: %T", watcher.Config())
+		return nil
+	}
+	return config
+}