Merge pull request apache#43 from wanhailong/master

Support RTC audit log

Author: wangtuanjie

go/src/pili.qiniu.com/dsync.v1/conf.go

@@ -13,12 +13,13 @@ import (
 )
 
 type Conf struct {
-	ConfigDir   string       `json:"config_dir"`
-	Host        string       `json:"host"`
-	IDC         string       `json:"idc"`
-	ScanSeconds int64        `json:"scan_seconds"`
-	OffsetDir   string       `json:"offset_dir"`
-	Sinker      flume.Config `json:"sinker"`
+	ConfigDir           string       `json:"config_dir"`
+	CustomizedConfigDir string       `json:"customized_config_dir"`
+	Host                string       `json:"host"`
+	IDC                 string       `json:"idc"`
+	ScanSeconds         int64        `json:"scan_seconds"`
+	OffsetDir           string       `json:"offset_dir"`
+	Sinker              flume.Config `json:"sinker"`
 }
 
 type Service struct {
@@ -65,6 +66,10 @@ func LoadConf(path string) (c Conf, err error) {
 		c.ConfigDir = "/home/qboxserver/logsync/_package/confs"
 	}
 
+	if c.CustomizedConfigDir == "" {
+		c.CustomizedConfigDir = "/home/qboxserver/pili-dsync/_package/confs"
+	}
+
 	return
 }
 

go/src/pili.qiniu.com/dsync.v1/main.go

@@ -22,6 +22,15 @@ func Main(configFile string) {
 			log.Fatalln("LoadServices failed :", err)
 		}
 
+		nowSrvs1, err := LoadServices(conf.CustomizedConfigDir)
+		if err != nil {
+			log.Fatalln("LoadServices failed :", err)
+		}
+
+		for k, v := range nowSrvs1 {
+			nowSrvs[k] = v
+		}
+
 		for k, v := range nowSrvs {
 			if _, ok := oldSrvs[k]; !ok {
 				go func(x Service) {

go/src/pili.qiniu.com/dsync.v1/parse.go

@@ -36,6 +36,7 @@ var (
 	StrWs     = "WS"
 	StrDocker = "DOCKER"
 	StrPili   = "PILI"
+	StrAUDIT  = "AUDIT"
 	BytesReq  = []byte("REQ")
 )
 
@@ -65,6 +66,8 @@ func ParseTime(line []byte, typ string) (ts int64, err error) {
 		t, err = parseOVS(line)
 	case Str3TEE:
 		t, err = parseTime3tee(line)
+	case StrAUDIT:
+		t, err = parseAudit(line)
 	default:
 		panic(typ)
 	}
@@ -265,6 +268,18 @@ func parsePili(line []byte) (t time.Time, err error) {
 	return
 }
 
+// {"time":1473345370,"duration":5,"room_id":"1378262837-5451333","user_id":"7938805"}
+func parseAudit(line []byte) (t time.Time, err error) {
+	d := struct {
+		T int64 `json:"time"`
+	}{}
+	if err = json.Unmarshal(line, &d); err != nil {
+		return
+	}
+	t = time.Unix(d.T, 0)
+	return
+}
+
 // https://pm.qbox.me/issues/25864
 // 2016-03-03T22:55:40.635Z|04473|connmgr|INFO|qcos-br0<->tcp:192.168.63.30:6603: 14 flow_mods in the 57 s starting 59 s ago (14 adds)
 func parseOVS(line []byte) (t time.Time, err error) {

go/src/pili.qiniu.com/dsync.v1/parse_test.go

@@ -107,6 +107,11 @@ func TestParseTime(t *testing.T) {
 			StrApp,
 			1444432800000000000,
 		},
+		{
+			`{"time":1473345375,"duration":5,"room_id":"1378262837-5451333","user_id":"6583953"}`,
+			StrAUDIT,
+			1473345375000000000,
+		},
 		{
 			`2015-10-10 07:20:00.09s [:0000001f  DEBUG http req: /rtc/room/create]`,
 			Str3TEE,