Replace user resources with ckecli resource set sub-command.

pkg/ckecli/cmd/resource_set.go

@@ -13,22 +13,16 @@ import (
 	k8sYaml "k8s.io/apimachinery/pkg/util/yaml"
 )
 
-func updateResource(ctx context.Context, data []byte) error {
-	key, err := cke.ParseResource(data)
-	if err != nil {
-		return err
-	}
-
-	return storage.SetResource(ctx, key, string(data))
-}
-
 var resourceSetCmd = &cobra.Command{
 	Use:   "set FILE",
 	Short: "register user-defined resources.",
 	Long: `Register user-defined resources.
 
 FILE should contain multiple Kubernetes resources in YAML or JSON format.
-If FILE is "-", then data is read from stdin.`,
+If FILE is "-", then data is read from stdin.
+
+If a resource with the same key as a registered resource is specified, the resource will be overwritten.
+If a resource exists in a registered resource but not in the specified resource, the resource will be deleted.`,
 
 	Args: cobra.ExactArgs(1),
 	RunE: func(cmd *cobra.Command, args []string) error {
@@ -43,21 +37,23 @@ If FILE is "-", then data is read from stdin.`,
 		}
 
 		well.Go(func(ctx context.Context) error {
+			newResources := make(map[string]string)
 			y := k8sYaml.NewYAMLReader(bufio.NewReader(r))
 			for {
 				data, err := y.Read()
 				if err == io.EOF {
-					return nil
+					break
 				}
 				if err != nil {
 					return err
 				}
-
-				err = updateResource(ctx, data)
+				key, err := cke.ParseResource(data)
 				if err != nil {
 					return err
 				}
+				newResources[key] = string(data)
 			}
+			return storage.ReplaceResources(ctx, newResources)
 		})
 		well.Stop()
 		return well.Wait()

storage.go

@@ -589,6 +589,53 @@ func (s Storage) DeleteResource(ctx context.Context, key string) error {
 	return err
 }
 
+// ReplaceResources replaces all user resources with new ones.
+func (s Storage) ReplaceResources(ctx context.Context, newResources map[string]string) error {
+RETRY:
+	current, err := s.GetAllResources(ctx)
+	if err != nil {
+		return err
+	}
+	currentResources := make(map[string]ResourceDefinition)
+	for _, r := range current {
+		currentResources[r.Key] = r
+	}
+
+	var ifOps []clientv3.Cmp
+	var thenOps []clientv3.Op
+
+	for key, value := range newResources {
+		cur := currentResources[key]
+		if value == string(cur.Definition) {
+			continue
+		}
+
+		ifOps = append(ifOps, clientv3.Compare(clientv3.ModRevision(KeyResourcePrefix+key), "=", cur.Revision))
+		thenOps = append(thenOps, clientv3.OpPut(KeyResourcePrefix+key, value))
+
+		if _, ok := currentResources[key]; ok {
+			delete(currentResources, key)
+		}
+	}
+	for key := range currentResources {
+		ifOps = append(ifOps, clientv3util.KeyMissing(KeyResourcePrefix+key))
+		thenOps = append(thenOps, clientv3.OpDelete(KeyResourcePrefix+key))
+	}
+
+	txnResp, err := s.Txn(ctx).
+		If(ifOps...).
+		Then(thenOps...).
+		Commit()
+
+	if err != nil {
+		return err
+	}
+	if !txnResp.Succeeded {
+		goto RETRY
+	}
+	return nil
+}
+
 // IsSabakanDisabled returns true if sabakan integration is disabled.
 func (s Storage) IsSabakanDisabled(ctx context.Context) (bool, error) {
 	resp, err := s.Get(ctx, KeySabakanDisabled)

storage_test.go

@@ -491,6 +491,23 @@ func testStorageResource(t *testing.T) {
 		t.Error(`err != ErrNotFound,`, err)
 	}
 
+	input := map[string]string{
+		"ServiceAccount/foo/sa1": "test",      // will not be changed
+		"ConfigMap/foo/conf1":    "overwrite", // will be overwritten
+		"Pod/foo/pod2":           "new",       // pod2 will be added, pod1 will be deleted
+	}
+	err = storage.ReplaceResources(ctx, input)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	resources, err = storage.GetAllResources(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !cmp.Equal(input, resources) {
+		t.Error("unexpected resources", cmp.Diff(input, resources))
+	}
 }
 
 func testStorageSabakan(t *testing.T) {