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

Ability to retain specialised pods for poolmanager functions (#2830)

Browse Source 
- added retainPods flag to take in the number of specialized pods to retain
- add retainPods in both the create function and update function command
- modify crd keys to be typed instead of string
- keep track of function generation in case of update function operation
- add delete handler function to make sure specialized pods are deleted in case function is deleted

---------

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>
Signed-off-by: Pranoy Kundu <pranoy1998k@gmail.com>
Co-authored-by: Pranoy Kundu <pranoy1998k@gmail.com>
This commit is contained in:
Sanket Sudake
2023-09-27 13:33:19 +05:30
committed by GitHub
co-authored by Pranoy Kundu
parent 657aee7cc2
commit 56b49dcee8
26 changed files with 405 additions and 211 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/executor/api.go
+3 -5
View File
@@ -31,7 +31,6 @@ import (
"go.uber.org/zap"
fv1 "github.com/fission/fission/pkg/apis/core/v1"
"github.com/fission/fission/pkg/crd"
ferror "github.com/fission/fission/pkg/error"
"github.com/fission/fission/pkg/executor/client"
"github.com/fission/fission/pkg/executor/fscache"
@@ -157,9 +156,9 @@ func (executor *Executor) getServiceForFunction(ctx context.Context, fn *fv1.Fun
return
}
if funcSvc != nil {
et.UnTapService(ctx, crd.CacheKey(funcSvc.Function), resp.funcSvc.Address)
et.UnTapService(ctx, funcSvc.Function, resp.funcSvc.Address)
} else {
et.MarkSpecializationFailure(ctx, crd.CacheKey(&fn.ObjectMeta))
et.MarkSpecializationFailure(ctx, &fn.ObjectMeta)
}
}
if errors.Is(ctx.Err(), context.Canceled) {
@@ -248,7 +247,6 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Failed to parse request", http.StatusBadRequest)
return
}
key := fmt.Sprintf("%v_%v", tapSvcReq.FnMetadata.UID, tapSvcReq.FnMetadata.ResourceVersion)
t := tapSvcReq.FnExecutorType
if t != fv1.ExecutorTypePoolmgr {
msg := fmt.Sprintf("Unknown executor type '%v'", t)
@@ -258,7 +256,7 @@ func (executor *Executor) unTapService(w http.ResponseWriter, r *http.Request) {
et := executor.executorTypes[t]
et.UnTapService(ctx, key, tapSvcReq.ServiceURL)
et.UnTapService(ctx, &tapSvcReq.FnMetadata, tapSvcReq.ServiceURL)
w.WriteHeader(http.StatusOK)
}
pkg/executor/executor.go
+3 -3
View File
@@ -138,13 +138,13 @@ func (executor *Executor) serveCreateFuncServices() {
}
// Cache miss -- is this first one to request the func?
wg, found := executor.fsCreateWg.Load(crd.CacheKey(fnMetadata))
wg, found := executor.fsCreateWg.Load(crd.CacheKeyURFromMeta(fnMetadata))
if !found {
// create a waitgroup for other requests for
// the same function to wait on
wg := &sync.WaitGroup{}
wg.Add(1)
executor.fsCreateWg.Store(crd.CacheKey(fnMetadata), wg)
executor.fsCreateWg.Store(crd.CacheKeyURFromMeta(fnMetadata), wg)
// launch a goroutine for each request, to parallelize
// the specialization of different functions
@@ -176,7 +176,7 @@ func (executor *Executor) serveCreateFuncServices() {
funcSvc: fsvc,
err: err,
}
executor.fsCreateWg.Delete(crd.CacheKey(fnMetadata))
executor.fsCreateWg.Delete(crd.CacheKeyURFromMeta(fnMetadata))
wg.Done()
}()
} else {
pkg/executor/executortype/container/containermgr.go
+2 -2
View File
@@ -176,12 +176,12 @@ func (caaf *Container) GetTotalAvailable(fn *fv1.Function) int {
}
// UnTapService has not been implemented for CaaF.
func (caaf *Container) UnTapService(ctx context.Context, key string, svcHost string) {
func (caaf *Container) UnTapService(ctx context.Context, fnMeta *metav1.ObjectMeta, svcHost string) {
// Not Implemented for CaaF.
}
// MarkSpecializationFailure has not been implemented for CaaF.
func (caaf *Container) MarkSpecializationFailure(ctx context.Context, key string) {
func (caaf *Container) MarkSpecializationFailure(ctx context.Context, fnMeta *metav1.ObjectMeta) {
// Not Implemented for CaaF.
}
pkg/executor/executortype/executortype.go
+4 -2
View File
@@ -21,6 +21,8 @@ import (
"go.uber.org/zap"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
fv1 "github.com/fission/fission/pkg/apis/core/v1"
"github.com/fission/fission/pkg/executor/fscache"
)
@@ -49,10 +51,10 @@ type ExecutorType interface {
TapService(ctx context.Context, serviceUrl string) error
// UnTapService updates the isActive to false
UnTapService(ctx context.Context, key string, svcHost string)
UnTapService(ctx context.Context, fnMeta *metav1.ObjectMeta, svcHost string)
// ReduceSpecializationInProgress updates the svcWaiting count in funcSvcGroup
MarkSpecializationFailure(ctx context.Context, key string)
MarkSpecializationFailure(ctx context.Context, fnMeta *metav1.ObjectMeta)
// IsValid returns true if a function service is valid. Different executor types
// use distinct ways to examine the function service.
pkg/executor/executortype/newdeploy/newdeploymgr.go
+2 -2
View File
@@ -201,12 +201,12 @@ func (deploy *NewDeploy) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscac
}
// UnTapService has not been implemented for NewDeployment.
func (deploy *NewDeploy) UnTapService(ctx context.Context, key string, svcHost string) {
func (deploy *NewDeploy) UnTapService(ctx context.Context, fnMeta *metav1.ObjectMeta, svcHost string) {
// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
}
// MarkSpecializationFailure has not been implemented for NewDeployment.
func (deploy *NewDeploy) MarkSpecializationFailure(ctx context.Context, key string) {
func (deploy *NewDeploy) MarkSpecializationFailure(ctx context.Context, fnMeta *metav1.ObjectMeta) {
// Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added.
}
pkg/executor/executortype/poolmgr/gp.go
+3 -3
View File
@@ -202,7 +202,7 @@ func (gp *GenericPool) updateCPUUtilizationSvc(ctx context.Context) {
if value, ok := gp.podFSVCMap.Load(val.ObjectMeta.Name); ok {
if valArray, ok1 := value.([]interface{}); ok1 {
function, address := valArray[0], valArray[1]
gp.fsCache.SetCPUUtilizaton(function.(string), address.(string), p)
gp.fsCache.SetCPUUtilizaton(function.(crd.CacheKeyURG), address.(string), p)
gp.logger.Info(fmt.Sprintf("updated function %s, address %s, cpuUsage %+v", function.(string), address.(string), p))
}
}
@@ -622,8 +622,8 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
}
gp.fsCache.PodToFsvc.Store(pod.GetObjectMeta().GetName(), fsvc)
gp.podFSVCMap.Store(pod.ObjectMeta.Name, []interface{}{crd.CacheKey(fsvc.Function), fsvc.Address})
gp.fsCache.AddFunc(ctx, *fsvc, fn.GetRequestPerPod())
gp.podFSVCMap.Store(pod.ObjectMeta.Name, []interface{}{crd.CacheKeyURGFromMeta(fsvc.Function), fsvc.Address})
gp.fsCache.AddFunc(ctx, *fsvc, fn.GetRequestPerPod(), fn.GetRetainPods())
logger.Info("added function service",
zap.String("pod", pod.ObjectMeta.Name),
pkg/executor/executortype/poolmgr/gpm.go
+13 -11
View File
@@ -71,7 +71,7 @@ type (
GenericPoolManager struct {
logger *zap.Logger
pools map[string]*GenericPool
pools map[k8sTypes.UID]*GenericPool
kubernetesClient kubernetes.Interface
metricsClient metricsclient.Interface
nsResolver *utils.NamespaceResolver
@@ -141,7 +141,7 @@ func MakeGenericPoolManager(ctx context.Context,
}
gpm := &GenericPoolManager{
logger: gpmLogger,
pools: make(map[string]*GenericPool),
pools: make(map[k8sTypes.UID]*GenericPool),
kubernetesClient: kubernetesClient,
nsResolver: utils.DefaultNSResolver(),
metricsClient: metricsClient,
@@ -237,9 +237,10 @@ func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(ctx context.Context, fsvc
gpm.fsCache.DeleteFunctionSvc(ctx, fsvc)
}
func (gpm *GenericPoolManager) UnTapService(ctx context.Context, key string, svcHost string) {
func (gpm *GenericPoolManager) UnTapService(ctx context.Context, fnMeta *metav1.ObjectMeta, svcHost string) {
key := crd.CacheKeyURGFromMeta(fnMeta)
otelUtils.SpanTrackEvent(ctx, "UnTapService",
attribute.KeyValue{Key: "key", Value: attribute.StringValue(key)},
attribute.KeyValue{Key: "key", Value: attribute.StringValue(key.String())},
attribute.KeyValue{Key: "svcHost", Value: attribute.StringValue(svcHost)})
gpm.fsCache.MarkAvailable(key, svcHost)
}
@@ -254,9 +255,10 @@ func (gpm *GenericPoolManager) TapService(ctx context.Context, svcHost string) e
return nil
}
func (gpm *GenericPoolManager) MarkSpecializationFailure(ctx context.Context, key string) {
func (gpm *GenericPoolManager) MarkSpecializationFailure(ctx context.Context, fnMeta *metav1.ObjectMeta) {
key := crd.CacheKeyURGFromMeta(fnMeta)
otelUtils.SpanTrackEvent(ctx, "MarkSpecializationFailure",
attribute.KeyValue{Key: "key", Value: attribute.StringValue(key)})
attribute.KeyValue{Key: "key", Value: attribute.StringValue(key.String())})
logger := otelUtils.LoggerWithTraceID(ctx, gpm.logger)
logger.Info("marking specialization failure", zap.Any("key", key))
gpm.fsCache.MarkSpecializationFailure(key)
@@ -503,7 +505,7 @@ func (gpm *GenericPoolManager) service() {
// just because they are missing in the cache, we end up creating another duplicate pool.
var err error
created := false
pool, ok := gpm.pools[crd.CacheKeyUID(&req.env.ObjectMeta)]
pool, ok := gpm.pools[crd.CacheKeyUIDFromMeta(&req.env.ObjectMeta)]
if !ok {
// To support backward compatibility, if envs are created in default ns, we go ahead
// and create pools in fission-function ns as earlier.
@@ -516,7 +518,7 @@ func (gpm *GenericPoolManager) service() {
req.responseChannel <- &response{error: err}
continue
}
gpm.pools[crd.CacheKeyUID(&req.env.ObjectMeta)] = pool
gpm.pools[crd.CacheKeyUIDFromMeta(&req.env.ObjectMeta)] = pool
created = true
}
req.responseChannel <- &response{pool: pool, created: created}
@@ -526,7 +528,7 @@ func (gpm *GenericPoolManager) service() {
zap.String("environment", env.ObjectMeta.Name),
zap.String("namespace", env.ObjectMeta.Namespace))
key := crd.CacheKeyUID(&req.env.ObjectMeta)
key := crd.CacheKeyUIDFromMeta(&req.env.ObjectMeta)
pool, ok := gpm.pools[key]
if !ok {
gpm.logger.Error("Could not find pool", zap.String("environment", env.ObjectMeta.Name), zap.String("namespace", env.ObjectMeta.Namespace))
@@ -573,7 +575,7 @@ func (gpm *GenericPoolManager) getFunctionEnv(ctx context.Context, fn *fv1.Funct
// Cached ?
// TODO: the cache should be able to search by <env name, fn namespace> instead of function metadata.
result, err := gpm.functionEnv.Get(crd.CacheKey(&fn.ObjectMeta))
result, err := gpm.functionEnv.Get(crd.CacheKeyURFromMeta(&fn.ObjectMeta))
if err == nil {
env = result.(*fv1.Environment)
return env, nil
@@ -591,7 +593,7 @@ func (gpm *GenericPoolManager) getFunctionEnv(ctx context.Context, fn *fv1.Funct
// cache for future lookups
m := fn.ObjectMeta
_, err = gpm.functionEnv.Set(crd.CacheKey(&m), env)
_, err = gpm.functionEnv.Set(crd.CacheKeyURFromMeta(&m), env)
if err != nil {
gpm.logger.Error(
"failed to set the key",
pkg/executor/executortype/poolmgr/poolpodcontroller.go
+14
View File
@@ -36,6 +36,7 @@ import (
"k8s.io/client-go/util/workqueue"
fv1 "github.com/fission/fission/pkg/apis/core/v1"
"github.com/fission/fission/pkg/crd"
"github.com/fission/fission/pkg/executor/fscache"
genInformer "github.com/fission/fission/pkg/generated/informers/externalversions"
flisterv1 "github.com/fission/fission/pkg/generated/listers/core/v1"
@@ -94,6 +95,14 @@ func NewPoolPodController(ctx context.Context, logger *zap.Logger,
}
}
}
for _, factory := range finformerFactory {
_, err := factory.Core().V1().Functions().Informer().AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
DeleteFunc: p.handleFuncDelete,
})
if err != nil {
return nil, err
}
}
for ns, informer := range finformerFactory {
_, err := informer.Core().V1().Environments().Informer().AddEventHandler(k8sCache.ResourceEventHandlerFuncs{
AddFunc: p.enqueueEnvAdd,
@@ -133,6 +142,11 @@ func IsPodActive(p *v1.Pod) bool {
p.DeletionTimestamp == nil
}
func (p *PoolPodController) handleFuncDelete(obj interface{}) {
fn := obj.(*fv1.Function)
p.gpm.fsCache.MarkFuncDeleted(crd.CacheKeyURGFromMeta(&fn.ObjectMeta))
}
func (p *PoolPodController) processRS(rs *apps.ReplicaSet) {
if *(rs.Spec.Replicas) != 0 {
return
pkg/executor/fscache/functionServiceCache.go
+21 -17
View File
@@ -134,9 +134,9 @@ func (fsc *FunctionServiceCache) service() {
funcObjects := make([]*FuncSvc, 0)
for _, funcSvc := range fscs {
mI := funcSvc.(metav1.ObjectMeta)
fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&mI))
fsvcI, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&mI))
if err != nil {
fsc.logger.Error("error while getting service", zap.Any("error", err))
fsc.logger.Error("error while getting service", zap.String("error", err.Error()))
return
}
fsvc := fsvcI.(*FuncSvc)
@@ -194,7 +194,7 @@ func (fsc *FunctionServiceCache) DumpDebugInfo(ctx context.Context) error {
// GetByFunction gets a function service from cache using function key.
func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, error) {
key := crd.CacheKey(m)
key := crd.CacheKeyURFromMeta(m)
fsvcI, err := fsc.byFunction.Get(key)
if err != nil {
@@ -211,7 +211,7 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc,
// GetFuncSvc gets a function service from pool cache using function key and returns number of active instances of function pod
func (fsc *FunctionServiceCache) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta, requestsPerPod int, concurrency int) (*FuncSvc, error) {
key := crd.CacheKey(m)
key := crd.CacheKeyURGFromMeta(m)
fsvc, err := fsc.connFunctionCache.GetSvcValue(ctx, key, requestsPerPod, concurrency)
if err != nil {
@@ -235,7 +235,7 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro
m := mI.(metav1.ObjectMeta)
fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m))
fsvcI, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&m))
if err != nil {
return nil, err
}
@@ -249,30 +249,34 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro
}
// AddFunc adds a function service to pool cache.
func (fsc *FunctionServiceCache) AddFunc(ctx context.Context, fsvc FuncSvc, requestsPerPod int) {
fsc.connFunctionCache.SetSvcValue(ctx, crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc, fsvc.CPULimit, requestsPerPod)
func (fsc *FunctionServiceCache) AddFunc(ctx context.Context, fsvc FuncSvc, requestsPerPod, svcsRetain int) {
fsc.connFunctionCache.SetSvcValue(ctx, crd.CacheKeyURGFromMeta(fsvc.Function), fsvc.Address, &fsvc, fsvc.CPULimit, requestsPerPod, svcsRetain)
now := time.Now()
fsvc.Ctime = now
fsvc.Atime = now
}
func (fsc *FunctionServiceCache) MarkFuncDeleted(key crd.CacheKeyURG) {
fsc.connFunctionCache.MarkFuncDeleted(key)
}
// SetCPUUtilizaton updates/sets CPUutilization in the pool cache
func (fsc *FunctionServiceCache) SetCPUUtilizaton(key string, svcHost string, cpuUsage resource.Quantity) {
func (fsc *FunctionServiceCache) SetCPUUtilizaton(key crd.CacheKeyURG, svcHost string, cpuUsage resource.Quantity) {
fsc.connFunctionCache.SetCPUUtilization(key, svcHost, cpuUsage)
}
// MarkAvailable marks the value at key [function][address] as available.
func (fsc *FunctionServiceCache) MarkAvailable(key string, svcHost string) {
func (fsc *FunctionServiceCache) MarkAvailable(key crd.CacheKeyURG, svcHost string) {
fsc.connFunctionCache.MarkAvailable(key, svcHost)
}
func (fsc *FunctionServiceCache) MarkSpecializationFailure(key string) {
func (fsc *FunctionServiceCache) MarkSpecializationFailure(key crd.CacheKeyURG) {
fsc.connFunctionCache.MarkSpecializationFailure(key)
}
// Add adds a function service to cache if it does not exist already.
func (fsc *FunctionServiceCache) Add(fsvc FuncSvc) (*FuncSvc, error) {
existing, err := fsc.byFunction.Set(crd.CacheKey(fsvc.Function), &fsvc)
existing, err := fsc.byFunction.Set(crd.CacheKeyURFromMeta(fsvc.Function), &fsvc)
if err != nil {
if IsNameExistError(err) {
f := existing.(*FuncSvc)
@@ -334,7 +338,7 @@ func (fsc *FunctionServiceCache) _touchByAddress(address string) error {
return err
}
m := mI.(metav1.ObjectMeta)
fsvcI, err := fsc.byFunction.Get(crd.CacheKey(&m))
fsvcI, err := fsc.byFunction.Get(crd.CacheKeyURFromMeta(&m))
if err != nil {
return err
}
@@ -346,7 +350,7 @@ func (fsc *FunctionServiceCache) _touchByAddress(address string) error {
// DeleteEntry deletes a function service from cache.
func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) {
msg := "error deleting function service"
err := fsc.byFunction.Delete(crd.CacheKey(fsvc.Function))
err := fsc.byFunction.Delete(crd.CacheKeyURFromMeta(fsvc.Function))
if err != nil {
fsc.logger.Error(
msg,
@@ -378,18 +382,18 @@ func (fsc *FunctionServiceCache) DeleteEntry(fsvc *FuncSvc) {
// DeleteFunctionSvc deletes a function service at key composed of [function][address].
func (fsc *FunctionServiceCache) DeleteFunctionSvc(ctx context.Context, fsvc *FuncSvc) {
err := fsc.connFunctionCache.DeleteValue(ctx, crd.CacheKey(fsvc.Function), fsvc.Address)
err := fsc.connFunctionCache.DeleteValue(ctx, crd.CacheKeyURGFromMeta(fsvc.Function), fsvc.Address)
if err != nil {
fsc.logger.Error(
"error deleting function service",
zap.Any("function", fsvc.Function.Name),
zap.Any("address", fsvc.Address),
zap.String("function", fsvc.Function.Name),
zap.String("address", fsvc.Address),
zap.Error(err),
)
}
}
func (fsc *FunctionServiceCache) SetCPUUtilization(key string, svcHost string, cpuUsage resource.Quantity) {
func (fsc *FunctionServiceCache) SetCPUUtilization(key crd.CacheKeyURG, svcHost string, cpuUsage resource.Quantity) {
fsc.connFunctionCache.SetCPUUtilization(key, svcHost, cpuUsage)
}
pkg/executor/fscache/functionServiceCache_test.go
+39 -60
View File
@@ -2,11 +2,11 @@ package fscache
import (
"context"
"fmt"
"log"
"testing"
"time"
"github.com/stretchr/testify/require"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
apiv1 "k8s.io/api/core/v1"
@@ -14,6 +14,7 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
fv1 "github.com/fission/fission/pkg/apis/core/v1"
"github.com/fission/fission/pkg/crd"
)
func panicIf(err error) {
@@ -29,9 +30,7 @@ func TestFunctionServiceCache(t *testing.T) {
panicIf(err)
fsc := MakeFunctionServiceCache(logger)
if fsc == nil {
log.Panicf("error creating cache")
}
require.NotNil(t, fsc)
var fsvc *FuncSvc
now := time.Now()
@@ -75,55 +74,30 @@ func TestFunctionServiceCache(t *testing.T) {
Atime: now,
}
_, err = fsc.Add(*fsvc)
if err != nil {
fsc.Log()
log.Panicf("Failed to add fsvc: %v", err)
}
require.NoError(t, err)
_, err = fsc.GetByFunction(fsvc.Function)
if err != nil {
fsc.Log()
log.Panicf("Failed to get fsvc: %v", err)
}
require.NoError(t, err)
f, err := fsc.GetByFunctionUID(fsvc.Function.UID)
if err != nil {
fsc.Log()
log.Panicf("Failed to get fsvc by function uid: %v", err)
}
require.NoError(t, err)
fsvc.Atime = f.Atime
fsvc.Ctime = f.Ctime
if f.Address != fsvc.Address {
fsc.Log()
log.Panicf("Incorrect fsvc \n(expected: %#v)\n (found: %#v)", fsvc, f)
}
require.Equal(t, fsvc.Address, f.Address)
err = fsc.TouchByAddress(fsvc.Address)
if err != nil {
fsc.Log()
log.Panicf("Failed to touch fsvc: %v", err)
}
require.NoError(t, err)
deleted, err := fsc.DeleteOld(fsvc, 0)
if err != nil {
fsc.Log()
log.Panicf("Failed to delete fsvc: %v", err)
}
if !deleted {
fsc.Log()
log.Panicf("Did not delete fsvc")
}
require.NoError(t, err)
require.False(t, deleted)
_, err = fsc.GetByFunction(fsvc.Function)
if err == nil {
fsc.Log()
log.Panicf("found fsvc while expecting empty cache: %v", err)
}
require.NoError(t, err)
_, err = fsc.GetByFunctionUID(fsvc.Function.UID)
if err == nil {
fsc.Log()
log.Panicf("found fsvc by function uid while expecting empty cache: %v", err)
}
require.NoError(t, err)
}
func TestFunctionServiceNewCache(t *testing.T) {
@@ -131,9 +105,7 @@ func TestFunctionServiceNewCache(t *testing.T) {
panicIf(err)
fsc := MakeFunctionServiceCache(logger)
if fsc == nil {
log.Panicf("error creating cache")
}
require.NotNil(t, fsc)
var fsvc *FuncSvc
now := time.Now()
@@ -174,8 +146,8 @@ func TestFunctionServiceNewCache(t *testing.T) {
Address: "xxx",
KubernetesObjects: objects,
CPULimit: resource.MustParse("5m"),
Ctime: now,
Atime: now,
Ctime: now.Add(-2 * time.Minute),
Atime: now.Add(-1 * time.Minute),
}
fn := &fv1.Function{
ObjectMeta: metav1.ObjectMeta{
@@ -187,27 +159,34 @@ func TestFunctionServiceNewCache(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
fsc.AddFunc(ctx, *fsvc, 10)
fsc.AddFunc(ctx, *fsvc, 10, fn.GetRetainPods())
concurrency := 10
_, err = fsc.GetFuncSvc(ctx, fsvc.Function, 5, concurrency)
if err != nil {
logger.Panic("received error while retrieving value from cache")
}
require.NoError(t, err)
key := fmt.Sprintf("%v_%v", fn.ObjectMeta.UID, fn.ObjectMeta.ResourceVersion)
//key := fmt.Sprintf("%v_%v", cancel.UID, fn.ObjectMeta.ResourceVersion)
key := crd.CacheKeyURGFromMeta(&fn.ObjectMeta)
fsc.MarkAvailable(key, fsvc.Address)
_, err = fsc.GetFuncSvc(ctx, fsvc.Function, 5, concurrency)
if err != nil {
logger.Panic("received error while retrieving value from cache")
}
require.NoError(t, err)
for i := 0; i < 2; i++ {
fsc.MarkAvailable(key, fsvc.Address)
}
vals, err := fsc.ListOldForPool(30 * time.Second)
if err != nil {
logger.Panic("received error while get list of old values")
}
if len(vals) != 0 {
logger.Panic(fmt.Sprintln("list of old values didn't matched the expected: 1", "received", len(vals)))
}
fsc.DeleteFunctionSvc(ctx, fsvc)
require.NoError(t, err)
require.Equal(t, 0, len(vals))
vals, err = fsc.ListOldForPool(0)
require.NoError(t, err)
require.Equal(t, 1, len(vals))
fsvc.Address = "xxx2"
fn.Spec.RetainPods = 2
fsc.AddFunc(ctx, *fsvc, 10, fn.GetRetainPods())
vals, err = fsc.ListOldForPool(0)
require.NoError(t, err)
require.Equal(t, 0, len(vals))
}
pkg/executor/fscache/poolcache.go
+67 -17
View File
@@ -24,7 +24,9 @@ import (
"go.uber.org/zap"
"k8s.io/apimachinery/pkg/api/resource"
"k8s.io/apimachinery/pkg/types"
"github.com/fission/fission/pkg/crd"
ferror "github.com/fission/fission/pkg/error"
otelUtils "github.com/fission/fission/pkg/utils/otel"
)
@@ -40,6 +42,7 @@ const (
setCPUUtilization
markSpecializationFailure
logFuncSvc
markDeleted
)
type (
@@ -52,14 +55,16 @@ type (
funcSvcGroup struct {
svcWaiting int
svcRetain int
svcs map[string]*funcSvcInfo
queue *Queue
deleted bool
}
// PoolCache implements a simple cache implementation having values mapped by two keys [function][address].
// As of now PoolCache is only used by poolmanager executor
PoolCache struct {
cache map[string]*funcSvcGroup
cache map[crd.CacheKeyURG]*funcSvcGroup
requestChannel chan *request
logger *zap.Logger
}
@@ -67,7 +72,7 @@ type (
request struct {
requestType
ctx context.Context
function string
function crd.CacheKeyURG
address string
dumpWriter io.Writer
value *FuncSvc
@@ -75,6 +80,7 @@ type (
cpuUsage resource.Quantity
responseChannel chan *response
concurrency int
svcsRetain int
}
response struct {
error
@@ -92,7 +98,7 @@ type (
func NewPoolCache(logger *zap.Logger) *PoolCache {
c := &PoolCache{
cache: make(map[string]*funcSvcGroup),
cache: make(map[crd.CacheKeyURG]*funcSvcGroup),
requestChannel: make(chan *request),
logger: logger,
}
@@ -131,7 +137,7 @@ func (c *PoolCache) service() {
// mark active
funcSvcGroup.svcs[addr].activeRequests++
if c.logger.Core().Enabled(zap.DebugLevel) {
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with getValue", zap.String("function", req.function), zap.String("address", addr), zap.Int("activeRequests", funcSvcGroup.svcs[addr].activeRequests))
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with getValue", zap.String("function", req.function.String()), zap.String("address", addr), zap.Int("activeRequests", funcSvcGroup.svcs[addr].activeRequests))
}
resp.value = funcSvcGroup.svcs[addr].val
found = true
@@ -172,6 +178,7 @@ func (c *PoolCache) service() {
if _, ok := c.cache[req.function].svcs[req.address]; !ok {
c.cache[req.function].svcs[req.address] = &funcSvcInfo{}
}
c.cache[req.function].svcRetain = req.svcsRetain
c.cache[req.function].svcs[req.address].val = req.value
c.cache[req.function].svcs[req.address].activeRequests++
if c.cache[req.function].svcWaiting > 0 {
@@ -196,22 +203,52 @@ func (c *PoolCache) service() {
}
}
if c.logger.Core().Enabled(zap.DebugLevel) {
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with setValue", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests))
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with setValue", zap.String("function", req.function.String()), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests))
}
c.cache[req.function].svcs[req.address].cpuLimit = req.cpuUsage
case markDeleted:
for key := range c.cache {
if key.UID == req.function.UID {
c.cache[key].deleted = true
break
}
}
case listAvailableValue:
vals := make([]*FuncSvc, 0)
latestFuncGen := make(map[types.UID]int64)
// find the latest generation of each function
for key := range c.cache {
if currentFuncGen, ok := latestFuncGen[key.UID]; ok {
if key.Generation > currentFuncGen {
latestFuncGen[key.UID] = key.Generation
}
} else {
latestFuncGen[key.UID] = key.Generation
}
}
for key1, values := range c.cache {
svcRetain := values.svcRetain
// if the function is not latest generation, then we don't need to retain any pods
if latestFuncGen[key1.UID] != key1.Generation || values.deleted {
svcRetain = 0
}
svcCleanQuota := len(values.svcs) - svcRetain
if svcCleanQuota <= 0 {
continue
}
for key2, value := range values.svcs {
debugLevel := c.logger.Core().Enabled(zap.DebugLevel)
if debugLevel {
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Reading active requests", zap.String("function", key1), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests))
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Reading active requests", zap.String("function", key1.String()), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests))
}
if value.activeRequests == 0 {
if value.activeRequests == 0 && svcCleanQuota > 0 {
if debugLevel {
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Function service with no active requests", zap.String("function", key1), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests))
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Function service with no active requests", zap.String("function", key1.String()), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests))
}
vals = append(vals, value.val)
svcCleanQuota--
}
}
}
@@ -230,10 +267,10 @@ func (c *PoolCache) service() {
if c.cache[req.function].svcs[req.address].activeRequests > 0 {
c.cache[req.function].svcs[req.address].activeRequests--
if c.logger.Core().Enabled(zap.DebugLevel) {
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Decrease active requests", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests))
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Decrease active requests", zap.String("function", req.function.String()), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests))
}
} else {
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Error("Invalid request to decrease active requests", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests))
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Error("Invalid request to decrease active requests", zap.String("function", req.function.String()), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests))
}
}
}
@@ -246,7 +283,12 @@ func (c *PoolCache) service() {
}
}
case deleteValue:
delete(c.cache[req.function].svcs, req.address)
if funcSvcGroup, ok := c.cache[req.function]; ok {
delete(c.cache[req.function].svcs, req.address)
if funcSvcGroup.deleted && len(c.cache[req.function].svcs) == 0 {
delete(c.cache, req.function)
}
}
req.responseChannel <- resp
case logFuncSvc:
datawriter := bufio.NewWriter(req.dumpWriter)
@@ -298,8 +340,15 @@ func (c *PoolCache) service() {
}
}
func (c *PoolCache) MarkFuncDeleted(function crd.CacheKeyURG) {
c.requestChannel <- &request{
requestType: markDeleted,
function: function,
}
}
// GetValue returns a function service with status in Active else return error
func (c *PoolCache) GetSvcValue(ctx context.Context, function string, requestsPerPod int, concurrency int) (*FuncSvc, error) {
func (c *PoolCache) GetSvcValue(ctx context.Context, function crd.CacheKeyURG, requestsPerPod int, concurrency int) (*FuncSvc, error) {
respChannel := make(chan *response)
c.requestChannel <- &request{
ctx: ctx,
@@ -334,7 +383,7 @@ func (c *PoolCache) ListAvailableValue() []*FuncSvc {
}
// SetValue marks the value at key [function][address] as active(begin used)
func (c *PoolCache) SetSvcValue(ctx context.Context, function, address string, value *FuncSvc, cpuLimit resource.Quantity, requestsPerPod int) {
func (c *PoolCache) SetSvcValue(ctx context.Context, function crd.CacheKeyURG, address string, value *FuncSvc, cpuLimit resource.Quantity, requestsPerPod, svcsRetain int) {
respChannel := make(chan *response)
c.requestChannel <- &request{
ctx: ctx,
@@ -344,12 +393,13 @@ func (c *PoolCache) SetSvcValue(ctx context.Context, function, address string, v
value: value,
cpuUsage: cpuLimit,
requestsPerPod: requestsPerPod,
svcsRetain: svcsRetain,
responseChannel: respChannel,
}
}
// SetCPUUtilization updates/sets the CPU utilization limit for the pod
func (c *PoolCache) SetCPUUtilization(function, address string, cpuUsage resource.Quantity) {
func (c *PoolCache) SetCPUUtilization(function crd.CacheKeyURG, address string, cpuUsage resource.Quantity) {
c.requestChannel <- &request{
requestType: setCPUUtilization,
function: function,
@@ -360,7 +410,7 @@ func (c *PoolCache) SetCPUUtilization(function, address string, cpuUsage resourc
}
// MarkAvailable marks the value at key [function][address] as available
func (c *PoolCache) MarkAvailable(function, address string) {
func (c *PoolCache) MarkAvailable(function crd.CacheKeyURG, address string) {
respChannel := make(chan *response)
c.requestChannel <- &request{
requestType: markAvailable,
@@ -371,7 +421,7 @@ func (c *PoolCache) MarkAvailable(function, address string) {
}
// DeleteValue deletes the value at key composed of [function][address]
func (c *PoolCache) DeleteValue(ctx context.Context, function, address string) error {
func (c *PoolCache) DeleteValue(ctx context.Context, function crd.CacheKeyURG, address string) error {
respChannel := make(chan *response)
c.requestChannel <- &request{
ctx: ctx,
@@ -385,7 +435,7 @@ func (c *PoolCache) DeleteValue(ctx context.Context, function, address string) e
}
// ReduceSpecializationInProgress reduces the svcWaiting count
func (c *PoolCache) MarkSpecializationFailure(function string) {
func (c *PoolCache) MarkSpecializationFailure(function crd.CacheKeyURG) {
c.requestChannel <- &request{
requestType: markSpecializationFailure,
function: function,
pkg/executor/fscache/poolcache_test.go
+143 -59
View File
@@ -12,6 +12,7 @@ import (
"github.com/stretchr/testify/require"
"k8s.io/apimachinery/pkg/api/resource"
"github.com/fission/fission/pkg/crd"
ferror "github.com/fission/fission/pkg/error"
"github.com/fission/fission/pkg/utils/loggerfactory"
)
@@ -26,77 +27,128 @@ func TestPoolCache(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
logger := loggerfactory.GetLogger()
c := NewPoolCache(logger)
concurrency := 5
requestsPerPod := 2
// should return err since no svc is present
_, err := c.GetSvcValue(ctx, "func", requestsPerPod, concurrency)
if err == nil {
log.Panicf("found value when expected it to be nil")
keyFunc := crd.CacheKeyURG{
UID: "func",
}
keyFunc2 := crd.CacheKeyURG{
UID: "func2",
}
c.SetSvcValue(ctx, "func", "ip", &FuncSvc{
Name: "value",
}, resource.MustParse("45m"), 10)
t.Run("Test create new svc ", func(t *testing.T) {
c1 := NewPoolCache(logger)
// should not return any error since we added a svc
_, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency)
checkErr(err)
// should return err since no svc is present
_, err := c1.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency)
if err == nil {
log.Panicf("found value when expected it to be nil")
}
c.SetSvcValue(ctx, "func", "ip", &FuncSvc{
Name: "value",
}, resource.MustParse("45m"), 10)
c1.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{
Name: "value",
}, resource.MustParse("45m"), 10, 0)
// should return err since all functions are busy
_, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency)
if err == nil {
log.Panicf("found value when expected it to be nil")
}
// should not return any error since we added a svc
_, err = c1.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency)
checkErr(err)
})
c.SetSvcValue(ctx, "func", "ip", &FuncSvc{
Name: "value",
}, resource.MustParse("45m"), 10)
t.Run("Test return error when functions are busy", func(t *testing.T) {
c2 := NewPoolCache(logger)
c2.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{
Name: "value",
}, resource.MustParse("45m"), 10, 0)
c2.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{
Name: "value",
}, resource.MustParse("45m"), 10, 0)
// should return err since all functions are busy
_, err := c2.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency)
if err == nil {
log.Panicf("found value when expected it to be nil")
}
})
c.SetSvcValue(ctx, "func2", "ip2", &FuncSvc{
Name: "value2",
}, resource.MustParse("50m"), 10)
t.Run("Test does not list available values when a function svc is deleted", func(t *testing.T) {
c3 := NewPoolCache(logger)
c3.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{
Name: "value",
}, resource.MustParse("45m"), 10, 0)
c.SetSvcValue(ctx, "func2", "ip22", &FuncSvc{
Name: "value22",
}, resource.MustParse("33m"), 10)
c3.SetSvcValue(ctx, keyFunc2, "ip2", &FuncSvc{
Name: "value2",
}, resource.MustParse("50m"), 10, 0)
checkErr(c.DeleteValue(ctx, "func2", "ip2"))
checkErr(c3.DeleteValue(ctx, keyFunc2, "ip2"))
cc := c.ListAvailableValue()
if len(cc) != 0 {
log.Panicf("expected 0 available items")
}
cc := c3.ListAvailableValue()
if len(cc) != 0 {
log.Panicf("expected 0 available items")
}
_, err := c3.GetSvcValue(ctx, keyFunc2, requestsPerPod, concurrency)
if err == nil {
log.Panicf("found deleted element")
}
})
c.MarkAvailable("func", "ip")
t.Run("Test return error when current CPU usage is more then permissible", func(t *testing.T) {
c4 := NewPoolCache(logger)
_, err := c4.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency)
if err == nil {
log.Panicf("found value when expected it to be nil")
}
checkErr(c.DeleteValue(ctx, "func", "ip"))
c4.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{
Name: "value",
}, resource.MustParse("45m"), 10, 0)
_, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency)
if err == nil {
log.Panicf("found deleted element")
}
// should not return any error since we added a svc
_, err = c4.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency)
checkErr(err)
c.SetSvcValue(ctx, "cpulimit", "100", &FuncSvc{
Name: "value",
}, resource.MustParse("3m"), 10)
c.SetCPUUtilization("cpulimit", "100", resource.MustParse("4m"))
c4.SetCPUUtilization(keyFunc, "ip", resource.MustParse("4m"))
_, err = c4.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency)
if err == nil {
log.Panicf("found value when expected it to be nil")
}
})
t.Run("Test function should not exist when mark deleted is called", func(t *testing.T) {
c5 := NewPoolCache(logger)
c5.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{
Name: "value",
}, resource.MustParse("45m"), 10, 0)
// should not return any error since we added a svc
_, err := c5.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency)
checkErr(err)
c5.MarkFuncDeleted(keyFunc)
checkErr(c5.DeleteValue(ctx, keyFunc, "ip"))
_, err = c5.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency)
if err == nil {
log.Panicf("found value when expected it to be nil")
}
})
}
func TestPoolCacheRequests(t *testing.T) {
key := crd.CacheKeyURG{
UID: "func",
Generation: 1,
}
type structForTest struct {
name string
requests int
concurrency int
rpp int
simultaneous int
failedRequests int
name string
requests int
concurrency int
rpp int
simultaneous int
failedRequests int
retainPods int
generationUpdate bool
}
for _, tt := range []structForTest{
@@ -118,7 +170,6 @@ func TestPoolCacheRequests(t *testing.T) {
concurrency: 5,
rpp: 60,
},
{
name: "test4",
requests: 6,
@@ -149,11 +200,19 @@ func TestPoolCacheRequests(t *testing.T) {
failedRequests: 10,
},
{
name: "test8",
requests: 10,
concurrency: 10,
rpp: 1,
simultaneous: 10,
name: "test8",
requests: 2,
concurrency: 2,
rpp: 1,
retainPods: 1,
},
{
name: "test9",
requests: 10,
concurrency: 5,
rpp: 2,
retainPods: 2,
generationUpdate: true,
},
} {
t.Run(fmt.Sprintf("scenario-%s", tt.name), func(t *testing.T) {
@@ -168,13 +227,13 @@ func TestPoolCacheRequests(t *testing.T) {
wg.Add(1)
go func(reqno int) {
defer wg.Done()
svc, err := p.GetSvcValue(context.Background(), "func", tt.rpp, tt.concurrency)
svc, err := p.GetSvcValue(context.Background(), key, tt.rpp, tt.concurrency)
if err != nil {
code, _ := ferror.GetHTTPError(err)
if code == http.StatusNotFound {
p.SetSvcValue(context.Background(), "func", fmt.Sprintf("svc-%d", svcCounter), &FuncSvc{
p.SetSvcValue(context.Background(), key, fmt.Sprintf("svc-%d", svcCounter), &FuncSvc{
Name: "value",
}, resource.MustParse("45m"), tt.rpp)
}, resource.MustParse("45m"), tt.rpp, tt.retainPods)
atomic.AddUint64(&svcCounter, 1)
} else {
t.Log(reqno, "=>", err)
@@ -195,6 +254,31 @@ func TestPoolCacheRequests(t *testing.T) {
require.Equal(t, tt.failedRequests, int(atomic.LoadUint64(&failedRequests)))
require.Equal(t, tt.concurrency, int(atomic.LoadUint64(&svcCounter)))
for i := 0; i < tt.concurrency; i++ {
for j := 0; j < tt.rpp; j++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
p.MarkAvailable(key, fmt.Sprintf("svc-%d", i))
}(i)
}
}
wg.Wait()
if tt.generationUpdate {
newKey := crd.CacheKeyURG{
UID: "func",
Generation: 2,
}
p.SetSvcValue(context.Background(), newKey, fmt.Sprintf("svc-%d", svcCounter), &FuncSvc{
Name: "value",
}, resource.MustParse("45m"), tt.rpp, tt.retainPods)
funcSvc := p.ListAvailableValue()
require.Equal(t, tt.concurrency, len(funcSvc))
} else {
funcSvc := p.ListAvailableValue()
require.Equal(t, tt.concurrency-tt.retainPods, len(funcSvc))
}
})
}
}