package fscache import ( "context" "fmt" "log" "net/http" "sync" "sync/atomic" "testing" "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" ) func checkErr(err error) { if err != nil { log.Panicf("err: %v", err) } } func TestPoolCache(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() logger := loggerfactory.GetLogger() concurrency := 5 requestsPerPod := 2 keyFunc := crd.CacheKeyURG{ UID: "func", } keyFunc2 := crd.CacheKeyURG{ UID: "func2", } t.Run("Test create new svc ", func(t *testing.T) { c1 := NewPoolCache(logger) // 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") } c1.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{ Name: "value", }, resource.MustParse("45m"), 10, 0) // should not return any error since we added a svc _, err = c1.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency) checkErr(err) }) 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") } }) 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) c3.SetSvcValue(ctx, keyFunc2, "ip2", &FuncSvc{ Name: "value2", }, resource.MustParse("50m"), 10, 0) checkErr(c3.DeleteValue(ctx, keyFunc2, "ip2")) 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") } }) 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") } c4.SetSvcValue(ctx, keyFunc, "ip", &FuncSvc{ Name: "value", }, resource.MustParse("45m"), 10, 0) // should not return any error since we added a svc _, err = c4.GetSvcValue(ctx, keyFunc, requestsPerPod, concurrency) checkErr(err) 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 retainPods int generationUpdate bool } for _, tt := range []structForTest{ { name: "test1", requests: 1, concurrency: 1, rpp: 1, }, { name: "test2", requests: 2, concurrency: 2, rpp: 1, }, { name: "test3", requests: 300, concurrency: 5, rpp: 60, }, { name: "test4", requests: 6, concurrency: 1, rpp: 5, failedRequests: 1, }, { name: "test5", requests: 6, concurrency: 5, rpp: 1, failedRequests: 1, }, { name: "test6", requests: 300, concurrency: 5, rpp: 60, simultaneous: 30, }, { name: "test7", requests: 310, concurrency: 5, rpp: 60, simultaneous: 30, failedRequests: 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) { var failedRequests, svcCounter uint64 p := NewPoolCache(loggerfactory.GetLogger()) wg := sync.WaitGroup{} simultaneous := tt.simultaneous if simultaneous == 0 { simultaneous = 1 } for i := 1; i <= tt.requests; i++ { reqno := i wg.Add(1) go func(reqno int) { defer wg.Done() svc, err := p.GetSvcValue(context.Background(), key, tt.rpp, tt.concurrency) if err != nil { code, _ := ferror.GetHTTPError(err) if code == http.StatusNotFound { atomic.AddUint64(&svcCounter, 1) address := fmt.Sprintf("svc-%d", atomic.LoadUint64(&svcCounter)) p.SetSvcValue(context.Background(), key, address, &FuncSvc{ Name: address, }, resource.MustParse("45m"), tt.rpp, tt.retainPods) } else { t.Log(reqno, "=>", err) atomic.AddUint64(&failedRequests, 1) } } else { if svc == nil { t.Log(reqno, "=>", "svc is nil") atomic.AddUint64(&failedRequests, 1) } // } else { // t.Log(reqno, "=>", svc.Name) // } } }(reqno) if reqno%simultaneous == 0 { wg.Wait() } } wg.Wait() 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) svcno := i go func(svcno int) { defer wg.Done() p.MarkAvailable(key, fmt.Sprintf("svc-%d", svcno+1)) }(svcno) } } wg.Wait() if tt.generationUpdate { newKey := crd.CacheKeyURG{ UID: "func", Generation: 2, } address := fmt.Sprintf("svc-%d", svcCounter) p.SetSvcValue(context.Background(), newKey, address, &FuncSvc{ Name: address, }, 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)) } }) } }