add admin handler environment feature and add relevant filters feature (

#373)

* add filter env

agent.go

@@ -13,6 +13,7 @@ import (
 	"path/filepath"
 	"runtime"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -75,6 +76,7 @@ type Agent struct {
 	httpProxyServer   *mserver.HTTPProxyServer
 
 	manageHandlers map[string]http.Handler
+	envHandlers    map[string]map[string]http.Handler
 
 	svcLock      sync.Mutex
 	clsLock      sync.Mutex
@@ -109,6 +111,7 @@ func NewAgent(extfactory motan.ExtensionFactory) *Agent {
 	agent.agentPortServer = make(map[int]motan.Server)
 	agent.serviceRegistries = motan.NewCopyOnWriteMap()
 	agent.manageHandlers = make(map[string]http.Handler)
+	agent.envHandlers = make(map[string]map[string]http.Handler)
 	agent.serviceMap = motan.NewCopyOnWriteMap()
 	return agent
 }
@@ -777,9 +780,13 @@ func fillDefaultReqInfo(r motan.Request, url *motan.URL) {
 		}
 	} else {
 		if r.GetAttachment(mpro.MSource) == "" {
-			application := url.GetParam(motan.ApplicationKey, "")
-			if application != "" {
-				r.SetAttachment(mpro.MSource, application)
+			if app := r.GetAttachment(motan.ApplicationKey); app != "" {
+				r.SetAttachment(mpro.MSource, app)
+			} else {
+				application := url.GetParam(motan.ApplicationKey, "")
+				if application != "" {
+					r.SetAttachment(mpro.MSource, application)
+				}
 			}
 		}
 		if r.GetAttachment(mpro.MGroup) == "" {
@@ -1090,6 +1097,12 @@ func (a *Agent) RegisterManageHandler(path string, handler http.Handler) {
 	}
 }
 
+func (a *Agent) RegisterEnvHandlers(envStr string, handlers map[string]http.Handler) {
+	if envStr != "" && handlers != nil {
+		a.envHandlers[envStr] = handlers // override
+	}
+}
+
 func (a *Agent) startMServer() {
 	handlers := make(map[string]http.Handler, 16)
 	for k, v := range GetDefaultManageHandlers() {
@@ -1098,6 +1111,16 @@ func (a *Agent) startMServer() {
 	for k, v := range a.manageHandlers {
 		handlers[k] = v
 	}
+	// register env handlers
+	extHandelrs := os.Getenv(motan.HandlerEnvironmentName)
+	for _, k := range strings.Split(extHandelrs, ",") {
+		if v, ok := a.envHandlers[strings.TrimSpace(k)]; ok {
+			for kk, vv := range v {
+				handlers[kk] = vv
+			}
+
+		}
+	}
 	for k, v := range handlers {
 		a.mhandle(k, v)
 	}

agent_test.go

@@ -91,6 +91,58 @@ motan-refer:
 	assert.Equal(t, "Hello jack from motan server", resp.GetValue())
 	assert.Equal(t, 100, server.GetProcessPoolSize())
 }
+
+func Test_envHandler(t *testing.T) {
+	t.Parallel()
+	time.Sleep(time.Second * 3)
+	// start client mesh
+	ext := GetDefaultExtFactory()
+	os.Remove("agent.sock")
+	config, _ := config.NewConfigFromReader(bytes.NewReader([]byte(`
+motan-agent:
+  mport: 13500
+  port: 14821
+  eport: 14281
+  htport: 25282
+
+motan-registry:
+  direct:
+    protocol: direct
+    address: 127.0.0.1:22991
+
+motan-refer:
+    recom-engine-refer:
+      group: hello
+      path: helloService
+      protocol: motan2
+      registry: direct
+      asyncInitConnection: false
+      serialization: breeze`)))
+	agent := NewAgent(ext)
+	agent.RegisterEnvHandlers("testHandler", map[string]http.Handler{
+		"/test/test": testHandler(),
+	})
+	os.Setenv(core.HandlerEnvironmentName, "testHandler")
+	go agent.StartMotanAgentFromConfig(config)
+	time.Sleep(time.Second * 3)
+	client := http.Client{
+		Timeout: time.Second,
+	}
+	resp, err := client.Get("http://127.0.0.1:13500/test/test")
+	assert.Nil(t, err)
+	defer resp.Body.Close()
+	b, err := ioutil.ReadAll(resp.Body)
+	assert.Nil(t, err)
+	assert.Equal(t, string(b), "OK")
+	os.Unsetenv(core.HandlerEnvironmentName)
+}
+
+func testHandler() http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		w.Write([]byte("OK"))
+	}
+}
+
 func Test_unixClientCall2(t *testing.T) {
 	t.Parallel()
 	startServer(t, "helloService", 22992)

core/constants.go

@@ -116,6 +116,7 @@ const (
 	GroupEnvironmentName     = "MESH_SERVICE_ADDITIONAL_GROUP"
 	DirectRPCEnvironmentName = "MESH_DIRECT_RPC"
 	FilterEnvironmentName    = "MESH_FILTERS"
+	HandlerEnvironmentName   = "MESH_ADMIN_EXT_HANDLERS"
 )
 
 // meta keys

core/globalContext.go

@@ -69,7 +69,8 @@ var (
 	defaultConfigPath = "./"
 	defaultFileSuffix = ".yaml"
 
-	urlFields = map[string]bool{"protocol": true, "host": true, "port": true, "path": true, "group": true}
+	urlFields  = map[string]bool{"protocol": true, "host": true, "port": true, "path": true, "group": true}
+	extFilters = make(map[string]bool)
 )
 
 // all env flag in motan-go
@@ -87,6 +88,17 @@ var (
 	Recover     = flag.Bool("recover", false, "recover from accidental exit")
 )
 
+func AddRelevantFilter(filterStr string) {
+	k := strings.TrimSpace(filterStr)
+	if k != "" {
+		extFilters[k] = true
+	}
+}
+
+func GetRelevantFilters() map[string]bool {
+	return extFilters
+}
+
 func (c *Context) confToURLs(section string) map[string]*URL {
 	urls := map[string]*URL{}
 	sectionConf, _ := c.Config.GetSection(section)
@@ -397,11 +409,12 @@ func (c *Context) basicConfToURLs(section string) map[string]*URL {
 			newURL = url
 		}
 
-		//final filters: defaultFilter + globalFilter + filters + envFilter
+		//final filters: defaultFilter + globalFilter + filters + envFilter + relevantFilters
 		finalFilters := c.MergeFilterSet(
 			c.GetDefaultFilterSet(newURL),
 			c.GetGlobalFilterSet(newURL),
 			c.GetEnvGlobalFilterSet(),
+			GetRelevantFilters(),
 			c.GetFilterSet(newURL.GetStringParamsWithDefault(FilterKey, ""), ""),
 		)
 		if len(finalFilters) > 0 {

dynamicConfig.go

@@ -328,6 +328,7 @@ func (h *DynamicConfigurerHandler) parseURL(url *core.URL) (*core.URL, error) {
 		h.agent.Context.GetDefaultFilterSet(url),
 		h.agent.Context.GetGlobalFilterSet(url),
 		h.agent.Context.GetEnvGlobalFilterSet(),
+		core.GetRelevantFilters(),
 		h.agent.Context.GetFilterSet(url.GetStringParamsWithDefault(core.FilterKey, ""), ""),
 	)
 	if len(finalFilters) > 0 {