修复recorder版本 依赖"https://github.com/v2pro/plz"无法编译的问题

go.mod

@@ -4,8 +4,7 @@ go 1.13
 
 require (
 	github.com/agiledragon/gomonkey v2.0.1+incompatible
-	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/stretchr/testify v1.5.1
-	github.com/v2pro/plz v0.0.0-20171129053502-03cb46d336e0
-	gopkg.in/yaml.v2 v2.2.8 // indirect
+	github.com/stretchr/testify v1.6.1
+	github.com/v2pro/plz v0.0.0-20221028024117-e5f9aec5b631
+	golang.org/x/text v0.1.0
 )

go.sum

@@ -1,19 +1,30 @@
+github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
+github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/agiledragon/gomonkey v2.0.1+incompatible h1:DIQT3ZshgGz9pTwBddRSZWDutIRPx2d7UzmjzgWo9q0=
 github.com/agiledragon/gomonkey v2.0.1+incompatible/go.mod h1:2NGfXu1a80LLr2cmWXGBDaHEjb1idR6+FVlX5T3D9hw=
-github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rakyll/statik v0.1.7/go.mod h1:AlZONWzMtEnMs7W4e/1LURLiI49pIMmp6V9Unghqrcc=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
-github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
-github.com/v2pro/plz v0.0.0-20171129053502-03cb46d336e0 h1:iDUJ1dPbPQ3RASxAhSUKSK4xC3YULCMoWbwlnY70md4=
-github.com/v2pro/plz v0.0.0-20171129053502-03cb46d336e0/go.mod h1:6xoYDIZTeCY25tlsJC/zNlCh84xCKwBSAXwKF32tdIg=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/v2pro/plz v0.0.0-20221028024117-e5f9aec5b631 h1:WYq/4UeJfAorBY7ncC31bVxI031x4MUCQvF+z12fIYA=
+github.com/v2pro/plz v0.0.0-20221028024117-e5f9aec5b631/go.mod h1:3gacX+hQo+xvl0vtLqCMufzxuNCwt4geAVOMt2LQYfE=
+golang.org/x/text v0.1.0 h1:LEnmSFmpuy9xPmlp2JeGQQOYbPv3TkQbuGJU3A0HegU=
+golang.org/x/text v0.1.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
-gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
-gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
-gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/natefinch/lumberjack.v2 v2.0.0 h1:1Lc07Kr7qY4U2YPouBjpCLxpiyxIVoxqXgkXLknAOE8=
+gopkg.in/natefinch/lumberjack.v2 v2.0.0/go.mod h1:l0ndWWf7gzL7RNwBG7wST/UCcT4T24xpD6X8LsfU/+k=
+gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
+gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

plugins/recorder_default.go

