Repository: openyurtio/openyurt
Pull request #258 (2021-04-13 09:16:22)

bugfix: list runtimeclass and csidriver from cache error when cloud-edge network disconnected
<!-- 
Please make sure you have read and understood the contributing guidelines;
https://github.com/openyurtio/openyurt/blob/master/CONTRIBUTING.md -->

### Ⅰ. Describe what this PR does
- background:
1. At present, we use `component/resource/namespace/name` as file path to store response objects from kube-apiserver on local disk.
2.  If response objects from kube-apiserver are empty(like kubelet list runtimeclass, csidriver), cache info will not be created on local disk.
3. when cloud-edge network becomes disconnected, kubelet will list runtimeclass, csidriver from local disk cache, and yurthub return 404 to kubele. so kubelet will re-list runtimeclass and csidriver frequently.

- solution:
 1. if response objects from kube-apiserver are empty(like kubelet list runtimeclass, csidriver), yurthub can create key info `component/resource` in local disk with `nil` data. 
 2. so when kubelet list runtimeclass, csidriver from local disk cache, yurthub will return empty list to kubelet.
 3. and when pods cache data on local disk is deleted by accident,  yurthub will return 404 error(not empty list) to kubelet in order to prevent killing pods on node.

### Ⅱ. Does this pull request fix one issue?
<!--If so, add "fixes #xxxx" below in the next line, for example, fixes #15. Otherwise, add "NONE" -->


### Ⅲ. List the added test cases (unit test/integration test) if any, please explain if no tests are needed.
=== RUN   TestCacheResponseForList/cache_response_for_list_nodes_with_fieldselector
=== RUN   TestCacheResponseForList/cache_response_for_list_runtimeclasses_with_no_objects
=== RUN   TestCreate/create_dir_key_with_no_data
=== RUN   TestCreate/create_dir_key_that_already_exists

### Ⅳ. Describe how to verify it
make all
and make sure when cloud-edge networ disconnected, kubelet list runtimeclass and no error has happened.

### Ⅴ. Special notes for reviews





