Add a basic concurrency test for the dataplane system tests (#832)

Add a basic concurrency test for the dataplane system tests. Also remove some spurious logging.

api/agent/lb_agent.go

@@ -30,8 +30,6 @@ func RequestReader(c *Call) (io.ReadCloser, error) {
 		return nil, errors.New("Call doesn't contain a request")
 	}
 
-	logrus.Info(cc.req)
-
 	return cc.req.Body, nil
 }
 

api/agent/nodepool/grpc/grpc_pool.go

@@ -311,7 +311,7 @@ func (r *gRPCRunner) TryExec(ctx context.Context, call agent.Call) (bool, error)
 	switch body := msg.Body.(type) {
 	case *pb.RunnerMsg_Acknowledged:
 		if !body.Acknowledged.Committed {
-			logrus.Errorf("Runner didn't commit invocation request: %v", body.Acknowledged.Details)
+			logrus.Debugf("Runner didn't commit invocation request: %v", body.Acknowledged.Details)
 			return false, nil
 			// Try the next runner
 		}

api/agent/pure_runner.go

@@ -544,8 +544,6 @@ func creds(cert string, key string, ca string) (credentials.TransportCredentials
 	}), nil
 }
 
-const megabyte uint64 = 1024 * 1024
-
 func createPureRunner(addr string, a Agent, creds credentials.TransportCredentials) (*pureRunner, error) {
 	var srv *grpc.Server
 	if creds != nil {
@@ -565,8 +563,10 @@ func createPureRunner(addr string, a Agent, creds credentials.TransportCredentia
 	return pr, nil
 }
 
+const megabyte uint64 = 1024 * 1024
+
 func getAvailableMemoryUnits() uint64 {
 	// To reuse code - but it's a bit of a hack. TODO: refactor the OS-specific get memory funcs out of that.
 	throwawayRT := NewResourceTracker().(*resourceTracker)
-	return throwawayRT.ramAsyncTotal
+	return throwawayRT.ramAsyncTotal / megabyte
 }

test/fn-system-tests/exec_test.go

@@ -2,7 +2,7 @@ package tests
 
 import (
 	"bytes"
-	//"fmt"
+	"fmt"
 	"net/url"
 	//"os"
 	"path"
@@ -50,3 +50,48 @@ func TestCanExecuteFunction(t *testing.T) {
 	}
 	apiutils.DeleteApp(t, s.Context, s.Client, s.AppName)
 }
+
+func TestBasicConcurrentExecution(t *testing.T) {
+	s := apiutils.SetupDefaultSuite()
+	apiutils.CreateApp(t, s.Context, s.Client, s.AppName, map[string]string{})
+	apiutils.CreateRoute(t, s.Context, s.Client, s.AppName, s.RoutePath, s.Image, "sync",
+		s.Format, s.Timeout, s.IdleTimeout, s.RouteConfig, s.RouteHeaders)
+
+	lb, err := LB()
+	if err != nil {
+		t.Fatalf("Got unexpected error: %v", err)
+	}
+	u := url.URL{
+		Scheme: "http",
+		Host:   lb,
+	}
+	u.Path = path.Join(u.Path, "r", s.AppName, s.RoutePath)
+
+	results := make(chan error)
+	concurrentFuncs := 10
+	for i := 0; i < concurrentFuncs; i++ {
+		go func() {
+			content := &bytes.Buffer{}
+			output := &bytes.Buffer{}
+			_, err = apiutils.CallFN(u.String(), content, output, "POST", []string{})
+			if err != nil {
+				results <- fmt.Errorf("Got unexpected error: %v", err)
+				return
+			}
+			expectedOutput := "Hello World!\n"
+			if !strings.Contains(expectedOutput, output.String()) {
+				results <- fmt.Errorf("Assertion error.\n\tExpected: %v\n\tActual: %v", expectedOutput, output.String())
+				return
+			}
+			results <- nil
+		}()
+	}
+	for i := 0; i < concurrentFuncs; i++ {
+		err := <-results
+		if err != nil {
+			t.Errorf("Error in basic concurrency execution test: %v", err)
+		}
+	}
+
+	apiutils.DeleteApp(t, s.Context, s.Client, s.AppName)
+}