Removes Progress and PartialData cache

Author: grussorusso

internal/api/server.go

@@ -52,7 +52,6 @@ func CacheSetup() {
 	// setup cache space
 	cache.Size = config.GetInt(config.CACHE_SIZE, 100)
 
-	cache.Persist = config.GetBool(config.CACHE_PERSISTENCE, true)
 	//setup cleanup interval
 	d := config.GetInt(config.CACHE_CLEANUP, 60)
 	interval := time.Duration(d)

internal/cache/cache_handler.go

@@ -14,7 +14,6 @@ var (
 	DefaultExp      time.Duration = 0 // default expiration
 	CleanupInterval time.Duration = 0 //cleanup interval
 	Size                          = 0
-	Persist                       = false // default value used to test progress/partial datas. If false, they are only saved in local cache
 )
 
 // GetCacheInstance : singleton implementation to retrieve THE cache

internal/test/partial_data_test.go

@@ -3,11 +3,9 @@ package test
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/serverledge-faas/serverledge/internal/cache"
 	"github.com/serverledge-faas/serverledge/internal/workflow"
 	u "github.com/serverledge-faas/serverledge/utils"
 	"testing"
-	"time"
 )
 
 func TestPartialDataMarshaling(t *testing.T) {
@@ -28,83 +26,3 @@ func TestPartialDataMarshaling(t *testing.T) {
 
 	u.AssertTrueMsg(t, retrieved.Equals(&partialData), fmt.Sprintf("retrieved partialData is not equal to initial partialData. Retrieved:\n%s,\nExpected:\n%s ", retrieved.String(), partialData.String()))
 }
-
-func TestPartialDataCache(t *testing.T) {
-	// it's an integration test because it needs etcd
-	if testing.Short() {
-		t.Skip()
-	}
-
-	request1 := workflow.ReqId("abc")
-	request2 := workflow.ReqId("zzz")
-
-	data := make(map[string]interface{})
-	data["num"] = 1
-	partialData1 := initPartialData(request1, "nodo1", "start", data)
-	data = make(map[string]interface{})
-	data["num"] = 2
-	partialData2 := initPartialData(request1, "nodo2", "nodo1", data)
-	data = make(map[string]interface{})
-	data["num"] = 3
-	partialData3 := initPartialData(request2, "start", "", data)
-	partialDatas := []*workflow.PartialData{partialData1, partialData2, partialData3}
-
-	// saving and retrieving partial datas one by one
-	for i := 0; i < len(partialDatas); i++ {
-		partialData := partialDatas[i]
-		err := workflow.SavePartialData(partialData, cache.Persist)
-		u.AssertNilMsg(t, err, "failed to save partialData")
-
-		retrievedPartialData, err := workflow.RetrievePartialData(partialData.ReqId, partialData.ForTask, cache.Persist)
-		u.AssertNilMsg(t, err, "partialData not found")
-		u.AssertTrueMsg(t, partialData.Equals(retrievedPartialData[0]), "progresses don't match")
-
-		_, err = workflow.DeleteAllPartialData(partialData.ReqId, cache.Persist)
-		u.AssertNilMsg(t, err, "failed to delete partialData")
-
-		_, err = workflow.RetrievePartialData(partialData.ReqId, partialData.ForTask, cache.Persist)
-		u.AssertNonNilMsg(t, err, "partialData should have been deleted")
-	}
-
-	requests := []workflow.ReqId{request1, request2}
-	partialDataMap := make(map[workflow.ReqId][]*workflow.PartialData)
-	partialDataMap[request1] = make([]*workflow.PartialData, 0, 2)
-	partialDataMap[request1] = append(partialDataMap[request1], partialData1, partialData2)
-	partialDataMap[request2] = make([]*workflow.PartialData, 0, 1)
-	partialDataMap[request2] = append(partialDataMap[request2], partialData3)
-
-	// saving, retrieving and deleting partial data request by request
-	for i := 0; i < len(requests); i++ {
-		request := requests[i]
-		partialDataList := partialDataMap[request]
-		for _, partialData := range partialDataList {
-			err := workflow.SavePartialData(partialData, cache.Persist)
-			u.AssertNilMsg(t, err, "failed to save partialData")
-		}
-
-		retrievedPartialData, err := workflow.RetrieveAllPartialData(request, cache.Persist)
-		u.AssertNil(t, err)
-		count := 0
-		retrievedPartialData.Range(func(key, value any) bool {
-			count++
-			return true
-		})
-		u.AssertEqualsMsg(t, len(partialDataList), count, "number of partial data for request  differs")
-
-		_, err = workflow.DeleteAllPartialData(request, cache.Persist)
-		u.AssertNilMsg(t, err, "failed to delete all partialData")
-
-		time.Sleep(200 * time.Millisecond)
-
-		numPartialData := workflow.NumberOfPartialDataFor(request, cache.Persist)
-		u.AssertEqualsMsg(t, 0, numPartialData, "retrieved partialData should have been 0")
-	}
-}
-
-func initPartialData(reqId workflow.ReqId, to, from workflow.TaskId, data map[string]interface{}) *workflow.PartialData {
-	return &workflow.PartialData{
-		ReqId:   reqId,
-		ForTask: to,
-		Data:    data,
-	}
-}

internal/test/progress_test.go

@@ -3,7 +3,6 @@ package test
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/serverledge-faas/serverledge/internal/cache"
 	"github.com/serverledge-faas/serverledge/internal/workflow"
 	u "github.com/serverledge-faas/serverledge/utils"
 	"testing"
@@ -120,29 +119,29 @@ func TestProgressCache(t *testing.T) {
 	for i := 0; i < len(workflows); i++ {
 		progress := progresses[i]
 		wflow := workflows[i]
-		err := workflow.SaveProgress(progress, cache.Persist)
+		err := workflow.SaveProgress(progress)
 		u.AssertNilMsg(t, err, "failed to save progress")
 
-		retrievedProgress, found := workflow.RetrieveProgress(progress.ReqId, cache.Persist)
-		u.AssertTrueMsg(t, found, "progress not found")
+		retrievedProgress, err := workflow.RetrieveProgress(progress.ReqId)
+		u.AssertNilMsg(t, err, "progress not found")
 		u.AssertTrueMsg(t, progress.Equals(retrievedProgress), "progresses don't match")
 
 		progress.Complete(wflow.Start.Id)
 		progress.Complete(wflow.Start.Next)
 
-		err = workflow.SaveProgress(progress, cache.Persist)
+		err = workflow.SaveProgress(progress)
 		u.AssertNilMsg(t, err, "failed to save after update")
 
-		retrievedProgress, found = workflow.RetrieveProgress(progress.ReqId, cache.Persist)
-		u.AssertTrueMsg(t, found, "progress not found after update")
+		retrievedProgress, err = workflow.RetrieveProgress(progress.ReqId)
+		u.AssertNilMsg(t, err, "progress not found after update")
 		u.AssertTrueMsg(t, progress.Equals(retrievedProgress), "progresses don't match after update")
 
-		err = workflow.DeleteProgress(progress.ReqId, cache.Persist)
+		err = workflow.DeleteProgress(progress.ReqId)
 		u.AssertNilMsg(t, err, "failed to delete progress")
 
 		time.Sleep(200 * time.Millisecond)
 
-		_, found = workflow.RetrieveProgress(progress.ReqId, cache.Persist)
-		u.AssertFalseMsg(t, found, "progress should have been deleted")
+		_, err = workflow.RetrieveProgress(progress.ReqId)
+		u.AssertNonNilMsg(t, err, "progress not found after delete")
 	}
 }

internal/test/workflow_integration_test.go

@@ -603,9 +603,9 @@ func TestResumeWorkflow(t *testing.T) {
 	progress := workflow.InitProgress(workflow.ReqId(request.Id), wflow)
 	pd := workflow.NewPartialData(workflow.ReqId(request.Id), wflow.Start.Id, request.Params)
 
-	err = workflow.SaveProgress(progress, true)
+	err = workflow.SaveProgress(progress)
 	u.AssertNil(t, err)
-	err = workflow.SavePartialData(pd, true)
+	err = workflow.SavePartialData(pd)
 	u.AssertNil(t, err)
 
 	resumedRequest := workflow.NewRequest(request.Id, wflow, params)