Optimize portlayer volume cache rebuild on startup

This commit modifies the portlayer volume cache's rebuildCache func to
only process the volumes from the volume store that is currently being
added to the cache. rebuildCache is invoked for every volume store
during portlayer startup.

Before this change, rebuildCache would process volumes from all volume
stores in the cache every time a volume store was added. This led to
unneeded extra computation which could slow down portlayer startup and
overwhelm NFS endpoints if NFS volume stores are being used.

Fixes #6991

lib/portlayer/storage/volume_cache.go

@@ -1,4 +1,4 @@
-// Copyright 2016-2017 VMware, Inc. All Rights Reserved.
+// Copyright 2016-2018 VMware, Inc. All Rights Reserved.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -59,7 +59,7 @@ func (v *VolumeLookupCache) GetVolumeStore(op trace.Operation, storeName string)
 	return u, nil
 }
 
-// AddStore adds a volumestore by name.  The url returned is the service url to the volume store.
+// AddStore adds a volumestore by name. The url returned is the service url to the volume store.
 func (v *VolumeLookupCache) AddStore(op trace.Operation, storeName string, vs VolumeStorer) (*url.URL, error) {
 	v.vlcLock.Lock()
 	defer v.vlcLock.Unlock()
@@ -70,12 +70,13 @@ func (v *VolumeLookupCache) AddStore(op trace.Operation, storeName string, vs Vo
 		return nil, err
 	}
 
-	if _, ok := v.volumeStores[u.String()]; ok {
-		return nil, fmt.Errorf("volumestore (%s) already added", u.String())
+	storeURLStr := u.String()
+	if _, ok := v.volumeStores[storeURLStr]; ok {
+		return nil, fmt.Errorf("volumestore (%s) already added", storeURLStr)
 	}
 
-	v.volumeStores[u.String()] = vs
-	return u, v.rebuildCache(op)
+	v.volumeStores[storeURLStr] = vs
+	return u, v.addVolumesToCache(op, storeURLStr, vs)
 }
 
 func (v *VolumeLookupCache) volumeStore(store *url.URL) (VolumeStorer, error) {
@@ -94,7 +95,6 @@ func (v *VolumeLookupCache) volumeStore(store *url.URL) (VolumeStorer, error) {
 
 // VolumeStoresList returns a list of volume store names
 func (v *VolumeLookupCache) VolumeStoresList(op trace.Operation) ([]string, error) {
-
 	v.vlcLock.RLock()
 	defer v.vlcLock.RUnlock()
 
@@ -201,21 +201,19 @@ func (v *VolumeLookupCache) VolumesList(op trace.Operation) ([]*Volume, error) {
 	return l, nil
 }
 
-// goto the volume store and repopulate the cache.
-func (v *VolumeLookupCache) rebuildCache(op trace.Operation) error {
-	op.Infof("Refreshing volume cache.")
+// addVolumesToCache finds the volumes in the input volume store and adds them to the cache.
+func (v *VolumeLookupCache) addVolumesToCache(op trace.Operation, storeURLStr string, vs VolumeStorer) error {
+	op.Infof("Adding volumes in volume store %s to volume cache", storeURLStr)
 
-	for _, vs := range v.volumeStores {
-		vols, err := vs.VolumesList(op)
-		if err != nil {
-			return err
-		}
+	vols, err := vs.VolumesList(op)
+	if err != nil {
+		return err
+	}
 
-		for _, vol := range vols {
-			log.Infof("Volumestore: Found vol %s on store %s.", vol.ID, vol.Store)
-			// Add it to the cache.
-			v.vlc[vol.ID] = *vol
-		}
+	for _, vol := range vols {
+		log.Infof("Volumestore: Found vol %s on store %s", vol.ID, vol.Store)
+		// Add it to the cache.
+		v.vlc[vol.ID] = *vol
 	}
 
 	return nil

lib/portlayer/storage/volume_cache_test.go

@@ -1,4 +1,4 @@
-// Copyright 2016-2017 VMware, Inc. All Rights Reserved.
+// Copyright 2016-2018 VMware, Inc. All Rights Reserved.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -95,7 +95,7 @@ func (m *MockVolumeStore) VolumeDestroy(op trace.Operation, vol *Volume) error {
 	return nil
 }
 
-// Lists all volumes on the given volume store`
+// VolumesList lists all volumes on the given volume store.
 func (m *MockVolumeStore) VolumesList(op trace.Operation) ([]*Volume, error) {
 	var i int
 	list := make([]*Volume, len(m.db))
@@ -214,6 +214,52 @@ func TestVolumeCreateGetListAndDelete(t *testing.T) {
 	}
 }
 
+// createVolumes is a test helper that creates a set of num volumes on the input volume cache and volume store.
+func createVolumes(t *testing.T, op trace.Operation, v *VolumeLookupCache, storeURL *url.URL, num int) map[string]*Volume {
+	vols := make(map[string]*Volume)
+	for i := 1; i <= num; i++ {
+		id := fmt.Sprintf("ID-%d", i)
+
+		// Write to the datastore
+		vol, err := v.VolumeCreate(op, id, storeURL, 0, nil)
+		if !assert.NoError(t, err) || !assert.NotNil(t, vol) {
+			return nil
+		}
+
+		vols[id] = vol
+	}
+
+	return vols
+}
+
+func TestAddVolumesToCache(t *testing.T) {
+	mvs1 := NewMockVolumeStore()
+	op := trace.NewOperation(context.Background(), "test")
+	v := NewVolumeLookupCache(op)
+
+	storeURL, err := util.VolumeStoreNameToURL("testStore")
+	assert.NotNil(t, storeURL)
+	storeURLStr := storeURL.String()
+	v.volumeStores[storeURLStr] = mvs1
+
+	// Create 50 volumes on the volume store.
+	vols := createVolumes(t, op, v, storeURL, 50)
+
+	// Clear the volume map after it has been filled during volume creation.
+	v.vlc = make(map[string]Volume)
+
+	err = v.addVolumesToCache(op, storeURLStr, mvs1)
+	assert.Nil(t, err)
+
+	// Check that the volume map is intact again in the cache.
+	for _, expectedVol := range vols {
+		vol, err := v.VolumeGet(op, expectedVol.ID)
+		if !assert.NoError(t, err) || !assert.Equal(t, expectedVol, vol) {
+			return
+		}
+	}
+}
+
 // Create 2 store caches but use the same backing datastore.  Create images
 // with the first cache, then get the image with the second.  This simulates
 // restart since the second cache is empty and has to go to the backing store.
@@ -227,26 +273,15 @@ func TestVolumeCacheRestart(t *testing.T) {
 		return
 	}
 
-	// Create a set of volumes
-	inVols := make(map[string]*Volume)
-	for i := 1; i < 50; i++ {
-		id := fmt.Sprintf("ID-%d", i)
-
-		// Write to the datastore
-		vol, err := firstCache.VolumeCreate(op, id, storeURL, 0, nil)
-		if !assert.NoError(t, err) || !assert.NotNil(t, vol) {
-			return
-		}
-
-		inVols[id] = vol
-	}
+	// Create a set of 50 volumes.
+	inVols := createVolumes(t, op, firstCache, storeURL, 50)
 
 	secondCache := NewVolumeLookupCache(op)
 	if !assert.NotNil(t, secondCache) {
 		return
 	}
 
-	_, err = secondCache.AddStore(op, "testStore", mvs)
+	storeURL, err = secondCache.AddStore(op, "testStore", mvs)
 	if !assert.NoError(t, err) || !assert.NotNil(t, storeURL) {
 		return
 	}