bugfix: query index by clude only get last record (#401)

* support time limit for m3db query

* bugfix: query index by clude only get last record

src/modules/transfer/backend/m3db/convert.go

@@ -88,7 +88,8 @@ func tagsIndexTagkvResp(tags *consolidators.CompleteTagsResult) *dataobj.IndexTa
 	return ret
 }
 
-func xcludeResp(iter ident.TagIterator) (ret dataobj.XcludeResp) {
+func xcludeResp(iter ident.TagIterator) *dataobj.XcludeResp {
+	ret := &dataobj.XcludeResp{}
 	tags := map[string]string{}
 	for iter.Next() {
 		tag := iter.Current()

src/modules/transfer/backend/m3db/m3db.go

@@ -221,18 +221,18 @@ func (p *Client) queryIndexByClude(session client.Session, input dataobj.CludeRe
 	}
 
 	// group by endpoint-metric
-	respMap := make(map[string]dataobj.XcludeResp)
+	respMap := make(map[string]*dataobj.XcludeResp)
 	for iter.Next() {
 		_, _, tagIter := iter.Current()
 
 		resp := xcludeResp(tagIter)
-		key := fmt.Sprintf("%s-%s", resp.Endpoint, resp.Metric)
-		if v, ok := respMap[key]; ok {
-			if len(resp.Tags) > 0 {
+		if len(resp.Tags) > 0 && len(resp.Tags[0]) > 0 {
+			key := fmt.Sprintf("%s-%s", resp.Endpoint, resp.Metric)
+			if v, ok := respMap[key]; ok {
 				v.Tags = append(v.Tags, resp.Tags[0])
+			} else {
+				respMap[key] = resp
 			}
-		} else {
-			respMap[key] = resp
 		}
 	}
 
@@ -243,7 +243,7 @@ func (p *Client) queryIndexByClude(session client.Session, input dataobj.CludeRe
 
 	resp := make([]dataobj.XcludeResp, 0, len(respMap))
 	for _, v := range respMap {
-		resp = append(resp, v)
+		resp = append(resp, *v)
 	}
 
 	return resp

src/modules/transfer/backend/m3db/query.go

@@ -184,7 +184,7 @@ func (cfg M3dbSection) queryIndexByCludeOptions(input dataobj.CludeRecv) (index.
 	if len(q) == 0 {
 		query = index.Query{idx.NewAllQuery()}
 	} else {
-		query = index.Query{idx.NewDisjunctionQuery(q...)}
+		query = index.Query{idx.NewConjunctionQuery(q...)}
 	}
 
 	return query, index.QueryOptions{
@@ -268,12 +268,8 @@ func includeTagsQuery2(in []dataobj.TagPair) idx.Query {
 func excludeTagsQuery(in []*dataobj.TagPair) idx.Query {
 	q := []idx.Query{}
 	for _, kvs := range in {
-		q1 := []idx.Query{}
 		for _, v := range kvs.Values {
-			q1 = append(q1, idx.NewNegationQuery(idx.NewTermQuery([]byte(kvs.Key), []byte(v))))
-		}
-		if len(q1) > 0 {
-			q = append(q, idx.NewConjunctionQuery(q1...))
+			q = append(q, idx.NewNegationQuery(idx.NewTermQuery([]byte(kvs.Key), []byte(v))))
 		}
 	}
 

src/modules/transfer/backend/m3db/test/06-query-index-by-clude.sh

@@ -1,5 +1,6 @@
 #!/bin/bash
 
+# data:[{Nid: Endpoint:10.86.76.13 Metric:disk.bytes.used.percent Tags:[mount=/tmp] Step:0 Dstype:}]
 
 # type CludeRecv struct {
 # 	Endpoints []string   `json:"endpoints"`
@@ -12,8 +13,8 @@
 curl -X POST  \
 	http://localhost:8008/api/index/counter/clude \
 -d '[{
-  "endpoints": [],
-  "metric": "test",
-  "include": [],
-  "exclude": [{"tagk":"city", "tagv": ["bjo"]}]
+  "endpoints": ["10.86.76.13"],
+  "metric": "disk.bytes.used.percent",
+  "exclude": [{"tagk":"mount", "tagv": ["/boot"]}],
+  "include": [{"tagk":"mount", "tagv": ["/", "/home"]}]
 }]' | jq .