Saving FunctionTask inputs to file for profiling

Co-authored-by: Matteo Nardelli <matnar@gmail.com>

Author: grussorusso

internal/config/keys.go

@@ -111,5 +111,8 @@ const WORKFLOW_THRESHOLD_BASED_POLICY_THRESHOLD = "workflow.offloading.policy.th
 // Max number of tasks offloaded at once in the threshold-based offloading policy
 const WORKFLOW_THRESHOLD_BASED_POLICY_MAX_OFFLOADED = "workflow.offloading.policy.threshold.offloaded.max"
 
+// For profiling: save function input to file
+const WORKFLOW_SAVE_FUNCTION_INPUT_TO_FILE = "workflow.profiling.saveinput"
+
 const LOAD_BALANCER_POLICY = "loadbalancer.policy"
 const LOAD_BALANCER_TARGET_UPDATE_INTERVAL = "loadbalancer.targets.interval"

internal/function/signature.go

@@ -190,6 +190,7 @@ func (s *Signature) CheckOrMatchInputs(inputMap map[string]interface{}) error {
 		err := def.CheckInput(inputMap)
 
 		if err != nil && len(s.Inputs) == 1 {
+			// TODO: Consider a better solution. If there is a single input parameter, we just try to match type
 			key, ok := def.FindEntryThatTypeChecks(inputMap)
 			if ok {
 				val := inputMap[key]
@@ -295,30 +296,3 @@ func datatypeToString(dataType DataTypeEnum) string {
 		return ""
 	}
 }
-
-// SignatureInference is a best-effort function that tries to infer signature from a function without a defined signature. Maybe we do not need it.
-func SignatureInference(params map[string]interface{}) *Signature {
-	signatureBuilder := NewSignature()
-
-	for k, v := range params {
-		typeList := []DataTypeEnum{
-			Float{},
-			Int{},
-			Bool{},
-			Text{},
-			Array[Float]{},
-			Array[Int]{},
-			Array[Bool]{},
-			Array[Text]{},
-			Void{},
-		}
-		for _, t := range typeList {
-			if t.TypeCheck(v) == nil {
-				signatureBuilder.AddInput(k, t)
-				break
-			}
-		}
-	}
-
-	return signatureBuilder.AddOutput("result", Text{}).Build()
-}

internal/workflow/function_task.go

@@ -4,6 +4,9 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/serverledge-faas/serverledge/internal/config"
+	"log"
+	"os"
 	"time"
 
 	"github.com/serverledge-faas/serverledge/internal/node"
@@ -40,7 +43,6 @@ func (s *FunctionTask) execute(input *TaskData, r *Request) (map[string]interfac
 
 	// FIXME: We are adding additional inputs from r.Params to match the function signature. This workaround should be
 	// dropped when we introduce the possibility to specify additional parameters for functions in a workflow.
-
 	funct, exists := function.GetFunction(s.Func)
 	if !exists {
 		return nil, fmt.Errorf("funtion %s doesn't exists", s.Func)
@@ -60,6 +62,28 @@ func (s *FunctionTask) execute(input *TaskData, r *Request) (map[string]interfac
 	if err != nil {
 		return nil, err
 	}
+
+	if config.GetBool(config.WORKFLOW_SAVE_FUNCTION_INPUT_TO_FILE, false) {
+		// Create directory if it doesn't exist
+		dirname := "saved-inputs"
+		err := os.MkdirAll(dirname, 0755)
+		if err != nil {
+			log.Printf("could not create directory: %v", err)
+		}
+
+		payload, err := json.Marshal(input.Data)
+		if err != nil {
+			log.Printf("could not marshal input data: %v", err)
+		} else {
+			// Write to file
+			filename := fmt.Sprintf("%s/%s-%s-%s.json", dirname, s.Id, s.Func, r.Id)
+			err = os.WriteFile(filename, payload, 0644)
+			if err != nil {
+				log.Printf("could not write input data to file: %v", err)
+			}
+		}
+	}
+
 	output, err := s.exec(r, input.Data)
 	if err != nil {
 		return nil, err