refactored cache

Signed-off-by: Viktor Kramarenko <[email protected]>

images/agent/src/pkg/cache/cache.go

@@ -1,25 +1,39 @@
 package cache
 
 import (
-	"agent/internal"
-	"agent/pkg/logger"
 	"bytes"
 	"fmt"
+	"sync"
+
+	"agent/internal"
+	"agent/pkg/logger"
+)
+
+const (
+	lvcount = 50
 )
 
 type Cache struct {
+	m          sync.RWMutex
 	devices    []internal.Device
 	deviceErrs bytes.Buffer
 	pvs        []internal.PVData
 	pvsErrs    bytes.Buffer
 	vgs        []internal.VGData
 	vgsErrs    bytes.Buffer
-	lvs        []internal.LVData
+	lvs        map[string]*LVData
 	lvsErrs    bytes.Buffer
 }
 
+type LVData struct {
+	Data  internal.LVData
+	Exist bool
+}
+
 func New() *Cache {
-	return &Cache{}
+	return &Cache{
+		lvs: make(map[string]*LVData, lvcount),
+	}
 }
 
 func (c *Cache) StoreDevices(devices []internal.Device, stdErr bytes.Buffer) {
@@ -59,37 +73,69 @@ func (c *Cache) GetVGs() ([]internal.VGData, bytes.Buffer) {
 }
 
 func (c *Cache) StoreLVs(lvs []internal.LVData, stdErr bytes.Buffer) {
-	c.lvs = lvs
+	lvsOnNode := make(map[string]internal.LVData, len(lvs))
+	for _, lv := range lvs {
+		lvsOnNode[c.configureLVKey(lv.VGName, lv.LVName)] = lv
+	}
+
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	for _, lv := range lvsOnNode {
+		k := c.configureLVKey(lv.VGName, lv.LVName)
+		if cachedLV, exist := c.lvs[k]; !exist || cachedLV.Exist {
+			c.lvs[k] = &LVData{
+				Data:  lv,
+				Exist: true,
+			}
+		}
+	}
+
+	for key, lv := range c.lvs {
+		if lv.Exist {
+			continue
+		}
+
+		if _, exist := lvsOnNode[key]; !exist {
+			delete(c.lvs, key)
+		}
+	}
+
 	c.lvsErrs = stdErr
 }
 
 func (c *Cache) GetLVs() ([]internal.LVData, bytes.Buffer) {
 	dst := make([]internal.LVData, len(c.lvs))
-	copy(dst, c.lvs)
 
-	return dst, c.lvsErrs
-}
-
-func (c *Cache) FindLV(vgName, lvName string) *internal.LVData {
+	c.m.RLock()
+	defer c.m.RUnlock()
 	for _, lv := range c.lvs {
-		if lv.VGName == vgName && lv.LVName == lvName {
-			return &lv
-		}
+		dst = append(dst, lv.Data)
 	}
 
-	return nil
+	return dst, c.lvsErrs
+}
+
+func (c *Cache) FindLV(vgName, lvName string) *LVData {
+	c.m.RLock()
+	defer c.m.RUnlock()
+	return c.lvs[c.configureLVKey(vgName, lvName)]
 }
 
 func (c *Cache) AddLV(vgName, lvName string) {
-	c.lvs = append(c.lvs, internal.LVData{VGName: vgName, LVName: lvName})
+	c.m.Lock()
+	defer c.m.Unlock()
+	c.lvs[c.configureLVKey(vgName, lvName)] = &LVData{
+		Data:  internal.LVData{VGName: vgName, LVName: lvName},
+		Exist: true,
+	}
 }
 
 func (c *Cache) RemoveLV(vgName, lvName string) {
-	for i, lv := range c.lvs {
-		if lv.VGName == vgName && lv.LVName == lvName {
-			c.lvs = append(c.lvs[:i], c.lvs[i+1:]...)
-		}
-	}
+	c.m.Lock()
+	defer c.m.Unlock()
+
+	c.lvs[c.configureLVKey(vgName, lvName)].Exist = false
 }
 
 func (c *Cache) FindVG(vgName string) *internal.VGData {
@@ -126,11 +172,17 @@ func (c *Cache) PrintTheCache(log logger.Logger) {
 	log.Cache(c.vgsErrs.String())
 	log.Cache("[VGs ENDS]")
 	log.Cache("[LVs BEGIN]")
+	c.m.RLock()
 	for _, lv := range c.lvs {
-		log.Cache(fmt.Sprintf("     LV Name: %s, VG name: %s, size: %s, tags: %s, attr: %s, pool: %s", lv.LVName, lv.VGName, lv.LVSize.String(), lv.LvTags, lv.LVAttr, lv.PoolName))
+		log.Cache(fmt.Sprintf("     Data Name: %s, VG name: %s, size: %s, tags: %s, attr: %s, pool: %s", lv.Data.LVName, lv.Data.VGName, lv.Data.LVSize.String(), lv.Data.LvTags, lv.Data.LVAttr, lv.Data.PoolName))
 	}
+	c.m.RUnlock()
 	log.Cache("[ERRS]")
 	log.Cache(c.lvsErrs.String())
 	log.Cache("[LVs ENDS]")
 	log.Cache("*****************CACHE ENDS*****************")
 }
+
+func (c *Cache) configureLVKey(vgName, lvName string) string {
+	return fmt.Sprintf("%s/%s", vgName, lvName)
+}

images/agent/src/pkg/controller/lvm_logical_volume_extender_watcher.go

@@ -149,7 +149,7 @@ func ReconcileLVMLogicalVolumeExtension(ctx context.Context, cl client.Client, m
 
 		lv := sdsCache.FindLV(lvg.Spec.ActualVGNameOnTheNode, llv.Spec.ActualLVNameOnTheNode)
 		if lv == nil {
-			err = fmt.Errorf("LV %s not found", llv.Spec.ActualLVNameOnTheNode)
+			err = fmt.Errorf("lv %s not found", llv.Spec.ActualLVNameOnTheNode)
 			log.Error(err, fmt.Sprintf("[ReconcileLVMLogicalVolumeExtension] unable to find LV %s of the LVMLogicalVolume %s", llv.Spec.ActualLVNameOnTheNode, llv.Name))
 			err = updateLVMLogicalVolumePhaseIfNeeded(ctx, cl, log, metrics, &llv, LLVStatusPhaseFailed, err.Error())
 			if err != nil {
@@ -159,13 +159,13 @@ func ReconcileLVMLogicalVolumeExtension(ctx context.Context, cl client.Client, m
 			continue
 		}
 
-		if utils.AreSizesEqualWithinDelta(llvRequestedSize, lv.LVSize, internal.ResizeDelta) {
+		if utils.AreSizesEqualWithinDelta(llvRequestedSize, lv.Data.LVSize, internal.ResizeDelta) {
 			log.Info(fmt.Sprintf("[ReconcileLVMLogicalVolumeExtension] the LVMLogicalVolume %s should not be extended", llv.Name))
 			continue
 		}
 
-		if llvRequestedSize.Value() < lv.LVSize.Value() {
-			log.Warning(fmt.Sprintf("[ReconcileLVMLogicalVolumeExtension] the LVMLogicalVolume %s requested size %s is less than actual one on the node %s", llv.Name, llvRequestedSize.String(), lv.LVSize.String()))
+		if llvRequestedSize.Value() < lv.Data.LVSize.Value() {
+			log.Warning(fmt.Sprintf("[ReconcileLVMLogicalVolumeExtension] the LVMLogicalVolume %s requested size %s is less than actual one on the node %s", llv.Name, llvRequestedSize.String(), lv.Data.LVSize.String()))
 			continue
 		}
 
@@ -207,18 +207,18 @@ func ReconcileLVMLogicalVolumeExtension(ctx context.Context, cl client.Client, m
 		)
 		for currentAttempts < maxAttempts {
 			lv = sdsCache.FindLV(lvg.Spec.ActualVGNameOnTheNode, llv.Spec.ActualLVNameOnTheNode)
-			if utils.AreSizesEqualWithinDelta(lv.LVSize, llvRequestedSize, internal.ResizeDelta) {
-				log.Debug(fmt.Sprintf("[ReconcileLVMLogicalVolumeExtension] LV %s of the LVMLogicalVolume %s was successfully updated in the cache", lv.LVName, llv.Name))
+			if utils.AreSizesEqualWithinDelta(lv.Data.LVSize, llvRequestedSize, internal.ResizeDelta) {
+				log.Debug(fmt.Sprintf("[ReconcileLVMLogicalVolumeExtension] LV %s of the LVMLogicalVolume %s was successfully updated in the cache", lv.Data.LVName, llv.Name))
 				break
 			}
 
-			log.Warning(fmt.Sprintf("[ReconcileLVMLogicalVolumeExtension] LV %s size of the LVMLogicalVolume %s was not yet updated in the cache, retry...", lv.LVName, llv.Name))
+			log.Warning(fmt.Sprintf("[ReconcileLVMLogicalVolumeExtension] LV %s size of the LVMLogicalVolume %s was not yet updated in the cache, retry...", lv.Data.LVName, llv.Name))
 			currentAttempts++
 			time.Sleep(1 * time.Second)
 		}
 
 		if currentAttempts == maxAttempts {
-			err = fmt.Errorf("LV %s is not updated in the cache", lv.LVName)
+			err = fmt.Errorf("LV %s is not updated in the cache", lv.Data.LVName)
 			log.Error(err, fmt.Sprintf("[ReconcileLVMLogicalVolumeExtension] unable to resize the LVMLogicalVolume %s", llv.Name))
 			shouldRetry = true
 
@@ -228,7 +228,7 @@ func ReconcileLVMLogicalVolumeExtension(ctx context.Context, cl client.Client, m
 			continue
 		}
 
-		updated, err := updateLLVPhaseToCreatedIfNeeded(ctx, cl, &llv, lv.LVSize)
+		updated, err := updateLLVPhaseToCreatedIfNeeded(ctx, cl, &llv, lv.Data.LVSize)
 		if err != nil {
 			log.Error(err, fmt.Sprintf("[ReconcileLVMLogicalVolumeExtension] unable to update the LVMLogicalVolume %s", llv.Name))
 			shouldRetry = true

images/agent/src/pkg/controller/lvm_logical_volume_watcher.go

@@ -145,10 +145,6 @@ func RunLVMLogicalVolumeWatcherController(
 				return reconcile.Result{RequeueAfter: cfg.LLVRequeueIntervalSec}, nil
 			}
 
-			// создаем LV и пытаемся получить его размер на 113278 строчке I0819 12:43:10.563596
-			// при этом у нас кэш начинает заполняться в I0819 12:43:09.486602 и заканчивает в I0819 12:43:22.070604, а потом только в I0819 12:43:22.081861
-			// при этом получаем ретрай в I0819 12:43:15.563851 (и на мо
-
 			log.Info(fmt.Sprintf("[RunLVMLogicalVolumeWatcherController] successfully ended reconciliation of the LVMLogicalVolume %s", request.Name))
 			return reconcile.Result{}, nil
 		}),

images/agent/src/pkg/controller/lvm_logical_volume_watcher_func.go

@@ -152,14 +152,14 @@ func updateLLVPhaseToCreatedIfNeeded(ctx context.Context, cl client.Client, llv
 
 func deleteLVIfNeeded(log logger.Logger, sdsCache *cache.Cache, vgName string, llv *v1alpha1.LVMLogicalVolume) error {
 	lv := sdsCache.FindLV(vgName, llv.Spec.ActualLVNameOnTheNode)
-	if lv == nil {
+	if lv == nil || !lv.Exist {
 		log.Warning(fmt.Sprintf("[deleteLVIfNeeded] did not find LV %s in VG %s", llv.Spec.ActualLVNameOnTheNode, vgName))
 		return nil
 	}
 
 	// this case prevents unexpected same-name LV deletions which does not actually belong to our LLV
-	if !checkIfLVBelongsToLLV(llv, lv) {
-		log.Warning(fmt.Sprintf("[deleteLVIfNeeded] no need to delete LV %s as it doesnt belong to LVMLogicalVolume %s", lv.LVName, llv.Name))
+	if !checkIfLVBelongsToLLV(llv, &lv.Data) {
+		log.Warning(fmt.Sprintf("[deleteLVIfNeeded] no need to delete LV %s as it doesnt belong to LVMLogicalVolume %s", lv.Data.LVName, llv.Name))
 		return nil
 	}
 
@@ -170,8 +170,8 @@ func deleteLVIfNeeded(log logger.Logger, sdsCache *cache.Cache, vgName string, l
 		return err
 	}
 
-	log.Debug(fmt.Sprintf("[deleteLVIfNeeded] removes LV %s from the cache", lv.LVName))
-	sdsCache.RemoveLV(lv.VGName, lv.LVName)
+	log.Debug(fmt.Sprintf("[deleteLVIfNeeded] removes LV %s from the cache", lv.Data.LVName))
+	sdsCache.RemoveLV(lv.Data.VGName, lv.Data.LVName)
 
 	return nil
 }
@@ -182,7 +182,7 @@ func getLVActualSize(sdsCache *cache.Cache, vgName, lvName string) resource.Quan
 		return resource.Quantity{}
 	}
 
-	result := resource.NewQuantity(lv.LVSize.Value(), resource.BinarySI)
+	result := resource.NewQuantity(lv.Data.LVSize.Value(), resource.BinarySI)
 
 	return *result
 }
@@ -298,10 +298,10 @@ func validateLVMLogicalVolume(sdsCache *cache.Cache, llv *v1alpha1.LVMLogicalVol
 	// if a specified Thick LV name matches the existing Thin one
 	lv := sdsCache.FindLV(lvg.Spec.ActualVGNameOnTheNode, llv.Spec.ActualLVNameOnTheNode)
 	if lv != nil {
-		if len(lv.LVAttr) == 0 {
-			reason.WriteString(fmt.Sprintf("LV %s was found on the node, but can't be validated due to its attributes is empty string. ", lv.LVName))
-		} else if !checkIfLVBelongsToLLV(llv, lv) {
-			reason.WriteString(fmt.Sprintf("Specified LV %s is already created and it is doesnt match the one on the node.", lv.LVName))
+		if len(lv.Data.LVAttr) == 0 {
+			// this means we found a LV added by the controller (it is empty template)
+		} else if !checkIfLVBelongsToLLV(llv, &lv.Data) {
+			reason.WriteString(fmt.Sprintf("Specified LV %s is already created and it is doesnt match the one on the node.", lv.Data.LVName))
 		}
 	}
 
@@ -347,7 +347,7 @@ func shouldReconcileByUpdateFunc(sdsCache *cache.Cache, vgName string, llv *v1al
 	}
 
 	lv := sdsCache.FindLV(vgName, llv.Spec.ActualLVNameOnTheNode)
-	return lv != nil
+	return lv != nil && lv.Exist
 }
 
 func isContiguous(llv *v1alpha1.LVMLogicalVolume) bool {

images/agent/src/pkg/controller/lvm_logical_volume_watcher_test.go

@@ -153,7 +153,7 @@ func TestLVMLogicaVolumeWatcher(t *testing.T) {
 
 			v, r := validateLVMLogicalVolume(sdsCache, llv, &v1alpha1.LvmVolumeGroup{})
 			if assert.False(t, v) {
-				assert.Equal(t, "Zero size for LV. Thin pool specified for Thick LV. LV test-lv was found on the node, but can't be validated due to its attributes is empty string. ", r)
+				assert.Equal(t, "Zero size for Data. Thin pool specified for Thick Data. ", r)
 			}
 		})
 
@@ -212,7 +212,7 @@ func TestLVMLogicaVolumeWatcher(t *testing.T) {
 
 			v, r := validateLVMLogicalVolume(sdsCache, llv, &v1alpha1.LvmVolumeGroup{})
 			if assert.False(t, v) {
-				assert.Equal(t, "No LV name specified. Zero size for LV. No thin pool specified. ", r)
+				assert.Equal(t, "No Data name specified. Zero size for Data. No thin pool specified. ", r)
 			}
 		})
 	})

images/agent/src/pkg/controller/lvm_volume_group_discover.go

@@ -145,7 +145,7 @@ func LVMVolumeGroupDiscoverReconcile(ctx context.Context, cl kclient.Client, met
 
 			if !hasLVMVolumeGroupDiff(log, lvg, candidate) {
 				log.Debug(fmt.Sprintf(`[RunLVMVolumeGroupDiscoverController] no data to update for LvmVolumeGroup, name: "%s"`, lvg.Name))
-				err = updateLVGConditionIfNeeded(ctx, cl, log, &lvg, metav1.ConditionTrue, internal.TypeVGReady, internal.ReasonUpdated, "ready to create LV")
+				err = updateLVGConditionIfNeeded(ctx, cl, log, &lvg, metav1.ConditionTrue, internal.TypeVGReady, internal.ReasonUpdated, "ready to create Data")
 				if err != nil {
 					log.Error(err, fmt.Sprintf("[RunLVMVolumeGroupDiscoverController] unable to add a condition %s to the LVMVolumeGroup %s", internal.TypeVGReady, lvg.Name))
 					shouldRequeue = true
@@ -178,7 +178,7 @@ func LVMVolumeGroupDiscoverReconcile(ctx context.Context, cl kclient.Client, met
 				continue
 			}
 
-			err = updateLVGConditionIfNeeded(ctx, cl, log, &lvg, metav1.ConditionTrue, internal.TypeVGReady, internal.ReasonUpdated, "ready to create LV")
+			err = updateLVGConditionIfNeeded(ctx, cl, log, &lvg, metav1.ConditionTrue, internal.TypeVGReady, internal.ReasonUpdated, "ready to create Data")
 			if err != nil {
 				log.Error(err, fmt.Sprintf("[RunLVMVolumeGroupDiscoverController] unable to add a condition %s to the LVMVolumeGroup %s", internal.TypeVGReady, lvg.Name))
 				shouldRequeue = true
@@ -447,7 +447,7 @@ func GetLVMVolumeGroupCandidates(log logger.Logger, sdsCache *cache.Cache, bds m
 	lvs, lvErrs := sdsCache.GetLVs()
 	var thinPools []internal.LVData
 	if len(lvs) > 0 {
-		// Filter LV to get only thin pools as we do not support thick for now.
+		// Filter Data to get only thin pools as we do not support thick for now.
 		thinPools = getThinPools(lvs)
 	}
 
@@ -458,7 +458,7 @@ func GetLVMVolumeGroupCandidates(log logger.Logger, sdsCache *cache.Cache, bds m
 		lvIssues = sortThinPoolIssuesByVG(log, thinPools)
 	}
 
-	// Sort PV,BlockDevices and LV by VG to fill needed information for LVMVolumeGroup resource further.
+	// Sort PV,BlockDevices and Data by VG to fill needed information for LVMVolumeGroup resource further.
 	sortedPVs := sortPVsByVG(pvs, vgWithTag)
 	sortedBDs := sortBlockDevicesByVG(bds, vgWithTag)
 	log.Trace(fmt.Sprintf("[GetLVMVolumeGroupCandidates] BlockDevices: %+v", bds))
@@ -732,7 +732,7 @@ func getStatusThinPools(log logger.Logger, thinPools, sortedLVs map[string][]int
 
 	for _, thinPool := range tps {
 		usedSize, err := getThinPoolUsedSize(thinPool)
-		log.Trace(fmt.Sprintf("[getStatusThinPools] LV %v for VG name %s", thinPool, vg.VGName))
+		log.Trace(fmt.Sprintf("[getStatusThinPools] Data %v for VG name %s", thinPool, vg.VGName))
 		if err != nil {
 			log.Error(err, "[getStatusThinPools] unable to getThinPoolUsedSize")
 		}
@@ -919,7 +919,7 @@ func UpdateLVMVolumeGroupByCandidate(
 		return fmt.Errorf(`[UpdateLVMVolumeGroupByCandidate] unable to update LVMVolumeGroup, name: "%s", err: %w`, lvg.Name, err)
 	}
 
-	err = updateLVGConditionIfNeeded(ctx, cl, log, lvg, metav1.ConditionTrue, internal.TypeVGReady, internal.ReasonUpdated, "ready to create LV")
+	err = updateLVGConditionIfNeeded(ctx, cl, log, lvg, metav1.ConditionTrue, internal.TypeVGReady, internal.ReasonUpdated, "ready to create Data")
 	if err != nil {
 		log.Error(err, fmt.Sprintf("[UpdateLVMVolumeGroupByCandidate] unable to add a condition %s to the LVMVolumeGroup %s", internal.TypeVGReady, lvg.Name))
 	}

images/agent/src/pkg/controller/lvm_volume_group_watcher.go

@@ -282,7 +282,7 @@ func reconcileLVGDeleteFunc(ctx context.Context, cl client.Client, log logger.Lo
 	if len(usedLVs) > 0 {
 		err := fmt.Errorf("VG %s uses LVs: %v. Delete used LVs first", lvg.Spec.ActualVGNameOnTheNode, usedLVs)
 		log.Error(err, fmt.Sprintf("[reconcileLVGDeleteFunc] unable to reconcile LVG %s", lvg.Name))
-		log.Debug(fmt.Sprintf("[reconcileLVGDeleteFunc] tries to add the condition %s status False to the LVMVolumeGroup %s due to LV does exist", internal.TypeVGConfigurationApplied, lvg.Name))
+		log.Debug(fmt.Sprintf("[reconcileLVGDeleteFunc] tries to add the condition %s status False to the LVMVolumeGroup %s due to Data does exist", internal.TypeVGConfigurationApplied, lvg.Name))
 		err = updateLVGConditionIfNeeded(ctx, cl, log, lvg, v1.ConditionFalse, internal.TypeVGConfigurationApplied, internal.ReasonTerminating, err.Error())
 		if err != nil {
 			log.Error(err, fmt.Sprintf("[reconcileLVGDeleteFunc] unable to add the condition %s to the LVMVolumeGroup %s", internal.TypeVGConfigurationApplied, lvg.Name))
@@ -292,7 +292,7 @@ func reconcileLVGDeleteFunc(ctx context.Context, cl client.Client, log logger.Lo
 		return true, nil
 	}
 
-	log.Debug(fmt.Sprintf("[reconcileLVGDeleteFunc] VG %s of the LVMVolumeGroup %s does not use any LV. Start to delete the VG", lvg.Spec.ActualVGNameOnTheNode, lvg.Name))
+	log.Debug(fmt.Sprintf("[reconcileLVGDeleteFunc] VG %s of the LVMVolumeGroup %s does not use any Data. Start to delete the VG", lvg.Spec.ActualVGNameOnTheNode, lvg.Name))
 	err := DeleteVGIfExist(log, metrics, sdsCache, lvg.Spec.ActualVGNameOnTheNode)
 	if err != nil {
 		log.Error(err, fmt.Sprintf("[reconcileLVGDeleteFunc] unable to delete VG %s", lvg.Spec.ActualVGNameOnTheNode))

images/agent/src/pkg/controller/lvm_volume_group_watcher_func.go

@@ -412,12 +412,12 @@ func validateLVGForUpdateFunc(log logger.Logger, sdsCache *cache.Cache, lvg *v1a
 		for _, tp := range lvg.Spec.ThinPools {
 			lv := sdsCache.FindLV(lvg.Spec.ActualVGNameOnTheNode, tp.Name)
 			if lv != nil {
-				if !isThinPool(*lv) {
-					reason.WriteString(fmt.Sprintf("LV %s is already created on the node and it is not a thin-pool", lv.LVName))
+				if !isThinPool(lv.Data) {
+					reason.WriteString(fmt.Sprintf("Data %s is already created on the node and it is not a thin-pool", lv.Data.LVName))
 					continue
 				}
 
-				actualThinPools[lv.LVName] = *lv
+				actualThinPools[lv.Data.LVName] = lv.Data
 			}
 		}
 
@@ -941,7 +941,7 @@ func ExtendThinPool(log logger.Logger, metrics monitoring.Metrics, lvg *v1alpha1
 	metrics.UtilsCommandsExecutionCount(LVMVolumeGroupWatcherCtrlName, "lvextend").Inc()
 	if err != nil {
 		metrics.UtilsCommandsErrorsCount(LVMVolumeGroupWatcherCtrlName, "lvextend").Inc()
-		log.Error(err, fmt.Sprintf("[ExtendThinPool] unable to extend LV, name: %s, cmd: %s", specThinPool.Name, cmd))
+		log.Error(err, fmt.Sprintf("[ExtendThinPool] unable to extend Data, name: %s, cmd: %s", specThinPool.Name, cmd))
 		return err
 	}