Ftr: AccessLogFilter support

common/constant/default.go

@@ -46,7 +46,7 @@ const (
 const (
 	DEFAULT_KEY               = "default"
 	PREFIX_DEFAULT_KEY        = "default."
-	DEFAULT_SERVICE_FILTERS   = "echo,token"
+	DEFAULT_SERVICE_FILTERS   = "echo,token,accesslog"
 	DEFAULT_REFERENCE_FILTERS = ""
 	GENERIC_REFERENCE_FILTERS = "generic"
 	GENERIC                   = "$invoke"

common/constant/key.go

@@ -38,6 +38,8 @@ const (
 	GENERIC_KEY            = "generic"
 	CLASSIFIER_KEY         = "classifier"
 	TOKEN_KEY              = "token"
+	LOCAL_ADDR             = "local-addr"
+	REMOTE_ADDR            = "remote-addr"
 )
 
 const (
@@ -58,6 +60,7 @@ const (
 	FORKS_KEY            = "forks"
 	DEFAULT_FORKS        = 2
 	DEFAULT_TIMEOUT      = 1000
+	ACCESS_LOG_KEY       = "accesslog"
 )
 
 const (

config/service_config.go

@@ -58,6 +58,7 @@ type ServiceConfig struct {
 	Retries       string            `yaml:"retries"  json:"retries,omitempty" property:"retries"`
 	Params        map[string]string `yaml:"params"  json:"params,omitempty" property:"params"`
 	Token         string            `yaml:"token" json:"token,omitempty" property:"token"`
+	AccessLog     string            `yaml:"accesslog" json:"accesslog,omitempty" property:"accesslog"`
 	unexported    *atomic.Bool
 	exported      *atomic.Bool
 	rpcService    common.RPCService
@@ -188,6 +189,9 @@ func (srvconfig *ServiceConfig) getUrlMap() url.Values {
 	//filter
 	urlMap.Set(constant.SERVICE_FILTER_KEY, mergeValue(providerConfig.Filter, srvconfig.Filter, constant.DEFAULT_SERVICE_FILTERS))
 
+	//filter special config
+	urlMap.Set(constant.ACCESS_LOG_KEY, srvconfig.AccessLog)
+
 	for _, v := range srvconfig.Methods {
 		urlMap.Set("methods."+v.Name+"."+constant.LOADBALANCE_KEY, v.Loadbalance)
 		urlMap.Set("methods."+v.Name+"."+constant.RETRIES_KEY, v.Retries)

examples/general/dubbo/go-server/profiles/dev/server.yml

@@ -25,7 +25,6 @@ registries :
     username: ""
     password: ""
 
-
 services:
   "UserProvider":
     # 可以指定多个registry，使用逗号隔开;不指定默认向所有注册中心注册
@@ -36,6 +35,10 @@ services:
     loadbalance: "random"
     warmup: "100"
     cluster: "failover"
+
+     # config the path to access log file. The value can be "true" or "default" which means that the access log will be log in general log file (which config in log.yml)
+#    accesslog: "/path/to/your/accesslog/file"
+
     methods:
     - name: "GetUser"
       retries: "1"

filter/impl/access_log_filter.go

@@ -0,0 +1,223 @@
+/*
+ * 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 impl
+
+import (
+	"os"
+	"reflect"
+	"strings"
+	"time"
+)
+
+import (
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/common/extension"
+	"github.com/apache/dubbo-go/common/logger"
+	"github.com/apache/dubbo-go/filter"
+	"github.com/apache/dubbo-go/protocol"
+)
+
+const (
+	//used in URL.
+	FileDateFormat    = "2006-01-02"
+	MessageDateLayout = "2006-01-02 15:04:05"
+	LogMaxBuffer      = 5000
+	LogFileMode       = 0600
+
+	// those fields are the data collected by this filter
+	Types     = "types"
+	Arguments = "arguments"
+)
+
+func init() {
+	extension.SetFilter(constant.ACCESS_LOG_KEY, GetAccessLogFilter)
+}
+
+/*
+ * Although the access log filter is a default filter,
+ * you should config "accesslog" in service's config to tell the filter where store the access log.
+ * for example:
+ * "UserProvider":
+    registry: "hangzhouzk"
+    protocol : "dubbo"
+    interface : "com.ikurento.user.UserProvider"
+    ... # other configuration
+    accesslog: "/your/path/to/store/the/log/", # it should be the path of file.
+
+ * the value of "accesslog" can be "true" or "default" too.
+ * If the value is one of them, the access log will be record in log file which defined in log.yml
+*/
+type AccessLogFilter struct {
+	logChan chan AccessLogData
+}
+
+func (ef *AccessLogFilter) Invoke(invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+	accessLog := invoker.GetUrl().GetParam(constant.ACCESS_LOG_KEY, "")
+	if len(accessLog) > 0 {
+		accessLogData := AccessLogData{data: ef.buildAccessLogData(invoker, invocation), accessLog: accessLog}
+		ef.logIntoChannel(accessLogData)
+	}
+	return invoker.Invoke(invocation)
+}
+
+// it won't block the invocation
+func (ef *AccessLogFilter) logIntoChannel(accessLogData AccessLogData) {
+	select {
+	case ef.logChan <- accessLogData:
+		return
+	default:
+		logger.Warn("The channel is full and the access logIntoChannel data will be dropped")
+		return
+	}
+}
+
+func (ef *AccessLogFilter) buildAccessLogData(invoker protocol.Invoker, invocation protocol.Invocation) map[string]string {
+	dataMap := make(map[string]string, 16)
+	attachments := invocation.Attachments()
+	dataMap[constant.INTERFACE_KEY] = attachments[constant.INTERFACE_KEY]
+	dataMap[constant.METHOD_KEY] = invocation.MethodName()
+	dataMap[constant.VERSION_KEY] = attachments[constant.VERSION_KEY]
+	dataMap[constant.GROUP_KEY] = attachments[constant.GROUP_KEY]
+	dataMap[constant.TIMESTAMP_KEY] = time.Now().Format(MessageDateLayout)
+	dataMap[constant.LOCAL_ADDR], _ = attachments[constant.LOCAL_ADDR]
+	dataMap[constant.REMOTE_ADDR], _ = attachments[constant.REMOTE_ADDR]
+
+	if len(invocation.Arguments()) > 0 {
+		builder := strings.Builder{}
+		// todo(after the paramTypes were set to the invocation. we should change this implementation)
+		typeBuilder := strings.Builder{}
+
+		builder.WriteString(reflect.ValueOf(invocation.Arguments()[0]).String())
+		typeBuilder.WriteString(reflect.TypeOf(invocation.Arguments()[0]).Name())
+		for idx := 1; idx < len(invocation.Arguments()); idx++ {
+			arg := invocation.Arguments()[idx]
+			builder.WriteString(",")
+			builder.WriteString(reflect.ValueOf(arg).String())
+
+			typeBuilder.WriteString(",")
+			typeBuilder.WriteString(reflect.TypeOf(arg).Name())
+		}
+		dataMap[Arguments] = builder.String()
+		dataMap[Types] = typeBuilder.String()
+	}
+
+	return dataMap
+}
+
+func (ef *AccessLogFilter) OnResponse(result protocol.Result, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+	return result
+}
+
+func (ef *AccessLogFilter) writeLogToFile(data AccessLogData) {
+	accessLog := data.accessLog
+	if isDefault(accessLog) {
+		logger.Info(data.toLogMessage())
+		return
+	}
+
+	logFile, err := ef.openLogFile(accessLog)
+	if err != nil {
+		logger.Warnf("Can not open 	the access log file: %s, %v", accessLog, err)
+		return
+	}
+	logger.Debugf("Append log to %s", accessLog)
+	message := data.toLogMessage()
+	message = message + "\n"
+	_, err = logFile.WriteString(message)
+	if err != nil {
+		logger.Warnf("Can not write the log into access log file: %s, %v", accessLog, err)
+	}
+}
+
+func (ef *AccessLogFilter) openLogFile(accessLog string) (*os.File, error) {
+	logFile, err := os.OpenFile(accessLog, os.O_CREATE|os.O_APPEND|os.O_RDWR, LogFileMode)
+	if err != nil {
+		logger.Warnf("Can not open the access log file: %s, %v", accessLog, err)
+		return nil, err
+	}
+	now := time.Now().Format(FileDateFormat)
+	fileInfo, err := logFile.Stat()
+	if err != nil {
+		logger.Warnf("Can not get the info of access log file: %s, %v", accessLog, err)
+		return nil, err
+	}
+	last := fileInfo.ModTime().Format(FileDateFormat)
+	if now != last {
+		err = os.Rename(fileInfo.Name(), fileInfo.Name()+"."+now)
+		if err != nil {
+			logger.Warnf("Can not rename access log file: %s, %v", fileInfo.Name(), err)
+			return nil, err
+		}
+		logFile, err = os.OpenFile(accessLog, os.O_CREATE|os.O_APPEND|os.O_RDWR, LogFileMode)
+	}
+	return logFile, err
+}
+
+func isDefault(accessLog string) bool {
+	return strings.EqualFold("true", accessLog) || strings.EqualFold("default", accessLog)
+}
+
+func GetAccessLogFilter() filter.Filter {
+	accessLogFilter := &AccessLogFilter{logChan: make(chan AccessLogData, LogMaxBuffer)}
+	go func() {
+		for accessLogData := range accessLogFilter.logChan {
+			accessLogFilter.writeLogToFile(accessLogData)
+		}
+	}()
+	return accessLogFilter
+}
+
+type AccessLogData struct {
+	accessLog string
+	data      map[string]string
+}
+
+func (ef *AccessLogData) toLogMessage() string {
+	builder := strings.Builder{}
+	builder.WriteString("[")
+	builder.WriteString(ef.data[constant.TIMESTAMP_KEY])
+	builder.WriteString("] ")
+	builder.WriteString(ef.data[constant.REMOTE_ADDR])
+	builder.WriteString(" -> ")
+	builder.WriteString(ef.data[constant.LOCAL_ADDR])
+	builder.WriteString(" - ")
+	if len(ef.data[constant.GROUP_KEY]) > 0 {
+		builder.WriteString(ef.data[constant.GROUP_KEY])
+		builder.WriteString("/")
+	}
+
+	builder.WriteString(ef.data[constant.INTERFACE_KEY])
+
+	if len(ef.data[constant.VERSION_KEY]) > 0 {
+		builder.WriteString(":")
+		builder.WriteString(ef.data[constant.VERSION_KEY])
+	}
+
+	builder.WriteString(" ")
+	builder.WriteString(ef.data[constant.METHOD_KEY])
+	builder.WriteString("(")
+	if len(ef.data[Types]) > 0 {
+		builder.WriteString(ef.data[Types])
+	}
+	builder.WriteString(") ")
+
+	if len(ef.data[Arguments]) > 0 {
+		builder.WriteString(ef.data[Arguments])
+	}
+	return builder.String()
+}

filter/impl/access_log_filter_test.go

@@ -0,0 +1,82 @@
+/*
+ * 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 impl
+
+import (
+	"context"
+	"testing"
+)
+
+import (
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/common"
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/protocol"
+	"github.com/apache/dubbo-go/protocol/invocation"
+)
+
+func TestAccessLogFilter_Invoke_Not_Config(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	url, _ := common.NewURL(context.Background(),
+		"dubbo://:20000/UserProvider?app.version=0.0.1&application=BDTService&bean.name=UserProvider"+
+			"&cluster=failover&environment=dev&group=&interface=com.ikurento.user.UserProvider&loadbalance=random&methods.GetUser."+
+			"loadbalance=random&methods.GetUser.retries=1&methods.GetUser.weight=0&module=dubbogo+user-info+server&name="+
+			"BDTService&organization=ikurento.com&owner=ZX&registry.role=3&retries=&"+
+			"service.filter=echo%2Ctoken%2Caccesslog&timestamp=1569153406&token=934804bf-b007-4174-94eb-96e3e1d60cc7&version=&warmup=100")
+	invoker := protocol.NewBaseInvoker(url)
+
+	attach := make(map[string]string, 10)
+	inv := invocation.NewRPCInvocation("MethodName", []interface{}{"OK", "Hello"}, attach)
+
+	accessLogFilter := GetAccessLogFilter()
+	result := accessLogFilter.Invoke(invoker, inv)
+	assert.Nil(t, result.Error())
+}
+
+func TestAccessLogFilter_Invoke_Default_Config(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	url, _ := common.NewURL(context.Background(),
+		"dubbo://:20000/UserProvider?app.version=0.0.1&application=BDTService&bean.name=UserProvider"+
+			"&cluster=failover&accesslog=true&environment=dev&group=&interface=com.ikurento.user.UserProvider&loadbalance=random&methods.GetUser."+
+			"loadbalance=random&methods.GetUser.retries=1&methods.GetUser.weight=0&module=dubbogo+user-info+server&name="+
+			"BDTService&organization=ikurento.com&owner=ZX&registry.role=3&retries=&"+
+			"service.filter=echo%2Ctoken%2Caccesslog&timestamp=1569153406&token=934804bf-b007-4174-94eb-96e3e1d60cc7&version=&warmup=100")
+	invoker := protocol.NewBaseInvoker(url)
+
+	attach := make(map[string]string, 10)
+	attach[constant.VERSION_KEY] = "1.0"
+	attach[constant.GROUP_KEY] = "MyGroup"
+	inv := invocation.NewRPCInvocation("MethodName", []interface{}{"OK", "Hello"}, attach)
+
+	accessLogFilter := GetAccessLogFilter()
+	result := accessLogFilter.Invoke(invoker, inv)
+	assert.Nil(t, result.Error())
+}
+
+func TestAccessLogFilter_OnResponse(t *testing.T) {
+	result := &protocol.RPCResult{}
+	accessLogFilter := GetAccessLogFilter()
+	response := accessLogFilter.OnResponse(result, nil, nil)
+	assert.Equal(t, result, response)
+}

go.sum

@@ -517,6 +517,7 @@ golang.org/x/tools v0.0.0-20190114222345-bf090417da8b/go.mod h1:n7NCudcB/nEzxVGm
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
+golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135 h1:5Beo0mZN8dRzgrMMkDp0jc8YXQKx9DiJ2k1dkvGsn5A=
 golang.org/x/tools v0.0.0-20190524140312-2c0ae7006135/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
 google.golang.org/api v0.0.0-20180829000535-087779f1d2c9 h1:z1TeLUmxf9ws9KLICfmX+KGXTs+rjm+aGWzfsv7MZ9w=
 google.golang.org/api v0.0.0-20180829000535-087779f1d2c9/go.mod h1:4mhQ8q/RsB7i+udVvVy5NUi08OU8ZlA0gRVgrF7VFY0=

protocol/dubbo/listener.go

@@ -246,8 +246,13 @@ func (h *RpcServerHandler) OnMessage(session getty.Session, pkg interface{}) {
 	}
 	invoker := exporter.(protocol.Exporter).GetInvoker()
 	if invoker != nil {
-		result := invoker.Invoke(invocation.NewRPCInvocation(p.Service.Method, p.Body.(map[string]interface{})["args"].([]interface{}),
-			p.Body.(map[string]interface{})["attachments"].(map[string]string)))
+		attachments := p.Body.(map[string]interface{})["attachments"].(map[string]string)
+		attachments[constant.LOCAL_ADDR] = session.LocalAddr()
+		attachments[constant.REMOTE_ADDR] = session.RemoteAddr()
+
+		args := p.Body.(map[string]interface{})["args"].([]interface{})
+		inv := invocation.NewRPCInvocation(p.Service.Method, args, attachments)
+		result := invoker.Invoke(inv)
 		if err := result.Error(); err != nil {
 			p.Header.ResponseStatus = hessian.Response_OK
 			p.Body = hessian.NewResponse(nil, err, result.Attachments())