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

64.79

/server/service/mongo/kv/kv_service.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"
        "reflect"
        "time"

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

//Service operate data in mongodb
type Service struct {
        timeout time.Duration
}

//CreateOrUpdate will create or update a key value record
//it first check label exists or not, and create labels if labels is first posted.
//if label exists, then get its latest revision, and update current revision,
//save the current label and its all key values to history collection
//then check key exists or not, then create or update it
func (s *Service) CreateOrUpdate(ctx context.Context, kv *model.KVDoc) (*model.KVDoc, error) {
        ctx, _ = context.WithTimeout(ctx, session.Timeout)
        if kv.Domain == "" {
                return nil, session.ErrMissingDomain
        }

        //check whether the project has certain labels or not
        labelID, err := label.Exist(ctx, kv.Domain, kv.Project, kv.Labels)
        if err != nil {
                if err == session.ErrLabelNotExists {
                        var l *model.LabelDoc
                        l, err = label.CreateLabel(ctx, kv.Domain, kv.Labels, kv.Project)
                        if err != nil {
                                openlogging.Error("create label failed", openlogging.WithTags(openlogging.Tags{
                                        "k":      kv.Key,
                                        "domain": kv.Domain,
                                }))
                                return nil, err
                        }
                        labelID = l.ID
                } else {
                        return nil, err
                }
        }
        kv.LabelID = string(labelID)
        if kv.ValueType == "" {
                kv.ValueType = session.DefaultValueType
        }
        keyID, err := s.Exist(ctx, kv.Domain, kv.Key, kv.Project, service.WithLabelID(kv.LabelID))
        if err != nil {
                if err == service.ErrKeyNotExists {
                        kv, err := createKey(ctx, kv)
                        if err != nil {
                                return nil, err
                        }
                        kv.Domain = ""
                        kv.Project = ""
                        return kv, nil
                }
                return nil, err
        }
        kv.ID = keyID
        revision, err := updateKeyValue(ctx, kv)
        if err != nil {
                return nil, err
        }
        kv.Revision = revision
        kv.Domain = ""
        kv.Project = ""
        return kv, nil

}

//Exist supports you query by label map or labels id
func (s *Service) Exist(ctx context.Context, domain, key string, project string, options ...service.FindOption) (id.ID, error) {
        ctx, _ = context.WithTimeout(context.Background(), session.Timeout)
        opts := service.FindOptions{}
        for _, o := range options {
                o(&opts)
        }
        if opts.LabelID != "" {
                kvs, err := findKVByLabelID(ctx, domain, opts.LabelID, key, project)
                if err != nil {
                        return "", err
                }
                return kvs[0].ID, nil
        }
        kvs, err := s.FindKV(ctx, domain, project, service.WithExactLabels(), service.WithLabels(opts.Labels), service.WithKey(key))
        if err != nil {
                return "", err
        }
        if len(kvs) != 1 {
                return "", session.ErrTooMany
        }

        return kvs[0].Data[0].ID, nil

}

