apache / servicecomb-kie / 214

Committed 28 Dec 2019 08:06AM UTC coverage: 53.976% (+7.3%) from 46.679%

Build # 214

Build Type

Pull #58

travis-ci

Committed by

web-flow

Commit Message

SCB-1549 peer to peer event notification

Pull Request Pull Request #58: SCB-1549 peer to peer event notification

Run Details

174 of 307 new or added lines in 10 files covered. (56.68%)

5 existing lines in 4 files now uncovered.

733 of 1358 relevant lines covered (53.98%)

1.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

66.45

/server/service/mongo/kv/kv_dao.go

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package kv

import (
        "context"
        "fmt"
        "github.com/apache/servicecomb-kie/server/id"
        "github.com/apache/servicecomb-kie/server/service"
        "github.com/apache/servicecomb-kie/server/service/mongo/label"
        "github.com/apache/servicecomb-kie/server/service/mongo/session"

        "github.com/apache/servicecomb-kie/pkg/model"
        "github.com/apache/servicecomb-kie/server/service/mongo/history"
        "github.com/go-mesh/openlogging"
        "go.mongodb.org/mongo-driver/bson"
        "go.mongodb.org/mongo-driver/bson/primitive"
        "go.mongodb.org/mongo-driver/mongo"
)

//createKey get latest revision from history
//and increase revision of label
//and insert key
func createKey(ctx context.Context, kv *model.KVDoc) (*model.KVDoc, error) {
        r, err := label.GetLatestLabel(ctx, kv.LabelID)
        if err != nil {
                if err != service.ErrRevisionNotExist {
                        openlogging.Error(fmt.Sprintf("get latest [%s][%s] in [%s],err: %s",
                                kv.Key, kv.Labels, kv.Domain, err.Error()))
                        return nil, err
                }
                //the first time labels is created, at this time, labels has no revision yet
                //after first key created, labels got revision 1
                r = &model.LabelRevisionDoc{Revision: 0}
        }
        if r != nil {
                r.Revision = r.Revision + 1
        }
        collection := session.GetDB().Collection(session.CollectionKV)
        res, err := collection.InsertOne(ctx, kv)
        if err != nil {
                return nil, err
        }
        objectID, _ := res.InsertedID.(primitive.ObjectID)
        kv.ID = id.ID(objectID.Hex())
        kvs, err := findKeys(ctx, bson.M{"label_id": kv.LabelID}, true)
        //Key may be empty When delete
        if err != nil && err != service.ErrKeyNotExists {
                return nil, err
        }
        revision, err := history.GetAndAddHistory(ctx, kv.LabelID, kv.Labels, kvs, kv.Domain, kv.Project)
        if err != nil {
                openlogging.Warn(
                        fmt.Sprintf("can not updateKeyValue version for [%s] [%s] in [%s]",
                                kv.Key, kv.Labels, kv.Domain))
        }
        openlogging.Debug(fmt.Sprintf("create %s with labels %s value [%s]", kv.Key, kv.Labels, kv.Value))
        kv.Revision = revision
        return kv, nil

}

//updateKeyValue get latest revision from history
//and increase revision of label
//and updateKeyValue and them add new revision
func updateKeyValue(ctx context.Context, kv *model.KVDoc) (int, error) {
        collection := session.GetDB().Collection(session.CollectionKV)
        ur, err := collection.UpdateOne(ctx, bson.M{"key": kv.Key, "label_id": kv.LabelID}, bson.D{
                {"$set", bson.D{
                        {"value", kv.Value},
                        {"checker", kv.Checker},
                }},
        })
        if err != nil {
                return 0, err
        }
        openlogging.Debug(
                fmt.Sprintf("updateKeyValue %s with labels %s value [%s] %d ",
                        kv.Key, kv.Labels, kv.Value, ur.ModifiedCount))
        kvs, err := findKeys(ctx, bson.M{"label_id": kv.LabelID}, true)
        //Key may be empty When delete
        if err != nil && err != service.ErrKeyNotExists {
                return 0, err
        }
        revision, err := history.GetAndAddHistory(ctx, kv.LabelID, kv.Labels, kvs, kv.Domain, kv.Project)
        if err != nil {
                openlogging.Warn(
                        fmt.Sprintf("can not label revision for [%s] [%s] in [%s],err: %s",
                                kv.Key, kv.Labels, kv.Domain, err))
        }
        openlogging.Debug(
                fmt.Sprintf("add history %s with labels %s value [%s] %d ",
                        kv.Key, kv.Labels, kv.Value, ur.ModifiedCount))
        return revision, nil

}

