Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make common cache typed with generics (#2896)

Browse Source 
Making typed common cache so that we don't use wrong types
across set/get methods and more higher-level methods can be
defined for cache.
Currently, we are not able to operate over all keys of the cache
due to generic types.
I also removed code comments around the cache.

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>
This commit is contained in:
Sanket Sudake
2024-01-04 10:46:32 +05:30
committed by GitHub
parent 12f8017d9b
commit 0c8573467b
8 changed files with 105 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/buildermgr/pkgwatcher.go
+35 -55
View File
@@ -30,6 +30,7 @@ import (
fv1 "github.com/fission/fission/pkg/apis/core/v1" fv1 "github.com/fission/fission/pkg/apis/core/v1"
"github.com/fission/fission/pkg/cache" "github.com/fission/fission/pkg/cache"
"github.com/fission/fission/pkg/crd"
"github.com/fission/fission/pkg/generated/clientset/versioned" "github.com/fission/fission/pkg/generated/clientset/versioned"
"github.com/fission/fission/pkg/utils" "github.com/fission/fission/pkg/utils"
"github.com/fission/fission/pkg/utils/manager" "github.com/fission/fission/pkg/utils/manager"
@@ -45,7 +46,7 @@ type (
podInformer map[string]k8sCache.SharedIndexInformer podInformer map[string]k8sCache.SharedIndexInformer
pkgInformer map[string]k8sCache.SharedIndexInformer pkgInformer map[string]k8sCache.SharedIndexInformer
storageSvcUrl string storageSvcUrl string
buildCache *cache.Cache buildCache *cache.Cache[crd.CacheKeyUR, *fv1.Package]
} }
) )
@@ -60,13 +61,13 @@ func makePackageWatcher(logger *zap.Logger, fissionClient versioned.Interface, k
podInformer: podInformer, podInformer: podInformer,
pkgInformer: pkgInformer, pkgInformer: pkgInformer,
storageSvcUrl: storageSvcUrl, storageSvcUrl: storageSvcUrl,
buildCache: cache.MakeCache(0, 0), buildCache: cache.MakeCache[crd.CacheKeyUR, *fv1.Package](0, 0),
} }
return pkgw return pkgw
} }
func (pkgw *packageWatcher) buildCacheKey(obj metav1.ObjectMeta) string { func (pkgw *packageWatcher) buildCacheKey(obj metav1.ObjectMeta) crd.CacheKeyUR {
return fmt.Sprintf("%s-%s-%s", obj.Namespace, obj.Name, obj.ResourceVersion) return crd.CacheKeyURFromMeta(&obj)
} }
func (pkgw *packageWatcher) buildWithCache(ctx context.Context, srcpkg *fv1.Package) { func (pkgw *packageWatcher) buildWithCache(ctx context.Context, srcpkg *fv1.Package) {
@@ -90,33 +91,33 @@ func (pkgw *packageWatcher) buildWithCache(ctx context.Context, srcpkg *fv1.Pack
// 6. Update package status to succeed state // 6. Update package status to succeed state
// *. Update package status to failed state,if any one of steps above failed/time out // *. Update package status to failed state,if any one of steps above failed/time out
func (pkgw *packageWatcher) build(ctx context.Context, srcpkg *fv1.Package) { func (pkgw *packageWatcher) build(ctx context.Context, srcpkg *fv1.Package) {
key := pkgw.buildCacheKey(srcpkg.ObjectMeta)
logger := pkgw.logger.With(zap.String("package", srcpkg.Name), zap.String("namespace", srcpkg.Namespace), zap.String("resource_version", srcpkg.ResourceVersion), zap.String("key", key.String()))
defer func() { defer func() {
key := pkgw.buildCacheKey(srcpkg.ObjectMeta)
err := pkgw.buildCache.Delete(key) err := pkgw.buildCache.Delete(key)
if err != nil { if err != nil {
pkgw.logger.Error("error deleting key from cache", zap.String("key", key), zap.Error(err)) logger.Error("error deleting key from cache", zap.Any("key", key), zap.Error(err))
} }
}() }()
pkgw.logger.Info("starting build for package", zap.String("package_name", srcpkg.ObjectMeta.Name), zap.String("resource_version", srcpkg.ObjectMeta.ResourceVersion)) logger.Info("starting build for package")
pkg, err := updatePackage(ctx, pkgw.logger, pkgw.fissionClient, srcpkg, fv1.BuildStatusRunning, "", nil) pkg, err := updatePackage(ctx, logger, pkgw.fissionClient, srcpkg, fv1.BuildStatusRunning, "", nil)
if err != nil { if err != nil {
pkgw.logger.Error("error setting package pending state", zap.Error(err)) logger.Error("error setting package pending state", zap.Error(err))
return return
} }
env, err := pkgw.fissionClient.CoreV1().Environments(pkg.Spec.Environment.Namespace).Get(ctx, pkg.Spec.Environment.Name, metav1.GetOptions{}) env, err := pkgw.fissionClient.CoreV1().Environments(pkg.Spec.Environment.Namespace).Get(ctx, pkg.Spec.Environment.Name, metav1.GetOptions{})
if k8serrors.IsNotFound(err) { if k8serrors.IsNotFound(err) {
e := "environment does not exist" e := "environment does not exist"
pkgw.logger.Error(e, zap.String("environment", pkg.Spec.Environment.Name)) logger.Error(e, zap.String("environment", pkg.Spec.Environment.Name))
_, er := updatePackage(ctx, pkgw.logger, pkgw.fissionClient, pkg, _, er := updatePackage(ctx, logger, pkgw.fissionClient, pkg,
fv1.BuildStatusFailed, fmt.Sprintf("%s: %q", e, pkg.Spec.Environment.Name), nil) fv1.BuildStatusFailed, fmt.Sprintf("%s: %q", e, pkg.Spec.Environment.Name), nil)
if er != nil { if er != nil {
pkgw.logger.Error( logger.Error(
"error updating package", "error updating package",
zap.String("package_name", pkg.ObjectMeta.Name),
zap.String("resource_version", pkg.ObjectMeta.ResourceVersion),
zap.Error(er), zap.Error(er),
) )
} }
@@ -127,6 +128,8 @@ func (pkgw *packageWatcher) build(ctx context.Context, srcpkg *fv1.Package) {
healthCheckBackOff := utils.NewDefaultBackOff() healthCheckBackOff := utils.NewDefaultBackOff()
builderNs := pkgw.nsResolver.GetBuilderNS(env.ObjectMeta.Namespace) builderNs := pkgw.nsResolver.GetBuilderNS(env.ObjectMeta.Namespace)
logger = logger.With(zap.String("environment", env.Name), zap.String("builder_namespace", builderNs), zap.String("environment_namespace", env.Namespace))
// if err != nil { // if err != nil {
// pkgw.logger.Error("Unable to create BackOff for Health Check", zap.Error(err)) // pkgw.logger.Error("Unable to create BackOff for Health Check", zap.Error(err))
//} //}
@@ -136,12 +139,12 @@ func (pkgw *packageWatcher) build(ctx context.Context, srcpkg *fv1.Package) {
// iterate all available environment builders. // iterate all available environment builders.
items := pkgw.podInformer[builderNs].GetStore().List() items := pkgw.podInformer[builderNs].GetStore().List()
if err != nil { if err != nil {
pkgw.logger.Error("error retrieving pod information for environment", zap.Error(err), zap.String("environment", env.ObjectMeta.Name)) logger.Error("error retrieving pod information for environment", zap.Error(err))
return return
} }
if len(items) == 0 { if len(items) == 0 {
pkgw.logger.Info("builder pod does not exist for environment, will retry again later", zap.String("environment", pkg.Spec.Environment.Name)) logger.Info("builder pod does not exist for environment, will retry again later")
time.Sleep(healthCheckBackOff.GetCurrentBackoffDuration()) time.Sleep(healthCheckBackOff.GetCurrentBackoffDuration())
continue continue
} }
@@ -165,27 +168,22 @@ func (pkgw *packageWatcher) build(ctx context.Context, srcpkg *fv1.Package) {
} }
if !podIsReady { if !podIsReady {
pkgw.logger.Info("builder pod is not ready for environment, will retry again later", zap.String("environment", pkg.Spec.Environment.Name)) logger.Info("builder pod is not ready for environment, will retry again later")
time.Sleep(healthCheckBackOff.GetCurrentBackoffDuration()) time.Sleep(healthCheckBackOff.GetCurrentBackoffDuration())
break break
} }
uploadResp, buildLogs, err := buildPackage(ctx, pkgw.logger, pkgw.fissionClient, builderNs, pkgw.storageSvcUrl, pkg) uploadResp, buildLogs, err := buildPackage(ctx, pkgw.logger, pkgw.fissionClient, builderNs, pkgw.storageSvcUrl, pkg)
if err != nil { if err != nil {
pkgw.logger.Error("error building package", zap.Error(err), zap.String("package_name", pkg.ObjectMeta.Name)) logger.Error("error building package", zap.Error(err))
_, er := updatePackage(ctx, pkgw.logger, pkgw.fissionClient, pkg, fv1.BuildStatusFailed, buildLogs, nil) _, er := updatePackage(ctx, logger, pkgw.fissionClient, pkg, fv1.BuildStatusFailed, buildLogs, nil)
if er != nil { if er != nil {
pkgw.logger.Error( logger.Error("error updating package", zap.Error(er))
"error updating package",
zap.String("package_name", pkg.ObjectMeta.Name),
zap.String("resource_version", pkg.ObjectMeta.ResourceVersion),
zap.Error(er),
)
} }
return return
} }
pkgw.logger.Info("starting package info update", zap.String("package_name", pkg.ObjectMeta.Name)) logger.Info("starting package info update")
fnList, err := pkgw.fissionClient.CoreV1(). fnList, err := pkgw.fissionClient.CoreV1().
Functions(pkg.Namespace).List(ctx, metav1.ListOptions{}) Functions(pkg.Namespace).List(ctx, metav1.ListOptions{})
@@ -197,8 +195,6 @@ func (pkgw *packageWatcher) build(ctx context.Context, srcpkg *fv1.Package) {
if er != nil { if er != nil {
pkgw.logger.Error( pkgw.logger.Error(
"error updating package", "error updating package",
zap.String("package_name", pkg.ObjectMeta.Name),
zap.String("resource_version", pkg.ObjectMeta.ResourceVersion),
zap.Error(er), zap.Error(er),
) )
} }
@@ -215,57 +211,41 @@ func (pkgw *packageWatcher) build(ctx context.Context, srcpkg *fv1.Package) {
_, err = pkgw.fissionClient.CoreV1().Functions(fn.ObjectMeta.Namespace).Update(ctx, &fn, metav1.UpdateOptions{}) _, err = pkgw.fissionClient.CoreV1().Functions(fn.ObjectMeta.Namespace).Update(ctx, &fn, metav1.UpdateOptions{})
if err != nil { if err != nil {
e := "error updating function package resource version" e := "error updating function package resource version"
pkgw.logger.Error(e, zap.Error(err)) logger.Error(e, zap.Error(err))
buildLogs += fmt.Sprintf("%s: %v\n", e, err) buildLogs += fmt.Sprintf("%s: %v\n", e, err)
_, er := updatePackage(ctx, pkgw.logger, pkgw.fissionClient, pkg, fv1.BuildStatusFailed, buildLogs, nil) _, er := updatePackage(ctx, logger, pkgw.fissionClient, pkg, fv1.BuildStatusFailed, buildLogs, nil)
if er != nil { if er != nil {
pkgw.logger.Error( logger.Error("error updating package", zap.Error(er))
"error updating package",
zap.String("package_name", pkg.ObjectMeta.Name),
zap.String("resource_version", pkg.ObjectMeta.ResourceVersion),
zap.Error(er),
)
} }
return return
} }
} }
} }
_, err = updatePackage(ctx, pkgw.logger, pkgw.fissionClient, pkg, _, err = updatePackage(ctx, logger, pkgw.fissionClient, pkg,
fv1.BuildStatusSucceeded, buildLogs, uploadResp) fv1.BuildStatusSucceeded, buildLogs, uploadResp)
if err != nil { if err != nil {
pkgw.logger.Error("error updating package info", zap.Error(err), zap.String("package_name", pkg.ObjectMeta.Name)) logger.Error("error updating package info", zap.Error(err))
_, er := updatePackage(ctx, pkgw.logger, pkgw.fissionClient, pkg, fv1.BuildStatusFailed, buildLogs, nil) _, er := updatePackage(ctx, logger, pkgw.fissionClient, pkg, fv1.BuildStatusFailed, buildLogs, nil)
if er != nil { if er != nil {
pkgw.logger.Error( logger.Error("error updating package", zap.Error(er))
"error updating package",
zap.String("package_name", pkg.ObjectMeta.Name),
zap.String("resource_version", pkg.ObjectMeta.ResourceVersion),
zap.Error(er),
)
} }
return return
} }
pkgw.logger.Info("completed package build request", zap.String("package_name", pkg.ObjectMeta.Name)) logger.Info("completed package build request")
return return
} }
time.Sleep(healthCheckBackOff.GetNext()) time.Sleep(healthCheckBackOff.GetNext())
} }
// build timeout // build timeout
_, err = updatePackage(ctx, pkgw.logger, pkgw.fissionClient, pkg, _, err = updatePackage(ctx, logger, pkgw.fissionClient, pkg,
fv1.BuildStatusFailed, "Build timeout due to environment builder not ready", nil) fv1.BuildStatusFailed, "Build timeout due to environment builder not ready", nil)
if err != nil { if err != nil {
pkgw.logger.Error( logger.Error("error updating package", zap.Error(err))
"error updating package",
zap.String("package_name", pkg.ObjectMeta.Name),
zap.String("resource_version", pkg.ObjectMeta.ResourceVersion),
zap.Error(err),
)
} }
pkgw.logger.Error("max retries exceeded in building source package, timeout due to environment builder not ready", logger.Error("max retries exceeded in building source package, timeout due to environment builder not ready")
zap.String("package", fmt.Sprintf("%s.%s", pkg.ObjectMeta.Name, pkg.ObjectMeta.Namespace)))
} }
func (pkgw *packageWatcher) packageInformerHandler(ctx context.Context) k8sCache.ResourceEventHandlerFuncs { func (pkgw *packageWatcher) packageInformerHandler(ctx context.Context) k8sCache.ResourceEventHandlerFuncs {
pkg/cache/cache.go
Vendored
+36 -36
View File
@@ -34,33 +34,33 @@ const (
) )
type ( type (
Value struct { Value[V any] struct {
ctime time.Time ctime time.Time
atime time.Time atime time.Time
value interface{} value V
} }
Cache struct { Cache[K comparable, V any] struct {
cache map[interface{}]*Value cache map[K]*Value[V]
ctimeExpiry time.Duration ctimeExpiry time.Duration
atimeExpiry time.Duration atimeExpiry time.Duration
requestChannel chan *request requestChannel chan *request[K, V]
} }
request struct { request[K comparable, V any] struct {
requestType requestType
key interface{} key K
value interface{} value V
responseChannel chan *response responseChannel chan *response[K, V]
} }
response struct { response[K comparable, V any] struct {
error error
existingValue interface{} existingValue V
mapCopy map[interface{}]interface{} mapCopy map[K]V
value interface{} value V
} }
) )
func (c *Cache) IsOld(v *Value) bool { func (c *Cache[K, V]) IsOld(v *Value[V]) bool {
if (c.ctimeExpiry != time.Duration(0)) && (time.Since(v.ctime) > c.ctimeExpiry) { if (c.ctimeExpiry != time.Duration(0)) && (time.Since(v.ctime) > c.ctimeExpiry) {
return true return true
} }
@@ -72,12 +72,12 @@ func (c *Cache) IsOld(v *Value) bool {
return false return false
} }
func MakeCache(ctimeExpiry, atimeExpiry time.Duration) *Cache { func MakeCache[K comparable, V any](ctimeExpiry, atimeExpiry time.Duration) *Cache[K, V] {
c := &Cache{ c := &Cache[K, V]{
cache: make(map[interface{}]*Value), cache: make(map[K]*Value[V]),
ctimeExpiry: ctimeExpiry, ctimeExpiry: ctimeExpiry,
atimeExpiry: atimeExpiry, atimeExpiry: atimeExpiry,
requestChannel: make(chan *request), requestChannel: make(chan *request[K, V]),
} }
go c.service() go c.service()
if ctimeExpiry != time.Duration(0) || atimeExpiry != time.Duration(0) { if ctimeExpiry != time.Duration(0) || atimeExpiry != time.Duration(0) {
@@ -86,10 +86,10 @@ func MakeCache(ctimeExpiry, atimeExpiry time.Duration) *Cache {
return c return c
} }
func (c *Cache) service() { func (c *Cache[K, V]) service() {
for { for {
req := <-c.requestChannel req := <-c.requestChannel
resp := &response{} resp := &response[K, V]{}
switch req.requestType { switch req.requestType {
case GET: case GET:
val, ok := c.cache[req.key] val, ok := c.cache[req.key]
@@ -115,7 +115,7 @@ func (c *Cache) service() {
resp.existingValue = val.value resp.existingValue = val.value
resp.error = ferror.MakeError(ferror.ErrorNameExists, "key already exists") resp.error = ferror.MakeError(ferror.ErrorNameExists, "key already exists")
} else { } else {
c.cache[req.key] = &Value{ c.cache[req.key] = &Value[V]{
value: req.value, value: req.value,
ctime: now, ctime: now,
atime: now, atime: now,
@@ -133,7 +133,7 @@ func (c *Cache) service() {
} }
// no response // no response
case COPY: case COPY:
resp.mapCopy = make(map[interface{}]interface{}) resp.mapCopy = make(map[K]V)
for k, v := range c.cache { for k, v := range c.cache {
resp.mapCopy[k] = v.value resp.mapCopy[k] = v.value
} }
@@ -146,9 +146,9 @@ func (c *Cache) service() {
} }
} }
func (c *Cache) Get(key interface{}) (interface{}, error) { func (c *Cache[K, V]) Get(key K) (V, error) {
respChannel := make(chan *response) respChannel := make(chan *response[K, V])
c.requestChannel <- &request{ c.requestChannel <- &request[K, V]{
requestType: GET, requestType: GET,
key: key, key: key,
responseChannel: respChannel, responseChannel: respChannel,
@@ -159,9 +159,9 @@ func (c *Cache) Get(key interface{}) (interface{}, error) {
// if key exists in the cache, the new value is NOT set; instead an // if key exists in the cache, the new value is NOT set; instead an
// error and the old value are returned // error and the old value are returned
func (c *Cache) Set(key interface{}, value interface{}) (interface{}, error) { func (c *Cache[K, V]) Set(key K, value V) (V, error) {
respChannel := make(chan *response) respChannel := make(chan *response[K, V])
c.requestChannel <- &request{ c.requestChannel <- &request[K, V]{
requestType: SET, requestType: SET,
key: key, key: key,
value: value, value: value,
@@ -171,9 +171,9 @@ func (c *Cache) Set(key interface{}, value interface{}) (interface{}, error) {
return resp.existingValue, resp.error return resp.existingValue, resp.error
} }
func (c *Cache) Delete(key interface{}) error { func (c *Cache[K, V]) Delete(key K) error {
respChannel := make(chan *response) respChannel := make(chan *response[K, V])
c.requestChannel <- &request{ c.requestChannel <- &request[K, V]{
requestType: DELETE, requestType: DELETE,
key: key, key: key,
responseChannel: respChannel, responseChannel: respChannel,
@@ -182,9 +182,9 @@ func (c *Cache) Delete(key interface{}) error {
return resp.error return resp.error
} }
func (c *Cache) Copy() map[interface{}]interface{} { func (c *Cache[K, V]) Copy() map[K]V {
respChannel := make(chan *response) respChannel := make(chan *response[K, V])
c.requestChannel <- &request{ c.requestChannel <- &request[K, V]{
requestType: COPY, requestType: COPY,
responseChannel: respChannel, responseChannel: respChannel,
} }
@@ -192,10 +192,10 @@ func (c *Cache) Copy() map[interface{}]interface{} {
return resp.mapCopy return resp.mapCopy
} }
func (c *Cache) expiryService() { func (c *Cache[K, V]) expiryService() {
for { for {
time.Sleep(time.Minute) time.Sleep(time.Minute)
c.requestChannel <- &request{ c.requestChannel <- &request[K, V]{
requestType: EXPIRE, requestType: EXPIRE,
} }
} }
pkg/cache/cache_test.go
Vendored
+1 -1
View File
@@ -29,7 +29,7 @@ func checkErr(err error) {
} }
func TestCache(t *testing.T) { func TestCache(t *testing.T) {
c := MakeCache(100*time.Millisecond, 100*time.Millisecond) c := MakeCache[string, string](100*time.Millisecond, 100*time.Millisecond)
_, err := c.Set("a", "b") _, err := c.Set("a", "b")
checkErr(err) checkErr(err)
pkg/canaryconfigmgr/canaryConfigCache.go
+3 -4
View File
@@ -27,7 +27,7 @@ import (
type ( type (
canaryConfigCancelFuncMap struct { canaryConfigCancelFuncMap struct {
cache *cache.Cache // map[metadataKey]*context.Context cache *cache.Cache[metadataKey, *CanaryProcessingInfo]
} }
// metav1.ObjectMeta is not hashable, so we make a hashable copy // metav1.ObjectMeta is not hashable, so we make a hashable copy
@@ -45,7 +45,7 @@ type (
func makecanaryConfigCancelFuncMap() *canaryConfigCancelFuncMap { func makecanaryConfigCancelFuncMap() *canaryConfigCancelFuncMap {
return &canaryConfigCancelFuncMap{ return &canaryConfigCancelFuncMap{
cache: cache.MakeCache(0, 0), cache: cache.MakeCache[metadataKey, *CanaryProcessingInfo](0, 0),
} }
} }
@@ -62,8 +62,7 @@ func (cancelFuncMap *canaryConfigCancelFuncMap) lookup(f *metav1.ObjectMeta) (*C
if err != nil { if err != nil {
return nil, err return nil, err
} }
value := item.(*CanaryProcessingInfo) return item, nil
return value, nil
} }
func (cancelFuncMap *canaryConfigCancelFuncMap) assign(f *metav1.ObjectMeta, value *CanaryProcessingInfo) error { func (cancelFuncMap *canaryConfigCancelFuncMap) assign(f *metav1.ObjectMeta, value *CanaryProcessingInfo) error {
pkg/executor/executortype/poolmgr/gpm.go
+3 -4
View File
@@ -78,7 +78,7 @@ type (
nsResolver *utils.NamespaceResolver nsResolver *utils.NamespaceResolver
fissionClient versioned.Interface fissionClient versioned.Interface
functionEnv *cache.Cache functionEnv *cache.Cache[crd.CacheKeyUR, *fv1.Environment]
fsCache *fscache.FunctionServiceCache fsCache *fscache.FunctionServiceCache
instanceID string instanceID string
requestChannel chan *request requestChannel chan *request
@@ -147,7 +147,7 @@ func MakeGenericPoolManager(ctx context.Context,
nsResolver: utils.DefaultNSResolver(), nsResolver: utils.DefaultNSResolver(),
metricsClient: metricsClient, metricsClient: metricsClient,
fissionClient: fissionClient, fissionClient: fissionClient,
functionEnv: cache.MakeCache(10*time.Second, 0), functionEnv: cache.MakeCache[crd.CacheKeyUR, *fv1.Environment](10*time.Second, 0),
fsCache: fscache.MakeFunctionServiceCache(gpmLogger), fsCache: fscache.MakeFunctionServiceCache(gpmLogger),
instanceID: instanceID, instanceID: instanceID,
requestChannel: make(chan *request), requestChannel: make(chan *request),
@@ -595,8 +595,7 @@ func (gpm *GenericPoolManager) getFunctionEnv(ctx context.Context, fn *fv1.Funct
// TODO: the cache should be able to search by <env name, fn namespace> instead of function metadata. // TODO: the cache should be able to search by <env name, fn namespace> instead of function metadata.
result, err := gpm.functionEnv.Get(crd.CacheKeyURFromMeta(&fn.ObjectMeta)) result, err := gpm.functionEnv.Get(crd.CacheKeyURFromMeta(&fn.ObjectMeta))
if err == nil { if err == nil {
env = result.(*fv1.Environment) return result, nil
return env, nil
} }
// Get env from controller // Get env from controller
pkg/executor/fscache/functionServiceCache.go
+19 -29
View File
@@ -68,12 +68,12 @@ type (
// FunctionServiceCache represents the function service cache // FunctionServiceCache represents the function service cache
FunctionServiceCache struct { FunctionServiceCache struct {
logger *zap.Logger logger *zap.Logger
byFunction *cache.Cache // function-key -> funcSvc : map[string]*funcSvc byFunction *cache.Cache[crd.CacheKeyUR, *FuncSvc]
byAddress *cache.Cache // address -> function : map[string]metav1.ObjectMeta byAddress *cache.Cache[string, metav1.ObjectMeta]
byFunctionUID *cache.Cache // function uid -> function : map[string]metav1.ObjectMeta byFunctionUID *cache.Cache[types.UID, metav1.ObjectMeta]
connFunctionCache *PoolCache // function-key -> funcSvc : map[string]*funcSvc connFunctionCache *PoolCache // function-key -> funcSvc : map[string]*funcSvc
PodToFsvc sync.Map // pod-name -> funcSvc: map[string]*FuncSvc PodToFsvc sync.Map // pod-name -> funcSvc: map[string]*FuncSvc
WebsocketFsvc sync.Map // funcSvc-name -> bool: map[string]bool WebsocketFsvc sync.Map // funcSvc-name -> bool: map[string]bool
requestChannel chan *fscRequest requestChannel chan *fscRequest
} }
@@ -110,9 +110,9 @@ func IsNameExistError(err error) bool {
func MakeFunctionServiceCache(logger *zap.Logger) *FunctionServiceCache { func MakeFunctionServiceCache(logger *zap.Logger) *FunctionServiceCache {
fsc := &FunctionServiceCache{ fsc := &FunctionServiceCache{
logger: logger.Named("function_service_cache"), logger: logger.Named("function_service_cache"),
byFunction: cache.MakeCache(0, 0), byFunction: cache.MakeCache[crd.CacheKeyUR, *FuncSvc](0, 0),
byAddress: cache.MakeCache(0, 0), byAddress: cache.MakeCache[string, metav1.ObjectMeta](0, 0),
byFunctionUID: cache.MakeCache(0, 0), byFunctionUID: cache.MakeCache[types.UID, metav1.ObjectMeta](0, 0),
connFunctionCache: NewPoolCache(logger.Named("conn_function_cache")), connFunctionCache: NewPoolCache(logger.Named("conn_function_cache")),
requestChannel: make(chan *fscRequest), requestChannel: make(chan *fscRequest),
} }
@@ -132,14 +132,12 @@ func (fsc *FunctionServiceCache) service() {
// get svcs idle for > req.age // get svcs idle for > req.age
fscs := fsc.byFunctionUID.Copy() fscs := fsc.byFunctionUID.Copy()
funcObjects := make([]*FuncSvc, 0) funcObjects := make([]*FuncSvc, 0)
for _, funcSvc := range fscs { for _, m := range fscs {
mI := funcSvc.(metav1.ObjectMeta) fsvc, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&m))
fsvcI, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&mI))
if err != nil { if err != nil {
fsc.logger.Error("error while getting service", zap.String("error", err.Error())) fsc.logger.Error("error while getting service", zap.String("error", err.Error()))
return return
} }
fsvc := fsvcI.(*FuncSvc)
if time.Since(fsvc.Atime) > req.age { if time.Since(fsvc.Atime) > req.age {
funcObjects = append(funcObjects, fsvc) funcObjects = append(funcObjects, fsvc)
} }
@@ -149,8 +147,7 @@ func (fsc *FunctionServiceCache) service() {
fsc.logger.Info("dumping function service cache") fsc.logger.Info("dumping function service cache")
funcCopy := fsc.byFunction.Copy() funcCopy := fsc.byFunction.Copy()
info := []string{} info := []string{}
for key, fsvcI := range funcCopy { for key, fsvc := range funcCopy {
fsvc := fsvcI.(*FuncSvc)
for _, kubeObj := range fsvc.KubernetesObjects { for _, kubeObj := range fsvc.KubernetesObjects {
info = append(info, fmt.Sprintf("%v\t%v\t%v", key, kubeObj.Kind, kubeObj.Name)) info = append(info, fmt.Sprintf("%v\t%v\t%v", key, kubeObj.Kind, kubeObj.Name))
} }
@@ -196,13 +193,12 @@ func (fsc *FunctionServiceCache) DumpDebugInfo(ctx context.Context) error {
func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, error) { func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, error) {
key := crd.CacheKeyURFromMeta(m) key := crd.CacheKeyURFromMeta(m)
fsvcI, err := fsc.byFunction.Get(key) fsvc, err := fsc.byFunction.Get(key)
if err != nil { if err != nil {
return nil, err return nil, err
} }
// update atime // update atime
fsvc := fsvcI.(*FuncSvc)
fsvc.Atime = time.Now() fsvc.Atime = time.Now()
fsvcCopy := *fsvc fsvcCopy := *fsvc
@@ -228,20 +224,17 @@ func (fsc *FunctionServiceCache) GetFuncSvc(ctx context.Context, m *metav1.Objec
// GetByFunctionUID gets a function service from cache using function UUID. // GetByFunctionUID gets a function service from cache using function UUID.
func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, error) { func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, error) {
mI, err := fsc.byFunctionUID.Get(uid) m, err := fsc.byFunctionUID.Get(uid)
if err != nil { if err != nil {
return nil, err return nil, err
} }
m := mI.(metav1.ObjectMeta) fsvc, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&m))
fsvcI, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&m))
if err != nil { if err != nil {
return nil, err return nil, err
} }
// update atime // update atime
fsvc := fsvcI.(*FuncSvc)
fsvc.Atime = time.Now() fsvc.Atime = time.Now()
fsvcCopy := *fsvc fsvcCopy := *fsvc
@@ -279,12 +272,11 @@ func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) {
existing, err := fsc.byFunction.Set(crd.CacheKeyURFromMeta(fsvc.Function), &fsvc) existing, err := fsc.byFunction.Set(crd.CacheKeyURFromMeta(fsvc.Function), &fsvc)
if err != nil { if err != nil {
if IsNameExistError(err) { if IsNameExistError(err) {
f := existing.(*FuncSvc) err2 := fsc.TouchByAddress(existing.Address)
err2 := fsc.TouchByAddress(f.Address)
if err2 != nil { if err2 != nil {
return nil, err2 return nil, err2
} }
fCopy := *f fCopy := *existing
return &fCopy, nil return &fCopy, nil
} }
return nil, err return nil, err
@@ -333,16 +325,14 @@ func (fsc *FunctionServiceCache) TouchByAddress(address string) error {
} }
func (fsc *FunctionServiceCache) _touchByAddress(address string) error { func (fsc *FunctionServiceCache) _touchByAddress(address string) error {
mI, err := fsc.byAddress.Get(address) m, err := fsc.byAddress.Get(address)
if err != nil { if err != nil {
return err return err
} }
m := mI.(metav1.ObjectMeta) fsvc, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&m))
fsvcI, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&m))
if err != nil { if err != nil {
return err return err
} }
fsvc := fsvcI.(*FuncSvc)
fsvc.Atime = time.Now() fsvc.Atime = time.Now()
return nil return nil
} }
pkg/router/functionReferenceResolver.go
+4 -11
View File
@@ -34,7 +34,7 @@ type (
// reference into a resolveResult // reference into a resolveResult
functionReferenceResolver struct { functionReferenceResolver struct {
// FunctionReference -> function metadata // FunctionReference -> function metadata
refCache *cache.Cache refCache *cache.Cache[namespacedTriggerReference, resolveResult]
funcInformer map[string]k8sCache.SharedIndexInformer funcInformer map[string]k8sCache.SharedIndexInformer
logger *zap.Logger logger *zap.Logger
// store k8sCache.Store // store k8sCache.Store
@@ -73,7 +73,7 @@ const (
func makeFunctionReferenceResolver(logger *zap.Logger, funcInformer map[string]k8sCache.SharedIndexInformer) *functionReferenceResolver { func makeFunctionReferenceResolver(logger *zap.Logger, funcInformer map[string]k8sCache.SharedIndexInformer) *functionReferenceResolver {
frr := &functionReferenceResolver{ frr := &functionReferenceResolver{
refCache: cache.MakeCache(time.Minute, 0), refCache: cache.MakeCache[namespacedTriggerReference, resolveResult](time.Minute, 0),
funcInformer: funcInformer, funcInformer: funcInformer,
logger: logger.Named("function_ref_resolver"), logger: logger.Named("function_ref_resolver"),
} }
@@ -89,9 +89,8 @@ func (frr *functionReferenceResolver) resolve(trigger fv1.HTTPTrigger) (*resolve
} }
// check cache // check cache
rrInt, err := frr.refCache.Get(nfr) result, err := frr.refCache.Get(nfr)
if err == nil { if err == nil {
result := rrInt.(resolveResult)
return &result, nil return &result, nil
} }
@@ -216,11 +215,5 @@ func (frr *functionReferenceResolver) delete(namespace string, triggerName, trig
} }
func (frr *functionReferenceResolver) copy() map[namespacedTriggerReference]resolveResult { func (frr *functionReferenceResolver) copy() map[namespacedTriggerReference]resolveResult {
cache := make(map[namespacedTriggerReference]resolveResult) return frr.refCache.Copy()
for k, v := range frr.refCache.Copy() {
key := k.(namespacedTriggerReference)
val := v.(resolveResult)
cache[key] = val
}
return cache
} }
pkg/router/functionServiceMap.go
+4 -5
View File
@@ -29,7 +29,7 @@ import (
type ( type (
functionServiceMap struct { functionServiceMap struct {
logger *zap.Logger logger *zap.Logger
cache *cache.Cache // map[metadataKey]*url.URL cache *cache.Cache[metadataKey, *url.URL]
} }
// metav1.ObjectMeta is not hashable, so we make a hashable copy // metav1.ObjectMeta is not hashable, so we make a hashable copy
@@ -44,7 +44,7 @@ type (
func makeFunctionServiceMap(logger *zap.Logger, expiry time.Duration) *functionServiceMap { func makeFunctionServiceMap(logger *zap.Logger, expiry time.Duration) *functionServiceMap {
return &functionServiceMap{ return &functionServiceMap{
logger: logger.Named("function_service_map"), logger: logger.Named("function_service_map"),
cache: cache.MakeCache(expiry, 0), cache: cache.MakeCache[metadataKey, *url.URL](expiry, 0),
} }
} }
@@ -62,15 +62,14 @@ func (fmap *functionServiceMap) lookup(f *metav1.ObjectMeta) (*url.URL, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
u := item.(*url.URL) return item, nil
return u, nil
} }
func (fmap *functionServiceMap) assign(f *metav1.ObjectMeta, serviceURL *url.URL) { func (fmap *functionServiceMap) assign(f *metav1.ObjectMeta, serviceURL *url.URL) {
mk := keyFromMetadata(f) mk := keyFromMetadata(f)
old, err := fmap.cache.Set(*mk, serviceURL) old, err := fmap.cache.Set(*mk, serviceURL)
if err != nil { if err != nil {
if *serviceURL == *(old.(*url.URL)) { if *serviceURL == *old {
return return
} }
fmap.logger.Error("error caching service url for function with a different value", zap.Error(err)) fmap.logger.Error("error caching service url for function with a different value", zap.Error(err))