Tweaked Driver to invoke vSwarm functions

.github/workflows/e2e.yaml

@@ -0,0 +1,104 @@
+name: End-to-End Tests
+
+on:
+ schedule:
+   - cron: "0 9 * * 1"
+ workflow_dispatch:
+ push:
+   branches: [main]
+ pull_request:
+   branches: [main]
+
+env:
+  GOOS: linux
+  GO111MODULE: on
+
+jobs:
+  test-knative:
+    name: Test Knative Deployment
+    env:
+      KIND_VERSION: v0.22.0
+      K8S_VERSION: v1.29
+      YAML_DIR: workloads/container
+
+    runs-on: ubuntu-20.04
+    strategy:
+      fail-fast: false
+      matrix:
+        service:
+          [
+            trace_func_go,
+          ]
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          lfs: "true"
+      - name: Checkout LFS objects
+        run: git lfs checkout
+
+      - uses: actions/setup-go@v5
+        with:
+          go-version: 1.22
+
+      - name: Create k8s Kind Cluster
+        run: bash ./scripts/konk-ci/01-kind.sh
+
+      - name: Install Serving
+        run: bash ./scripts/konk-ci/02-serving.sh
+
+      - name: Install Kourier
+        run: bash ./scripts/konk-ci/02-kourier.sh
+
+      - name: Setup domain and autoscaler
+        run: |
+          INGRESS_HOST="127.0.0.1"
+          KNATIVE_DOMAIN=$INGRESS_HOST.sslip.io
+          kubectl patch configmap -n knative-serving config-domain -p "{\"data\": {\"$KNATIVE_DOMAIN\": \"\"}}"
+          kubectl patch configmap -n knative-serving config-autoscaler -p "{\"data\": {\"allow-zero-initial-scale\": \"true\"}}"
+          kubectl patch configmap -n knative-serving config-features -p "{\"data\": {\"kubernetes.podspec-affinity\": \"enabled\"}}"
+          kubectl label node knative-control-plane loader-nodetype=worker
+
+      - name: Build and run loader
+        run: go run cmd/loader.go --config pkg/config/test_config.json
+
+      - name: Check the output
+        run: test -f "data/out/experiment_duration_2.csv" && test $(cat data/out/experiment_duration_2.csv | wc -l) -gt 1 && test $(grep true data/out/experiment_duration_2.csv | wc -l) -eq 0 # test the output file for errors (true means failure to invoke)
+
+      - name: Print logs
+        if: ${{ always() }}
+        run: |
+          set -x
+          container_list=$(kubectl get pods -n default -o jsonpath="{.items[*].spec.containers[*].name}")
+          for container_name in $container_list
+          do
+            kubectl logs -n default -c $container_name -l serving.knative.dev/service=${{ matrix.service }}
+          done
+      - name: Down
+        if: ${{ always() }}
+        run: |
+          kn service delete --all
+
+      - name: Untar vSwarm YAMLs
+        if: ${{ always() }}
+        run: |
+          tar -xzvf workloads/container/yamls.tar.gz -C workloads/container/ 
+      - name: Run vSwarm loader
+        run: go run cmd/loader.go --config pkg/config/test_vswarm_config.json
+
+      - name: Check vSwarm output
+        run: test -f "data/out/experiment_duration_2.csv" && test $(cat data/out/experiment_duration_2.csv | wc -l) -gt 1 && test $(grep true data/out/experiment_duration_2.csv | wc -l) -le 1 # test the output file for errors (true means failure to invoke)
+
+      - name: Print logs
+        if: ${{ always() }}
+        run: |
+          set -x
+          container_list=$(kubectl get pods -n default -o jsonpath="{.items[*].spec.containers[*].name}")
+          for container_name in $container_list
+          do
+            kubectl logs -n default -c $container_name -l serving.knative.dev/service=${{ matrix.service }}
+          done
+      - name: Down
+        if: ${{ always() }}
+        run: |
+          kn service delete --all
+

.github/workflows/e2e_loader.yaml

