Merge branch 'develop' into simplifyConf

.github/workflows/go.sum

@@ -0,0 +1 @@
+github.com/dubbogo/dubbo-go-pixiu-filter v0.1.3/go.mod h1:d6SDK5BHl/QCvg84BN+g6LZS9QzVqnI2+yw0NBu0uac=

go.mod

@@ -13,9 +13,9 @@ require (
 	github.com/emirpasic/gods v1.12.0
 	github.com/ghodss/yaml v1.0.1-0.20190212211648-25d852aebe32
 	github.com/goinggo/mapstructure v0.0.0-20140717182941-194205d9b4a9
-	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/golang/protobuf v1.5.2
 	github.com/google/uuid v1.2.0 // indirect
-	github.com/hashicorp/consul/api v1.5.0
+	github.com/jhump/protoreflect v1.9.0
 	github.com/mitchellh/mapstructure v1.4.1
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/common v0.29.0 // indirect
@@ -25,21 +25,23 @@ require (
 	github.com/stretchr/testify v1.7.0
 	github.com/tklauser/go-sysconf v0.3.5 // indirect
 	go.etcd.io/etcd/api/v3 v3.5.0-alpha.0
-	go.opentelemetry.io/otel v1.0.0-RC1
+	go.opentelemetry.io/otel v1.0.0-RC2
+	go.opentelemetry.io/otel/exporters/jaeger v1.0.0-RC2
 	go.opentelemetry.io/otel/exporters/prometheus v0.21.0
 	go.opentelemetry.io/otel/metric v0.21.0
-	go.opentelemetry.io/otel/sdk v1.0.0-RC1
+	go.opentelemetry.io/otel/sdk v1.0.0-RC2
 	go.opentelemetry.io/otel/sdk/export/metric v0.21.0
 	go.opentelemetry.io/otel/sdk/metric v0.21.0
+	go.opentelemetry.io/otel/trace v1.0.0-RC2
 	go.uber.org/zap v1.16.0
 	golang.org/x/crypto v0.0.0-20210415154028-4f45737414dc // indirect
 	golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c // indirect
-	google.golang.org/protobuf v1.27.1 // indirect
+	google.golang.org/grpc v1.40.0
+	google.golang.org/protobuf v1.27.1
 	gopkg.in/yaml.v2 v2.4.0
 )
 
 replace (
-	github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473 => github.com/envoyproxy/go-control-plane v0.8.0
+	github.com/envoyproxy/go-control-plane => github.com/envoyproxy/go-control-plane v0.8.0
 	google.golang.org/api => google.golang.org/api v0.13.0
-	google.golang.org/grpc => google.golang.org/grpc v1.27.0
-)
+)

igt/Makefile

@@ -91,6 +91,7 @@ config:
 	$(info   >  Setting up config files)
 	@mkdir -p $(OUT_DIR)/server
 	@mkdir -p $(OUT_DIR)/pixiuconf
+	@mkdir -p $(OUT_DIR)/proto
 	@-test -f $(PROJECT_DIR)/server/profiles/dev/log.yml && cat $(PROJECT_DIR)/server/profiles/dev/log.yml | sed "s#\$$HOST_IP#$(DOCKER_HOST_IP)#g" > $(OUT_DIR)/server/log.yml && echo "  > $(OUT_DIR)/conf/log.yml"
 	@-test -f $(PROJECT_DIR)/server/profiles/dev/server.yml && cat $(PROJECT_DIR)/server/profiles/dev/server.yml | sed "s#\$$HOST_IP#$(DOCKER_HOST_IP)#g" > $(OUT_DIR)/server/server.yml && echo "  > $(OUT_DIR)/conf/server.yml"
 	@-test -f $(PROJECT_DIR)/pixiu/api_config.yaml && cat $(PROJECT_DIR)/pixiu/api_config.yaml | sed "s#\$$HOST_IP#$(DOCKER_HOST_IP)#g" > $(OUT_DIR)/pixiuconf/api_config.yaml && echo "  > $(OUT_DIR)/pixiuconf/api_config.yaml"
@@ -100,13 +101,13 @@ config:
 .PHONY: docker-up
 docker-up:
 	$(info   >  Starting dependency services with $(PROJECT_DIR)/docker/docker-compose.yml)
-	@docker-compose -f $(PROJECT_DIR)/docker/docker-compose.yml up -d
+	@-test -f $(PROJECT_DIR)/docker/docker-compose.yml && docker-compose -f $(PROJECT_DIR)/docker/docker-compose.yml up -d
 
 ## docker-down: Shutdown dependency services on docker
 .PHONY: docker-down
 docker-down:
 	$(info   >  Stopping dependency services with $(PROJECT_DIR)/docker/docker-compose.yml)
