add value filter in ListKV API

pkg/common/common.go

@@ -27,6 +27,7 @@ const (
 	QueryParamRev          = "revision"
 	QueryParamMatch        = "match"
 	QueryParamKey          = "key"
+	QueryParamValue        = "value"
 	QueryParamLabel        = "label"
 	QueryParamStatus       = "status"
 	QueryParamOffset       = "offset"

pkg/model/db_schema.go

@@ -96,6 +96,7 @@ type ListKVRequest struct {
 	Project string            `json:"project,omitempty" yaml:"project,omitempty" validate:"min=1,max=256,commonName"`
 	Domain  string            `json:"domain,omitempty" yaml:"domain,omitempty" validate:"min=1,max=256,commonName"` //redundant
 	Key     string            `json:"key" yaml:"key" validate:"max=128,getKey"`
+	Value   string            `json:"value" yaml:"value" validate:"max=128"`
 	Labels  map[string]string `json:"labels,omitempty" yaml:"labels,omitempty" validate:"max=8,dive,keys,labelK,endkeys,labelV"` //redundant
 	Offset  int64             `validate:"min=0"`
 	Limit   int64             `validate:"min=0,max=100"`

server/datasource/etcd/kv/kv_cache.go

@@ -55,6 +55,10 @@ func NewKvCache() *Cache {
 	}
 }
 
+func GetKvCache() *Cache {
+	return kvCache
+}
+
 func Enabled() bool {
 	return kvCache != nil
 }
@@ -220,9 +224,9 @@ func (kc *Cache) DeleteKvDoc(kvID string) {
 	kc.kvDocCache.Delete(kvID)
 }
 
-func Search(ctx context.Context, req *CacheSearchReq) (*model.KVResponse, bool, error) {
+func Search(ctx context.Context, req *CacheSearchReq) (*model.KVResponse, bool) {
 	if !req.Opts.ExactLabels {
-		return nil, false, nil
+		return nil, false
 	}
 
 	openlog.Debug(fmt.Sprintf("using cache to search kv, domain %v, project %v, opts %+v", req.Domain, req.Project, *req.Opts))
@@ -233,7 +237,7 @@ func Search(ctx context.Context, req *CacheSearchReq) (*model.KVResponse, bool,
 	kvIds, ok := kvCache.LoadKvIDSet(cacheKey)
 	if !ok {
 		kvCache.StoreKvIDSet(cacheKey, IDSet{})
-		return result, true, nil
+		return result, true
 	}
 
 	var docs []*model.KVDoc
@@ -257,7 +261,7 @@ func Search(ctx context.Context, req *CacheSearchReq) (*model.KVResponse, bool,
 		}
 	}
 	result.Total = len(result.Data)
-	return result, true, nil
+	return result, true
 }
 
 func (kc *Cache) getKvFromEtcd(ctx context.Context, req *CacheSearchReq, kvIdsLeft []string) []*model.KVDoc {
@@ -304,6 +308,9 @@ func isMatch(req *CacheSearchReq, doc *model.KVDoc) bool {
 	if req.Regex != nil && !req.Regex.MatchString(doc.Key) {
 		return false
 	}
+	if req.Opts.Value != "" && !strings.Contains(doc.Value, req.Opts.Value) {
+		return false
+	}
 	return true
 }
 

server/datasource/etcd/kv/kv_dao.go

@@ -524,18 +524,15 @@ func (s *Dao) listData(ctx context.Context, project, domain string, options ...d
 	}
 
 	if Enabled() {
-		result, useCache, err := Search(ctx, &CacheSearchReq{
+		result, useCache := Search(ctx, &CacheSearchReq{
 			Domain:  domain,
 			Project: project,
 			Opts:    &opts,
 			Regex:   regex,
 		})
-		if useCache && err == nil {
+		if useCache {
 			return result, opts, nil
 		}
-		if useCache && err != nil {
-			openlog.Error("using cache to search kv failed: " + err.Error())
-		}
 	}
 
 	result, err := matchLabelsSearch(ctx, domain, project, regex, opts)
@@ -646,5 +643,8 @@ func filterMatch(doc *model.KVDoc, opts datasource.FindOptions, regex *regexp.Re
 	if opts.LabelFormat != "" && doc.LabelFormat != opts.LabelFormat {
 		return false
 	}
+	if opts.Value != "" && !strings.Contains(doc.Value, opts.Value) {
+		return false
+	}
 	return true
 }

server/datasource/mongo/kv/kv_dao.go

@@ -271,6 +271,9 @@ func findKV(ctx context.Context, domain string, project string, opts datasource.
 			filter["key"] = bson.M{"$regex": "^" + value + "$", "$options": "$i"}
 		}
 	}
+	if opts.Value != "" {
+		filter["value"] = bson.M{"$regex": opts.Value}
+	}
 	if len(opts.Labels) != 0 {
 		for k, v := range opts.Labels {
 			filter["labels."+k] = v

server/datasource/options.go

@@ -63,6 +63,7 @@ type FindOptions struct {
 	Depth       int
 	ID          string
 	Key         string
+	Value       string
 	Labels      map[string]string
 	LabelFormat string
 	ClearLabel  bool
@@ -115,6 +116,13 @@ func WithKey(key string) FindOption {
 	}
 }
 
+// WithValue find by value
+func WithValue(value string) FindOption {
+	return func(o *FindOptions) {
+		o.Value = value
+	}
+}
+
 // WithStatus enabled/disabled
 func WithStatus(status string) FindOption {
 	return func(o *FindOptions) {

server/resource/v1/kv_resource.go

@@ -175,6 +175,7 @@ func (r *KVResource) List(rctx *restful.Context) {
 		Project: rctx.ReadPathParameter(common.PathParameterProject),
 		Domain:  ReadDomain(rctx.Ctx),
 		Key:     rctx.ReadQueryParameter(common.QueryParamKey),
+		Value:   rctx.ReadQueryParameter(common.QueryParamValue),
 		Status:  rctx.ReadQueryParameter(common.QueryParamStatus),
 		Match:   getMatchPattern(rctx),
 	}

server/resource/v1/kv_resource_test.go

@@ -253,6 +253,38 @@ func TestKVResource_List(t *testing.T) {
 		assert.NoError(t, err)
 		assert.Equal(t, 3, len(result.Data))
 	})
+	t.Run("list kv by value, should return 1 kv", func(t *testing.T) {
+		r, _ := http.NewRequest("GET", "/v1/kv_test/kie/kv?value=aaa", nil)
+		r.Header.Set("Content-Type", "application/json")
+		kvr := &v1.KVResource{}
+		c, err := restfultest.New(kvr, nil)
+		assert.NoError(t, err)
+		resp := httptest.NewRecorder()
+		c.ServeHTTP(resp, r)
+		body, err := ioutil.ReadAll(resp.Body)
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusOK, resp.Code, string(body))
+		result := &model.KVResponse{}
+		err = json.Unmarshal(body, result)
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(result.Data))
+	})
+	t.Run("list kv by value, should return 1 kv", func(t *testing.T) {
+		r, _ := http.NewRequest("GET", "/v1/kv_test/kie/kv?value=AAA", nil)
+		r.Header.Set("Content-Type", "application/json")
+		kvr := &v1.KVResource{}
+		c, err := restfultest.New(kvr, nil)
+		assert.NoError(t, err)
+		resp := httptest.NewRecorder()
+		c.ServeHTTP(resp, r)
+		body, err := ioutil.ReadAll(resp.Body)
+		assert.NoError(t, err)
+		assert.Equal(t, http.StatusOK, resp.Code, string(body))
+		result := &model.KVResponse{}
+		err = json.Unmarshal(body, result)
+		assert.NoError(t, err)
+		assert.Equal(t, 0, len(result.Data))
+	})
 	var rev string
 	t.Run("list kv by service label, exact match,should return 2 kv", func(t *testing.T) {
 		r, _ := http.NewRequest("GET", "/v1/kv_test/kie/kv?label=service:utService&match=exact", nil)

server/service/kv/kv_svc.go

@@ -20,6 +20,7 @@ package kv
 import (
 	"context"
 	"crypto/sha256"
+	"errors"
 	"fmt"
 	"strings"
 	"time"
@@ -45,6 +46,7 @@ var listSema = concurrency.NewSemaphore(concurrency.DefaultConcurrency)
 func ListKV(ctx context.Context, request *model.ListKVRequest) (int64, *model.KVResponse, *errsvc.Error) {
 	opts := []datasource.FindOption{
 		datasource.WithKey(request.Key),
+		datasource.WithValue(request.Value),
 		datasource.WithLabels(request.Labels),
 		datasource.WithOffset(request.Offset),
 		datasource.WithLimit(request.Limit),
@@ -126,6 +128,9 @@ func Create(ctx context.Context, kv *model.KVDoc) (*model.KVDoc, *errsvc.Error)
 	kv, err = datasource.GetBroker().GetKVDao().Create(ctx, kv, datasource.WithSync(sync.FromContext(ctx)))
 	if err != nil {
 		openlog.Error(fmt.Sprintf("post err:%s", err.Error()))
+		if errors.Is(err, datasource.ErrKVAlreadyExists) {
+			err = config.NewError(config.ErrRecordAlreadyExists, datasource.ErrKVAlreadyExists.Error())
+		}
 		return nil, util.SvcErr(err)
 	}
 	err = datasource.GetBroker().GetHistoryDao().AddHistory(ctx, kv)