@@ -39,7 +39,7 @@ jobs:
         run: go run cmd/loader.go --config pkg/config/test_config.json
 
       - name: Check the output
-        run: test -f "data/out/experiment_duration_5.csv" && test $(cat data/out/experiment_duration_5.csv | wc -l) -gt 1 && test $(grep true data/out/experiment_duration_5.csv | wc -l) -eq 0 # test the output file for errors (true means failure to invoke)
+        run: test -f "data/out/experiment_duration_2.csv" && test $(cat data/out/experiment_duration_2.csv | wc -l) -gt 1 && test $(grep true data/out/experiment_duration_2.csv | wc -l) -eq 0 # test the output file for errors (true means failure to invoke)
 
       - name: Print logs
         if: ${{ always() }}
@@ -54,4 +54,31 @@ jobs:
       - name: Down
         if: ${{ always() }}
         run: |
-          kn service delete --all
+          kn service delete --all
+
+      - name: Untar vSwarm YAMLs
+        if: ${{ always() }}
+        run: |
+          tar -xzvf workloads/container/yamls.tar.gz -C workloads/container/ 
+      - name: Run vSwarm loader
+        run: go run cmd/loader.go --config pkg/config/test_vswarm_config.json
+
+      - name: Check vSwarm output
+        run: test -f "data/out/experiment_duration_2.csv" && test $(cat data/out/experiment_duration_2.csv | wc -l) -gt 1 && test $(grep true data/out/experiment_duration_2.csv | wc -l) -le 1 # test the output file for errors (true means failure to invoke)
+
+      - name: Print logs
+        if: ${{ always() }}
+        run: |
+          set -x
+          container_list=$(kubectl get pods -n default -o jsonpath="{.items[*].spec.containers[*].name}")
+          for container_name in $container_list
+          do
+            kubectl logs -n default -c $container_name -l serving.knative.dev/service=${{ matrix.service }}
+          done
+      - name: Down
+        if: ${{ always() }}
+        run: |
+          kn service delete --all
+
+
+

cmd/config_vswarm_trace.json

@@ -0,0 +1,32 @@
+{
+  "Seed": 42,
+
+  "Platform": "Knative",
+  "InvokeProtocol" : "grpc",
+  "YAMLSelector": "container",
+  "EndpointPort": 80,
+
+  "BusyLoopOnSandboxStartup": false,
+
+  "TracePath": "data/traces/example/",
+  "Granularity": "minute",
+  "OutputPathPrefix": "data/out/experiment",
+  "IATDistribution": "exponential",
+  "CPULimit": "1vCPU",
+  "ExperimentDuration": 10,
+  "WarmupDuration": 0,
+
+  "IsPartiallyPanic": false,
+  "EnableZipkinTracing": false,
+  "EnableMetricsScrapping": false,
+  "MetricScrapingPeriodSeconds": 15,
+  "AutoscalingMetric": "concurrency",
+
+  "GRPCConnectionTimeoutSeconds": 15,
+  "GRPCFunctionTimeoutSeconds": 900,
+  "DAGMode": false,
+  "EnableDAGDataset": true,
+  "Width": 2,
+  "Depth": 2,
+  "VSwarm": true
+}

cmd/loader.go