-	@docker-compose -f $(PROJECT_DIR)/docker/docker-compose.yml down
+	@-test -f $(PROJECT_DIR)/docker/docker-compose.yml && docker-compose -f $(PROJECT_DIR)/docker/docker-compose.yml down
 
 ## clean: Clean up the output and the binary of the application
 .PHONY: clean

pkg/client/dubbo/dubbo.go

@@ -32,6 +32,9 @@ import (
 	fc "github.com/dubbogo/dubbo-go-pixiu-filter/pkg/api/config"
 
 	"github.com/pkg/errors"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 import (
@@ -48,6 +51,13 @@ const (
 
 const (
 	defaultDubboProtocol = "zookeeper"
+
+	traceNameDubbogoClient = "dubbogo-client"
+	spanNameDubbogoClient  = "DUBBOGO CLIENT"
+
+	spanTagMethod = "method"
+	spanTagType   = "type"
+	spanTagValues = "values"
 )
 
 var (
@@ -165,8 +175,15 @@ func (dc *Client) Call(req *client.Request) (res interface{}, err error) {
 	logger.Debugf("[dubbo-go-pixiu] dubbo invoke, method:%s, types:%s, reqData:%v", method, val.Types, val.Values)
 
 	gs := dc.Get(dm)
-
-	rst, err := gs.Invoke(req.Context, []interface{}{method, val.Types, val.Values})
+	tr := otel.Tracer(traceNameDubbogoClient)
+	_, span := tr.Start(req.Context, spanNameDubbogoClient)
+	trace.SpanFromContext(req.Context).SpanContext()
+	span.SetAttributes(attribute.Key(spanTagMethod).String(method))
+	span.SetAttributes(attribute.Key(spanTagType).Array(val.Types))
+	span.SetAttributes(attribute.Key(spanTagValues).Array(val.Values))
+	defer span.End()
+	ctx := context.WithValue(req.Context, constant.TRACING_REMOTE_SPAN_CTX, trace.SpanFromContext(req.Context).SpanContext())
+	rst, err := gs.Invoke(ctx, []interface{}{method, val.Types, val.Values})
 	if err != nil {
 		return nil, err
 	}

pkg/client/http/http.go

@@ -23,16 +23,16 @@ import (
 	"strings"
 	"sync"
 	"time"
-)
 
-import (
-	"github.com/pkg/errors"
-)
+	contexthttp "github.com/apache/dubbo-go-pixiu/pkg/context/http"
 
-import (
 	"github.com/apache/dubbo-go-pixiu/pkg/client"
 	"github.com/apache/dubbo-go-pixiu/pkg/common/constant"
 	"github.com/apache/dubbo-go-pixiu/pkg/router"
+	"github.com/pkg/errors"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 // RestMetadata http metadata, api config
@@ -54,6 +54,15 @@ var (
 	countDown  = sync.Once{}
 )
 
+const (
+	traceNameHTTPClient = "http-client"
+	spanNameHTTPClient  = "HTTP CLIENT"
+
+	spanTagMethod = "method"
+	spanTagURL    = "url"
+	spanTagBody   = "body"
+)
+
 // Client client to generic invoke dubbo
 type Client struct{}
 
@@ -103,6 +112,16 @@ func (dc *Client) Call(req *client.Request) (resp interface{}, err error) {
 	newReq, _ := http.NewRequest(req.IngressRequest.Method, targetURL, params.Body)
 	newReq.Header = params.Header
 	httpClient := &http.Client{Timeout: 5 * time.Second}
+
+	tr := otel.Tracer(traceNameHTTPClient)
+	_, span := tr.Start(req.Context, spanNameHTTPClient)
+	trace.SpanFromContext(req.Context).SpanContext()
+	span.SetAttributes(attribute.Key(spanTagMethod).String(req.IngressRequest.Method))
+	span.SetAttributes(attribute.Key(spanTagURL).String(targetURL))
+	body := contexthttp.ExtractRequestBody(newReq)
+	span.SetAttributes(attribute.Key(spanTagBody).String(string(body)))
+	defer span.End()
+
 	tmpRet, err := httpClient.Do(newReq)
 
 	return tmpRet, err

pkg/common/constant/http.go

@@ -18,8 +18,13 @@
 package constant
 
 const (
-	HeaderKeyContextType              = "Content-Type"
-	HeaderKeyAccessControlAllowOrigin = "Access-Control-Allow-Origin"
+	HeaderKeyContextType = "Content-Type"
+
+	HeaderKeyAccessControlAllowOrigin      = "Access-Control-Allow-Origin"
+	HeaderKeyAccessControlExposeHeaders    = "Access-Control-Expose-Headers"
+	HeaderKeyAccessControlAllowMethods     = "Access-Control-Allow-Methods"
+	HeaderKeyAccessControlMaxAge           = "Access-Control-Max-Age"
+	HeaderKeyAccessControlAllowCredentials = "Access-Control-Allow-Credentials"
 
 	HeaderValueJsonUtf8  = "application/json;charset=UTF-8"
 	HeaderValueTextPlain = "text/plain"

pkg/common/constant/key.go

@@ -33,6 +33,8 @@ const (
 	HTTPDubboProxyFilter = "dgp.filter.http.dubboproxy"
 	HTTPApiConfigFilter  = "dgp.filter.http.apiconfig"
 	HTTPTimeoutFilter    = "dgp.filter.http.timeout"
+	TracingFilter        = "dgp.filters.tracing"
+	HTTPCorsFilter       = "dgp.filter.http.cors"
 )
 
 const (

pkg/common/extension/filter/filter_manager.go

@@ -57,18 +57,18 @@ func (fm *FilterManager) GetFilters() []HttpFilter {
 	return fm.filters
 }
 
-// Load load the filter from config
+// Load the filter from config
 func (fm *FilterManager) Load() {
 	fm.ReLoad(fm.filterConfigs)
 }
 
-// ReLoad reload filter configs
+// ReLoad filter configs
 func (fm *FilterManager) ReLoad(filters []*model.HTTPFilter) {
 	tmp := make([]HttpFilter, 0, len(filters))
 	for _, f := range filters {
 		apply, err := fm.Apply(f.Name, f.Config)
 		if err != nil {
-			logger.Errorf("apply [%s] init fail, %s", err)
+			logger.Errorf("apply [%s] init fail, %s", f.Name, err.Error())
 		}
 		tmp = append(tmp, apply)
 	}

pkg/common/http/manager.go

@@ -37,13 +37,13 @@ import (
 
 // HttpConnectionManager network filter for http
 type HttpConnectionManager struct {
-	config            *model.HttpConnectionManager
+	config            *model.HttpConnectionManagerConfig
 	routerCoordinator *router2.RouterCoordinator
 	filterManager     *filter.FilterManager
 }
 
 // CreateHttpConnectionManager create http connection manager
-func CreateHttpConnectionManager(hcmc *model.HttpConnectionManager, bs *model.Bootstrap) *HttpConnectionManager {
+func CreateHttpConnectionManager(hcmc *model.HttpConnectionManagerConfig, bs *model.Bootstrap) *HttpConnectionManager {
 	hcm := &HttpConnectionManager{config: hcmc}
 	hcm.routerCoordinator = router2.CreateRouterCoordinator(hcmc)
 	hcm.filterManager = filter.NewFilterManager(hcmc.HTTPFilters)

pkg/common/http/manager_test.go

@@ -89,7 +89,7 @@ func (f *DemoFilter) Apply() error {
 func TestCreateHttpConnectionManager(t *testing.T) {
 	filter.RegisterHttpFilter(&Plugin{})
 
-	hcmc := model.HttpConnectionManager{
+	hcmc := model.HttpConnectionManagerConfig{
 		RouteConfig: model.RouteConfiguration{
 			Routes: []*model.Router{
 				{

pkg/common/router/router.go

@@ -37,7 +37,7 @@ type (
 )
 
 // CreateRouterCoordinator create coordinator for http connection manager
-func CreateRouterCoordinator(hcmc *model.HttpConnectionManager) *RouterCoordinator {
+func CreateRouterCoordinator(hcmc *model.HttpConnectionManagerConfig) *RouterCoordinator {
 
 	rc := &RouterCoordinator{activeConfig: &hcmc.RouteConfig}
 	if hcmc.RouteConfig.Dynamic {

pkg/common/router/router_test.go

@@ -33,7 +33,7 @@ import (
 )
 
 func TestCreateRouterCoordinator(t *testing.T) {
-	hcmc := model.HttpConnectionManager{
+	hcmc := model.HttpConnectionManagerConfig{
 		RouteConfig: model.RouteConfiguration{
 			Routes: []*model.Router{
 				{

pkg/config/config_load_test.go

@@ -38,7 +38,7 @@ var b model.Bootstrap
 func TestMain(m *testing.M) {
 	log.Println("Prepare Bootstrap")
 
-	hcmc := model.HttpConnectionManager{
+	hcmc := model.HttpConnectionManagerConfig{
 		RouteConfig: model.RouteConfiguration{
 			Routes: []*model.Router{
 				{

pkg/context/http/context.go

@@ -54,7 +54,7 @@ type HttpContext struct {
 	// happen error
 	Err error
 
-	HttpConnectionManager model.HttpConnectionManager
+	HttpConnectionManager model.HttpConnectionManagerConfig
 	Listener              *model.Listener
 	Route                 *model.RouteAction
 	Api                   router.API

pkg/context/http/util.go

@@ -18,6 +18,10 @@
 package http
 
 import (
+	"bytes"
+	"io/ioutil"
+	"mime"
+	"net/http"
 	"regexp"
 	"strings"
 )
@@ -60,3 +64,31 @@ func HttpRouteActionMatch(c *HttpContext, ra model.RouteAction) bool {
 
 	return true
 }
+
+// ExtractRequestBody extract body of http request
+func ExtractRequestBody(req *http.Request) []byte {
+	isUpload := isUpload(req)
+	if isUpload {
+		return nil
+	}
+	body, err := ioutil.ReadAll(req.Body)
+	if err != nil {
+		return nil
+	}
+	req.Body = ioutil.NopCloser(bytes.NewBuffer(body))
+	return body
+}
+
+// isUpload determine whether it is a file upload request
+func isUpload(req *http.Request) bool {
+	var ret bool
+	v := req.Header.Get("Content-Type")
+	if v == "" {
+		return ret
+	}
+	d, _, err := mime.ParseMediaType(v)
+	if err == nil && d == "multipart/form-data" {
+		ret = true
+	}
+	return ret
+}

pkg/filter/accesslog/access_log.go

@@ -38,28 +38,28 @@ const (
 )
 
 func init() {
-	filter.RegisterHttpFilter(&AccessPlugin{})
+	filter.RegisterHttpFilter(&Plugin{})
 }
 
 type (
-	// AccessPlugin is http filter plugin.
-	AccessPlugin struct {
+	// Plugin is http filter plugin.
+	Plugin struct {
 	}
-	// AccessFilter is http filter instance
-	AccessFilter struct {
+	// Filter is http filter instance
+	Filter struct {
 		conf *AccessLogConfig
 		alw  *AccessLogWriter
 	}
 )
 
 // Kind return plugin kind
-func (ap *AccessPlugin) Kind() string {
+func (p *Plugin) Kind() string {
 	return Kind
 }
 
 // CreateFilter create filter
-func (ap *AccessPlugin) CreateFilter() (filter.HttpFilter, error) {
-	return &AccessFilter{
+func (p *Plugin) CreateFilter() (filter.HttpFilter, error) {
+	return &Filter{
 		conf: &AccessLogConfig{},
 		alw: &AccessLogWriter{
 			AccessLogDataChan: make(chan AccessLogData, constant.LogDataBuffer),
@@ -68,32 +68,32 @@ func (ap *AccessPlugin) CreateFilter() (filter.HttpFilter, error) {
 }
 
 // PrepareFilterChain prepare chain when http context init
-func (af *AccessFilter) PrepareFilterChain(ctx *http.HttpContext) error {
-	ctx.AppendFilterFunc(af.Handle)
+func (f *Filter) PrepareFilterChain(ctx *http.HttpContext) error {
+	ctx.AppendFilterFunc(f.Handle)
 	return nil
 }
 
-// Handle handle http context
-func (af *AccessFilter) Handle(c *http.HttpContext) {
+// Handle process http context
+func (f *Filter) Handle(c *http.HttpContext) {
 	start := time.Now()
 	c.Next()
 	latency := time.Since(start)
 	// build access_log message
 	accessLogMsg := buildAccessLogMsg(c, latency)
 	if len(accessLogMsg) > 0 {
-		af.alw.Writer(AccessLogData{AccessLogConfig: *af.conf, AccessLogMsg: accessLogMsg})
+		f.alw.Writer(AccessLogData{AccessLogConfig: *f.conf, AccessLogMsg: accessLogMsg})
 	}
 }
 
 // Config return config of filter
-func (af *AccessFilter) Config() interface{} {
-	return af.conf
+func (f *Filter) Config() interface{} {
+	return f.conf
 }
 
 // Apply init after config set
-func (af *AccessFilter) Apply() error {
+func (f *Filter) Apply() error {
 	// init
-	af.alw.Write()
+	f.alw.Write()
 	return nil
 }