ETCD/Redis supports "BatchPut"

engine/factory/factory.go

@@ -57,7 +57,9 @@ func GetEngineFromCache(ctx context.Context, config types.Config, endpoint, ca,
 
 // RemoveEngineFromCache .
 func RemoveEngineFromCache(endpoint, ca, cert, key string) {
-	engineCache.Delete(getEngineCacheKey(endpoint, ca, cert, key))
+	cacheKey := getEngineCacheKey(endpoint, ca, cert, key)
+	log.Debugf(context.TODO(), "[RemoveEngineFromCache] remove %v, key %v", endpoint, cacheKey)
+	engineCache.Delete(cacheKey)
 }
 
 // GetEngine get engine
@@ -68,7 +70,9 @@ func GetEngine(ctx context.Context, config types.Config, nodename, endpoint, ca,
 
 	defer func() {
 		if err == nil && client != nil {
-			engineCache.Set(getEngineCacheKey(endpoint, ca, cert, key), client)
+			cacheKey := getEngineCacheKey(endpoint, ca, cert, key)
+			log.Debugf(ctx, "[GetEngine] store engine of %v in cache, key: %v", endpoint, cacheKey)
+			engineCache.Set(cacheKey, client)
 		}
 	}()
 

store/etcdv3/meta/etcd.go

@@ -240,6 +240,11 @@ func (e *ETCD) BatchUpdate(ctx context.Context, data map[string]string, opts ...
 	return e.batchUpdate(ctx, data, opts...)
 }
 
+// BatchPut .
+func (e *ETCD) BatchPut(ctx context.Context, data map[string]string, opts ...clientv3.OpOption) (*clientv3.TxnResponse, error) {
+	return e.batchPut(ctx, data, nil, opts...)
+}
+
 // isTTLChanged returns true if there is a lease with a different ttl bound to the key
 func (e *ETCD) isTTLChanged(ctx context.Context, key string, ttl int64) (bool, error) {
 	resp, err := e.GetOne(ctx, key)

store/etcdv3/meta/meta.go

@@ -30,6 +30,7 @@ type KV interface {
 	BatchCreate(ctx context.Context, data map[string]string, opts ...clientv3.OpOption) (*clientv3.TxnResponse, error)
 	BatchUpdate(ctx context.Context, data map[string]string, opts ...clientv3.OpOption) (*clientv3.TxnResponse, error)
 	BatchDelete(ctx context.Context, keys []string, opts ...clientv3.OpOption) (*clientv3.TxnResponse, error)
+	BatchPut(ctx context.Context, data map[string]string, opts ...clientv3.OpOption) (*clientv3.TxnResponse, error)
 
 	StartEphemeral(ctx context.Context, path string, heartbeat time.Duration) (<-chan struct{}, func(), error)
 	CreateLock(key string, ttl time.Duration) (lock.DistributedLock, error)

store/etcdv3/node.go

@@ -165,7 +165,7 @@ func (m *Mercury) UpdateNodes(ctx context.Context, nodes ...*types.Node) error {
 		enginefactory.RemoveEngineFromCache(node.Endpoint, node.Ca, node.Cert, node.Key)
 	}
 
-	resp, err := m.BatchUpdate(ctx, data)
+	resp, err := m.BatchPut(ctx, data)
 	if err != nil {
 		return err
 	}

store/etcdv3/node_test.go

@@ -207,7 +207,7 @@ func TestUpdateNode(t *testing.T) {
 			Key:      "hh",
 		},
 	}
-	assert.Error(t, m.UpdateNodes(ctx, fakeNode))
+	assert.NoError(t, m.UpdateNodes(ctx, fakeNode))
 	assert.NoError(t, m.UpdateNodes(ctx, node))
 	node.Available = false
 	assert.NoError(t, m.UpdateNodes(ctx, node))

store/redis/node.go

@@ -161,7 +161,7 @@ func (r *Rediaron) UpdateNodes(ctx context.Context, nodes ...*types.Node) error
 		addIfNotEmpty(fmt.Sprintf(nodeCertKey, node.Name), node.Cert)
 		addIfNotEmpty(fmt.Sprintf(nodeKeyKey, node.Name), node.Key)
 	}
-	return errors.WithStack(r.BatchUpdate(ctx, data))
+	return errors.WithStack(r.BatchPut(ctx, data))
 }
 
 // UpdateNodeResource update cpu and memory on a node, either add or subtract

store/redis/node_test.go

@@ -200,7 +200,7 @@ func (s *RediaronTestSuite) TestUpdateNode() {
 			Key:      "hh",
 		},
 	}
-	s.Error(s.rediaron.UpdateNodes(ctx, fakeNode))
+	s.NoError(s.rediaron.UpdateNodes(ctx, fakeNode))
 	s.NoError(s.rediaron.UpdateNodes(ctx, node))
 }
 

store/redis/rediaron.go

@@ -238,6 +238,28 @@ func (r *Rediaron) BatchCreate(ctx context.Context, data map[string]string) erro
 	return nil
 }
 
+// BatchPut is wrapper to adapt etcd batch replace
+func (r *Rediaron) BatchPut(ctx context.Context, data map[string]string) error {
+	replace := func(pipe redis.Pipeliner) error {
+		for key, value := range data {
+			pipe.Set(ctx, key, value, 0)
+		}
+		return nil
+	}
+
+	cmds, err := r.cli.TxPipelined(ctx, replace)
+	if err != nil {
+		return err
+	}
+
+	for _, cmd := range cmds {
+		if err := cmd.Err(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // BatchCreateAndDecr decr processing and add workload
 func (r *Rediaron) BatchCreateAndDecr(ctx context.Context, data map[string]string, decrKey string) (err error) {
 	batchCreateAndDecr := func(pipe redis.Pipeliner) error {