func findKV(ctx context.Context, domain string, project string, opts service.FindOptions) (*mongo.Cursor, error) {
        collection := session.GetDB().Collection(session.CollectionKV)
        ctx, _ = context.WithTimeout(ctx, opts.Timeout)
        filter := bson.M{"domain": domain, "project": project}
        if opts.Key != "" {
                filter["key"] = opts.Key
        }
        if len(opts.Labels) != 0 {
                for k, v := range opts.Labels {
                        filter["labels."+k] = v
                }
        } else {
                filter["labels"] = ""
        }

        cur, err := collection.Find(ctx, filter)
        if err != nil {
                if err.Error() == context.DeadlineExceeded.Error() {
                        openlogging.Error("find kv failed, deadline exceeded", openlogging.WithTags(openlogging.Tags{
                                "timeout": opts.Timeout,
                        }))
                        return nil, fmt.Errorf("can not find kv in %s", opts.Timeout)
                }
                return nil, err
        }
        return cur, err
}
func findOneKey(ctx context.Context, filter bson.M) ([]*model.KVDoc, error) {
        collection := session.GetDB().Collection(session.CollectionKV)
        sr := collection.FindOne(ctx, filter)
        if sr.Err() != nil {
                return nil, sr.Err()
        }
        curKV := &model.KVDoc{}
        err := sr.Decode(curKV)
        if err != nil {
                if err == mongo.ErrNoDocuments {
                        return nil, service.ErrKeyNotExists
                }
                openlogging.Error("decode error: " + err.Error())
                return nil, err
        }
        return []*model.KVDoc{curKV}, nil
}

//deleteKV by kvID
func deleteKV(ctx context.Context, hexID primitive.ObjectID, project string) error {
        collection := session.GetDB().Collection(session.CollectionKV)
        dr, err := collection.DeleteOne(ctx, bson.M{"_id": hexID, "project": project})
        //check error and delete number
        if err != nil {
                openlogging.Error(fmt.Sprintf("delete [%s] failed : [%s]", hexID, err))
                return err
        }
        if dr.DeletedCount != 1 {
                openlogging.Warn(fmt.Sprintf("Failed,May have been deleted,kvID=%s", hexID))
        } else {
                openlogging.Info(fmt.Sprintf("delete success,kvID=%s", hexID))
        }
        return err
}
func findKeys(ctx context.Context, filter bson.M, withoutLabel bool) ([]*model.KVDoc, error) {
        collection := session.GetDB().Collection(session.CollectionKV)
        cur, err := collection.Find(ctx, filter)
        if err != nil {
                if err.Error() == context.DeadlineExceeded.Error() {
                        openlogging.Error("find kvs failed, dead line exceeded", openlogging.WithTags(openlogging.Tags{
                                "timeout": session.Timeout,
                        }))
                        return nil, fmt.Errorf("can not find keys due to timout")
                }
                return nil, err
        }
        defer cur.Close(ctx)
        if cur.Err() != nil {
                return nil, err
        }
        kvs := make([]*model.KVDoc, 0)
        for cur.Next(ctx) {
                curKV := &model.KVDoc{}
                if err := cur.Decode(curKV); err != nil {
                        openlogging.Error("decode to KVs error: " + err.Error())
                        return nil, err
                }
                if withoutLabel {
                        curKV.Labels = nil
                }
                kvs = append(kvs, curKV)

        }
        if len(kvs) == 0 {
                return nil, service.ErrKeyNotExists
        }
        return kvs, nil
}

//findKVByLabelID get kvs by key and label id
//key can be empty, then it will return all key values
//if key is given, will return 0-1 key value
func findKVByLabelID(ctx context.Context, domain, labelID, key string, project string) ([]*model.KVDoc, error) {
        filter := bson.M{"label_id": labelID, "domain": domain, "project": project}
        if key != "" {
                filter["key"] = key
                return findOneKey(ctx, filter)
        }
        return findKeys(ctx, filter, true)

}

