Skip to content

Commit

Permalink
* add TTLCache
Browse files Browse the repository at this point in the history
Signed-off-by: Ivan.Makeev <[email protected]>
  • Loading branch information
Ranger-X committed Nov 29, 2024
1 parent b7c2850 commit 3d3d242
Show file tree
Hide file tree
Showing 4 changed files with 207 additions and 25 deletions.
40 changes: 35 additions & 5 deletions images/storage-network-controller/src/internal/config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,16 +19,19 @@ package config
import (
"flag"
"fmt"
"log"
"net"
"os"
"strconv"

"storage-network-controller/internal/logger"
)

const (
ControllerNamespaceEnv = "CONTROLLER_NAMESPACE"
DiscoverySecEnv = "DISCOVERY_INTERVAL_SEC"
CacheSecEnv = "CACHE_TTL_SEC"
DefaultDiscoverySec = 15
DefaultCacheTTLSec = 60
HardcodedControllerNS = "d8-sds-replicated-volume"
LogLevelEnv = "LOG_LEVEL"
)
Expand All @@ -38,6 +41,7 @@ type StorageNetworkCIDR []string
type Options struct {
ControllerNamespace string
DiscoverySec int
CacheTTLSec int
DiscoveryMode bool
Loglevel logger.Verbosity
StorageNetworkCIDR StorageNetworkCIDR
Expand Down Expand Up @@ -74,17 +78,43 @@ func NewConfig() (*Options, error) {
opts.Loglevel = logger.Verbosity(loglevel)
}

opts.DiscoverySec = DefaultDiscoverySec
discoverySec := os.Getenv(DiscoverySecEnv)
if discoverySec == "" {
opts.DiscoverySec = DefaultDiscoverySec
} else {
i, err := strconv.Atoi(discoverySec)
if err != nil {
fmt.Printf("Failed to convert value of env var %s to integer: %s", DiscoverySecEnv, err.Error())
fmt.Printf("Using default %d seconds", DefaultDiscoverySec)
opts.DiscoverySec = DefaultDiscoverySec
} else {
opts.DiscoverySec = i
}
}

cacheTTLSec := os.Getenv(CacheSecEnv)
if cacheTTLSec == "" {
opts.CacheTTLSec = DefaultCacheTTLSec
} else {
i, err := strconv.Atoi(cacheTTLSec)
if err != nil {
fmt.Printf("Failed to convert value of env var %s to integer: %s", CacheSecEnv, err.Error())
fmt.Printf("Using default %d seconds", DefaultCacheTTLSec)
opts.CacheTTLSec = DefaultCacheTTLSec
} else {
opts.CacheTTLSec = i
}
}

opts.ControllerNamespace = os.Getenv(ControllerNamespaceEnv)
if opts.ControllerNamespace == "" {
namespace, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace")
if err != nil {
log.Printf("Failed to get namespace from filesystem: %v", err)
log.Printf("Using hardcoded namespace: %s", HardcodedControllerNS)
fmt.Printf("Failed to get namespace from filesystem: %s", err.Error())
fmt.Printf("Using hardcoded namespace: %s", HardcodedControllerNS)
opts.ControllerNamespace = HardcodedControllerNS
} else {
log.Printf("Got namespace from filesystem: %s", string(namespace))
fmt.Printf("Got namespace from filesystem: %s", string(namespace))
opts.ControllerNamespace = string(namespace)
}
}
Expand Down
16 changes: 16 additions & 0 deletions images/storage-network-controller/src/pkg/cache/cache.go
Original file line number Diff line number Diff line change
@@ -1,3 +1,19 @@
/*
Copyright 2024 Flant JSC
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package cache

import (
Expand Down
131 changes: 131 additions & 0 deletions images/storage-network-controller/src/pkg/cache/ttl_cache.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,131 @@
/*
Copyright 2024 Flant JSC
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package cache

import (
"sync"
"time"
)

// item represents a cache item with a value and an expiration time.
type item[V any] struct {
value V
expiry time.Time
}

// isExpired checks if the cache item has expired.
func (i item[V]) isExpired() bool {
return time.Now().After(i.expiry)
}

// TTLCache is a generic cache implementation with support for time-to-live
// (TTL) expiration.
type TTLCache[K comparable, V any] struct {
items map[K]item[V] // The map storing cache items.
mu sync.Mutex // Mutex for controlling concurrent access to the cache.
}

// NewTTL creates a new TTLCache instance and starts a goroutine to periodically
// remove expired items every 5 seconds.
func NewTTL[K comparable, V any]() *TTLCache[K, V] {
c := &TTLCache[K, V]{
items: make(map[K]item[V]),
}

go func() {
for range time.Tick(5 * time.Second) {
c.mu.Lock()

// Iterate over the cache items and delete expired ones.
for key, item := range c.items {
if item.isExpired() {
delete(c.items, key)
}
}

c.mu.Unlock()
}
}()

return c
}

// Set adds a new item to the cache with the specified key, value, and
// time-to-live (TTL).
func (c *TTLCache[K, V]) Set(key K, value V, ttl time.Duration) {
c.mu.Lock()
defer c.mu.Unlock()

c.items[key] = item[V]{
value: value,
expiry: time.Now().Add(ttl),
}
}

// Get retrieves the value associated with the given key from the cache.
func (c *TTLCache[K, V]) Get(key K) (V, bool) {
c.mu.Lock()
defer c.mu.Unlock()

item, found := c.items[key]
if !found {
// If the key is not found, return the zero value for V and false.
return item.value, false
}

if item.isExpired() {
// If the item has expired, remove it from the cache and return the
// value and false.
delete(c.items, key)
return item.value, false
}

// Otherwise return the value and true.
return item.value, true
}

// Remove removes the item with the specified key from the cache.
func (c *TTLCache[K, V]) Remove(key K) {
c.mu.Lock()
defer c.mu.Unlock()

// Delete the item with the given key from the cache.
delete(c.items, key)
}

// Pop removes and returns the item with the specified key from the cache.
func (c *TTLCache[K, V]) Pop(key K) (V, bool) {
c.mu.Lock()
defer c.mu.Unlock()

item, found := c.items[key]
if !found {
// If the key is not found, return the zero value for V and false.
return item.value, false
}

// If the key is found, delete the item from the cache.
delete(c.items, key)

if item.isExpired() {
// If the item has expired, return the value and false.
return item.value, false
}

// Otherwise return the value and true.
return item.value, true
}
45 changes: 25 additions & 20 deletions images/storage-network-controller/src/pkg/discoverer/discoverer.go
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,8 @@ import (

type discoveredIPs []string

var DiscoveryCache cache.TTLCache[string, string]

func DiscoveryLoop(cfg config.Options, mgr manager.Manager, log *logger.Logger) error {
storageNetworks, err := parseCIDRs(cfg.StorageNetworkCIDR)
if err != nil {
Expand All @@ -55,8 +57,8 @@ func DiscoveryLoop(cfg config.Options, mgr manager.Manager, log *logger.Logger)
}
log.Info("Shared informer cache has been intialized")

// initialize in-memory node IPs cache
discoveredIPs := make(discoveredIPs, 0)
// create a new discoveryCache with TTL (item expiring) capabilities
// DiscoveryCache := cache.NewTTL[string, string]()

ctx, cancel := context.WithCancel(context.Background())
defer cancel()
Expand All @@ -74,8 +76,8 @@ func DiscoveryLoop(cfg config.Options, mgr manager.Manager, log *logger.Logger)

// discoverer loop
for {
if err := discovery(storageNetworks, &discoveredIPs, ctx, &cl, log); err != nil {
log.Error(err, "Discovery error occured")
if err := discovery(ctx, storageNetworks, &cl, log, cfg); err != nil {
log.Error(err, "Discovery error occurred")
cancel()
return err
}
Expand All @@ -98,14 +100,14 @@ func parseCIDRs(cidrs config.StorageNetworkCIDR) ([]netip.Prefix, error) {
return networks, nil
}

func discovery(storageNetworks []netip.Prefix, cachedIPs *discoveredIPs, ctx context.Context, cl *client.Client, log *logger.Logger) error {
func discovery(ctx context.Context, storageNetworks []netip.Prefix, cl *client.Client, log *logger.Logger, cfg config.Options) error {
select {
case <-ctx.Done():
// do nothing in case of cancel
return nil

default:
log.Trace(fmt.Sprintf("[discovery] storageNetworks: %s, cachedIPs: %s", storageNetworks, *cachedIPs))
log.Trace(fmt.Sprintf("[discovery] storageNetworks: %s", storageNetworks))

addrs, err := net.InterfaceAddrs()
if err != nil {
Expand All @@ -131,24 +133,27 @@ func discovery(storageNetworks []netip.Prefix, cachedIPs *discoveredIPs, ctx con
}
}

// if we found any IP that match with storageNetwork CIDRs that not already in cache
// TODO: theoretically, there is a possible case where the IP changes, but the length does not change
if len(foundedIP) != len(*cachedIPs) {
if len(foundedIP) > 0 {
log.Info(fmt.Sprintf("Founded %d storage network IPs: %s", len(foundedIP), strings.Join(foundedIP, ", ")))
*cachedIPs = foundedIP

// TODO: what if there is 2 or more IPs founded?
// for now we get only FIRST IP in founded list
ip := foundedIP[0]

// update node status only if no IP in cache
if _, found := DiscoveryCache.Get(ip); !found {
node, err := getMyNode()
if err != nil {
log.Error(err, "cannot get my node info")
return err
}

node, err := getMyNode()
if err != nil {
log.Error(err, "cannot get my node info")
return err
}

err = updateNodeStatusWithIP(ctx, node, foundedIP[0], *cl, log)
if err != nil {
log.Error(err, "cannot update node status field")
return err
err = updateNodeStatusWithIP(ctx, node, ip, *cl, log)
if err != nil {
log.Error(err, "cannot update node status field")
return err
}
DiscoveryCache.Set(ip, "", time.Duration(cfg.CacheTTLSec)*time.Second)
}
}
}
Expand Down

0 comments on commit 3d3d242

Please sign in to comment.