apache#79 消息推送跟踪

新增TTL索引来实现推送轨迹有效期控制，通过revision和session_ID来区分记录的唯一性

deployments/db.js

@@ -83,15 +83,15 @@ db.createCollection( "polling_detail", {
     max: 100,
     validator: { $jsonSchema: {
             bsonType: "object",
-            required: [ "id","session_id","domain","url_path" ],
+            required: [ "id","session_id","revision","url_path"],
             properties: {
                 id: {
                     bsonType: "string",
                 },
                 domain: {
                     bsonType: "string",
                 },
-                params: {
+                polling_data: {
                     bsonType: "object"
                 },
                 ip: {
@@ -103,8 +103,8 @@ db.createCollection( "polling_detail", {
                 response_body: {
                     bsonType: "object"
                 },
-                response_header: {
-                    bsonType: "object"
+                Revision: {
+                    bsonType: "string"
                 },
                 response_code: {
                     bsonType: "number"
@@ -119,8 +119,7 @@ db.kv.createIndex({key: 1, label_format: 1,domain:1,project:1},{ unique: true })
 db.kv_revision.createIndex( { "delete_time": 1 }, { expireAfterSeconds: 7 * 24 * 3600 } );
 db.label.createIndex({"id": 1}, { unique: true } );
 db.label.createIndex({format: 1,domain:1,project:1},{ unique: true });
-db.polling_detail.createIndex({"id": 1}, { unique: true } );
-db.polling_detail.createIndex({session_id:1,domain:1}, { unique: true } );
+db.polling_detail.createIndex({timestamp: 1}, { expireAfterSeconds: 7 * 24 * 3600});
 db.counter.createIndex({name: 1,domain:1},{ unique: true });
 db.view.createIndex({"id": 1}, { unique: true } );
 db.view.createIndex({display:1,domain:1,project:1},{ unique: true });

go.mod

@@ -5,9 +5,7 @@ require (
 	github.com/emicklei/go-restful v2.12.0+incompatible
 	github.com/go-chassis/go-archaius v1.3.6-0.20201103103813-43dd1680ebfb
 	github.com/go-chassis/go-chassis/v2 v2.1.1-0.20201208062518-9c2e86bd7a6c
-
 	github.com/go-chassis/openlog v1.1.2
-	github.com/go-chassis/paas-lager v1.1.1
 	github.com/go-chassis/seclog v1.3.0
 	github.com/go-playground/universal-translator v0.17.0
 	github.com/go-playground/validator v9.31.0+incompatible

pkg/model/db_schema.go

@@ -17,6 +17,8 @@
 
 package model
 
+import "time"
+
 //LabelDoc is database struct to store labels
 type LabelDoc struct {
 	ID      string            `json:"id,omitempty" bson:"id,omitempty" yaml:"id,omitempty" swag:"string"`
@@ -59,14 +61,16 @@ type ViewDoc struct {
 type PollingDetail struct {
 	ID             string                 `json:"id,omitempty" yaml:"id,omitempty"`
 	SessionID      string                 `json:"session_id,omitempty" bson:"session_id," yaml:"session_id,omitempty"`
+	SessionGroup   string                 `json:"session_group,omitempty" bson:"session_group," yaml:"session_group,omitempty"`
 	Domain         string                 `json:"domain,omitempty" yaml:"domain,omitempty"`
 	PollingData    map[string]interface{} `json:"polling_data,omitempty" yaml:"polling_data,omitempty"`
+	Revision       string                 `json:"revision,omitempty" yaml:"revision,omitempty"`
 	IP             string                 `json:"ip,omitempty" yaml:"ip,omitempty"`
 	UserAgent      string                 `json:"user_agent,omitempty" bson:"user_agent," yaml:"user_agent,omitempty"`
 	URLPath        string                 `json:"url_path,omitempty"  bson:"url_path,"  yaml:"url_path,omitempty"`
 	ResponseBody   []*KVDoc               `json:"kv,omitempty"  bson:"kv,"  yaml:"kv,omitempty"`
-	ResponseHeader map[string][]string    `json:"response_header,omitempty"  bson:"response_header,"  yaml:"response_header,omitempty"`
 	ResponseCode   int                    `json:"response_code,omitempty"  bson:"response_code,"  yaml:"response_code,omitempty"`
+	Timestamp     time.Time              `json:"timestamp,omitempty" yaml:"timestamp,omitempty"`
 }
 
 // UpdateKVRequest is db struct, it contains kv update request params

server/handler/track_handler.go

@@ -29,6 +29,7 @@ import (
 	"github.com/go-chassis/openlog"
 	"net/http"
 	"strings"
+	"time"
 )
 
 //const of noop auth handler
@@ -66,13 +67,15 @@ func (h *TrackHandler) Handle(chain *handler.Chain, inv *invocation.Invocation,
 		data := &model.PollingDetail{}
 		data.URLPath = req.Request.Method + " " + req.Request.URL.Path
 		data.SessionID = sessionID
+		data.SessionGroup = req.HeaderParameter(v1.HeaderSessionGroup)
 		data.UserAgent = req.HeaderParameter(v1.HeaderUserAgent)
 		data.Domain = v1.ReadDomain(req.Request.Context())
 		data.IP = iputil.ClientIP(req.Request)
 		data.ResponseBody = req.Attribute(common.RespBodyContextKey).([]*model.KVDoc)
+		data.Timestamp = time.Now()
 		data.ResponseCode = ir.Status
 		if resp != nil {
-			data.ResponseHeader = resp.Header()
+			data.Revision = resp.Header().Get(common.HeaderRevision)
 		}
 		data.PollingData = map[string]interface{}{
 			"revision": revStr,

server/resource/v1/common.go

@@ -43,6 +43,7 @@ import (
 const (
 	HeaderUserAgent    = "User-Agent"
 	HeaderSessionID    = "X-Session-Id"
+	HeaderSessionGroup    = "X-Session-Group"
 	AttributeDomainKey = "domain"
 
 	FmtReadRequestError = "decode request body failed: %v"

server/service/mongo/session/session.go

@@ -222,19 +222,13 @@ func ensureKVLongPolling(session *mgo.Session) {
 	c := session.DB(DBName).C(CollectionPollingDetail)
 	err := c.Create(&mgo.CollectionInfo{Validator: bson.M{
 		"id":         bson.M{"$exists": true},
-		"params":     bson.M{"$exists": true},
+		"revision":   bson.M{"$exists": true},
 		"session_id": bson.M{"$exists": true},
 		"url_path":   bson.M{"$exists": true},
 	}})
-	wrapError(err, MsgDBExists)
-	err = c.EnsureIndex(mgo.Index{
-		Key:    []string{"id"},
-		Unique: true,
-	})
-	wrapError(err)
 	err = c.EnsureIndex(mgo.Index{
-		Key:    []string{"session_id", "domain"},
-		Unique: true,
+		Key:         []string{"timestamp"},
+		ExpireAfter: 7 * 24 * time.Hour,
 	})
 	wrapError(err)
 }

server/service/mongo/track/polling_detail_dao.go

@@ -29,9 +29,10 @@ import (
 )
 
 //CreateOrUpdate create a record or update exist record
+//If revision and session_id is exit: update else: insert
 func CreateOrUpdate(ctx context.Context, detail *model.PollingDetail) (*model.PollingDetail, error) {
 	collection := session.GetDB().Collection(session.CollectionPollingDetail)
-	queryFilter := bson.M{"domain": detail.Domain, "session_id": detail.SessionID}
+	queryFilter := bson.M{"revision": detail.Revision, "session_id": detail.SessionID}
 	res := collection.FindOne(ctx, queryFilter)
 	if res.Err() != nil {
 		if res.Err() == mongo.ErrNoDocuments {
@@ -67,6 +68,9 @@ func Get(ctx context.Context, detail *model.PollingDetail) ([]*model.PollingDeta
 	if detail.URLPath != "" {
 		queryFilter["url_path"] = detail.URLPath
 	}
+	if detail.Revision != "" {
+		queryFilter["revision"] = detail.Revision
+	}
 	cur, err := collection.Find(ctx, queryFilter)
 	if err != nil {
 		return nil, err