1	/*
2	* Licensed to the Apache Software Foundation (ASF) under one or more
3	* contributor license agreements. See the NOTICE file distributed with
4	* this work for additional information regarding copyright ownership.
5	* The ASF licenses this file to You under the Apache License, Version 2.0
6	* (the "License"); you may not use this file except in compliance with
7	* the License. You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17
18	package kv
19
20	import (
21	"context"
22	"fmt"
23	"github.com/apache/servicecomb-kie/server/id"
24	"github.com/apache/servicecomb-kie/server/service"
25	"github.com/apache/servicecomb-kie/server/service/mongo/label"
26	"github.com/apache/servicecomb-kie/server/service/mongo/session"
27
28	"github.com/apache/servicecomb-kie/pkg/model"
29	"github.com/apache/servicecomb-kie/server/service/mongo/history"
30	"github.com/go-mesh/openlogging"
31	"go.mongodb.org/mongo-driver/bson"
32	"go.mongodb.org/mongo-driver/bson/primitive"
33	"go.mongodb.org/mongo-driver/mongo"
34	)
35
36	//createKey get latest revision from history
37	//and increase revision of label
38	//and insert key
39	func createKey(ctx context.Context, kv model.KVDoc) (model.KVDoc, error) {	5✔
40	r, err := label.GetLatestLabel(ctx, kv.LabelID)	5✔
41	if err != nil {	8✔
42	if err != service.ErrRevisionNotExist {	3✔
43	openlogging.Error(fmt.Sprintf("get latest [%s][%s] in [%s],err: %s",	×
44	kv.Key, kv.Labels, kv.Domain, err.Error()))	×
45	return nil, err	×
46	}	×
47	//the first time labels is created, at this time, labels has no revision yet
48	//after first key created, labels got revision 1
49	r = &model.LabelRevisionDoc{Revision: 0}	3✔
50	}
51	if r != nil {	10✔
52	r.Revision = r.Revision + 1	5✔
53	}	5✔
54	collection := session.GetDB().Collection(session.CollectionKV)	5✔
55	res, err := collection.InsertOne(ctx, kv)	5✔
56	if err != nil {	5✔
57	return nil, err	×
58	}	×
59	objectID, _ := res.InsertedID.(primitive.ObjectID)	5✔
60	kv.ID = id.ID(objectID.Hex())	5✔
61	kvs, err := findKeys(ctx, bson.M{"label_id": kv.LabelID}, true)	5✔
62	//Key may be empty When delete	5✔
63	if err != nil && err != service.ErrKeyNotExists {	5✔
64	return nil, err	×
65	}	×
66	revision, err := history.GetAndAddHistory(ctx, kv.LabelID, kv.Labels, kvs, kv.Domain, kv.Project)	5✔
67	if err != nil {	5✔
68	openlogging.Warn(	×
69	fmt.Sprintf("can not updateKeyValue version for [%s] [%s] in [%s]",	×
70	kv.Key, kv.Labels, kv.Domain))	×
71	}	×
72	openlogging.Debug(fmt.Sprintf("create %s with labels %s value [%s]", kv.Key, kv.Labels, kv.Value))	5✔
73	kv.Revision = revision	5✔
74	return kv, nil	5✔
75
76	}
77
78	//updateKeyValue get latest revision from history
79	//and increase revision of label
80	//and updateKeyValue and them add new revision
81	func updateKeyValue(ctx context.Context, kv *model.KVDoc) (int, error) {	1✔
82	collection := session.GetDB().Collection(session.CollectionKV)	1✔
83	ur, err := collection.UpdateOne(ctx, bson.M{"key": kv.Key, "label_id": kv.LabelID}, bson.D{	1✔
84	{"$set", bson.D{	1✔
85	{"value", kv.Value},	1✔
86	{"checker", kv.Checker},	1✔
87	}},	1✔
88	})	1✔
89	if err != nil {	1✔
90	return 0, err	×
91	}	×
92	openlogging.Debug(	1✔
93	fmt.Sprintf("updateKeyValue %s with labels %s value [%s] %d ",	1✔
94	kv.Key, kv.Labels, kv.Value, ur.ModifiedCount))	1✔
95	kvs, err := findKeys(ctx, bson.M{"label_id": kv.LabelID}, true)	1✔
96	//Key may be empty When delete	1✔
97	if err != nil && err != service.ErrKeyNotExists {	1✔
98	return 0, err	×
99	}	×
100	revision, err := history.GetAndAddHistory(ctx, kv.LabelID, kv.Labels, kvs, kv.Domain, kv.Project)	1✔
101	if err != nil {	1✔
102	openlogging.Warn(	×
103	fmt.Sprintf("can not label revision for [%s] [%s] in [%s],err: %s",	×
104	kv.Key, kv.Labels, kv.Domain, err))	×
105	}	×
106	openlogging.Debug(	1✔
107	fmt.Sprintf("add history %s with labels %s value [%s] %d ",	1✔
108	kv.Key, kv.Labels, kv.Value, ur.ModifiedCount))	1✔
109	return revision, nil	1✔
110
111	}
112
113	func findKV(ctx context.Context, domain string, project string, opts service.FindOptions) (*mongo.Cursor, error) {	9✔
114	collection := session.GetDB().Collection(session.CollectionKV)	9✔
115	ctx, _ = context.WithTimeout(ctx, opts.Timeout)	9✔
116	filter := bson.M{"domain": domain, "project": project}	9✔
117	if opts.Key != "" {	16✔
118	filter["key"] = opts.Key	7✔
119	}	7✔
120	if len(opts.Labels) != 0 {	18✔
121	for k, v := range opts.Labels {	21✔
122	filter["labels."+k] = v	12✔
123	}	12✔
NEW 124	} else {	×
NEW 125	filter["labels"] = ""	×
UNCOV 126	}	×
127
128	cur, err := collection.Find(ctx, filter)	9✔
129	if err != nil {	9✔
130	if err.Error() == context.DeadlineExceeded.Error() {	×
131	openlogging.Error("find kv failed, deadline exceeded", openlogging.WithTags(openlogging.Tags{	×
132	"timeout": opts.Timeout,	×
133	}))	×
134	return nil, fmt.Errorf("can not find kv in %s", opts.Timeout)	×
135	}	×
136	return nil, err	×
137	}
138	return cur, err	9✔
139	}
140	func findOneKey(ctx context.Context, filter bson.M) ([]*model.KVDoc, error) {	7✔
141	collection := session.GetDB().Collection(session.CollectionKV)	7✔
142	sr := collection.FindOne(ctx, filter)	7✔
143	if sr.Err() != nil {	7✔
144	return nil, sr.Err()	×
145	}	×
146	curKV := &model.KVDoc{}	7✔
147	err := sr.Decode(curKV)	7✔
148	if err != nil {	12✔
149	if err == mongo.ErrNoDocuments {	10✔
150	return nil, service.ErrKeyNotExists	5✔
151	}	5✔
152	openlogging.Error("decode error: " + err.Error())	×
153	return nil, err	×
154	}
155	return []*model.KVDoc{curKV}, nil	2✔
156	}
157
158	//deleteKV by kvID
159	func deleteKV(ctx context.Context, hexID primitive.ObjectID, project string) error {	2✔
160	collection := session.GetDB().Collection(session.CollectionKV)	2✔
161	dr, err := collection.DeleteOne(ctx, bson.M{"_id": hexID, "project": project})	2✔
162	//check error and delete number	2✔
163	if err != nil {	2✔
164	openlogging.Error(fmt.Sprintf("delete [%s] failed : [%s]", hexID, err))	×
165	return err	×
166	}	×
167	if dr.DeletedCount != 1 {	2✔
168	openlogging.Warn(fmt.Sprintf("Failed,May have been deleted,kvID=%s", hexID))	×
169	} else {	2✔
170	openlogging.Info(fmt.Sprintf("delete success,kvID=%s", hexID))	2✔
171	}	2✔
172	return err	2✔
173	}
174	func findKeys(ctx context.Context, filter bson.M, withoutLabel bool) ([]*model.KVDoc, error) {	8✔
175	collection := session.GetDB().Collection(session.CollectionKV)	8✔
176	cur, err := collection.Find(ctx, filter)	8✔
177	if err != nil {	8✔
178	if err.Error() == context.DeadlineExceeded.Error() {	×
179	openlogging.Error("find kvs failed, dead line exceeded", openlogging.WithTags(openlogging.Tags{	×
180	"timeout": session.Timeout,	×
181	}))	×
182	return nil, fmt.Errorf("can not find keys due to timout")	×
183	}	×
184	return nil, err	×
185	}
186	defer cur.Close(ctx)	8✔
187	if cur.Err() != nil {	8✔
188	return nil, err	×
189	}	×
190	kvs := make([]*model.KVDoc, 0)	8✔
191	for cur.Next(ctx) {	14✔
192	curKV := &model.KVDoc{}	6✔
193	if err := cur.Decode(curKV); err != nil {	6✔
194	openlogging.Error("decode to KVs error: " + err.Error())	×
195	return nil, err	×
196	}	×
197	if withoutLabel {	12✔
198	curKV.Labels = nil	6✔
199	}	6✔
200	kvs = append(kvs, curKV)	6✔
201
202	}
203	if len(kvs) == 0 {	10✔
204	return nil, service.ErrKeyNotExists	2✔
205	}	2✔
206	return kvs, nil	6✔
207	}
208
209	//findKVByLabelID get kvs by key and label id
210	//key can be empty, then it will return all key values
211	//if key is given, will return 0-1 key value
212	func findKVByLabelID(ctx context.Context, domain, labelID, key string, project string) ([]*model.KVDoc, error) {	6✔
213	filter := bson.M{"label_id": labelID, "domain": domain, "project": project}	6✔
214	if key != "" {	12✔
215	filter["key"] = key	6✔
216	return findOneKey(ctx, filter)	6✔
217	}	6✔
218	return findKeys(ctx, filter, true)	×
219
220	}

apache / servicecomb-kie / 214

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous