New gRPC interface

Signed-off-by: Lazar Cvetković <[email protected]>
vhive-serverless · Sep 3, 2024 · b46e2b2 · b46e2b2
1 parent 684e8d1
commit b46e2b2
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 16 deletions.
diff --git a/.github/configs/wordlist.txt b/.github/configs/wordlist.txt
@@ -763,4 +763,8 @@ RpsMemoryMB
 RpsRuntimeMs
 RpsTarget
 SQRT
-RpsImage
+RpsImage
+BusyLoopOnSandboxStartup
+DirigentControlPlaneIP
+InvokeProtocol
+autoscaler
diff --git a/pkg/config/test_config.json b/pkg/config/test_config.json
@@ -2,6 +2,7 @@
   "Seed": 42,
 
   "Platform": "Knative",
+  "InvokeProtocol": "grpc",
   "YAMLSelector": "container",
   "EndpointPort": 80,
 

diff --git a/pkg/driver/clients/grpc_client.go b/pkg/driver/clients/grpc_client.go
@@ -33,7 +33,7 @@ import (
 	"github.com/vhive-serverless/loader/pkg/config"
 	"github.com/vhive-serverless/loader/pkg/workload/proto"
 
-	log "github.com/sirupsen/logrus"
+	"github.com/sirupsen/logrus"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
@@ -51,7 +51,7 @@ func newGRPCInvoker(cfg *config.LoaderConfiguration) *grpcInvoker {
 }
 
 func (i *grpcInvoker) Invoke(function *common.Function, runtimeSpec *common.RuntimeSpecification) (bool, *mc.ExecutionRecord) {
-	log.Tracef("(Invoke)\t %s: %d[ms], %d[MiB]", function.Name, runtimeSpec.Runtime, runtimeSpec.Memory)
+	logrus.Tracef("(Invoke)\t %s: %d[ms], %d[MiB]", function.Name, runtimeSpec.Runtime, runtimeSpec.Memory)
 
 	record := &mc.ExecutionRecord{
 		ExecutionRecordBase: mc.ExecutionRecordBase{
@@ -65,30 +65,25 @@ func (i *grpcInvoker) Invoke(function *common.Function, runtimeSpec *common.Runt
 	start := time.Now()
 	record.StartTime = start.UnixMicro()
 
-	dialContext, cancelDialing := context.WithTimeout(context.Background(), time.Duration(i.cfg.GRPCConnectionTimeoutSeconds)*time.Second)
-	defer cancelDialing()
-
 	var dialOptions []grpc.DialOption
 	dialOptions = append(dialOptions, grpc.WithTransportCredentials(insecure.NewCredentials()))
-	dialOptions = append(dialOptions, grpc.WithBlock())
 
 	grpcStart := time.Now()
 
-	conn, err := grpc.DialContext(dialContext, function.Endpoint, dialOptions...)
-	defer gRPCConnectionClose(conn)
+	conn, err := grpc.NewClient(function.Endpoint, dialOptions...)
 	if err != nil {
-		log.Debugf("Failed to establish a gRPC connection - %v\n", err)
+		logrus.Debugf("Failed to establish a gRPC connection - %v\n", err)
 
 		record.ResponseTime = time.Since(start).Microseconds()
 		record.ConnectionTimeout = true
 
 		return false, record
 	}
+	defer gRPCConnectionClose(conn)
 
 	record.GRPCConnectionEstablishTime = time.Since(grpcStart).Microseconds()
 
 	grpcClient := proto.NewExecutorClient(conn)
-
 	executionCxt, cancelExecution := context.WithTimeout(context.Background(), time.Duration(i.cfg.GRPCFunctionTimeoutSeconds)*time.Second)
 	defer cancelExecution()
 
@@ -99,9 +94,10 @@ func (i *grpcInvoker) Invoke(function *common.Function, runtimeSpec *common.Runt
 	})
 
 	if err != nil {
-		log.Debugf("gRPC timeout exceeded for function %s - %s", function.Name, err)
+		logrus.Debugf("gRPC timeout exceeded for function %s - %s", function.Name, err)
 
 		record.ResponseTime = time.Since(start).Microseconds()
+		record.ConnectionTimeout = true // WithBlock deprecated in new gRPC interface
 		record.FunctionTimeout = true
 
 		return false, record
@@ -117,9 +113,9 @@ func (i *grpcInvoker) Invoke(function *common.Function, runtimeSpec *common.Runt
 		record.ActualMemoryUsage = common.Kib2Mib(response.MemoryUsageInKb)
 	}
 
-	log.Tracef("(Replied)\t %s: %s, %.2f[ms], %d[MiB]", function.Name, response.Message,
+	logrus.Tracef("(Replied)\t %s: %s, %.2f[ms], %d[MiB]", function.Name, response.Message,
 		float64(response.DurationInMicroSec)/1e3, common.Kib2Mib(response.MemoryUsageInKb))
-	log.Tracef("(E2E Latency) %s: %.2f[ms]\n", function.Name, float64(record.ResponseTime)/1e3)
+	logrus.Tracef("(E2E Latency) %s: %.2f[ms]\n", function.Name, float64(record.ResponseTime)/1e3)
 
 	return true, record
 }
@@ -139,6 +135,6 @@ func gRPCConnectionClose(conn *grpc.ClientConn) {
 	}
 
 	if err := conn.Close(); err != nil {
-		log.Warnf("Error while closing gRPC connection - %s\n", err)
+		logrus.Warnf("Error while closing gRPC connection - %s\n", err)
 	}
 }
diff --git a/tools/driver/README.md b/tools/driver/README.md
@@ -46,7 +46,6 @@ is then the duration of the measurement phase + the duration of the warmup phase
 "outputDir": The local output directory on the machine running the experiment driver, this is where experiment results
 will be saved.  
 "YAMLSelector": Which yaml specification to use, supported values are 
-"[wimpy](/workloads/container/wimpy.yaml)", 
 "[container](/workloads/container/trace_func_go.yaml)" and 
 "[firecracker](/workloads/firecracker/trace_func_go.yaml)".  
 "IATDistribution": The IAT distribution that the loader will use when sending invocations to the worker node(s).