//Delete delete kv,If the labelID is "", query the collection kv to get it
//domain=tenant
//1.delete kv;2.add history
func (s *Service) Delete(kvID string, labelID string, domain string, project string) error {
        ctx, _ := context.WithTimeout(context.Background(), session.Timeout)
        if domain == "" {
                return session.ErrMissingDomain
        }
        if project == "" {
                return session.ErrMissingProject
        }
        hex, err := primitive.ObjectIDFromHex(kvID)
        if err != nil {
                return err
        }
        //if labelID == "",get labelID by kvID
        var kv *model.KVDoc
        if labelID == "" {
                kvArray, err := findOneKey(ctx, bson.M{"_id": hex, "project": project})
                if err != nil {
                        return err
                }
                if len(kvArray) > 0 {
                        kv = kvArray[0]
                        labelID = kv.LabelID
                }
        }
        //get Label and check labelID
        r, err := label.GetLatestLabel(ctx, labelID)
        if err != nil {
                if err == service.ErrRevisionNotExist {
                        openlogging.Warn(fmt.Sprintf("failed,kvID and labelID do not match"))
                        return session.ErrKvIDAndLabelIDNotMatch
                }
                return err
        }
        //delete kv
        err = deleteKV(ctx, hex, project)
        if err != nil {
                return err
        }
        kvs, err := findKeys(ctx, bson.M{"label_id": labelID, "project": project}, true)
        //Key may be empty When delete
        if err != nil && err != service.ErrKeyNotExists {
                return err
        }
        //Labels will not be empty when deleted
        revision, err := history.AddHistory(ctx, r, labelID, kvs)
        if err != nil {
                openlogging.Warn("add history failed ,", openlogging.WithTags(openlogging.Tags{
                        "kvID":    kvID,
                        "labelID": labelID,
                        "error":   err.Error(),
                }))
        } else {
                openlogging.Info("add history success,", openlogging.WithTags(openlogging.Tags{
                        "kvID":     kvID,
                        "labelID":  labelID,
                        "revision": revision,
                }))
        }
        return nil
}

//List get kv list by key and criteria
func (s *Service) List(ctx context.Context, domain, project, key string, labels map[string]string, limit, offset int) (*model.KVResponse, error) {
        opts := service.NewDefaultFindOpts()
        opts.Labels = labels
        opts.Key = key
        cur, err := findKV(ctx, domain, project, opts)
        if err != nil {
                return nil, err
        }
        defer cur.Close(ctx)
        result := &model.KVResponse{}
        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
                }
                clearPart(curKV)
                result.Data = append(result.Data, curKV)
        }
        if len(result.Data) == 0 {
                return nil, service.ErrKeyNotExists
        }
        return result, nil
}