Patch:
diff --git a/pkg/yurthub/cachemanager/cache_manager.go b/pkg/yurthub/cachemanager/cache_manager.go
index b1052d93..c4796be4 100644
--- a/pkg/yurthub/cachemanager/cache_manager.go
+++ b/pkg/yurthub/cachemanager/cache_manager.go
@@ -370,6 +370,15 @@ func (cm *cacheManager) saveListObject(ctx context.Context, info *apirequest.Req
 	accessor := meta.NewAccessor()
 
 	comp, _ := util.ClientComponentFrom(ctx)
+	// even if no objects in cloud cluster, we need to
+	// make up a storage that represents the no resources
+	// in local disk, so when cloud-edge network disconnected,
+	// yurthub can return empty objects instead of 404 code(not found)
+	if len(items) == 0 {
+		key, _ := util.KeyFunc(comp, info.Resource, info.Namespace, "")
+		return cm.storage.Create(key, nil)
+	}
+
 	var errs []error
 	for i := range items {
 		name, err := accessor.Name(items[i])
diff --git a/pkg/yurthub/cachemanager/fake_storage_wrapper.go b/pkg/yurthub/cachemanager/fake_storage_wrapper.go
index d949ab13..fedf425c 100644
--- a/pkg/yurthub/cachemanager/fake_storage_wrapper.go
+++ b/pkg/yurthub/cachemanager/fake_storage_wrapper.go
@@ -76,7 +76,7 @@ func (fsw *fakeStorageWrapper) ListKeys(key string) ([]string, error) {
 func (fsw *fakeStorageWrapper) List(key string) ([]runtime.Object, error) {
 	objs := make([]runtime.Object, 0, len(fsw.data))
 	for k, obj := range fsw.data {
-		if strings.HasPrefix(k, key) {
+		if strings.HasPrefix(k, key) && obj != nil {
 			objs = append(objs, obj)
 		}
 	}
diff --git a/pkg/yurthub/cachemanager/storage_wrapper.go b/pkg/yurthub/cachemanager/storage_wrapper.go
index fdeb9c63..15cf6d58 100644
--- a/pkg/yurthub/cachemanager/storage_wrapper.go
+++ b/pkg/yurthub/cachemanager/storage_wrapper.go
@@ -59,18 +59,23 @@ func NewStorageWrapper(storage storage.Store) StorageWrapper {
 }
 
 // Create store runtime object into backend storage
+// if obj is nil, the storage used to represent the key
+// will be created. for example: for disk storage,
+// a directory that indicates the key will be created.
 func (sw *storageWrapper) Create(key string, obj runtime.Object) error {
 	var buf bytes.Buffer
-	if err := sw.backendSerializer.Encode(obj, &buf); err != nil {
-		klog.Errorf("failed to encode object in create for %s, %v", key, err)
-		return err
+	if obj != nil {
+		if err := sw.backendSerializer.Encode(obj, &buf); err != nil {
+			klog.Errorf("failed to encode object in create for %s, %v", key, err)
+			return err
+		}
 	}
 
 	if err := sw.store.Create(key, buf.Bytes()); err != nil {
 		return err
 	}
 
-	if isCacheKey(key) {
+	if obj != nil && isCacheKey(key) {
 		sw.Lock()
 		sw.cache[key] = obj
 		sw.Unlock()
@@ -141,6 +146,14 @@ func (sw *storageWrapper) List(key string) ([]runtime.Object, error) {
 		klog.Errorf("could not list objects for %s, %v", key, err)
 		return nil, err
 	} else if len(bb) == 0 {
+		if isPodKey(key) {
+			// because at least there will be yurt-hub pod on the node.
+			// if no pods in cache, maybe all of pods have been deleted by accident,
+			// if empty object is returned, pods on node will be deleted by kubelet.
+			// in order to prevent the influence to business, return error here so pods
+			// will be kept on node.
+			return objects, storage.ErrStorageNotFound
+		}
 		return objects, nil
 	}
 
@@ -200,3 +213,9 @@ func isCacheKey(key string) bool {
 
 	return false
 }
+
+// isPodKey verify the key is kubelet/pods or not
+func isPodKey(key string) bool {
+	comp, resource, _, _ := util.SplitKey(key)
+	return comp == "kubelet" && resource == "pods"
+}
diff --git a/pkg/yurthub/storage/disk/storage.go b/pkg/yurthub/storage/disk/storage.go
index af74e24b..d9f1e888 100644
--- a/pkg/yurthub/storage/disk/storage.go
+++ b/pkg/yurthub/storage/disk/storage.go
@@ -63,10 +63,11 @@ func NewDiskStorage(dir string) (storage.Store, error) {
 	return ds, nil
 }
 
-// Create new a file with key and contents
+// Create new a file with key and contents or create dir only
+// when contents are empty.
 func (ds *diskStorage) Create(key string, contents []byte) error {
-	if key == "" || len(contents) == 0 {
-		return nil
+	if key == "" {
+		return storage.ErrKeyIsEmpty
 	}
 
 	if !ds.lockKey(key) {
@@ -74,6 +75,32 @@ func (ds *diskStorage) Create(key string, contents []byte) error {
 	}
 	defer ds.unLockKey(key)
 
+	// no contents, create key dir only
+	if len(contents) == 0 {
+		keyPath := filepath.Join(ds.baseDir, key)
+		if info, err := os.Stat(keyPath); err != nil {
+			if os.IsNotExist(err) {
+				if err = os.MkdirAll(keyPath, 0755); err == nil {
+					return nil
+				}
+			}
+			return err
+		} else if info.IsDir() {
+			return nil
+		} else {
+			return storage.ErrKeyHasNoContent
+		}
+	}
+
+	return ds.create(key, contents)
+}
+
+// create will make up a file with key as file path and contents as file contents.
+func (ds *diskStorage) create(key string, contents []byte) error {
+	if key == "" {
+		return storage.ErrKeyIsEmpty
+	}
+
 	keyPath := filepath.Join(ds.baseDir, key)
 	dir, _ := filepath.Split(keyPath)
 	if _, err := os.Stat(dir); err != nil {
@@ -107,9 +134,14 @@ func (ds *diskStorage) Create(key string, contents []byte) error {
 // Delete delete file that specified by key
 func (ds *diskStorage) Delete(key string) error {
 	if key == "" {
-		return nil
+		return storage.ErrKeyIsEmpty
 	}
 
+	if !ds.lockKey(key) {
+		return storage.ErrStorageAccessConflict
+	}
+	defer ds.unLockKey(key)
+
 	errs := make([]error, 0)
 	if err := ds.delete(key); err != nil {
 		errs = append(errs, err)
@@ -128,14 +160,9 @@ func (ds *diskStorage) Delete(key string) error {
 
 func (ds *diskStorage) delete(key string) error {
 	if key == "" {
-		return nil
+		return storage.ErrKeyIsEmpty
 	}
 
-	if !ds.lockKey(key) {
-		return storage.ErrStorageAccessConflict
-	}
-	defer ds.unLockKey(key)
-
 	absKey := filepath.Join(ds.baseDir, key)
 	info, err := os.Stat(absKey)
 	if err != nil {
@@ -154,40 +181,49 @@ func (ds *diskStorage) delete(key string) error {
 
 // Get get contents from the file that specified by key
 func (ds *diskStorage) Get(key string) ([]byte, error) {
-	return ds.get(filepath.Join(ds.baseDir, key))
-}
-
-func (ds *diskStorage) get(path string) ([]byte, error) {
-	if path == "" {
-		return nil, nil
+	if key == "" {
+		return []byte{}, storage.ErrKeyIsEmpty
 	}
 
-	key := strings.TrimPrefix(path, ds.baseDir)
 	if !ds.lockKey(key) {
 		return nil, storage.ErrStorageAccessConflict
 	}
 	defer ds.unLockKey(key)
+	return ds.get(filepath.Join(ds.baseDir, key))
+}
+
+// get returns contents from the file of path
+func (ds *diskStorage) get(path string) ([]byte, error) {
+	if path == "" {
+		return []byte{}, storage.ErrKeyIsEmpty
+	}
 
 	info, err := os.Stat(path)
 	if err != nil {
 		if os.IsNotExist(err) {
 			return []byte{}, storage.ErrStorageNotFound
 		}
-		return nil, fmt.Errorf("failed to get bytes for %s, %v", key, err)
+		return nil, fmt.Errorf("failed to get bytes from %s, %v", path, err)
 	} else if info.Mode().IsRegular() {
 		b, err := ioutil.ReadFile(path)
 		if err != nil {
-			return nil, err
+			return []byte{}, err
 		}
 
 		return b, nil
+	} else if info.IsDir() {
+		return []byte{}, storage.ErrKeyHasNoContent
 	}
 
-	return nil, fmt.Errorf("%s is exist, but not recognized, %v", key, info.Mode())
+	return nil, fmt.Errorf("%s is exist, but not recognized, %v", path, info.Mode())
 }
 
 // ListKeys list all of keys for files
 func (ds *diskStorage) ListKeys(key string) ([]string, error) {
+	if key == "" {
+		return []string{}, storage.ErrKeyIsEmpty
+	}
+
 	keys := make([]string, 0)
 	absPath := filepath.Join(ds.baseDir, key)
 	if info, err := os.Stat(absPath); err != nil {
@@ -202,8 +238,7 @@ func (ds *diskStorage) ListKeys(key string) ([]string, error) {
 			}
 
 			if info.Mode().IsRegular() {
-				_, file := filepath.Split(path)
-				if !strings.HasPrefix(file, tmpPrefix) {
+				if !isTmpFile(path) {
 					keys = append(keys, strings.TrimPrefix(path, ds.baseDir))
 				}
 			}
@@ -223,9 +258,14 @@ func (ds *diskStorage) ListKeys(key string) ([]string, error) {
 // List get all of contents for local files
 func (ds *diskStorage) List(key string) ([][]byte, error) {
 	if key == "" {
-		return nil, fmt.Errorf("key for list is empty")
+		return [][]byte{}, storage.ErrKeyIsEmpty
 	}
 
+	if !ds.lockKey(key) {
+		return nil, storage.ErrStorageAccessConflict
+	}
+	defer ds.unLockKey(key)
+
 	bb := make([][]byte, 0)
 	absKey := filepath.Join(ds.baseDir, key)
 	info, err := os.Stat(absKey)
@@ -249,7 +289,7 @@ func (ds *diskStorage) List(key string) ([][]byte, error) {
 				return err
 			}
 
-			if info.Mode().IsRegular() {
+			if info.Mode().IsRegular() && !isTmpFile(path) {
 				b, err := ds.get(path)
 				if err != nil {
 					klog.Warningf("failed to get bytes for %s when listing bytes, %v", path, err)
@@ -272,10 +312,12 @@ func (ds *diskStorage) List(key string) ([][]byte, error) {
 	return nil, fmt.Errorf("%s is exist, but not recognized, %v", key, info.Mode())
 }
 
-// Update update local file that specified by key with contents
+// Update will update local file that specified by key with contents
 func (ds *diskStorage) Update(key string, contents []byte) error {
-	if key == "" || len(contents) == 0 {
-		return nil
+	if key == "" {
+		return storage.ErrKeyIsEmpty
+	} else if len(contents) == 0 {
+		return storage.ErrKeyHasNoContent
 	}
 
 	if !ds.lockKey(key) {
@@ -283,34 +325,31 @@ func (ds *diskStorage) Update(key string, contents []byte) error {
 	}
 	defer ds.unLockKey(key)
 
-	dir, file := filepath.Split(key)
-	tmpKey := filepath.Join(dir, fmt.Sprintf("%s%s", tmpPrefix, file))
-
-	err := ds.Create(tmpKey, contents)
+	// 1. create new file with tmpKey
+	tmpKey := getTmpKey(key)
+	err := ds.create(tmpKey, contents)
 	if err != nil {
 		return err
 	}
 
-	tmpPath := filepath.Join(ds.baseDir, tmpKey)
-	absKey := filepath.Join(ds.baseDir, key)
-	info, err := os.Stat(absKey)
+	// 2. delete old file by key
+	err = ds.delete(key)
 	if err != nil {
-		if !os.IsNotExist(err) {
-			os.Remove(tmpPath)
-			return err
-		}
-	} else if info.Mode().IsRegular() {
-		if err := os.Remove(absKey); err != nil {
-			os.Remove(tmpPath)
-			return err
-		}
+		ds.delete(tmpKey)
+		return err
 	}
 
-	return os.Rename(tmpPath, absKey)
+	// 3. rename tmpKey file to key file
+	return os.Rename(filepath.Join(ds.baseDir, tmpKey), filepath.Join(ds.baseDir, key))
 }
 
 // Recover recover storage error
 func (ds *diskStorage) Recover(key string) error {
+	if !ds.lockKey(key) {
+		return nil
+	}
+	defer ds.unLockKey(key)
+
 	dir := filepath.Join(ds.baseDir, key)
 	err := filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
 		if err != nil {
@@ -318,16 +357,10 @@ func (ds *diskStorage) Recover(key string) error {
 		}
 
 		if info.Mode().IsRegular() {
-			_, file := filepath.Split(path)
-			if strings.HasPrefix(file, tmpPrefix) {
+			if isTmpFile(path) {
 				tmpKey := strings.TrimPrefix(path, ds.baseDir)
 				key := getKey(tmpKey)
 				keyPath := filepath.Join(ds.baseDir, key)
-				if !ds.lockKey(key) {
-					return nil
-				}
-				defer ds.unLockKey(key)
-
 				iErr := os.Rename(path, keyPath)
 				if iErr != nil {
 					klog.V(2).Infof("failed to recover bytes %s, %v", tmpKey, err)
@@ -344,13 +377,24 @@ func (ds *diskStorage) Recover(key string) error {
 }
 
 func (ds *diskStorage) lockKey(key string) bool {
-	if ds.isKeyPending(key) {
+	ds.RLock()
+	defer ds.RUnlock()
+	if _, ok := ds.keyPendingStatus[key]; ok {
 		return false
 	}
 
-	if !ds.setKeyPending(key) {
-		return false
+	for pendingKey := range ds.keyPendingStatus {
+		if len(key) > len(pendingKey) {
+			if strings.Contains(key, pendingKey) {
+				return false
+			}
+		} else {
+			if strings.Contains(pendingKey, key) {
+				return false
+			}
+		}
 	}
+	ds.keyPendingStatus[key] = struct{}{}
 
 	return true
 }
@@ -361,31 +405,19 @@ func (ds *diskStorage) unLockKey(key string) {
 	delete(ds.keyPendingStatus, key)
 }
 
-func (ds *diskStorage) isKeyPending(key string) bool {
-	ds.RLock()
-	defer ds.RUnlock()
-	if _, ok := ds.keyPendingStatus[key]; ok {
-		return true
-	}
-	return false
-}
-
-func (ds *diskStorage) setKeyPending(key string) bool {
-	ds.Lock()
-	defer ds.Unlock()
-	if _, ok := ds.keyPendingStatus[key]; ok {
-		return false
-	}
-
-	ds.keyPendingStatus[key] = struct{}{}
-	return true
-}
-
 func getTmpKey(key string) string {
 	dir, file := filepath.Split(key)
 	return filepath.Join(dir, fmt.Sprintf("%s%s", tmpPrefix, file))
 }
 
+func isTmpFile(path string) bool {
+	_, file := filepath.Split(path)
+	if strings.HasPrefix(file, tmpPrefix) {
+		return true
+	}
+	return false
+}
+
 func getKey(tmpKey string) string {
 	dir, file := filepath.Split(tmpKey)
 	return filepath.Join(dir, strings.TrimPrefix(file, tmpPrefix))
diff --git a/pkg/yurthub/storage/store.go b/pkg/yurthub/storage/store.go
index a7191dd0..7cea9efd 100644
--- a/pkg/yurthub/storage/store.go
+++ b/pkg/yurthub/storage/store.go
@@ -26,6 +26,12 @@ var ErrStorageAccessConflict = errors.New("specified key is under accessing")
 // ErrStorageNotFound is an error for not found accessing key
 var ErrStorageNotFound = errors.New("specified key is not found")
 
+// ErrorKeyHasNoContent is an error for file key that has no contents
+var ErrKeyHasNoContent = errors.New("specified key has no contents")
+
+// ErrorKeyWithoutContent is an error for key is empty
+var ErrKeyIsEmpty = errors.New("specified key is empty")
+
 // Store is an interface for caching data into backend storage
 type Store interface {
 	Create(key string, contents []byte) error


Test patch:
diff --git a/pkg/yurthub/cachemanager/cache_manager_test.go b/pkg/yurthub/cachemanager/cache_manager_test.go
index 20990be8..88d03368 100644
--- a/pkg/yurthub/cachemanager/cache_manager_test.go
+++ b/pkg/yurthub/cachemanager/cache_manager_test.go
@@ -32,6 +32,7 @@ import (
 	proxyutil "github.com/openyurtio/openyurt/pkg/yurthub/proxy/util"
 	"github.com/openyurtio/openyurt/pkg/yurthub/util"
 	v1 "k8s.io/api/core/v1"
+	nodev1beta1 "k8s.io/api/node/v1beta1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -642,6 +643,71 @@ func TestCacheResponseForList(t *testing.T) {
 				},
 			},
 		},
+		{
+			desc:    "cache response for list nodes with fieldselector",
+			group:   "",
+			version: "v1",
+			key:     "kubelet/nodes",
+			inputObj: runtime.Object(
+				&v1.NodeList{
+					TypeMeta: metav1.TypeMeta{
+						APIVersion: "v1",
+						Kind:       "NodeList",
+					},
+					ListMeta: metav1.ListMeta{
+						ResourceVersion: "12",
+					},
+					Items: []v1.Node{
+						{
+							TypeMeta: metav1.TypeMeta{
+								APIVersion: "v1",
+								Kind:       "Node",
+							},
+							ObjectMeta: metav1.ObjectMeta{
+								Name:            "mynode",
+								ResourceVersion: "12",
+							},
+						},
+					},
+				},
+			),
+			userAgent:  "kubelet",
+			accept:     "application/json",
+			verb:       "GET",
+			path:       "/api/v1/nodes?fieldselector=meatadata.name=mynode",
+			namespaced: false,
+			expectResult: expectData{
+				data: map[string]struct{}{
+					"node-mynode-12": {},
+				},
+			},
+		},
+		{
+			desc:    "cache response for list runtimeclasses with no objects",
+			group:   "node.k8s.io",
+			version: "v1beta1",
+			key:     "kubelet/runtimeclass",
+			inputObj: runtime.Object(
+				&nodev1beta1.RuntimeClassList{
+					TypeMeta: metav1.TypeMeta{
+						APIVersion: "node.k8s.io/v1beta1",
+						Kind:       "RuntimeClassList",
+					},
+					ListMeta: metav1.ListMeta{
+						ResourceVersion: "12",
+					},
+					Items: []nodev1beta1.RuntimeClass{},
+				},
+			),
+			userAgent:  "kubelet",
+			accept:     "application/json",
+			verb:       "GET",
+			path:       "/apis/node.k8s.io/v1beta1/runtimeclasses",
+			namespaced: false,
+			expectResult: expectData{
+				data: map[string]struct{}{},
+			},
+		},
 	}
 
 	accessor := meta.NewAccessor()
@@ -683,7 +749,7 @@ func TestCacheResponseForList(t *testing.T) {
 
 			if tt.expectResult.err {
 				if err == nil {
-					t.Errorf("Got no error, but expect err")
+					t.Error("Got no error, but expect err")
 				}
 			} else {
 				if err != nil {
@@ -691,8 +757,8 @@ func TestCacheResponseForList(t *testing.T) {
 				}
 
 				objs, err := storage.List(tt.key)
-				if err != nil || len(objs) == 0 {
-					t.Errorf("failed to get object from storage")
+				if err != nil {
+					t.Errorf("failed to list objects from storage, %v", err)
 				}
 
 				if len(objs) != len(tt.expectResult.data) {
@@ -717,6 +783,7 @@ func TestCacheResponseForList(t *testing.T) {
 					}
 				}
 			}
+			resetStorage(storage, tt.key)
 		})
 	}
 }
diff --git a/pkg/yurthub/storage/disk/storage_test.go b/pkg/yurthub/storage/disk/storage_test.go
index 4e98764b..78dad349 100644
--- a/pkg/yurthub/storage/disk/storage_test.go
+++ b/pkg/yurthub/storage/disk/storage_test.go
@@ -17,11 +17,7 @@ limitations under the License.
 package disk
 
 import (
-	"bytes"
-	"fmt"
-	"io/ioutil"
 	"os"
-	"path/filepath"
 	"testing"
 
 	"github.com/openyurtio/openyurt/pkg/yurthub/storage"
@@ -29,103 +25,147 @@ import (
 
 var (
 	testDir = "/tmp/cache/"
-	tempDir = "kubelet/default/pods"
-	tempKey = "kubelet/default/pods/test-pod"
 )
 
 func TestCreate(t *testing.T) {
-	s, err := NewDiskStorage(testDir)
-	if err != nil {
-		t.Fatalf("unable to new disk storage, %v", err)
-	}
-
-	err = s.Create(tempKey, []byte("test-pod"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s", err, tempKey)
-	}
-
-	createdFile := filepath.Join(testDir, tempKey)
-	if fi, err := os.Stat(createdFile); err != nil {
-		t.Errorf("Got error %v, wanted file %q to be there", err, createdFile)
-	} else if !fi.Mode().IsRegular() {
-		t.Errorf("Got %q not a regular file", createdFile)
-	}
-
-	b, err := ioutil.ReadFile(createdFile)
-	if err != nil {
-		t.Errorf("Got error %v, unable read regular file %q", err, createdFile)
-	} else if !bytes.Equal(b, []byte("test-pod")) {
-		t.Errorf("Wanted string: test-pod but got %s", string(b))
-	}
-
-	if err = os.RemoveAll(testDir); err != nil {
-		t.Errorf("Got error %v, unable remove path %s", err, testDir)
-	}
-}
-
-func TestCreateFileExist(t *testing.T) {
-	s, err := NewDiskStorage(testDir)
-	if err != nil {
-		t.Fatalf("unable to new disk storage, %v", err)
-	}
-
-	err = s.Create(tempKey, []byte("test-pod1"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s", err, tempKey)
-	}
-
-	err = s.Create(tempKey, []byte("test-pod2"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s witch contents test-pod2", err, tempKey)
-	}
-
-	createdFile := filepath.Join(testDir, tempKey)
-	if fi, err := os.Stat(createdFile); err != nil {
-		t.Errorf("Got error %v, wanted file %q to be there", err, createdFile)
-	} else if !fi.Mode().IsRegular() {
-		t.Errorf("Got %q not a regular file", createdFile)
-	}
-
-	b, err := ioutil.ReadFile(createdFile)
-	if err != nil {
-		t.Errorf("Got error %v, unable read regular file %q", err, createdFile)
-	} else if !bytes.Equal(b, []byte("test-pod2")) {
-		t.Errorf("Wanted string: test-pod2 but got %s", string(b))
-	}
-
-	if err = os.RemoveAll(testDir); err != nil {
-		t.Errorf("Got error %v, unable remove path %s", err, testDir)
+	testcases := map[string]struct {
+		rootDir       string
+		preCreatedKey map[string]string
+		keysData      map[string]string
+		result        map[string]struct {
+			err  error
+			data string
+		}
+		createErr error
+	}{
+		"create dir key with no data": {
+			keysData: map[string]string{
+				"/kubelet/default/pods": "",
+			},
+			result: map[string]struct {
+				err  error
+				data string
+			}{
+				"/kubelet/default/pods": {
+					err: storage.ErrKeyHasNoContent,
+				},
+			},
+		},
+		"create dir key that already exists": {
+			preCreatedKey: map[string]string{
+				"/kubelet/default/pods/foo": "test-pod1",
+			},
+			keysData: map[string]string{
+				"/kubelet/default/pods": "",
+			},
+			result: map[string]struct {
+				err  error
+				data string
+			}{
+				"/kubelet/default/pods/foo": {
+					data: "test-pod1",
+				},
+			},
+		},
+		"create key normally": {
+			keysData: map[string]string{
+				"/kubelet/default/pods/foo": "test-pod",
+			},
+			result: map[string]struct {
+				err  error
+				data string
+			}{
+				"/kubelet/default/pods/foo": {
+					data: "test-pod",
+				},
+			},
+		},
+		"create key when key already exists": {
+			preCreatedKey: map[string]string{
+				"/kubelet/default/pods/bar": "test-pod1",
+			},
+			keysData: map[string]string{
+				"/kubelet/default/pods/bar": "test-pod2",
+			},
+			result: map[string]struct {
+				err  error
+				data string
+			}{
+				"/kubelet/default/pods/bar": {
+					data: "test-pod2",
+				},
+			},
+		},
+		"create key when parent dir already exists": {
+			preCreatedKey: map[string]string{
+				"/kubelet/default/pods": "",
+			},
+			keysData: map[string]string{
+				"/kubelet/default/pods/bar": "test-pod1",
+			},
+			result: map[string]struct {
+				err  error
+				data string
+			}{
+				"/kubelet/default/pods/bar": {
+					data: "test-pod1",
+				},
+			},
+		},
+		"create key that already exists with no data": {
+			preCreatedKey: map[string]string{
+				"/kubelet/default/pods/foo": "test-pod1",
+			},
+			keysData: map[string]string{
+				"/kubelet/default/pods/foo": "",
+			},
+			createErr: storage.ErrKeyHasNoContent,
+		},
 	}
-}
-
-func TestCreateDirExist(t *testing.T) {
 	s, err := NewDiskStorage(testDir)
 	if err != nil {
 		t.Fatalf("unable to new disk storage, %v", err)
 	}
 
-	createdFile := filepath.Join(testDir, tempKey)
-	dir, _ := filepath.Split(createdFile)
-	if err = os.MkdirAll(dir, 0755); err != nil {
-		t.Errorf("Got error %v, unable make dir %s", err, dir)
-	}
+	for k, tc := range testcases {
+		t.Run(k, func(t *testing.T) {
+			for key, data := range tc.preCreatedKey {
+				err = s.Create(key, []byte(data))
+				if err != nil {
+					t.Errorf("%s Got error %v, wanted successful create %s", k, err, key)
+				}
+			}
 
-	err = s.Create(tempKey, []byte("test-pod"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s", err, tempKey)
-	}
+			for key, data := range tc.keysData {
+				err = s.Create(key, []byte(data))
+				if err != nil {
+					if tc.createErr != err {
+						t.Errorf("%s: expect create error %v, but got %v", k, tc.createErr, err)
+					}
+				}
+			}
 
-	if fi, err := os.Stat(createdFile); err != nil {
-		t.Errorf("Got error %v, wanted file %q to be there", err, createdFile)
-	} else if !fi.Mode().IsRegular() {
-		t.Errorf("Got %q not a regular file", createdFile)
-	}
+			for key, result := range tc.result {
+				b, err := s.Get(key)
+				if result.err != nil {
+					if result.err != err {
+						t.Errorf("%s(key=%s) expect error %v, but got error %v", k, key, result.err, err)
+					}
+				}
+
+				if result.data != "" {
+					if result.data != string(b) {
+						t.Errorf("%s(key=%s) expect result %s, but got data %s", k, key, result.data, string(b))
+					}
+				}
+			}
 
-	b, err := ioutil.ReadFile(createdFile)
-	if err != nil {
-		t.Errorf("Got error %v, unable read regular file %q", err, createdFile)
-	} else if !bytes.Equal(b, []byte("test-pod")) {
-		t.Errorf("Wanted string: test-pod but got %s", string(b))
+			for key := range tc.result {
+				if err = s.Delete(key); err != nil {
+					t.Errorf("%s failed to delete key %s, %v", k, key, err)
+				}
+			}
+		})
 	}
 
 	if err = os.RemoveAll(testDir); err != nil {
@@ -134,75 +174,108 @@ func TestCreateDirExist(t *testing.T) {
 }
 
 func TestDelete(t *testing.T) {
-	s, err := NewDiskStorage(testDir)
-	if err != nil {
-		t.Fatalf("unable to new disk storage, %v", err)
-	}
-
-	err = s.Create(tempKey, []byte("test-pod"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s", err, tempKey)
-	}
-
-	createdFile := filepath.Join(testDir, tempKey)
-	if fi, err := os.Stat(createdFile); err != nil {
-		t.Errorf("Got error %v, wanted file %q to be there", err, createdFile)
-	} else if !fi.Mode().IsRegular() {
-		t.Errorf("Got %q not a regular file", createdFile)
-	}
-
-	err = s.Delete(tempKey)
-	if err != nil {
-		t.Errorf("Got error %v, unable delete key %q", err, tempKey)
-	}
-
-	if _, err := os.Stat(createdFile); err == nil || !os.IsNotExist(err) {
-		t.Errorf("want %q is deleted, but it still exist", createdFile)
-	}
-
-	if err = os.RemoveAll(testDir); err != nil {
-		t.Errorf("Got error %v, unable remove path %s", err, testDir)
+	testcases := map[string]struct {
+		preCreatedKeys map[string]string
+		deleteKeys     []string
+		result         map[string]struct {
+			beforeDelete string
+			deleteErr    error
+			getErr       error
+		}
+	}{
+		"normally delete key": {
+			preCreatedKeys: map[string]string{
+				"kubelet/nodes/foo": "node-test1",
+			},
+			deleteKeys: []string{"kubelet/nodes/foo"},
+			result: map[string]struct {
+				beforeDelete string
+				deleteErr    error
+				getErr       error
+			}{
+				"kubelet/nodes/foo": {
+					beforeDelete: "node-test1",
+					deleteErr:    nil,
+					getErr:       storage.ErrStorageNotFound,
+				},
+			},
+		},
+		"delete key that not exist": {
+			deleteKeys: []string{"kubelet/nodes/foo"},
+			result: map[string]struct {
+				beforeDelete string
+				deleteErr    error
+				getErr       error
+			}{
+				"kubelet/nodes/foo": {
+					deleteErr: nil,
+				},
+			},
+		},
+		"delete dir key": {
+			preCreatedKeys: map[string]string{
+				"kubelet/nodes": "",
+			},
+			deleteKeys: []string{"kubelet/nodes"},
+			result: map[string]struct {
+				beforeDelete string
+				deleteErr    error
+				getErr       error
+			}{
+				"kubelet/nodes/": {
+					deleteErr: nil,
+					getErr:    storage.ErrKeyHasNoContent,
+				},
+			},
+		},
 	}
-}
-
-func TestDeleteFileNotExist(t *testing.T) {
 	s, err := NewDiskStorage(testDir)
 	if err != nil {
 		t.Fatalf("unable to new disk storage, %v", err)
 	}
 
-	createdFile := filepath.Join(testDir, tempKey)
-	err = s.Delete(tempKey)
-	if err != nil {
-		t.Errorf("Got error %v, delete not exist file(%q) returned error", err, createdFile)
-	}
-
-	if err = os.RemoveAll(testDir); err != nil {
-		t.Errorf("Got error %v, unable remove path %s", err, testDir)
-	}
-}
-
-func TestDeleteDir(t *testing.T) {
-	s, err := NewDiskStorage(testDir)
-	if err != nil {
-		t.Fatalf("unable to new disk storage, %v", err)
-	}
+	for k, tc := range testcases {
+		t.Run(k, func(t *testing.T) {
+			for key, data := range tc.preCreatedKeys {
+				err = s.Create(key, []byte(data))
+				if err != nil {
+					t.Errorf("%s: Got error %v, wanted successful create %s", k, err, key)
+				}
+			}
 
-	err = s.Create(tempKey, []byte("test-pod"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s", err, tempKey)
-	}
+			for key, result := range tc.result {
+				if result.beforeDelete != "" {
+					b, err := s.Get(key)
+					if err != nil {
+						t.Errorf("%s: got error %v when get key %s before deletion", k, err, key)
+					}
+
+					if result.beforeDelete != string(b) {
+						t.Errorf("%s: expect data %s, but got %s before deletion", k, result.beforeDelete, string(b))
+					}
+				}
+			}
 
-	err = s.Delete(tempDir)
-	if err != nil {
-		t.Errorf("Got error %v, unable delete dir key %q", err, tempDir)
-	}
+			for _, key := range tc.deleteKeys {
+				err = s.Delete(key)
+				if result, ok := tc.result[key]; ok {
+					if result.deleteErr != err {
+						t.Errorf("%s: delete key(%s) expect error %v, but got %v", k, key, result.deleteErr, err)
+					}
+				} else if err != nil {
+					t.Errorf("%s: Got error %v, unable delete key %q", k, err, key)
+				}
+			}
 
-	createdFile := filepath.Join(testDir, tempKey)
-	if fi, err := os.Stat(createdFile); err != nil {
-		t.Errorf("Got error %v, wanted file %q to be there", err, createdFile)
-	} else if !fi.Mode().IsRegular() {
-		t.Errorf("Got %q not a regular file", createdFile)
+			for key, result := range tc.result {
+				_, err := s.Get(key)
+				if result.getErr != nil {
+					if result.getErr != err {
+						t.Errorf("%s: expect error %v, but got error %v", k, result.getErr, err)
+					}
+				}
+			}
+		})
 	}
 
 	if err = os.RemoveAll(testDir); err != nil {
@@ -211,60 +284,85 @@ func TestDeleteDir(t *testing.T) {
 }
 
 func TestGet(t *testing.T) {
-	s, err := NewDiskStorage(testDir)
-	if err != nil {
-		t.Fatalf("unable to new disk storage, %v", err)
-	}
-
-	err = s.Create(tempKey, []byte("test-pod"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s", err, tempKey)
-	}
-
-	b, err := s.Get(tempKey)
-	if err != nil {
-		t.Errorf("Got error %v, get key %q", err, tempKey)
-	} else if !bytes.Equal(b, []byte("test-pod")) {
-		t.Errorf("Wanted string: test-pod but got %s", string(b))
-	}
-
-	if err = os.RemoveAll(testDir); err != nil {
-		t.Errorf("Got error %v, unable remove path %s", err, testDir)
+	testcases := map[string]struct {
+		preCreatedKeys map[string]string
+		result         map[string]struct {
+			err  error
+			data string
+		}
+	}{
+		"normally get key": {
+			preCreatedKeys: map[string]string{
+				"kubelet/nodes/foo": "test-node1",
+			},
+			result: map[string]struct {
+				err  error
+				data string
+			}{
+				"kubelet/nodes/foo": {
+					data: "test-node1",
+				},
+			},
+		},
+		"get key that is not exist": {
+			result: map[string]struct {
+				err  error
+				data string
+			}{
+				"kubelet/nodes/foo": {
+					err: storage.ErrStorageNotFound,
+				},
+			},
+		},
+		"get key that is not regular file": {
+			preCreatedKeys: map[string]string{
+				"kubelet/nodes": "",
+			},
+			result: map[string]struct {
+				err  error
+				data string
+			}{
+				"kubelet/nodes": {
+					err: storage.ErrKeyHasNoContent,
+				},
+			},
+		},
 	}
-}
-
-func TestGetFileNotExist(t *testing.T) {
 	s, err := NewDiskStorage(testDir)
 	if err != nil {
 		t.Fatalf("unable to new disk storage, %v", err)
 	}
 
-	b, err := s.Get(tempKey)
-	if err != storage.ErrStorageNotFound {
-		t.Errorf("Got error %v, expect error %v", err, storage.ErrStorageNotFound)
-	} else if len(b) != 0 {
-		t.Errorf("Wanted empty string got %s", string(b))
-	}
-
-	if err = os.RemoveAll(testDir); err != nil {
-		t.Errorf("Got error %v, unable remove path %s", err, testDir)
-	}
-}
-
-func TestGetNotRegularFile(t *testing.T) {
-	s, err := NewDiskStorage(testDir)
-	if err != nil {
-		t.Fatalf("unable to new disk storage, %v", err)
-	}
+	for k, tc := range testcases {
+		t.Run(k, func(t *testing.T) {
+			for key, data := range tc.preCreatedKeys {
+				err := s.Create(key, []byte(data))
+				if err != nil {
+					t.Errorf("%s Got error %v, wanted successful create %s", k, err, key)
+				}
+			}
 
-	err = s.Create(tempKey, []byte("test-pod"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s", err, tempKey)
-	}
+			for key, result := range tc.result {
+				b, err := s.Get(key)
+				if result.err != nil {
+					if result.err != err {
+						t.Errorf("%s: expect error %v, but got error %v", k, result.err, err)
+					}
+				}
+
+				if result.data != "" {
+					if result.data != string(b) {
+						t.Errorf("%s: expect result data %s, but got data %s", k, result.data, string(b))
+					}
+				}
+			}
 
-	_, err = s.Get(tempDir)
-	if err == nil {
-		t.Errorf("Got not error for dir key %q", tempDir)
+			for key := range tc.result {
+				if err = s.Delete(key); err != nil {
+					t.Errorf("%s failed to delete key %s, %v", k, key, err)
+				}
+			}
+		})
 	}
 
 	if err = os.RemoveAll(testDir); err != nil {
@@ -273,89 +371,88 @@ func TestGetNotRegularFile(t *testing.T) {
 }
 
 func TestListKeys(t *testing.T) {
-	s, err := NewDiskStorage(testDir)
-	if err != nil {
-		t.Fatalf("unable to new disk storage, %v", err)
-	}
-
-	tempKeys := make([]string, 5)
-	for i := 0; i < 5; i++ {
-		tempKeys[i] = fmt.Sprintf("%s-%d", tempKey, i)
-		err = s.Create(tempKeys[i], []byte("test-pod"))
-		if err != nil {
-			t.Errorf("Got error %v, wanted successful create %s", err, tempKeys[i])
-		}
-	}
-
-	keys, err := s.ListKeys(tempDir)
-	if err != nil {
-		t.Errorf("Got error %v, unable list keys for %s", err, tempDir)
+	testcases := map[string]struct {
+		preCreatedKeys map[string]string
+		listKey        string
+		result         map[string]struct{}
+		listErr        error
+	}{
+		"normally list keys": {
+			preCreatedKeys: map[string]string{
+				"kubelet/pods/default/foo1":         "pod1",
+				"kubelet/pods/default/foo2":         "pod2",
+				"kubelet/pods/kube-system/foo3":     "pod3",
+				"kubelet/pods/kube-system/foo4":     "pod4",
+				"kubelet/pods/kube-system/tmp_foo5": "pod5",
+				"kubelet/pods/foo5":                 "",
+			},
+			listKey: "kubelet/pods",
+			result: map[string]struct{}{
+				"kubelet/pods/default/foo1":     {},
+				"kubelet/pods/default/foo2":     {},
+				"kubelet/pods/kube-system/foo3": {},
+				"kubelet/pods/kube-system/foo4": {},
+			},
+		},
+		"list keys for dir only": {
+			preCreatedKeys: map[string]string{
+				"kubelet/pods/default":     "",
+				"kubelet/pods/kube-system": "",
+				"kubelet/pods/foo5":        "",
+			},
+			listKey: "kubelet/pods",
+			result:  map[string]struct{}{},
+		},
+		"list keys for regular file": {
+			preCreatedKeys: map[string]string{
+				"kubelet/pods/default/foo1": "pod1",
+			},
+			listKey: "kubelet/pods/default/foo1",
+			result: map[string]struct{}{
+				"kubelet/pods/default/foo1": {},
+			},
+		},
+		"list keys for not exist file": {
+			listKey: "kubelet/pods/default/foo5",
+			result:  map[string]struct{}{},
+		},
 	}
-
-	if len(tempKeys) != len(keys) {
-		t.Errorf("expect %d keys, but got %d keys", len(tempKeys), len(keys))
-	}
-
-	for _, key := range tempKeys {
-		found := false
-		for _, cachedKey := range keys {
-			if key == cachedKey {
-				found = true
-				break
-			}
-		}
-		if !found {
-			t.Errorf("key %s is not found by list keys %v", key, keys)
-		}
-	}
-
-	if err = os.RemoveAll(testDir); err != nil {
-		t.Errorf("Got error %v, unable remove path %s", err, testDir)
-	}
-}
-
-func TestListKeysForEmptyDir(t *testing.T) {
 	s, err := NewDiskStorage(testDir)
 	if err != nil {
 		t.Fatalf("unable to new disk storage, %v", err)
 	}
 
-	keys, err := s.ListKeys(tempDir)
-	if err != nil {
-		t.Errorf("Got error %v, unable list keys for empty dir %s", err, tempDir)
-	}
-
-	if len(keys) != 0 {
-		t.Errorf("expect 0 key, but got %d keys", len(keys))
-	}
-
-	if err = os.RemoveAll(testDir); err != nil {
-		t.Errorf("Got error %v, unable remove path %s", err, testDir)
-	}
-}
-
-func TestListKeysForRegularFile(t *testing.T) {
-	s, err := NewDiskStorage(testDir)
-	if err != nil {
-		t.Fatalf("unable to new disk storage, %v", err)
-	}
+	for k, tc := range testcases {
+		t.Run(k, func(t *testing.T) {
+			for key, data := range tc.preCreatedKeys {
+				err = s.Create(key, []byte(data))
+				if err != nil {
+					t.Errorf("%s: Got error %v, wanted successful create %s", k, err, key)
+				}
+			}
 
-	err = s.Create(tempKey, []byte("test-pod"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s", err, tempKey)
-	}
+			keys, err := s.ListKeys(tc.listKey)
+			if err != nil {
+				t.Errorf("%s: Got error %v, unable list keys for %s", k, err, tc.listKey)
+			}
 
-	keys, err := s.ListKeys(tempKey)
-	if err != nil {
-		t.Errorf("Got error %v, unable list keys for empty dir %s", err, tempDir)
-	}
+			if len(tc.result) != len(keys) {
+				t.Errorf("%s: expect %d keys, but got %d keys", k, len(tc.result), len(keys))
+			}
 
-	if len(keys) != 1 {
-		t.Errorf("listKeys: expect 1 key, but got %d keys", len(keys))
-	}
+			for _, key := range keys {
+				if _, ok := tc.result[key]; !ok {
+					t.Errorf("%s: got key %s not in result %v", k, key, tc.result)
+				}
+			}
 
-	if keys[0] != tempKey {
-		t.Errorf("listKeys: expect %s key, but got %s key", tempKey, keys[0])
+			for key := range tc.preCreatedKeys {
+				err = s.Delete(key)
+				if err != nil {
+					t.Errorf("%s: failed to delete key(%s), %v", k, key, err)
+				}
+			}
+		})
 	}
 
 	if err = os.RemoveAll(testDir); err != nil {
@@ -364,89 +461,91 @@ func TestListKeysForRegularFile(t *testing.T) {
 }
 
 func TestList(t *testing.T) {
-	s, err := NewDiskStorage(testDir)
-	if err != nil {
-		t.Fatalf("unable to new disk storage, %v", err)
-	}
-
-	tempContents := make([]string, 5)
-	for i := 0; i < 5; i++ {
-		tempContents[i] = fmt.Sprintf("test-pod-%d", i)
-		err = s.Create(fmt.Sprintf("%s-%d", tempKey, i), []byte(tempContents[i]))
-		if err != nil {
-			t.Errorf("Got error %v, wanted successful create %s", err, fmt.Sprintf("%s-%d", tempKey, i))
-		}
-	}
-
-	contents, err := s.List(tempDir)
-	if err != nil {
-		t.Errorf("Got error %v, unable list for %s", err, tempDir)
-	}
-
-	if len(tempContents) != len(contents) {
-		t.Errorf("expect %d number of contents, but got %d number of contents", len(tempContents), len(contents))
-	}
-
-	for _, content := range tempContents {
-		found := false
-		for _, cachedContent := range contents {
-			if content == string(cachedContent) {
-				found = true
-				break
-			}
-		}
-		if !found {
-			t.Errorf("content %s is not found by list", content)
-		}
+	testcases := map[string]struct {
+		preCreatedKeys map[string]string
+		listKey        string
+		result         map[string]struct{}
+		listErr        error
+	}{
+		"normally list": {
+			preCreatedKeys: map[string]string{
+				"kubelet/pods/default/foo1":         "pod1",
+				"kubelet/pods/default/foo2":         "pod2",
+				"kubelet/pods/kube-system/foo3":     "pod3",
+				"kubelet/pods/kube-system/foo4":     "pod4",
+				"kubelet/pods/kube-system/tmp_foo5": "pod5",
+				"kubelet/pods/foo5":                 "",
+			},
+			listKey: "kubelet/pods",
+			result: map[string]struct{}{
+				"pod1": {},
+				"pod2": {},
+				"pod3": {},
+				"pod4": {},
+			},
+		},
+		"list for empty dir": {
+			preCreatedKeys: map[string]string{
+				"kubelet/pods/default":     "",
+				"kubelet/pods/kube-system": "",
+				"kubelet/pods/foo5":        "",
+			},
+			listKey: "kubelet/pods",
+			result:  map[string]struct{}{},
+		},
+		"list for regular file": {
+			preCreatedKeys: map[string]string{
+				"kubelet/pods/default/foo1": "pod1",
+			},
+			listKey: "kubelet/pods/default/foo1",
+			result: map[string]struct{}{
+				"pod1": {},
+			},
+		},
+		"list for not exist file": {
+			listKey: "kubelet/pods/default/foo5",
+			result:  map[string]struct{}{},
+			listErr: storage.ErrStorageNotFound,
+		},
 	}
-
-	if err = os.RemoveAll(testDir); err != nil {
-		t.Errorf("Got error %v, unable remove path %s", err, testDir)
-	}
-}
-
-func TestListEmptyDir(t *testing.T) {
 	s, err := NewDiskStorage(testDir)
 	if err != nil {
 		t.Fatalf("unable to new disk storage, %v", err)
 	}
 
-	contents, err := s.List(tempDir)
-	if err != storage.ErrStorageNotFound {
-		t.Errorf("Got error %v, expect error %v", err, storage.ErrStorageNotFound)
-	}
-
-	if len(contents) != 0 {
-		t.Errorf("expect no contents, but got %d number of contents", len(contents))
-	}
-
-	if err = os.RemoveAll(testDir); err != nil {
-		t.Errorf("Got error %v, unable remove path %s", err, testDir)
-	}
-}
-
-func TestListSpecifiedFile(t *testing.T) {
-	s, err := NewDiskStorage(testDir)
-	if err != nil {
-		t.Fatalf("unable to new disk storage, %v", err)
-	}
+	for k, tc := range testcases {
+		t.Run(k, func(t *testing.T) {
+			for key, data := range tc.preCreatedKeys {
+				err = s.Create(key, []byte(data))
+				if err != nil {
+					t.Errorf("%s: Got error %v, wanted successful create %s", k, err, key)
+				}
+			}
 
-	err = s.Create(tempKey, []byte("test-pod"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s", err, tempKey)
-	}
+			data, err := s.List(tc.listKey)
+			if err != nil {
+				if tc.listErr != err {
+					t.Errorf("%s: list(%s) expect error %v, but got error %v", k, tc.listKey, tc.listErr, err)
+				}
+			}
 
-	contents, err := s.List(tempKey)
-	if err != nil {
-		t.Errorf("Got error %v, unable list for %s", err, tempKey)
-	}
+			if len(tc.result) != len(data) {
+				t.Errorf("%s: list expect %d objects, but got %d objects", k, len(tc.result), len(data))
+			}
 
-	if len(contents) != 1 {
-		t.Errorf("expect 1 contents, but got %d number of contents", len(contents))
-	}
+			for i := range data {
+				if _, ok := tc.result[string(data[i])]; !ok {
+					t.Errorf("%s: list data %s not in result %v", k, string(data[i]), tc.result)
+				}
+			}
 
-	if string(contents[0]) != "test-pod" {
-		t.Errorf("expect content: test-pod, but got content: %s", contents[0])
+			for key := range tc.preCreatedKeys {
+				err = s.Delete(key)
+				if err != nil {
+					t.Errorf("%s: failed to delete key(%s), %v", k, key, err)
+				}
+			}
+		})
 	}
 
 	if err = os.RemoveAll(testDir); err != nil {
@@ -455,33 +554,93 @@ func TestListSpecifiedFile(t *testing.T) {
 }
 
 func TestUpdate(t *testing.T) {
+	testcases := map[string]struct {
+		preCreatedKeys map[string]string
+		updateKeys     map[string]string
+		result         map[string]string
+		updateErr      error
+	}{
+		"normally update keys": {
+			preCreatedKeys: map[string]string{
+				"kubelet/pods/default/foo1": "test-pod1",
+				"kubelet/pods/default/foo2": "test-pod2",
+			},
+			updateKeys: map[string]string{
+				"kubelet/pods/default/foo1": "test-pod11",
+				"kubelet/pods/default/foo2": "test-pod21",
+			},
+			result: map[string]string{
+				"kubelet/pods/default/foo1": "test-pod11",
+				"kubelet/pods/default/foo2": "test-pod21",
+			},
+		},
+		"update keys that not exist": {
+			updateKeys: map[string]string{
+				"kubelet/pods/default/foo1": "test-pod11",
+				"kubelet/pods/default/foo2": "test-pod21",
+			},
+			result: map[string]string{
+				"kubelet/pods/default/foo1": "test-pod11",
+				"kubelet/pods/default/foo2": "test-pod21",
+			},
+		},
+		"update keys with empty string": {
+			preCreatedKeys: map[string]string{
+				"kubelet/pods/default/foo1": "test-pod1",
+				"kubelet/pods/default/foo2": "test-pod2",
+			},
+			updateKeys: map[string]string{
+				"kubelet/pods/default/foo1": "",
+				"kubelet/pods/default/foo2": "",
+			},
+			result: map[string]string{
+				"kubelet/pods/default/foo1": "test-pod1",
+				"kubelet/pods/default/foo2": "test-pod2",
+			},
+			updateErr: storage.ErrKeyHasNoContent,
+		},
+	}
 	s, err := NewDiskStorage(testDir)
 	if err != nil {
 		t.Fatalf("unable to new disk storage, %v", err)
 	}
 
-	err = s.Create(tempKey, []byte("test-pod"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s", err, tempKey)
-	}
+	for k, tc := range testcases {
+		t.Run(k, func(t *testing.T) {
+			for key, data := range tc.preCreatedKeys {
+				err = s.Create(key, []byte(data))
+				if err != nil {
+					t.Errorf("%s: Got error %v, wanted successful create %s", k, err, key)
+				}
+			}
 
-	err = s.Update(tempKey, []byte("test-pod1"))
-	if err != nil {
-		t.Errorf("Got error %v, unable update key %s", err, tempKey)
-	}
+			for key, data := range tc.updateKeys {
+				err = s.Update(key, []byte(data))
+				if err != nil {
+					if tc.updateErr != err {
+						t.Errorf("%s: expect error %v, but got %v", k, tc.updateErr, err)
+					}
+				}
+			}
 
-	createdFile := filepath.Join(testDir, tempKey)
-	if fi, err := os.Stat(createdFile); err != nil {
-		t.Errorf("Got error %v, wanted file %q to be there", err, createdFile)
-	} else if !fi.Mode().IsRegular() {
-		t.Errorf("Got %q not a regular file", createdFile)
-	}
+			for key, data := range tc.result {
+				b, err := s.Get(key)
+				if err != nil {
+					t.Errorf("%s: Got error %v, unable get key %s", k, err, key)
+				}
 
-	b, err := ioutil.ReadFile(createdFile)
-	if err != nil {
-		t.Errorf("Got error %v, unable read regular file %q", err, createdFile)
-	} else if !bytes.Equal(b, []byte("test-pod1")) {
-		t.Errorf("Wanted string: test-pod1 but got %s", string(b))
+				if data != string(b) {
+					t.Errorf("%s: expect updated data %s, but got %s", k, data, string(b))
+				}
+			}
+
+			for key := range testcases {
+				err = s.Delete(key)
+				if err != nil {
+					t.Errorf("%s: failed to delete key(%s), %v", k, key, err)
+				}
+			}
+		})
 	}
 
 	if err = os.RemoveAll(testDir); err != nil {
@@ -489,37 +648,70 @@ func TestUpdate(t *testing.T) {
 	}
 }
 
-func TestUpdateEmptyString(t *testing.T) {
-	s, err := NewDiskStorage(testDir)
-	if err != nil {
-		t.Fatalf("unable to new disk storage, %v", err)
-	}
-
-	err = s.Create(tempKey, []byte("test-pod"))
-	if err != nil {
-		t.Errorf("Got error %v, wanted successful create %s", err, tempKey)
-	}
-
-	err = s.Update(tempKey, []byte(""))
-	if err != nil {
-		t.Errorf("Got error %v, unable update key %s", err, tempKey)
-	}
+func TestLockKey(t *testing.T) {
+	testcases := map[string]struct {
+		preLockedKeys    []string
+		lockKey          string
+		lockResult       bool
+		lockVerifyResult bool
+	}{
+		"normal lock key": {
+			preLockedKeys:    []string{},
+			lockKey:          "kubelet/pods/kube-system/foo",
+			lockResult:       true,
+			lockVerifyResult: false,
+		},
+		"lock pre-locked key with same key": {
+			preLockedKeys:    []string{"kubelet/pods/kube-system/foo"},
+			lockKey:          "kubelet/pods/kube-system/foo",
+			lockResult:       false,
+			lockVerifyResult: false,
+		},
+		"lock child key of pre-locked key": {
+			preLockedKeys:    []string{"kubelet/pods/kube-system"},
+			lockKey:          "kubelet/pods/kube-system/foo",
+			lockResult:       false,
+			lockVerifyResult: false,
+		},
+		"lock parent key of pre-locked key ": {
+			preLockedKeys:    []string{"kubelet/pods/kube-system/foo"},
+			lockKey:          "kubelet/pods/kube-system",
+			lockResult:       false,
+			lockVerifyResult: false,
+		},
+		"lock different key with pre-locked key ": {
+			preLockedKeys:    []string{"kubelet/pods/kube-system/foo1"},
+			lockKey:          "kubelet/pods/kube-system/foo2",
+			lockResult:       true,
+			lockVerifyResult: false,
+		},
+	}
+
+	s := &diskStorage{
+		keyPendingStatus: make(map[string]struct{}),
+		baseDir:          testDir,
+	}
+
+	for k, tc := range testcases {
+		t.Run(k, func(t *testing.T) {
+			for _, key := range tc.preLockedKeys {
+				s.lockKey(key)
+			}
 
-	createdFile := filepath.Join(testDir, tempKey)
-	if fi, err := os.Stat(createdFile); err != nil {
-		t.Errorf("Got error %v, wanted file %q to be there", err, createdFile)
-	} else if !fi.Mode().IsRegular() {
-		t.Errorf("Got %q not a regular file", createdFile)
-	}
+			lResult := s.lockKey(tc.lockKey)
+			if lResult != tc.lockResult {
+				t.Errorf("%s: expect lock result %v, but got %v", k, tc.lockResult, lResult)
+			}
 
-	b, err := ioutil.ReadFile(createdFile)
-	if err != nil {
-		t.Errorf("Got error %v, unable read regular file %q", err, createdFile)
-	} else if len(b) == 0 {
-		t.Errorf("Wanted string: empty string but got %s", string(b))
-	}
+			lResult2 := s.lockKey(tc.lockKey)
+			if lResult2 != tc.lockVerifyResult {
+				t.Errorf("%s: expect lock verify result %v, but got %v", k, tc.lockVerifyResult, lResult2)
+			}
 
-	if err = os.RemoveAll(testDir); err != nil {
-		t.Errorf("Got error %v, unable remove path %s", err, testDir)
+			s.unLockKey(tc.lockKey)
+			for _, key := range tc.preLockedKeys {
+				s.unLockKey(key)
+			}
+		})
 	}
 }