handle rate limit config update event

cmd/pixiu/pixiu.go

@@ -30,15 +30,6 @@ import (
 	"github.com/urfave/cli"
 )
 
-import (
-	_ "github.com/apache/dubbo-go-pixiu/pkg/filter/accesslog"
-	_ "github.com/apache/dubbo-go-pixiu/pkg/filter/logger"
-	_ "github.com/apache/dubbo-go-pixiu/pkg/filter/recovery"
-	_ "github.com/apache/dubbo-go-pixiu/pkg/filter/remote"
-	_ "github.com/apache/dubbo-go-pixiu/pkg/filter/response"
-	_ "github.com/apache/dubbo-go-pixiu/pkg/filter/timeout"
-)
-
 // Version pixiu version
 var Version = "0.3.0"
 

pkg/config/api_config.go

@@ -27,6 +27,7 @@ import (
 
 import (
 	fc "github.com/dubbogo/dubbo-go-pixiu-filter/pkg/api/config"
+	fr "github.com/dubbogo/dubbo-go-pixiu-filter/pkg/api/config/ratelimit"
 	etcdv3 "github.com/dubbogo/gost/database/kv/etcd/v3"
 	perrors "github.com/pkg/errors"
 	"go.etcd.io/etcd/clientv3"
@@ -47,6 +48,12 @@ var (
 	lock      sync.RWMutex
 )
 
+var (
+	BASE_INFO_NAME = "name"
+	BASE_INFO_DESC = "description"
+	BASE_INFO_PFP  = "pluginFilePath"
+)
+
 // APIConfigResourceListener defines api resource and method config listener interface
 type APIConfigResourceListener interface {
 	// ResourceChange handle modify resource event
@@ -61,6 +68,12 @@ type APIConfigResourceListener interface {
 	MethodAdd(res fc.Resource, method fc.Method) bool
 	// MethodDelete handle delete method event
 	MethodDelete(res fc.Resource, method fc.Method) bool
+
+	PluginPathChange(filePath string)
+
+	PluginGroupChange(group []fc.PluginsGroup)
+
+	RateLimitChange(*fr.Config)
 }
 
 // LoadAPIConfigFromFile load the api config from file
@@ -105,11 +118,21 @@ func LoadAPIConfig(metaConfig *model.APIMetaConfig) (*fc.APIConfig, error) {
 
 func initAPIConfigFromKVList(kList, vList []string) error {
 	var skList, svList, mkList, mvList []string
+	var baseInfo string
+	var pluginGroup []string
+	var rateLimit string
 
 	for i, k := range kList {
 		v := vList[i]
+		//handle base info
+		re := getCheckBaseInfoRegexp()
+		if m := re.Match([]byte(k)); m {
+			baseInfo = v
+			continue
+		}
+
 		// handle resource
-		re := getCheckResourceRegexp()
+		re = getCheckResourceRegexp()
 		if m := re.Match([]byte(k)); m {
 			skList = append(skList, k)
 			svList = append(svList, v)
@@ -122,12 +145,30 @@ func initAPIConfigFromKVList(kList, vList []string) error {
 			mvList = append(mvList, v)
 			continue
 		}
+
+		//handle plugin group
+		re = getCheckPluginsGroupRegexp()
+		if m := re.Match([]byte(k)); m {
+			pluginGroup = append(pluginGroup, v)
+			continue
+		}
+
+		//handle rate limit config
+		re = getCheckRatelimitRegexp()
+		if m := re.Match([]byte(k)); m {
+			rateLimit = v
+			continue
+		}
 	}
 
 	lock.Lock()
 	defer lock.Unlock()
 
 	tmpApiConf := &fc.APIConfig{}
+	if err := initBaseInfoFromString(tmpApiConf, baseInfo); err != nil {
+		logger.Error("initBaseInfoFromString error %v", err.Error())
+		return err
+	}
 	if err := initAPIConfigServiceFromKvList(tmpApiConf, skList, svList); err != nil {
 		logger.Error("initAPIConfigServiceFromKvList error %v", err.Error())
 		return err
@@ -136,11 +177,61 @@ func initAPIConfigFromKVList(kList, vList []string) error {
 		logger.Error("initAPIConfigMethodFromKvList error %v", err.Error())
 		return err
 	}
+	if err := initAPIConfigPluginsFromStringList(tmpApiConf, pluginGroup); err != nil {
+		logger.Error("initAPIConfigPluginsFromStringList error %v", err.Error())
+		return err
+	}
+	if err := initAPIConfigRatelimitFromString(tmpApiConf, rateLimit); err != nil {
+		logger.Error("initAPIConfigRatelimitFromString error %v", err.Error())
+		return err
+	}
 
 	apiConfig = tmpApiConf
 	return nil
 }
 
+func initBaseInfoFromString(conf *fc.APIConfig, str string) error {
+	properties := make(map[string]string, 8)
+	if err := yaml.UnmarshalYML([]byte(str), properties); err != nil {
+		logger.Errorf("unmarshalYmlConfig error %v", err.Error())
+		return err
+	}
+	if v, ok := properties[BASE_INFO_NAME]; ok {
+		conf.Name = v
+	}
+	if v, ok := properties[BASE_INFO_DESC]; ok {
+		conf.Description = v
+	}
+	if v, ok := properties[BASE_INFO_PFP]; ok {
+		conf.PluginFilePath = v
+	}
+	return nil
+}
+
+func initAPIConfigRatelimitFromString(conf *fc.APIConfig, str string) error {
+	c := fr.Config{}
+	if err := yaml.UnmarshalYML([]byte(str), &c); err != nil {
+		logger.Errorf("unmarshalYmlConfig error %v", err.Error())
+		return err
+	}
+	conf.RateLimit = c
+	return nil
+}
+
+func initAPIConfigPluginsFromStringList(conf *fc.APIConfig, plugins []string) error {
+	var groups []fc.PluginsGroup
+	for _, v := range plugins {
+		g := fc.PluginsGroup{}
+		if err := yaml.UnmarshalYML([]byte(v), &g); err != nil {
+			logger.Errorf("unmarshalYmlConfig error %v", err.Error())
+			return err
+		}
+		groups = append(groups, g)
+	}
+	conf.PluginsGroup = groups
+	return nil
+}
+
 func initAPIConfigMethodFromKvList(config *fc.APIConfig, kList, vList []string) error {
 	for i, _ := range kList {
 		v := vList[i]
@@ -304,6 +395,12 @@ func handleDeleteEvent(key, val []byte) {
 		}
 		deleteApiConfigMethod(resourceId, methodId)
 	}
+
+	re = getCheckRatelimitRegexp()
+	if m := re.Match(key); m {
+		empty := &fr.Config{}
+		listener.RateLimitChange(empty)
+	}
 }
 
 func handlePutEvent(key, val []byte) {
@@ -332,6 +429,27 @@ func handlePutEvent(key, val []byte) {
 		}
 		mergeApiConfigMethod(res.ResourcePath, *res)
 	}
+
+	//handle base info
+	re = getCheckBaseInfoRegexp()
+	if m := re.Match(key); m {
+		mergeBaseInfo(val)
+		return
+	}
+
+	//handle plugins group
+	re = getCheckPluginsGroupRegexp()
+	if m := re.Match(key); m {
+		mergePluginGroup(val)
+		return
+	}
+
+	//handle ratelimit
+	re = getCheckRatelimitRegexp()
+	if m := re.Match(key); m {
+		mergeRatelimit(val)
+		return
+	}
 }
 
 func deleteApiConfigResource(resourceId int) {
@@ -361,6 +479,36 @@ func mergeApiConfigResource(val fc.Resource) {
 	listener.ResourceAdd(val)
 }
 
+func mergeRatelimit(val []byte) {
+	c := &fr.Config{}
+	if err := yaml.UnmarshalYML(val, c); err != nil {
+		logger.Errorf("unmarshalYmlConfig error %v", err.Error())
+		return
+	}
+	apiConfig.RateLimit = *c
+	listener.RateLimitChange(c)
+}
+
+func mergePluginGroup(val []byte) {
+	g := &fc.PluginsGroup{}
+	if err := yaml.UnmarshalYML(val, g); err != nil {
+		logger.Errorf("unmarshalYmlConfig error %v", err.Error())
+		return
+	}
+	for i, v := range apiConfig.PluginsGroup {
+		if v.GroupName == g.GroupName {
+			apiConfig.PluginsGroup[i] = *g
+		}
+	}
+	listener.PluginGroupChange(apiConfig.PluginsGroup)
+}
+
+func mergeBaseInfo(val []byte) {
+	_ = initBaseInfoFromString(apiConfig, string(val))
+
+	listener.PluginPathChange(apiConfig.PluginFilePath)
+}
+
 func deleteApiConfigMethod(resourceId, methodId int) {
 	for _, resource := range apiConfig.Resources {
 		if resource.ID != resourceId {
@@ -402,7 +550,7 @@ func mergeApiConfigMethod(path string, val fc.Method) {
 }
 
 func getCheckBaseInfoRegexp() *regexp.Regexp {
-	return regexp.MustCompile(".+/base$")
+	return regexp.MustCompile(".+/Base$")
 }
 
 func getCheckResourceRegexp() *regexp.Regexp {
@@ -413,6 +561,14 @@ func getExtractMethodRegexp() *regexp.Regexp {
 	return regexp.MustCompile("Resources/([^/]+)/Method/[^/]+/?$")
 }
 
+func getCheckPluginsGroupRegexp() *regexp.Regexp {
+	return regexp.MustCompile(".+/PluginGroup/[^/]+/?$")
+}
+
+func getCheckRatelimitRegexp() *regexp.Regexp {
+	return regexp.MustCompile(".+/Ratelimit/[^/]+/?$")
+}
+
 // RegisterConfigListener register APIConfigListener
 func RegisterConfigListener(li APIConfigResourceListener) {
 	listener = li

pkg/filter/accesslog/access_log.go

@@ -40,7 +40,7 @@ import (
 
 var accessLogWriter = &model.AccessLogWriter{AccessLogDataChan: make(chan model.AccessLogData, constant.LogDataBuffer)}
 
-func init() {
+func Init() {
 	extension.SetFilterFunc(constant.AccessLogFilter, accessLog())
 	accessLogWriter.Write()
 }

pkg/filter/plugins/plugins.go

@@ -38,6 +38,7 @@ var (
 	// url path -> filter chain
 	filterChainCache = make(map[string]FilterChain)
 	groupCache       = make(map[string]map[string]WithFunc)
+	localFilePath    = ""
 
 	errEmptyConfig = errors.New("Empty plugin config")
 )
@@ -60,24 +61,33 @@ type WithFunc struct {
 	fn       context.FilterFunc
 }
 
+func OnFilePathChange(filePath string) {
+	if len(filePath) == 0 {
+		return
+	}
+	localFilePath = filePath
+}
+
 // OnResourceUpdate update plugins cache map when api-resource update
 func OnResourceUpdate(resource *config.Resource) {
 	InitFilterChainForResource(resource, "", nil)
 }
 
 // OnGroupUpdate update group cache
-func OnGroupUpdate(groups []config.PluginsGroup, filePath string) {
-	InitPluginsGroup(groups, filePath)
+func OnGroupUpdate(groups []config.PluginsGroup) {
+	InitPluginsGroup(groups, "")
 }
 
 // InitPluginsGroup prase api_config.yaml(pluginsGroup) to map[string][]PluginsWithFunc
 func InitPluginsGroup(groups []config.PluginsGroup, filePath string) {
-	if "" == filePath || len(groups) == 0 {
+	OnFilePathChange(filePath)
+
+	if "" == localFilePath || len(groups) == 0 {
 		return
 	}
 
 	// load file.so
-	pls, err := plugin.Open(filePath)
+	pls, err := plugin.Open(localFilePath)
 	if nil != err {
 		panic(err)
 	}

pkg/initialize/init.go

@@ -18,6 +18,7 @@
 package initialize
 
 import (
+	"github.com/apache/dubbo-go-pixiu/pkg/filter/accesslog"
 	"github.com/apache/dubbo-go-pixiu/pkg/filter/api"
 	"github.com/apache/dubbo-go-pixiu/pkg/filter/authority"
 	"github.com/apache/dubbo-go-pixiu/pkg/filter/logger"
@@ -41,6 +42,7 @@ func Run(config config.APIConfig) {
 }
 
 func filterInit(config *config.APIConfig) {
+	accesslog.Init()
 	api.Init()
 	authority.Init()
 	logger.Init()

pkg/service/api/discovery_service.go

@@ -27,12 +27,15 @@ import (
 	"github.com/apache/dubbo-go-pixiu/pkg/common/constant"
 	"github.com/apache/dubbo-go-pixiu/pkg/common/extension"
 	pc "github.com/apache/dubbo-go-pixiu/pkg/config"
+	"github.com/apache/dubbo-go-pixiu/pkg/filter/plugins"
+	"github.com/apache/dubbo-go-pixiu/pkg/filter/ratelimit"
 	"github.com/apache/dubbo-go-pixiu/pkg/router"
 	"github.com/apache/dubbo-go-pixiu/pkg/service"
 )
 
 import (
 	"github.com/dubbogo/dubbo-go-pixiu-filter/pkg/api/config"
+	ratelimitConf "github.com/dubbogo/dubbo-go-pixiu-filter/pkg/api/config/ratelimit"
 	fr "github.com/dubbogo/dubbo-go-pixiu-filter/pkg/router"
 )
 
@@ -269,3 +272,15 @@ func loadAPIFromMethods(fullPath string, methods []config.Method, headers map[st
 	}
 	return nil
 }
+
+func (ads *LocalMemoryAPIDiscoveryService) PluginPathChange(filePath string) {
+	plugins.OnFilePathChange(filePath)
+}
+
+func (ads *LocalMemoryAPIDiscoveryService) PluginGroupChange(group []config.PluginsGroup) {
+	plugins.OnGroupUpdate(group)
+}
+
+func (ads *LocalMemoryAPIDiscoveryService) RateLimitChange(c *ratelimitConf.Config) {
+	ratelimit.OnUpdate(c)
+}