Skip to content

Commit

Permalink
first draft
Browse files Browse the repository at this point in the history
Signed-off-by: Viktor Kramarenko <[email protected]>
  • Loading branch information
ViktorKram committed Sep 26, 2024
1 parent dc89220 commit 88e4f55
Show file tree
Hide file tree
Showing 2 changed files with 54 additions and 15 deletions.
60 changes: 51 additions & 9 deletions images/sds-local-volume-scheduler-extender/src/pkg/cache/cache.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"errors"
"fmt"
"sync"
"time"

snc "github.com/deckhouse/sds-node-configurator/api/v1alpha1"
v1 "k8s.io/api/core/v1"
Expand All @@ -21,10 +22,11 @@ const (
)

type Cache struct {
lvgs sync.Map // map[string]*lvgCache
pvcLVGs sync.Map // map[string][]string
nodeLVGs sync.Map // map[string][]string
log logger.Logger
lvgs sync.Map // map[string]*lvgCache
pvcLVGs sync.Map // map[string][]string
nodeLVGs sync.Map // map[string][]string
log logger.Logger
expiredDuration time.Duration
}

type lvgCache struct {
Expand All @@ -44,17 +46,57 @@ type pvcCache struct {

// NewCache initialize new cache.
func NewCache(logger logger.Logger) *Cache {
return &Cache{
log: logger,
ch := &Cache{
log: logger,
expiredDuration: 30 * time.Second,
}

go func() {
timer := time.NewTimer(ch.expiredDuration)

for {
select {
case <-timer.C:
ch.clearBoundExpiredPVC()
timer.Reset(ch.expiredDuration)
}
}
}()
return ch
}

func (c *Cache) clearBoundExpiredPVC() {
c.log.Debug("[clearBoundExpiredPVC] starts to clear expired PVC")
c.lvgs.Range(func(lvgName, _ any) bool {
pvcs, err := c.GetAllPVCForLVG(lvgName.(string))
if err != nil {
c.log.Error(err, fmt.Sprintf("[clearBoundExpiredPVC] unable to get PVCs for the LVMVolumeGroup %s", lvgName.(string)))
return false
}

for _, pvc := range pvcs {
if pvc.Status.Phase != v1.ClaimBound {
c.log.Trace(fmt.Sprintf("[clearBoundExpiredPVC] PVC %s is not in a Bound state", pvc.Name))
continue
}

if time.Now().Sub(pvc.CreationTimestamp.Time) > c.expiredDuration {
c.log.Warning(fmt.Sprintf("[clearBoundExpiredPVC] PVC %s is in a Bound state and expired, remove it from the cache", pvc.Name))
c.RemovePVCFromTheCache(pvc)
} else {
c.log.Trace(fmt.Sprintf("[clearBoundExpiredPVC] PVC %s is in a Bound state but not expired yet.", pvc.Name))
}
}

return true
})
c.log.Debug("[clearBoundExpiredPVC] finished the expired PVC clearing")
}

// AddLVG adds selected LVMVolumeGroup resource to the cache. If it is already stored, does nothing.
func (c *Cache) AddLVG(lvg *snc.LVMVolumeGroup) {
_, loaded := c.lvgs.LoadOrStore(lvg.Name, &lvgCache{
lvg: lvg,
thickPVCs: sync.Map{},
thinPools: sync.Map{},
lvg: lvg,
})
if loaded {
c.log.Debug(fmt.Sprintf("[AddLVG] the LVMVolumeGroup %s has been already added to the cache", lvg.Name))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,20 +3,18 @@ package controller
import (
"context"
"fmt"
"reflect"

snc "github.com/deckhouse/sds-node-configurator/api/v1alpha1"
v1 "k8s.io/api/core/v1"
"k8s.io/client-go/util/workqueue"
"reflect"
"sds-local-volume-scheduler-extender/pkg/cache"
"sds-local-volume-scheduler-extender/pkg/logger"
"sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/event"
"sigs.k8s.io/controller-runtime/pkg/handler"
"sigs.k8s.io/controller-runtime/pkg/manager"
"sigs.k8s.io/controller-runtime/pkg/reconcile"
"sigs.k8s.io/controller-runtime/pkg/source"

"sds-local-volume-scheduler-extender/pkg/cache"
"sds-local-volume-scheduler-extender/pkg/logger"
)

const (
Expand Down Expand Up @@ -90,7 +88,6 @@ func RunLVGWatcherCacheController(
log.Info(fmt.Sprintf("[RunCacheWatcherController] UpdateFunc starts the cache reconciliation for the LVMVolumeGroup %s", e.ObjectNew.GetName()))
oldLvg := e.ObjectOld
newLvg := e.ObjectNew

err := cache.UpdateLVG(newLvg)
if err != nil {
log.Error(err, fmt.Sprintf("[RunLVGWatcherCacheController] unable to update the LVMVolumeGroup %s cache", newLvg.Name))
Expand Down

0 comments on commit 88e4f55

Please sign in to comment.