@@ -62,8 +62,7 @@ type DefaultRecorder struct {
 func (r *DefaultRecorder) Record(session *recording.Session) {
 	defer func() {
 		if err := recover(); err != nil {
-			countlog.Fatal("event!kafka_recorder.panic", "err", err,
-				"stacktrace", countlog.ProvideStacktrace)
+			countlog.LogPanic(err, "kafka_recorder.panic")
 		}
 	}()
 

recorder-agent/common/httpclient/init.go

@@ -0,0 +1,47 @@
+package httpclient
+
+import (
+	"bytes"
+	"context"
+	"io/ioutil"
+	"net/http"
+	"time"
+
+	"github.com/didi/sharingan/recorder-agent/common/zap"
+)
+
+type HttpClient struct {
+}
+
+var Handler HttpClient
+
+func Init() {
+	Handler = HttpClient{}
+}
+
+//Post http post
+func (hc *HttpClient) Post(ctx context.Context, url string, jsonBytes []byte, timeout time.Duration) (*http.Response, []byte, error) {
+	req, err := http.NewRequest("POST", url, bytes.NewBuffer(jsonBytes))
+	if err != nil {
+		zap.Logger.Error(zap.Format(ctx, "ERROR", err.Error()))
+		return nil, nil, err
+	}
+	req.SetBasicAuth("username", "password")
+	//默认 application/json
+	req.Header.Set("Content-Type", "application/json;charset=utf-8")
+
+	client := &http.Client{}
+	if timeout > 0 {
+		client.Timeout = timeout
+	}
+	resp, err := client.Do(req)
+	zap.Logger.Info(zap.Format(ctx, "INFO", "resp=%v", resp))
+	if err != nil {
+		zap.Logger.Error(zap.Format(ctx, "ERROR", err.Error()))
+		return nil, nil, err
+	}
+	defer resp.Body.Close()
+
+	body, _ := ioutil.ReadAll(resp.Body)
+	return resp, body, nil
+}

recorder-agent/common/zap/init.go

@@ -1,7 +1,11 @@
 package zap
 
 import (
+	"context"
+	"fmt"
 	"log"
+	"runtime"
+	"strings"
 	"time"
 
 	"github.com/didi/sharingan/recorder-agent/common/conf"
@@ -49,3 +53,25 @@ func init() {
 
 	Logger = zap.New(core)
 }
+
+func Format(ctx context.Context, level string, format string, args ...interface{}) string {
+	// time
+	ts := time.Now().Format("2006-01-02 15:04:05.000000")
+
+	// file, line
+	_, file, line, ok := runtime.Caller(2)
+	if !ok {
+		file = "file"
+		line = -1
+	}
+
+	// igonre dir
+	file = strings.TrimPrefix(file, path.Root+"/")
+
+	var ctxString string
+	if t, ok := ctx.Value(tracerKey).(Tracer); ok {
+		ctxString = t.Format()
+	}
+
+	return fmt.Sprintf("[%s][%s][%s:%d] %s%s", level, ts, file, line, ctxString, fmt.Sprintf(format, args...))
+}

recorder-agent/common/zap/trace.go

@@ -0,0 +1,35 @@
+package zap
+
+import (
+	"bytes"
+	"context"
+)
+
+type Tracer interface {
+	Format() string
+}
+
+var NewTracer = func(data map[string]string) *tracer {
+	return &tracer{data: data}
+}
+
+type tracer struct {
+	data map[string]string
+}
+
+func (t *tracer) Format() string {
+	var buf bytes.Buffer
+	for k, v := range t.data {
+		buf.WriteString(k)
+		buf.WriteString("=")
+		buf.WriteString(v)
+		buf.WriteString("||")
+	}
+	return buf.String()
+}
+
+const tracerKey = "::tracer::"
+
+func TraceContext(ctx context.Context, t Tracer) context.Context {
+	return context.WithValue(ctx, tracerKey, t)
+}

recorder-agent/conf/app.toml

@@ -10,4 +10,7 @@ idleTimeout = 5000       # Keep-alive timeout(ms)
 filename = "log/recorder.log.%Y%m%d%H"
 linkname = "log/recorder.log"
 maxHourAge    = 4 # 默认日志保留4小时
-maxHourRotate = 1 # 默认日志按小时切分
+maxHourRotate = 1 # 默认日志按小时切分
+
+[es_url]
+# 确保录制机器可以访问，check：`curl http://127.0.0.1:9002/xxx/recorder_index/_search`能写入数据

recorder-agent/main.go

@@ -5,17 +5,27 @@ import (
 	"log"
 	"net/http"
 	"runtime/debug"
+	"time"
 
 	"github.com/didi/sharingan/recorder-agent/common/conf"
+	"github.com/didi/sharingan/recorder-agent/common/httpclient"
 	"github.com/didi/sharingan/recorder-agent/common/zap"
 	"github.com/didi/sharingan/recorder-agent/record"
 	"github.com/didi/sharingan/recorder-agent/server"
 )
 
+const (
+	timeOut = 10 * time.Second
+)
+
 var (
 	svr = server.New()
 )
 
+func init() {
+	httpclient.Init()
+}
+
 func main() {
 	defer func() {
 		if r := recover(); r != nil {
@@ -57,8 +67,16 @@ func indexHandler(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// TO log
-	w.Write([]byte("OK"))
-	zap.Logger.Info(string(buf)) // 日志收集，最终入ES
+	// 日志收集，最终入ES
+	url := conf.Handler.GetString("es_url.default")
+	if _, _, err := httpclient.Handler.Post(r.Context(), url, buf, timeOut); err != nil {
+		zap.Logger.Error(zap.Format(r.Context(), "ERROR", "send data to es err: %v", err))
+
+		// TO log
+		w.Write([]byte("OK"))
+		zap.Logger.Info(string(buf))
+		return
+	}
+
 	return
 }

recorder/koala/logger/logger.go

@@ -5,6 +5,10 @@ import (
 	"strings"
 
 	"github.com/v2pro/plz/countlog"
+	"github.com/v2pro/plz/countlog/output"
+	"github.com/v2pro/plz/countlog/output/async"
+	"github.com/v2pro/plz/countlog/output/json"
+	"github.com/v2pro/plz/countlog/output/lumberjack"
 )
 
 var logFile = "STDOUT"
@@ -21,14 +25,17 @@ func Init() {
 		SetLogFile(file)
 	}
 
-	logWriter := countlog.NewAsyncLogWriter(logLevel, countlog.NewFileLogOutput(logFile))
-	logWriter.LogFormatter = &countlog.HumanReadableFormat{
-		ContextPropertyNames: []string{"threadID", "outboundSrc"},
-		StringLengthCap:      512,
-	}
+	writer := async.NewAsyncWriter(async.AsyncWriterConfig{
+		Writer: &lumberjack.Logger{
+			Filename: logFile,
+		},
+	})
+	defer writer.Close()
 
-	logWriter.Start()
-	countlog.LogWriters = append(countlog.LogWriters, logWriter)
+	countlog.EventWriter = output.NewEventWriter(output.EventWriterConfig{
+		Format: &json.JsonFormat{},
+		Writer: writer,
+	})
 }
 
 // SetLogFile set log file, default: STDOUT

recorder/koala/recording/action.go

@@ -1,9 +1,24 @@
 package recording
 
 import (
+	"bufio"
+	"bytes"
+	"compress/gzip"
 	"encoding/json"
+	"fmt"
+	"io/ioutil"
 	"net"
+	"net/http"
+	"strings"
 	"unicode/utf8"
+
+	"golang.org/x/text/encoding/simplifiedchinese"
+	"golang.org/x/text/transform"
+)
+
+const (
+	callOutboundAction  = "callOutbound"
+	returnInboundAction = "returnInbound"
 )
 
 // Action Action
@@ -63,7 +78,7 @@ func (returnInbound *ReturnInbound) MarshalJSON() ([]byte, error) {
 		Response json.RawMessage
 	}{
 		ReturnInbound: *returnInbound,
-		Response:      EncodeAnyByteArray(returnInbound.Response),
+		Response:      ParesResponse(returnInbound.Response, returnInboundAction),
 	})
 }
 
@@ -90,7 +105,7 @@ func (callOutbound *CallOutbound) MarshalJSON() ([]byte, error) {
 	}{
 		CallOutbound: *callOutbound,
 		Request:      EncodeAnyByteArray(callOutbound.Request),
-		Response:     EncodeAnyByteArray(callOutbound.Response),
+		Response:     ParesResponse(callOutbound.Response, callOutboundAction),
 		CSpanID:      EncodeAnyByteArray(callOutbound.CSpanID),
 	})
 }
@@ -310,3 +325,59 @@ func EncodeAnyByteArray(s []byte) json.RawMessage {
 	encoded = append(encoded, '"')
 	return json.RawMessage(encoded)
 }
+
+// ParesResponse ...
+func ParesResponse(s []byte, action string) json.RawMessage {
+	encoded := []byte{'"'}
+	if !bytes.Contains(s, []byte("Content-Encoding: gzip")) {
+		return EncodeAnyByteArray(s)
+	}
+
+	// handle gzip response
+	reader := bufio.NewReader(strings.NewReader(string(s)))
+	resp, err := http.ReadResponse(reader, nil)
+	if err != nil {
+		fmt.Println("反序列化HTTP响应出错：", err)
+		return encoded
+	}
+	defer resp.Body.Close()
+
+	// 解析响应体
+	bodyBytes, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		fmt.Println("读取响应体出错：", err)
+		return encoded
+	}
+
+	// 检查Content-Encoding，并解压缩响应体
+	if resp.Header.Get("Content-Encoding") == "gzip" {
+		reader, err := gzip.NewReader(bytes.NewReader(bodyBytes))
+		if err != nil {
+			fmt.Println("创建gzip解压缩读取器出错：", err)
+			return encoded
+		}
+		defer reader.Close()
+
+		bodyBytes, err = ioutil.ReadAll(reader)
+		if err != nil {
+			fmt.Println("读取解压缩后的内容出错：", err)
+			return encoded
+		}
+
+		switch action {
+		case returnInboundAction:
+			return bodyBytes
+		case callOutboundAction:
+			// 将GBK编码转换为UTF-8编码
+			utf8Bytes, err := ioutil.ReadAll(transform.NewReader(bytes.NewReader(bodyBytes), simplifiedchinese.GBK.NewDecoder()))
+			if err != nil {
+				fmt.Println("err: ", err)
+				return encoded
+			}
+
+			return utf8Bytes
+		}
+	}
+
+	return encoded
+}

recorder/koala/recording/async.go

@@ -31,12 +31,8 @@ func (recorder *AsyncRecorder) Start() {
 // backgroundRecord backgroundRecord
 func (recorder *AsyncRecorder) backgroundRecord() {
 	defer func() {
-		recovered := recover()
-		if recovered != nil {
-			countlog.Error("event!recording.panic",
-				"err", recovered,
-				"ctx", recorder.Context,
-				"stacktrace", countlog.ProvideStacktrace)
+		if recovered := recover(); recovered != nil {
+			countlog.LogPanic(recovered, "recording.panic")
 		}
 	}()
 	for {