Workflow extention of Serverledge

Co-authored-by: Valerio Crecco <valerio.crecco@alumni.uniroma2.eu>
Co-authored-by: Ludovico De Santis <ludovico.desantis@alumni.uniroma2.eu>

Author: 3 people

cmd/lb/main.go

@@ -47,7 +47,13 @@ func main() {
 	// TODO: split Area in Region + Type (e.g., cloud/lb/edge)
 	region := config.GetString(config.REGISTRY_AREA, "ROME")
 	registry := &registration.Registry{Area: "lb/" + region}
-	hostport := fmt.Sprintf("http://%s:%d", utils.GetIpAddress().String(), config.GetInt(config.API_PORT, 1323))
+	ipAddress, err := utils.GetOutboundIp()
+	if err != nil {
+		log.Printf("Could not get ip address: %v\n", err)
+		return
+	}
+
+	hostport := fmt.Sprintf("http://%s:%d", ipAddress.String(), config.GetInt(config.API_PORT, 1323))
 	if _, err := registry.RegisterToEtcd(hostport); err != nil {
 		log.Printf("Could not register to Etcd: %v\n", err)
 	}

cmd/serverledge/main.go

@@ -119,7 +119,11 @@ func main() {
 		log.Fatal(err)
 	}
 
-	ip := config.GetString(config.API_IP, utils.GetIpAddress().String())
+	address, err := utils.GetOutboundIp()
+	if err != nil {
+		log.Fatalf("failed to get ip address: %v", err)
+	}
+	ip := config.GetString(config.API_IP, address.String())
 	url := fmt.Sprintf("http://%s:%d", ip, config.GetInt(config.API_PORT, 1323))
 	myKey, err := registry.RegisterToEtcd(url)
 	if err != nil {

internal/api/api.go

@@ -203,7 +203,13 @@ func GetServerStatus(c echo.Context) error {
 	node.Resources.RLock()
 	defer node.Resources.RUnlock()
 	portNumber := config.GetInt("api.port", 1323)
-	url := fmt.Sprintf("http://%s:%d", utils.GetIpAddress().String(), portNumber)
+
+	address, err := utils.GetOutboundIp()
+	if err != nil {
+		return c.String(http.StatusServiceUnavailable, err.Error())
+	}
+
+	url := fmt.Sprintf("http://%s:%d", address.String(), portNumber)
 	response := registration.StatusInformation{
 		Url:                     url,
 		AvailableWarmContainers: node.WarmStatus(),

internal/api/composition.go

@@ -5,6 +5,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
+	"net/http"
+	"time"
+
 	"github.com/cornelk/hashmap"
 	"github.com/grussorusso/serverledge/internal/client"
 	"github.com/grussorusso/serverledge/internal/container"
@@ -14,9 +18,6 @@ import (
 	"github.com/grussorusso/serverledge/internal/node"
 	"github.com/labstack/echo/v4"
 	"github.com/labstack/gommon/log"
-	"io"
-	"net/http"
-	"time"
 )
 
 // ===== Function Composition =====
@@ -31,7 +32,7 @@ func CreateFunctionCompositionFromASL(e echo.Context) error {
 
 	err := json.Unmarshal(body, &creationRequest)
 	if err != nil && err != io.EOF {
-		log.Printf("Could not parse request: %v", err)
+		log.Printf("Could not parse compose request - error during unmarshal: %v", err)
 		return err
 	}
 
@@ -49,7 +50,8 @@ func CreateFunctionCompositionFromASL(e echo.Context) error {
 		log.Printf("Could not decode composition source ASL: %v", err)
 		return e.JSON(http.StatusBadRequest, "composition already exists")
 	}
-	comp, err := fc.FromASL(creationRequest.Name, decodedSrc[:])
+
+	comp, err := fc.FromASL(creationRequest.Name, creationRequest.RmFnOnDeletionFlag, decodedSrc[:])
 	if err != nil {
 		log.Printf("Could not parse composition from ASL: %v", err)
 		return e.JSON(http.StatusBadRequest, "composition already exists")
@@ -76,7 +78,7 @@ func CreateFunctionComposition(e echo.Context) error {
 
 	err := json.Unmarshal(body, &comp)
 	if err != nil && err != io.EOF {
-		log.Printf("Could not parse request: %v", err)
+		log.Printf("Could not parse composition request - error during unmarshal: %v", err)
 		return err
 	}
 	// checking if the function already exists. If exists we return an error
@@ -97,6 +99,10 @@ func CreateFunctionComposition(e echo.Context) error {
 				log.Printf("Dropping request for composition with non-existing function '%s'", fName)
 				return e.JSON(http.StatusBadRequest, "composition with non-existing function")
 			}
+			if f.Signature == nil {
+				return e.JSON(http.StatusBadRequest, "function "+fName+"has nil signature")
+			}
+
 			funcs[fName] = f
 		}
 		comp.Functions = funcs
@@ -137,7 +143,7 @@ func DeleteFunctionComposition(c echo.Context) error {
 	// here we only need the name of the function composition (and if all function should be deleted with it)
 	err := json.NewDecoder(c.Request().Body).Decode(&comp)
 	if err != nil && err != io.EOF {
-		log.Printf("Could not parse request: %v", err)
+		log.Printf("Could not parse delete request - error during decoding: %v", err)
 		return err
 	}
 
@@ -189,7 +195,7 @@ func InvokeFunctionComposition(e echo.Context) error {
 	var fcInvocationRequest client.CompositionInvocationRequest
 	err := json.NewDecoder(e.Request().Body).Decode(&fcInvocationRequest)
 	if err != nil && err != io.EOF {
-		log.Printf("Could not parse request: %v", err)
+		log.Printf("Could not parse invoke request - error during decoding: %v", err)
 		return e.JSON(http.StatusInternalServerError, "failed to parse composition invocation request. Check parameters and composition definition")
 	}
 	// gets a fc.CompositionRequest from the pool goroutine-safe cache.
@@ -214,11 +220,6 @@ func InvokeFunctionComposition(e echo.Context) error {
 			OffloadLatency: 0,
 			SchedAction:    "",
 		})
-
-		//fcReq.ExecReport.Reports[execReportId] = &function.ExecutionReport{
-		//	SchedAction:    "",
-		//	OffloadLatency: 0.0,
-		//}
 	}
 
 	if fcReq.Async {
@@ -242,6 +243,17 @@ func InvokeFunctionComposition(e echo.Context) error {
 		}
 		return e.JSON(http.StatusInternalServerError, v)
 	} else {
-		return e.JSON(http.StatusOK, fc.CompositionResponse{Success: true, CompositionExecutionReport: fcReq.ExecReport})
+		reports := make(map[string]*function.ExecutionReport)
+		fcReq.ExecReport.Reports.Range(func(id fc.ExecutionReportId, report *function.ExecutionReport) bool {
+			reports[string(id)] = report
+			return true
+		})
+
+		return e.JSON(http.StatusOK, fc.CompositionResponse{
+			Success:      true,
+			Result:       fcReq.ExecReport.Result,
+			Reports:      reports,
+			ResponseTime: fcReq.ExecReport.ResponseTime,
+		})
 	}
 }

internal/api/server.go

@@ -4,18 +4,19 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"os/signal"
+	"time"
+
 	"github.com/grussorusso/serverledge/internal/cache"
 	"github.com/grussorusso/serverledge/internal/config"
 	"github.com/grussorusso/serverledge/internal/node"
 	"github.com/grussorusso/serverledge/internal/registration"
 	"github.com/grussorusso/serverledge/internal/scheduling"
 	"github.com/labstack/echo/v4"
 	"github.com/labstack/echo/v4/middleware"
-	"log"
-	"net/http"
-	"os"
-	"os/signal"
-	"time"
 )
 
 func StartAPIServer(e *echo.Echo) {

internal/asl/catch.go

@@ -0,0 +1,29 @@
+package asl
+
+import (
+	"github.com/grussorusso/serverledge/internal/types"
+	"golang.org/x/exp/slices"
+)
+
+// Catch is a field in Task, Parallel and Map states. When a state reports an error and either there is no Retrier, or retries have failed to resolve the error, the interpreter scans through the Catchers in array order, and when the Error Name appears in the value of a Catcher’s "ErrorEquals" field, transitions the machine to the state named in the value of the "Next" field. The reserved name "States.ALL" appearing in a Retrier’s "ErrorEquals" field is a wildcard and matches any Error Name.
+type Catch struct {
+	ErrorEquals []string
+	ResultPath  string
+	Next        string
+}
+
+func (c *Catch) Equals(cmp types.Comparable) bool {
+	c2 := cmp.(*Catch)
+
+	return slices.Equal(c.ErrorEquals, c2.ErrorEquals) &&
+		c.ResultPath == c2.ResultPath &&
+		c.Next == c2.Next
+}
+
+type CanCatch interface {
+	GetCatchOpt() Catch
+}
+
+func NoCatch() *Catch {
+	return &Catch{}
+}

internal/asl/choice.go

@@ -0,0 +1,123 @@
+package asl
+
+import (
+	"fmt"
+
+	"github.com/buger/jsonparser"
+	"github.com/grussorusso/serverledge/internal/types"
+	"github.com/labstack/gommon/log"
+)
+
+type ChoiceState struct {
+	Type       StateType    // Necessary
+	Choices    []ChoiceRule // Necessary. All DataTestExpression must be State Machine with an end, like fc.ChoiceNode(s).
+	InputPath  Path         // Optional
+	OutputPath Path         // Optional
+	// Default is the default state to execute when no other DataTestExpression matches
+	Default string // Optional, but to avoid errors it is highly recommended.
+}
+
+func (c *ChoiceState) ParseFrom(jsonData []byte) (State, error) {
+	c.Type = StateType(JsonExtractStringOrDefault(jsonData, "Type", "Choice"))
+	c.InputPath = JsonExtractRefPathOrDefault(jsonData, "InputPath", "")
+	c.OutputPath = JsonExtractRefPathOrDefault(jsonData, "OutputPath", "")
+	c.Default = JsonExtractStringOrDefault(jsonData, "Default", "")
+
+	choiceRules := make([]ChoiceRule, 0)
+
+	choices, errChoice := JsonExtract(jsonData, "Choices")
+	if errChoice != nil {
+		return nil, fmt.Errorf("failed to parse Choices %v", errChoice)
+	}
+
+	_, _ = jsonparser.ArrayEach(choices, func(value []byte, dataType jsonparser.ValueType, offset int, err error) {
+		cr, errR := ParseRule(value)
+		if errR != nil {
+			log.Errorf("failed to parse choice rule %d: %v", offset, err)
+			return
+		}
+		choiceRules = append(choiceRules, cr)
+	})
+	//if errArr != nil {
+	//	return nil, fmt.Errorf("error %v when parsing choice rule %s", errArr, choices[:offset])
+	//}
+	c.Choices = choiceRules
+
+	return c, nil
+}
+
+func (c *ChoiceState) Validate(stateNames []string) error {
+	if c.Default == "" {
+		log.Warn("Default choice not specified")
+	}
+	return nil
+}
+
+func NewEmptyChoice() *ChoiceState {
+	return &ChoiceState{
+		Type:       Choice,
+		Choices:    []ChoiceRule{},
+		InputPath:  "",
+		OutputPath: "",
+		Default:    "",
+	}
+}
+
+func (c *ChoiceState) GetType() StateType {
+	return Choice
+}
+
+// GetNext for ChoiceState returns the Default branch instead of next
+func (c *ChoiceState) GetNext() (string, bool) {
+	if c.Default != "" {
+		return c.Default, true
+	}
+	return "", false
+}
+
+// IsEndState always returns true for a ChoiceState, because it is always a terminal state.
+func (c *ChoiceState) IsEndState() bool {
+	return true
+}
+
+func (c *ChoiceState) Equals(cmp types.Comparable) bool {
+	c2 := cmp.(*ChoiceState)
+
+	if len(c.Choices) != len(c2.Choices) {
+		return false
+	}
+
+	for i, c1 := range c.Choices {
+		if !c1.Equals(c2.Choices[i]) {
+			return false
+		}
+	}
+
+	return c.Type == c2.Type &&
+		c.InputPath == c2.InputPath &&
+		c.OutputPath == c2.OutputPath &&
+		c.Default == c2.Default
+}
+
+func (c *ChoiceState) String() string {
+	str := fmt.Sprint("{",
+		"\n\t\t\tType: ", c.Type,
+		"\n\t\t\tDefault: ", c.Default,
+		"\n\t\t\tChoices: [")
+	for i, c1 := range c.Choices {
+		str += c1.String()
+		if i < len(c.Choices)-1 {
+			str += ","
+		}
+	}
+	str += "\n\t\t\t]\n"
+
+	if c.InputPath != "" {
+		str += fmt.Sprintf("\t\t\tInputPath: %s\n", c.InputPath)
+	}
+	if c.OutputPath != "" {
+		str += fmt.Sprintf("\t\t\tOutputPath: %s\n", c.OutputPath)
+	}
+	str += "\t\t}"
+	return str
+}