@@ -175,11 +175,15 @@ func parseTraceGranularity(cfg *config.LoaderConfiguration) common.TraceGranular
 func runTraceMode(cfg *config.LoaderConfiguration, readIATFromFile bool, writeIATsToFile bool) {
 	durationToParse := determineDurationToParse(cfg.ExperimentDuration, cfg.WarmupDuration)
 	yamlPath := parseYAMLSpecification(cfg)
-
+	var functions []*common.Function
 	// Azure trace parsing
-	traceParser := trace.NewAzureParser(cfg.TracePath, durationToParse)
-	functions := traceParser.Parse()
-
+	if !cfg.VSwarm {
+		traceParser := trace.NewAzureParser(cfg.TracePath, durationToParse, yamlPath)
+		functions = traceParser.Parse()
+	} else {
+		traceParser := trace.NewMapperParser(cfg.TracePath, durationToParse)
+		functions = traceParser.Parse()
+	}
 	// Dirigent metadata parsing
 	dirigentMetadataParser := trace.NewDirigentMetadataParser(cfg.TracePath, functions, yamlPath, cfg.Platform)
 	dirigentMetadataParser.Parse()
@@ -200,7 +204,7 @@ func runTraceMode(cfg *config.LoaderConfiguration, readIATFromFile bool, writeIA
 		TraceGranularity: parseTraceGranularity(cfg),
 		TraceDuration:    durationToParse,
 
-		YAMLPath: yamlPath,
+		YAMLPath: yamlPath, //make a slice of YAMLPaths instead
 		TestMode: false,
 
 		Functions: functions,

data/traces/example/mapper_output.json

@@ -0,0 +1,14 @@
+{
+    "c13acdc7567b225971cef2416a3a2b03c8a4d8d154df48afe75834e2f5c59ddf": {
+        "proxy-function": "video-processing-python-10"
+    },
+    "a2faad786b3c813b12ce57d349d5e62f6d0f22ceecfa86cd72a962853383b600": {
+        "proxy-function": "image-rotate-go-11"
+    },
+    "7dc5aeabc131669912e8c793c8925cc9928321f45f13a4af031592b4611630d7": {
+        "proxy-function": "video-processing-python-100"
+    },
+    "ae8a1640fa932024f59b38a0b001808b5c64612bd60c6f3eb80ba9461ba2d091": {
+        "proxy-function": "video-processing-python-10"
+    }
+}

pkg/common/trace_types.go

@@ -99,11 +99,12 @@ type Function struct {
 
 	ColdStartBusyLoopMs int
 
-	CPURequestsMilli  int
-	MemoryRequestsMiB int
-	CPULimitsMilli    int
-
-	Specification *FunctionSpecification
+	CPURequestsMilli      int
+	MemoryRequestsMiB     int
+	CPULimitsMilli        int
+	YAMLPath              string
+	PreDeploymentCommands []string
+	Specification         *FunctionSpecification
 }
 
 type Node struct {

pkg/config/parser_test.go

@@ -53,7 +53,7 @@ func TestConfigParser(t *testing.T) {
 		!strings.HasPrefix(config.OutputPathPrefix, "data/out/experiment") ||
 		config.IATDistribution != "equidistant" ||
 		config.CPULimit != "1vCPU" ||
-		config.ExperimentDuration != 5 ||
+		config.ExperimentDuration != 2 ||
 		config.WarmupDuration != 0 ||
 		config.IsPartiallyPanic != false ||
 		config.EnableZipkinTracing != false ||

pkg/config/test_config.json

@@ -11,7 +11,7 @@
   "OutputPathPrefix": "data/out/experiment",
   "IATDistribution": "equidistant",
   "CPULimit": "1vCPU",
-  "ExperimentDuration": 5,
+  "ExperimentDuration": 2,
   "WarmupDuration": 0,
 
   "IsPartiallyPanic": false,

pkg/config/test_vswarm_config.json

@@ -0,0 +1,32 @@
+{
+  "Seed": 42,
+
+  "Platform": "Knative",
+  "InvokeProtocol" : "grpc",
+  "YAMLSelector": "container",
+  "EndpointPort": 80,
+
+  "BusyLoopOnSandboxStartup": false,
+
+  "TracePath": "data/traces/example/",
+  "Granularity": "minute",
+  "OutputPathPrefix": "data/out/experiment",
+  "IATDistribution": "exponential",
+  "CPULimit": "1vCPU",
+  "ExperimentDuration": 2,
+  "WarmupDuration": 0,
+
+  "IsPartiallyPanic": false,
+  "EnableZipkinTracing": false,
+  "EnableMetricsScrapping": false,
+  "MetricScrapingPeriodSeconds": 15,
+  "AutoscalingMetric": "concurrency",
+
+  "GRPCConnectionTimeoutSeconds": 15,
+  "GRPCFunctionTimeoutSeconds": 900,
+  "DAGMode": false,
+  "EnableDAGDataset": true,
+  "Width": 2,
+  "Depth": 2,
+  "VSwarm": true
+}

pkg/driver/clients/grpc_client_test.go

@@ -25,19 +25,16 @@
 package clients
 
 import (
-	"context"
 	"fmt"
+	"os"
+	"testing"
+	"time"
+
 	"github.com/sirupsen/logrus"
 	"github.com/vhive-serverless/loader/pkg/common"
 	"github.com/vhive-serverless/loader/pkg/config"
 	"github.com/vhive-serverless/loader/pkg/workload/standard"
-	helloworld "github.com/vhive-serverless/vSwarm/utils/protobuf/helloworld"
-	"google.golang.org/grpc"
-	"google.golang.org/grpc/reflection"
-	"net"
-	"os"
-	"testing"
-	"time"
+	"github.com/vhive-serverless/loader/pkg/workload/vswarm"
 )
 
 func createFakeLoaderConfiguration() *config.LoaderConfiguration {
@@ -71,29 +68,6 @@ var testRuntimeSpecs = common.RuntimeSpecification{
 	Memory:  128,
 }
 
-type vSwarmServer struct {
-	helloworld.UnimplementedGreeterServer
-}
-
-func (s *vSwarmServer) SayHello(_ context.Context, req *helloworld.HelloRequest) (*helloworld.HelloReply, error) {
-	return &helloworld.HelloReply{
-		Message: "Reply message",
-	}, nil
-}
-
-func startVSwarmGRPCServer(serverAddress string, serverPort int) {
-	lis, err := net.Listen("tcp", fmt.Sprintf("%s:%d", serverAddress, serverPort))
-	if err != nil {
-		logrus.Fatalf("failed to listen: %v", err)
-	}
-
-	grpcServer := grpc.NewServer()
-
-	reflection.Register(grpcServer) // gRPC Server Reflection is used by gRPC CLI
-	helloworld.RegisterGreeterServer(grpcServer, &vSwarmServer{})
-	_ = grpcServer.Serve(lis)
-}
-
 func TestGRPCClientWithServerUnreachable(t *testing.T) {
 	cfg := createFakeLoaderConfiguration()
 	cfg.EnableZipkinTracing = true
@@ -161,7 +135,7 @@ func TestVSwarmClientWithServerReachable(t *testing.T) {
 	address, port := "localhost", 18081
 	testFunction.Endpoint = fmt.Sprintf("%s:%d", address, port)
 
-	go startVSwarmGRPCServer(address, port)
+	go vswarm.StartVSwarmGRPCServer(address, port)
 	time.Sleep(2 * time.Second)
 
 	cfgSwarm := createFakeVSwarmLoaderConfiguration()

pkg/driver/deployment/knative.go

@@ -26,7 +26,6 @@ var (
 type knativeDeployer struct{}
 
 type knativeDeploymentConfiguration struct {
-	YamlPath          string
 	IsPartiallyPanic  bool
 	EndpointPort      int
 	AutoscalingMetric string
@@ -38,7 +37,6 @@ func newKnativeDeployer() *knativeDeployer {
 
 func newKnativeDeployerConfiguration(cfg *config.Configuration) knativeDeploymentConfiguration {
 	return knativeDeploymentConfiguration{
-		YamlPath:          cfg.YAMLPath,
 		IsPartiallyPanic:  cfg.LoaderConfiguration.IsPartiallyPanic,
 		EndpointPort:      cfg.LoaderConfiguration.EndpointPort,
 		AutoscalingMetric: cfg.LoaderConfiguration.AutoscalingMetric,
@@ -61,7 +59,7 @@ func (*knativeDeployer) Deploy(cfg *config.Configuration) {
 
 			knativeDeploySingleFunction(
 				cfg.Functions[i],
-				knativeConfig.YamlPath,
+				cfg.Functions[i].YAMLPath,
 				knativeConfig.IsPartiallyPanic,
 				knativeConfig.EndpointPort,
 				knativeConfig.AutoscalingMetric,