Adding logic for creating and managing DAG invocations

Signed-off-by: Kway Yi Shen <[email protected]>

.github/configs/wordlist.txt

@@ -23,6 +23,7 @@ workerNodeNum
 durations
 dur
 ACM
+Acyclic
 addr
 adservice
 AdService
@@ -151,8 +152,10 @@ Ctrl
 currencyservice
 Daglis
 DAGMode
+DAGEntryFunction
 datacenter
 Datacenter
+DAGs
 dataflows
 dataset
 david
@@ -710,6 +713,7 @@ cgroups
 noop
 YAMLs
 cgo
+EnableDAGDataset
 EnableMetricsScrapping
 EnableZipkinTracing
 EndpointPort

cmd/config_knative_trace.json

@@ -24,5 +24,9 @@
 
   "GRPCConnectionTimeoutSeconds": 15,
   "GRPCFunctionTimeoutSeconds": 900,
-  "DAGMode": false
+  "DAGMode": false,
+  "DAGEntryFunction": 0,
+  "EnableDAGDataset": true,
+  "Width": 2,
+  "Depth": 2
 }

cmd/loader.go

@@ -31,11 +31,12 @@ import (
 	"strings"
 	"time"
 
+	"golang.org/x/exp/slices"
+
 	"github.com/vhive-serverless/loader/pkg/common"
 	"github.com/vhive-serverless/loader/pkg/config"
 	"github.com/vhive-serverless/loader/pkg/driver"
 	"github.com/vhive-serverless/loader/pkg/trace"
-	"golang.org/x/exp/slices"
 
 	log "github.com/sirupsen/logrus"
 	tracer "github.com/vhive-serverless/vSwarm/utils/tracing/go"

data/traces/example/dag_structure.csv

@@ -0,0 +1,6 @@
+Width,Width - Percentile,Depth,Depth - Percentile,Total Nodes,Total Nodes
+1,0.00%,1,0.00%,2,0.00%
+1,78.66%,1,12.51%,2,55.98%
+2,92.13%,2,67.96%,3,79.20%
+3,95.24%,3,86.84%,4,86.63%
+4,100.00%,4,100.00%,5,100.00%

data/traces/reference/.gitattributes

@@ -1 +1,2 @@
 *.tar.gz filter=lfs diff=lfs merge=lfs -text
+dag_structure.csv filter=lfs diff=lfs merge=lfs -text

docs/configuration.md

@@ -30,7 +30,11 @@
 | MetricScrapingPeriodSeconds  | int       | > 0                                                                 | 15                  | Period of Prometheus metrics scrapping                                               |
 | GRPCConnectionTimeoutSeconds | int       | > 0                                                                 | 60                  | Timeout for establishing a gRPC connection                                           |
 | GRPCFunctionTimeoutSeconds   | int       | > 0                                                                 | 90                  | Maximum time given to function to execute[^5]                                        |
-| DAGMode                      | bool      | true/false                                                          | false               | Sequential invocation of all functions one after another                             |
+| DAGMode                      | bool      | true/false                                                          | false               | Generates a dedicated DAG workflow[^6] with a specified function as the entry point. Frequency and IAT of the DAG follows the entry function, while Duration and Memory of each function will follow their respective values in TracePath.                                                                                                              |                            
+| DAGEntryFunction                      | int      | >=0                                                          | 0               | The index of the entry function in the DAG. Frequency of the DAG will follow this function.                                                                        |
+| EnableDAGDataset             | bool      | true/false                                                          | true                |  Generate width and depth from data/traces/example/dag_structure.csv[^7]                                                                                                      |
+| Width                        | int       | > 0                                                                 | 2                   | Default width of DAG                                                                 |
+| Depth                        | int       | > 0                                                                 | 2                   | Default depth of DAG                                                                 |
 
 [^1]: To run RPS experiments add suffix `-RPS`.
 
@@ -50,6 +54,10 @@ this [table](https://cloud.google.com/functions/pricing#compute_time) for Google
 [^5]: Function can execute for at most 15 minutes as in AWS
 Lambda; https://aws.amazon.com/about-aws/whats-new/2018/10/aws-lambda-supports-functions-that-can-run-up-to-15-minutes/
 
+[^6]: The generated DAG consists of unique functions. The shape of each DAG is determined either ```Width,Depth``` or randomly generated based on ```EnableDAGDAtaset```.
+
+[^7]: A [data sample](https://github.com/icanforce/Orion-OSDI22/blob/main/Public_Dataset/dag_structure.xlsx) of DAG structures has been created based on past Microsoft Azure traces. Width and Depth are determined based on probabilities of this sample.
+
 ---
 
 InVitro can cause failure on cluster manager components. To do so, please configure the `cmd/failure.json`. Make sure

docs/loader.md

@@ -179,6 +179,44 @@ For more options, please see the `Makefile`.
 
 For instructions on how to use the loader with OpenWhisk go to `openwhisk_setup/README.md`.
 
+## Workflow Invocation
+Generation of a Directed Acyclic Graph (DAG) workflow is supported by setting `"DAGMode: true"` in `cmd/config_knative_trace.json` (as specified in [`docs/configuration.md`](../docs/configuration.md)). Before invocation, a single DAG containing unique functions will be generated, with entry function index in `"DAGEntryFunction"`. The invocation frequency of the DAG will follow the specified entry function in `data/traces/example/invocations.csv`.
+
+The DAG consists of functions, linked in numerical order, until the desired shape is achieved. If the number of functions provided in `data/traces/example` exceed the required number for DAG generation, the excess functions will be ignored. 
+
+ An example of the generated workflow can be seen here:
+
+DAG of Width = 3, Depth = 4 with `DAGEntryFunction`: 0:
+
+```bash
+f(0) -> f(1) -> f(3) -> f(5)
+  \        
+   \ -> f(2) -> f(4) -> f(6)
+                  \
+                   \ -> f(7)
+```
+In this DAG, each invocation will result in 8 total functions invoked, with parallel invocations per branch. Functions not in the DAG will not be invoked.
+
+To obtain [reference traces](https://github.com/vhive-serverless/invitro/blob/main/docs/sampler.md#reference-traces) for DAG execution, use the following command:
+```bash
+git lfs pull 
+tar -xzf data/traces/reference/sampled_150.tar.gz -C data/traces 
+```
+Microsoft has publicly released Microsoft Azure traces of function invocations from 10/18/2021 to 10/31/2021. From this trace, a [data sample](https://github.com/icanforce/Orion-OSDI22/blob/main/Public_Dataset) of DAG structures, representing the cumulative distribution of width and depth of DAGs during that period, was generated. Probabilities were applied to the data to derive the shape of the DAGs. The file `data/traces/example/dag_structure.csv` provides a simplified sample of the publicly released traces.
+
+By default, the shape of the DAG is automatically calculated using the above mentioned cumulative distribution.
+To manually set the shape of the DAG, change the following parameters in `cmd/config_knative_trace.json`. Note that the number of functions in `TracePath` must be large enough to support the maximum size of the DAG.
+```bash
+"EnableDAGDataset": false,
+"Width": <width>,
+"Depth": <depth>
+```
+
+Lastly, start the experiment. This invokes the DAG with the entry function's invocation frequency.
+```bash
+go run cmd/loader.go --config cmd/config_knative_trace.json
+```
+
 ## Running on Cloud Using Serverless Framework
 
 **Currently supported vendors:** AWS
@@ -204,7 +242,6 @@ For instructions on how to use the loader with OpenWhisk go to `openwhisk_setup/
     ```bash
     go run cmd/loader.go --config cmd/config_knative_trace.json
     ```
-
 ---
 Note:
 - Current deployment is via container image.

pkg/common/trace_types.go

@@ -24,6 +24,8 @@
 
 package common
 
+import "container/list"
+
 type FunctionInvocationStats struct {
 	HashOwner    string
 	HashApp      string
@@ -103,3 +105,9 @@ type Function struct {
 
 	Specification *FunctionSpecification
 }
+
+type Node struct {
+	Function *Function
+	Branches []*list.List
+	Depth    int
+}

pkg/common/utilities.go

@@ -135,3 +135,15 @@ func SumNumberOfInvocations(withWarmup bool, totalDuration int, functions []*Fun
 
 	return result
 }
+
+func GetName(function *Function) int {
+	parts := strings.Split(function.Name, "-")
+	if parts[0] == "test" {
+		return 0
+	}
+	functionId, err := strconv.Atoi(parts[2])
+	if err != nil {
+		log.Fatal(err)
+	}
+	return functionId
+}

pkg/config/parser.go

@@ -80,6 +80,10 @@ type LoaderConfiguration struct {
 	GRPCConnectionTimeoutSeconds int  `json:"GRPCConnectionTimeoutSeconds"`
 	GRPCFunctionTimeoutSeconds   int  `json:"GRPCFunctionTimeoutSeconds"`
 	DAGMode                      bool `json:"DAGMode"`
+	EnableDAGDataset             bool `json:"EnableDAGDataset"`
+	DAGEntryFunction             int  `json:"DAGEntryFunction"`
+	Width                        int  `json:"Width"`
+	Depth                        int  `json:"Depth"`
 }
 
 func ReadConfigurationFile(path string) LoaderConfiguration {

pkg/config/parser_test.go

@@ -61,7 +61,12 @@ func TestConfigParser(t *testing.T) {
 		config.MetricScrapingPeriodSeconds != 15 ||
 		config.AutoscalingMetric != "concurrency" ||
 		config.GRPCConnectionTimeoutSeconds != 15 ||
-		config.GRPCFunctionTimeoutSeconds != 900 {
+		config.GRPCFunctionTimeoutSeconds != 900 ||
+		config.DAGMode != false ||
+		config.EnableDAGDataset != true ||
+		config.DAGEntryFunction != 0 ||
+		config.Width != 2 ||
+		config.Depth != 2 {
 
 		t.Error("Unexpected configuration read.")
 	}

pkg/config/test_config.json

@@ -22,5 +22,9 @@
 
   "GRPCConnectionTimeoutSeconds": 15,
   "GRPCFunctionTimeoutSeconds": 900,
-  "DAGMode": false
+  "DAGMode": false,
+  "EnableDAGDataset": true,
+  "DAGEntryFunction": 0,
+  "Width": 2,
+  "Depth": 2
 }

pkg/config/test_config_aws.json

@@ -21,5 +21,9 @@
 
   "GRPCConnectionTimeoutSeconds": 15,
   "GRPCFunctionTimeoutSeconds": 900,
-  "DAGMode": false
-}
+  "DAGMode": false,
+  "EnableDAGDataset": true,
+  "DAGEntryFunction": 0,
+  "Width": 2,
+  "Depth": 2
+}