//FindKV get kvs by key, labels
//because labels has a a lot of combination,
//you can use WithDepth(0) to return only one kv which's labels exactly match the criteria
func (s *Service) FindKV(ctx context.Context, domain string, project string, options ...service.FindOption) ([]*model.KVResponse, error) {
        opts := service.FindOptions{}
        for _, o := range options {
                o(&opts)
        }
        if opts.Timeout == 0 {
                opts.Timeout = session.DefaultTimeout
        }
        if domain == "" {
                return nil, session.ErrMissingDomain
        }
        if project == "" {
                return nil, session.ErrMissingProject
        }

        cur, err := findKV(ctx, domain, project, opts)
        if err != nil {
                return nil, err
        }
        defer cur.Close(ctx)

        kvResp := make([]*model.KVResponse, 0)
        if opts.Depth == 0 && opts.Key != "" {
                openlogging.Debug("find one key", openlogging.WithTags(
                        map[string]interface{}{
                                "key":    opts.Key,
                                "label":  opts.Labels,
                                "domain": domain,
                        },
                ))
                return cursorToOneKV(ctx, cur, opts.Labels)
        }
        openlogging.Debug("find more", openlogging.WithTags(openlogging.Tags{
                "depth":  opts.Depth,
                "k":      opts.Key,
                "labels": opts.Labels,
        }))
        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 (len(curKV.Labels) - len(opts.Labels)) > opts.Depth {
                        //because it is query by labels, so result can not be minus
                        //so many labels,then continue
                        openlogging.Debug("so deep, skip this key")
                        continue
                }
                openlogging.Debug(fmt.Sprintf("%v", curKV))
                var groupExist bool
                var labelGroup *model.KVResponse
                for _, labelGroup = range kvResp {
                        if reflect.DeepEqual(labelGroup.LabelDoc.Labels, curKV.Labels) {
                                groupExist = true
                                clearAll(curKV)
                                labelGroup.Data = append(labelGroup.Data, curKV)
                                break
                        }

                }
                if !groupExist {
                        labelGroup = &model.KVResponse{
                                LabelDoc: &model.LabelDocResponse{
                                        Labels:  curKV.Labels,
                                        LabelID: curKV.LabelID,
                                },
                                Data: []*model.KVDoc{curKV},
                        }
                        clearAll(curKV)
                        openlogging.Debug("add new label group")
                        kvResp = append(kvResp, labelGroup)
                }

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

}

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	"reflect"
25	"time"
26
27	"github.com/apache/servicecomb-kie/pkg/model"
28	"github.com/apache/servicecomb-kie/server/service"
29	"github.com/apache/servicecomb-kie/server/service/mongo/history"
30	"github.com/apache/servicecomb-kie/server/service/mongo/label"
31	"github.com/apache/servicecomb-kie/server/service/mongo/session"
32	"github.com/go-mesh/openlogging"
33	"go.mongodb.org/mongo-driver/bson"
34	"go.mongodb.org/mongo-driver/bson/primitive"
35	)
36
37	//Service operate data in mongodb
38	type Service struct {
39	timeout time.Duration
40	}
41
42	//CreateOrUpdate will create or update a key value record
43	//it first check label exists or not, and create labels if labels is first posted.
44	//if label exists, then get its latest revision, and update current revision,
45	//save the current label and its all key values to history collection
46	//then check key exists or not, then create or update it
47	func (s Service) CreateOrUpdate(ctx context.Context, kv model.KVDoc) (*model.KVDoc, error) {	6✔
48	ctx, _ = context.WithTimeout(ctx, session.Timeout)	6✔
49	if kv.Domain == "" {	6✔
50	return nil, session.ErrMissingDomain	×
51	}	×
52
53	//check whether the project has certain labels or not
54	labelID, err := label.Exist(ctx, kv.Domain, kv.Project, kv.Labels)	6✔
55	if err != nil {	9✔
56	if err == session.ErrLabelNotExists {	6✔
57	var l *model.LabelDoc	3✔
58	l, err = label.CreateLabel(ctx, kv.Domain, kv.Labels, kv.Project)	3✔
59	if err != nil {	3✔
60	openlogging.Error("create label failed", openlogging.WithTags(openlogging.Tags{	×
61	"k": kv.Key,	×
62	"domain": kv.Domain,	×
63	}))	×
64	return nil, err	×
65	}	×
66	labelID = l.ID	3✔
67	} else {	×
68	return nil, err	×
69	}	×
70	}
71	kv.LabelID = string(labelID)	6✔
72	if kv.ValueType == "" {	12✔
73	kv.ValueType = session.DefaultValueType	6✔
74	}	6✔
75	keyID, err := s.Exist(ctx, kv.Domain, kv.Key, kv.Project, service.WithLabelID(kv.LabelID))	6✔
76	if err != nil {	11✔
77	if err == service.ErrKeyNotExists {	10✔
78	kv, err := createKey(ctx, kv)	5✔
79	if err != nil {	5✔
80	return nil, err	×
81	}	×
82	kv.Domain = ""	5✔
83	kv.Project = ""	5✔
84	return kv, nil	5✔
85	}
86	return nil, err	×
87	}
88	kv.ID = keyID	1✔
89	revision, err := updateKeyValue(ctx, kv)	1✔
90	if err != nil {	1✔
91	return nil, err	×
92	}	×
93	kv.Revision = revision	1✔
94	kv.Domain = ""	1✔
95	kv.Project = ""	1✔
96	return kv, nil	1✔
97
98	}
99
100	//Exist supports you query by label map or labels id
101	func (s *Service) Exist(ctx context.Context, domain, key string, project string, options ...service.FindOption) (id.ID, error) {	8✔
102	ctx, _ = context.WithTimeout(context.Background(), session.Timeout)	8✔
103	opts := service.FindOptions{}	8✔
104	for _, o := range options {	16✔
105	o(&opts)	8✔
106	}	8✔
107	if opts.LabelID != "" {	14✔
108	kvs, err := findKVByLabelID(ctx, domain, opts.LabelID, key, project)	6✔
109	if err != nil {	11✔
110	return "", err	5✔
111	}	5✔
112	return kvs[0].ID, nil	1✔
113	}
114	kvs, err := s.FindKV(ctx, domain, project, service.WithExactLabels(), service.WithLabels(opts.Labels), service.WithKey(key))	2✔
115	if err != nil {	2✔
116	return "", err	×
117	}	×
118	if len(kvs) != 1 {	2✔
119	return "", session.ErrTooMany	×
120	}	×
121
122	return kvs[0].Data[0].ID, nil	2✔
123
124	}
125
126	//Delete delete kv,If the labelID is "", query the collection kv to get it
127	//domain=tenant
128	//1.delete kv;2.add history
129	func (s *Service) Delete(kvID string, labelID string, domain string, project string) error {	5✔
130	ctx, _ := context.WithTimeout(context.Background(), session.Timeout)	5✔
131	if domain == "" {	6✔
132	return session.ErrMissingDomain	1✔
133	}	1✔
134	if project == "" {	4✔
135	return session.ErrMissingProject	×
136	}	×
137	hex, err := primitive.ObjectIDFromHex(kvID)	4✔
138	if err != nil {	6✔
139	return err	2✔
140	}	2✔
141	//if labelID == "",get labelID by kvID
142	var kv *model.KVDoc	2✔
143	if labelID == "" {	3✔
144	kvArray, err := findOneKey(ctx, bson.M{"_id": hex, "project": project})	1✔
145	if err != nil {	1✔
146	return err	×
147	}	×
148	if len(kvArray) > 0 {	2✔
149	kv = kvArray[0]	1✔
150	labelID = kv.LabelID	1✔
151	}	1✔
152	}
153	//get Label and check labelID
154	r, err := label.GetLatestLabel(ctx, labelID)	2✔
155	if err != nil {	2✔
156	if err == service.ErrRevisionNotExist {	×
157	openlogging.Warn(fmt.Sprintf("failed,kvID and labelID do not match"))	×
158	return session.ErrKvIDAndLabelIDNotMatch	×
159	}	×
160	return err	×
161	}
162	//delete kv
163	err = deleteKV(ctx, hex, project)	2✔
164	if err != nil {	2✔
165	return err	×
166	}	×
167	kvs, err := findKeys(ctx, bson.M{"label_id": labelID, "project": project}, true)	2✔
168	//Key may be empty When delete	2✔
169	if err != nil && err != service.ErrKeyNotExists {	2✔
170	return err	×
171	}	×
172	//Labels will not be empty when deleted
173	revision, err := history.AddHistory(ctx, r, labelID, kvs)	2✔
174	if err != nil {	2✔
175	openlogging.Warn("add history failed ,", openlogging.WithTags(openlogging.Tags{	×
176	"kvID": kvID,	×
177	"labelID": labelID,	×
178	"error": err.Error(),	×
179	}))	×
180	} else {	2✔
181	openlogging.Info("add history success,", openlogging.WithTags(openlogging.Tags{	2✔
182	"kvID": kvID,	2✔
183	"labelID": labelID,	2✔
184	"revision": revision,	2✔
185	}))	2✔
186	}	2✔
187	return nil	2✔
188	}
189
190	//List get kv list by key and criteria
191	func (s Service) List(ctx context.Context, domain, project, key string, labels map[string]string, limit, offset int) (model.KVResponse, error) {	×
192	opts := service.NewDefaultFindOpts()	×
193	opts.Labels = labels	×
194	opts.Key = key	×
195	cur, err := findKV(ctx, domain, project, opts)	×
196	if err != nil {	×
197	return nil, err	×
198	}	×
199	defer cur.Close(ctx)	×
200	result := &model.KVResponse{}	×
201	for cur.Next(ctx) {	×
202	curKV := &model.KVDoc{}	×
203	if err := cur.Decode(curKV); err != nil {	×
204	openlogging.Error("decode to KVs error: " + err.Error())	×
205	return nil, err	×
206	}	×
207	clearPart(curKV)	×
208	result.Data = append(result.Data, curKV)	×
209	}
NEW 210	if len(result.Data) == 0 {	×
NEW 211	return nil, service.ErrKeyNotExists	×
NEW 212	}	×
UNCOV 213	return result, nil	×
214	}
215
216	//FindKV get kvs by key, labels
217	//because labels has a a lot of combination,
218	//you can use WithDepth(0) to return only one kv which's labels exactly match the criteria
219	func (s Service) FindKV(ctx context.Context, domain string, project string, options ...service.FindOption) ([]model.KVResponse, error) {	9✔
220	opts := service.FindOptions{}	9✔
221	for _, o := range options {	32✔
222	o(&opts)	23✔
223	}	23✔
224	if opts.Timeout == 0 {	18✔
225	opts.Timeout = session.DefaultTimeout	9✔
226	}	9✔
227	if domain == "" {	9✔
228	return nil, session.ErrMissingDomain	×
229	}	×
230	if project == "" {	9✔
231	return nil, session.ErrMissingProject	×
232	}	×
233
234	cur, err := findKV(ctx, domain, project, opts)	9✔
235	if err != nil {	9✔
236	return nil, err	×
237	}	×
238	defer cur.Close(ctx)	9✔
239		9✔
240	kvResp := make([]*model.KVResponse, 0)	9✔
241	if opts.Depth == 0 && opts.Key != "" {	15✔
242	openlogging.Debug("find one key", openlogging.WithTags(	6✔
243	map[string]interface{}{	6✔
244	"key": opts.Key,	6✔
245	"label": opts.Labels,	6✔
246	"domain": domain,	6✔
247	},	6✔
248	))	6✔
249	return cursorToOneKV(ctx, cur, opts.Labels)	6✔
250	}	6✔
251	openlogging.Debug("find more", openlogging.WithTags(openlogging.Tags{	3✔
252	"depth": opts.Depth,	3✔
253	"k": opts.Key,	3✔
254	"labels": opts.Labels,	3✔
255	}))	3✔
256	for cur.Next(ctx) {	11✔
257	curKV := &model.KVDoc{}	8✔
258		8✔
259	if err := cur.Decode(curKV); err != nil {	8✔
260	openlogging.Error("decode to KVs error: " + err.Error())	×
261	return nil, err	×
262	}	×
263	if (len(curKV.Labels) - len(opts.Labels)) > opts.Depth {	11✔
264	//because it is query by labels, so result can not be minus	3✔
265	//so many labels,then continue	3✔
266	openlogging.Debug("so deep, skip this key")	3✔
267	continue	3✔
268	}
269	openlogging.Debug(fmt.Sprintf("%v", curKV))	5✔
270	var groupExist bool	5✔
271	var labelGroup *model.KVResponse	5✔
272	for _, labelGroup = range kvResp {	8✔
273	if reflect.DeepEqual(labelGroup.LabelDoc.Labels, curKV.Labels) {	3✔
274	groupExist = true	×
275	clearAll(curKV)	×
276	labelGroup.Data = append(labelGroup.Data, curKV)	×
277	break	×
278	}
279
280	}
281	if !groupExist {	10✔
282	labelGroup = &model.KVResponse{	5✔
283	LabelDoc: &model.LabelDocResponse{	5✔
284	Labels: curKV.Labels,	5✔
285	LabelID: curKV.LabelID,	5✔
286	},	5✔
287	Data: []*model.KVDoc{curKV},	5✔
288	}	5✔
289	clearAll(curKV)	5✔
290	openlogging.Debug("add new label group")	5✔
291	kvResp = append(kvResp, labelGroup)	5✔
292	}	5✔
293
294	}
295	if len(kvResp) == 0 {	3✔
296	return nil, service.ErrKeyNotExists	×
297	}	×
298	return kvResp, nil	3✔
299
300	}

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