merge dev

agent.go

@@ -44,6 +44,7 @@ const (
 )
 
 var (
+	startTime                 = time.Now()
 	initParamLock             sync.Mutex
 	setAgentLock              sync.Mutex
 	notFoundProviderCount     int64 = 0
@@ -134,6 +135,14 @@ func (a *Agent) OnAfterStart(f func(a *Agent)) {
 	a.onAfterStart = append(a.onAfterStart, f)
 }
 
+func (a *Agent) GetRuntimeInfo() map[string]interface{} {
+	info := map[string]interface{}{}
+	info[motan.RuntimeStartTimeKey] = startTime
+	info[motan.RuntimeCpuPercentKey] = GetCpuPercent()
+	info[motan.RuntimeRssMemoryKey] = GetRssMemory()
+	return info
+}
+
 func (a *Agent) RegisterCommandHandler(f CommandHandler) {
 	a.commandHandlers = append(a.commandHandlers, f)
 }

agent_test.go

@@ -6,6 +6,7 @@ import (
 	"flag"
 	"fmt"
 	_ "fmt"
+	"github.com/weibocom/motan-go/cluster"
 	"github.com/weibocom/motan-go/config"
 	"github.com/weibocom/motan-go/endpoint"
 	vlog "github.com/weibocom/motan-go/log"
@@ -16,6 +17,7 @@ import (
 	_ "golang.org/x/net/context"
 	"io/ioutil"
 	"math/rand"
+	"mime/multipart"
 	"net"
 	"net/http"
 	"net/url"
@@ -1064,18 +1066,169 @@ func TestRuntimeHandler(t *testing.T) {
 	err = json.Unmarshal(bodyBytes, &runtimeInfo)
 	assert.Nil(t, err)
 
-	for _, s := range []string{
-		core.RuntimeInstanceTypeKey,
+	for _, s := range defaultKeys {
+		info, ok := runtimeInfo[s]
+		assert.True(t, ok)
+		assert.NotNil(t, info)
+		t.Logf("key: %s", s)
+	}
+
+	// test param keys
+	keys := []string{
 		core.RuntimeExportersKey,
-		core.RuntimeClustersKey,
-		core.RuntimeHttpClustersKey,
-		core.RuntimeExtensionFactoryKey,
 		core.RuntimeServersKey,
-		core.RuntimeBasicKey,
-	} {
+		core.RuntimeHttpClustersKey,
+	}
+	resp, err = http.Get("http://127.0.0.1:8002/runtime/info" + "?keys=" + strings.Join(keys, ","))
+	assert.Nil(t, err)
+	bodyBytes, err = ioutil.ReadAll(resp.Body)
+	assert.Nil(t, err)
+	err = resp.Body.Close()
+	assert.Nil(t, err)
+	err = json.Unmarshal(bodyBytes, &runtimeInfo)
+	assert.Nil(t, err)
+	for _, s := range keys {
 		info, ok := runtimeInfo[s]
 		assert.True(t, ok)
 		assert.NotNil(t, info)
 		t.Logf("key: %s", s)
 	}
 }
+
+func TestClusterRefersFilterHandler(t *testing.T) {
+	completeConfigBody := `[{"service":"com.weibo.api.test1","mode":"exclude","rule":"127.93.0.0/16"},{"service":"com.weibo.api.test2","mode":"exclude","rule":"127.93.0.0/16"}]`
+	var completeConfig cluster.RefersFilterConfigList
+	_ = json.Unmarshal([]byte(completeConfigBody), &completeConfig)
+	completeConfigExpect, _ := json.Marshal(completeConfig)
+	rewriteConfigBody := `[{"mode":"exclude","rule":"127.93.0.0/16"},{"service":"com.weibo.api.test2","mode":"exclude","rule":"127.93.0.0/16"}]`
+	var rewriteConfig cluster.RefersFilterConfigList
+	_ = json.Unmarshal([]byte(rewriteConfigBody), &rewriteConfig)
+	rewriteConfigExpect, _ := json.Marshal(rewriteConfig)
+	emptyConfigExpect := `[]`
+	cases := []struct {
+		desc       string
+		request    *http.Request
+		assertFunc func(t *testing.T, resp *http.Response, respErr error)
+	}{
+		{
+			desc: "verify initial value",
+			request: func() *http.Request {
+				req, _ := http.NewRequest("GET", "http://127.0.0.1:8002/refers/filter/get", nil)
+				return req
+			}(),
+			assertFunc: func(t *testing.T, resp *http.Response, respErr error) {
+				assert.Nil(t, respErr)
+				assert.Equal(t, http.StatusOK, resp.StatusCode)
+				bodyBytes, err := ioutil.ReadAll(resp.Body)
+				assert.Nil(t, err)
+				assert.Equal(t, fmt.Sprintf(`{"result":"ok","data":%s}`, emptyConfigExpect), string(bodyBytes))
+			},
+		},
+		{
+			desc: "set empty config",
+			request: func() *http.Request {
+				payload := &bytes.Buffer{}
+				writer := multipart.NewWriter(payload)
+				_ = writer.WriteField("config", "[]")
+				_ = writer.Close()
+				req, _ := http.NewRequest("POST", "http://127.0.0.1:8002/refers/filter/set", payload)
+				req.Header.Set("Content-Type", writer.FormDataContentType())
+				return req
+			}(),
+			assertFunc: func(t *testing.T, resp *http.Response, respErr error) {
+				assert.Nil(t, respErr)
+				assert.Equal(t, http.StatusOK, resp.StatusCode)
+				bodyBytes, err := ioutil.ReadAll(resp.Body)
+				assert.Nil(t, err)
+				assert.Equal(t, `{"result":"ok","data":"ok"}`, string(bodyBytes))
+			},
+		},
+		{
+			desc: "get empty config",
+			request: func() *http.Request {
+				req, _ := http.NewRequest("GET", "http://127.0.0.1:8002/refers/filter/get", nil)
+				return req
+			}(),
+			assertFunc: func(t *testing.T, resp *http.Response, respErr error) {
+				assert.Nil(t, respErr)
+				assert.Equal(t, http.StatusOK, resp.StatusCode)
+				bodyBytes, err := ioutil.ReadAll(resp.Body)
+				assert.Nil(t, err)
+				assert.Equal(t, fmt.Sprintf(`{"result":"ok","data":%s}`, emptyConfigExpect), string(bodyBytes))
+			},
+		},
+		{
+			desc: "set complete config",
+			request: func() *http.Request {
+				payload := &bytes.Buffer{}
+				writer := multipart.NewWriter(payload)
+				_ = writer.WriteField("config", completeConfigBody)
+				_ = writer.Close()
+				req, _ := http.NewRequest("POST", "http://127.0.0.1:8002/refers/filter/set", payload)
+				req.Header.Set("Content-Type", writer.FormDataContentType())
+				return req
+			}(),
+			assertFunc: func(t *testing.T, resp *http.Response, respErr error) {
+				assert.Nil(t, respErr)
+				assert.Equal(t, http.StatusOK, resp.StatusCode)
+				bodyBytes, err := ioutil.ReadAll(resp.Body)
+				assert.Nil(t, err)
+				assert.Equal(t, `{"result":"ok","data":"ok"}`, string(bodyBytes))
+			},
+		},
+		{
+			desc: "get complete config",
+			request: func() *http.Request {
+				req, _ := http.NewRequest("POST", "http://127.0.0.1:8002/refers/filter/get", nil)
+				return req
+			}(),
+			assertFunc: func(t *testing.T, resp *http.Response, respErr error) {
+				assert.Nil(t, respErr)
+				assert.Equal(t, http.StatusOK, resp.StatusCode)
+				bodyBytes, err := ioutil.ReadAll(resp.Body)
+				assert.Nil(t, err)
+				assert.Equal(t, fmt.Sprintf(`{"result":"ok","data":%s}`, completeConfigExpect), string(bodyBytes))
+			},
+		},
+		{
+			desc: "rewrite config",
+			request: func() *http.Request {
+				payload := &bytes.Buffer{}
+				writer := multipart.NewWriter(payload)
+				_ = writer.WriteField("config", string(rewriteConfigBody))
+				_ = writer.Close()
+				req, _ := http.NewRequest("POST", "http://127.0.0.1:8002/refers/filter/set", payload)
+				req.Header.Set("Content-Type", writer.FormDataContentType())
+				return req
+			}(),
+			assertFunc: func(t *testing.T, resp *http.Response, respErr error) {
+				assert.Nil(t, respErr)
+				assert.Equal(t, http.StatusOK, resp.StatusCode)
+				bodyBytes, err := ioutil.ReadAll(resp.Body)
+				assert.Nil(t, err)
+				assert.Equal(t, `{"result":"ok","data":"ok"}`, string(bodyBytes))
+			},
+		},
+		{
+			desc: "get rewrite config",
+			request: func() *http.Request {
+				req, _ := http.NewRequest("POST", "http://127.0.0.1:8002/refers/filter/get", nil)
+				return req
+			}(),
+			assertFunc: func(t *testing.T, resp *http.Response, respErr error) {
+				assert.Nil(t, respErr)
+				assert.Equal(t, http.StatusOK, resp.StatusCode)
+				bodyBytes, err := ioutil.ReadAll(resp.Body)
+				assert.Nil(t, err)
+				assert.Equal(t, fmt.Sprintf(`{"result":"ok","data":%s}`, rewriteConfigExpect), string(bodyBytes))
+			},
+		},
+	}
+
+	for _, c := range cases {
+		t.Logf("case %s start", c.desc)
+		resp, err := http.DefaultClient.Do(c.request)
+		c.assertFunc(t, resp, err)
+		t.Logf("case %s finish", c.desc)
+	}
+}

cluster/RefersFilter.go

@@ -0,0 +1,161 @@
+package cluster
+
+import (
+	"fmt"
+	motan "github.com/weibocom/motan-go/core"
+	vlog "github.com/weibocom/motan-go/log"
+	"net"
+	"strings"
+)
+
+const (
+	FilterModeInclude = "include"
+	FilterModeExclude = "exclude"
+)
+
+type RefersFilterConfigList []RefersFilterConfig
+
+type RefersFilterConfig struct {
+	Group   string `json:"group"`
+	Service string `json:"service"`
+	Mode    string `json:"mode"`
+	Rule    string `json:"rule"`
+}
+
+func (list RefersFilterConfigList) Verify() error {
+	for _, filter := range list {
+		if filter.Rule == "" {
+			return fmt.Errorf("exist empty rule in filter config")
+		}
+		if filter.Mode != FilterModeInclude && filter.Mode != FilterModeExclude {
+			return fmt.Errorf("invalid mode(%s) for rule(%s)", filter.Mode, filter.Rule)
+		}
+	}
+	return nil
+}
+
+func (list RefersFilterConfigList) ParseRefersFilters(clusterURL *motan.URL) RefersFilter {
+	var rules []RefersFilterConfig
+	for _, filter := range list {
+		if filter.Group != "" && filter.Group != clusterURL.Group {
+			continue
+		}
+		if filter.Service != "" && filter.Service != clusterURL.Path {
+			continue
+		}
+		rules = append(rules, filter)
+		vlog.Infof("add refer filter for group: %s, service: %s, rule: %s, mode: %s, filter group: %s, filter service: %s", clusterURL.Group, clusterURL.Path, filter.Rule, filter.Mode, filter.Group, filter.Service)
+	}
+	if len(rules) == 0 {
+		return nil
+	}
+	return NewDefaultRefersFilter(rules)
+}
+
+type RefersFilter interface {
+	Filter([]motan.EndPoint) []motan.EndPoint
+}
+
+type filterRule struct {
+	mode     string
+	prefixes []string
+	subnets  []*net.IPNet
+}
+
+func (fr filterRule) IsMatch(refer motan.EndPoint) bool {
+	for _, prefix := range fr.prefixes {
+		if strings.HasPrefix(refer.GetURL().Host, prefix) {
+			if fr.mode == FilterModeExclude {
+				vlog.Infof("filter refer: %s, prefix rule: %s, mode: %s", refer.GetURL().GetIdentity(), prefix, fr.mode)
+			}
+			return true
+		}
+	}
+	if len(fr.subnets) == 0 {
+		return false
+	}
+	ip := net.ParseIP(refer.GetURL().Host)
+	if ip == nil {
+		vlog.Errorf("invalid refer ip: %s", refer.GetURL().Host)
+		return false
+	}
+	for _, ipNet := range fr.subnets {
+		if ipNet.Contains(ip) {
+			if fr.mode == FilterModeExclude {
+				vlog.Infof("filter refer: %s, subnet rule: %s, mode: %s", refer.GetURL().GetIdentity(), ipNet.String(), fr.mode)
+			}
+			return true
+		}
+	}
+	return false
+}
+
+type DefaultRefersFilter struct {
+	excludeRules []filterRule
+	includeRules []filterRule
+}
+
+func NewDefaultRefersFilter(filterConfig []RefersFilterConfig) *DefaultRefersFilter {
+	var includeRules []filterRule
+	var excludeRules []filterRule
+	for _, config := range filterConfig {
+		rules := motan.TrimSplit(config.Rule, ",")
+		fr := filterRule{
+			mode:     config.Mode,
+			prefixes: []string{},
+			subnets:  []*net.IPNet{},
+		}
+		for _, item := range rules {
+			if strings.Contains(item, "/") {
+				_, subnet, err := net.ParseCIDR(item)
+				if err != nil {
+					vlog.Errorf("invalid subnet rule: %s", item)
+					continue
+				}
+				fr.subnets = append(fr.subnets, subnet)
+			} else {
+				fr.prefixes = append(fr.prefixes, item)
+			}
+		}
+		if config.Mode == FilterModeExclude {
+			excludeRules = append(excludeRules, fr)
+		} else {
+			includeRules = append(includeRules, fr)
+		}
+	}
+	return &DefaultRefersFilter{includeRules: includeRules, excludeRules: excludeRules}
+}
+
+func (f *DefaultRefersFilter) Filter(refers []motan.EndPoint) []motan.EndPoint {
+	var newRefers []motan.EndPoint
+	for _, refer := range refers {
+		// discard refer if hit an exclude rule
+		excludeRefer := false
+		for _, excludeRule := range f.excludeRules {
+			excludeRefer = excludeRule.IsMatch(refer)
+			if excludeRefer {
+				break
+			}
+		}
+		if excludeRefer {
+			continue
+		}
+		// retained refer if hit an include rule
+		var includeRefer bool
+		if len(f.includeRules) == 0 {
+			includeRefer = true
+		}
+		for _, includeRule := range f.includeRules {
+			includeRefer = includeRule.IsMatch(refer)
+			if includeRefer {
+				break
+			}
+		}
+		if includeRefer {
+			newRefers = append(newRefers, refer)
+		} else {
+			vlog.Infof("no include rule hit. filter refer: %s", refer.GetURL().GetIdentity())
+		}
+	}
+	return newRefers
+}