functions: fix route timeout (#349)

* functions: add route-level timeout configuration options

* functions: harmonize defaults

api/datastore/bolt/bolt.go

@@ -342,6 +342,9 @@ func (ds *BoltDatastore) UpdateRoute(ctx context.Context, newroute *models.Route
 		if route.Type != "" {
 			route.Type = newroute.Type
 		}
+		if newroute.Timeout != 0 {
+			route.Timeout = newroute.Timeout
+		}
 		route.Format = newroute.Format
 		route.MaxConcurrency = newroute.MaxConcurrency
 		if newroute.Headers != nil {

api/datastore/postgres/postgres.go

@@ -22,6 +22,7 @@ CREATE TABLE IF NOT EXISTS routes (
 	format character varying(16) NOT NULL,
 	maxc integer NOT NULL,
 	memory integer NOT NULL,
+	timeout integer NOT NULL,
 	type character varying(16) NOT NULL,
 	headers text NOT NULL,
 	config text NOT NULL,
@@ -38,7 +39,7 @@ const extrasTableCreate = `CREATE TABLE IF NOT EXISTS extras (
 	value character varying(256) NOT NULL
 );`
 
-const routeSelector = `SELECT app_name, path, image, format, maxc, memory, type, headers, config FROM routes`
+const routeSelector = `SELECT app_name, path, image, format, maxc, memory, type, timeout, headers, config FROM routes`
 
 type rowScanner interface {
 	Scan(dest ...interface{}) error
@@ -267,10 +268,11 @@ func (ds *PostgresDatastore) InsertRoute(ctx context.Context, route *models.Rout
 			maxc,
 			memory,
 			type,
+			timeout,
 			headers,
 			config
 		)
-		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9);`,
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10);`,
 		route.AppName,
 		route.Path,
 		route.Image,
@@ -278,6 +280,7 @@ func (ds *PostgresDatastore) InsertRoute(ctx context.Context, route *models.Rout
 		route.MaxConcurrency,
 		route.Memory,
 		route.Type,
+		route.Timeout,
 		string(hbyte),
 		string(cbyte),
 	)
@@ -314,8 +317,9 @@ func (ds *PostgresDatastore) UpdateRoute(ctx context.Context, route *models.Rout
 			memory = $5,
 			maxc = $6,
 			type = $7,
-			headers = $8,
-			config = $9
+			timeout = $8,
+			headers = $9,
+			config = $10
 		WHERE app_name = $1 AND path = $2;`,
 		route.AppName,
 		route.Path,
@@ -324,6 +328,7 @@ func (ds *PostgresDatastore) UpdateRoute(ctx context.Context, route *models.Rout
 		route.Memory,
 		route.MaxConcurrency,
 		route.Type,
+		route.Timeout,
 		string(hbyte),
 		string(cbyte),
 	)
@@ -386,6 +391,7 @@ func scanRoute(scanner rowScanner, route *models.Route) error {
 		&route.Memory,
 		&route.MaxConcurrency,
 		&route.Type,
+		&route.Timeout,
 		&headerStr,
 		&configStr,
 	)

api/models/route.go

@@ -10,6 +10,10 @@ import (
 	apiErrors "github.com/go-openapi/errors"
 )
 
+const (
+	defaultRouteTimeout = 30 // seconds
+)
+
 var (
 	ErrRoutesCreate        = errors.New("Could not create route")
 	ErrRoutesUpdate        = errors.New("Could not update route")
@@ -33,6 +37,7 @@ type Route struct {
 	Type           string      `json:"type,omitempty"`
 	Format         string      `json:"format,omitempty"`
 	MaxConcurrency int         `json:"max_concurrency,omitempty"`
+	Timeout        int32       `json:"timeout,omitempty"`
 	Config         `json:"config"`
 }
 
@@ -47,6 +52,7 @@ var (
 	ErrRoutesValidationMissingPath     = errors.New("Missing route Path")
 	ErrRoutesValidationMissingType     = errors.New("Missing route Type")
 	ErrRoutesValidationPathMalformed   = errors.New("Path malformed")
+	ErrRoutesValidationNegativeTimeout = errors.New("Negative timeout")
 )
 
 func (r *Route) Validate() error {
@@ -93,6 +99,12 @@ func (r *Route) Validate() error {
 		r.MaxConcurrency = 1
 	}
 
+	if r.Timeout == 0 {
+		r.Timeout = defaultRouteTimeout
+	} else if r.Timeout < 0 {
+		res = append(res, ErrRoutesValidationNegativeTimeout)
+	}
+
 	if len(res) > 0 {
 		return apiErrors.CompositeValidationError(res...)
 	}

api/server/runner.go

@@ -179,7 +179,7 @@ func (s *Server) serve(ctx context.Context, c *gin.Context, appName string, foun
 		Memory:         found.Memory,
 		Stdin:          payload,
 		Stdout:         &stdout,
-		Timeout:        30 * time.Second,
+		Timeout:        time.Duration(found.Timeout) * time.Second,
 	}
 
 	switch found.Type {
@@ -216,12 +216,14 @@ func (s *Server) serve(ctx context.Context, c *gin.Context, appName string, foun
 			c.Header(k, v[0])
 		}
 
-		if result.Status() == "success" {
+		switch result.Status() {
+		case "success":
 			c.Data(http.StatusOK, "", stdout.Bytes())
-		} else {
+		case "timeout":
+			c.AbortWithStatus(http.StatusGatewayTimeout)
+		default:
 			c.AbortWithStatus(http.StatusInternalServerError)
 		}
-
 	}
 
 	return true

api/server/runner_test.go

@@ -126,18 +126,19 @@ func TestRouteRunnerExecution(t *testing.T) {
 	for i, test := range []struct {
 		path            string
 		body            string
+		method          string
 		expectedCode    int
 		expectedHeaders map[string][]string
 	}{
-		{"/r/myapp/myroute", ``, http.StatusOK, map[string][]string{"X-Function": {"Test"}}},
-		{"/r/myapp/myerror", ``, http.StatusInternalServerError, map[string][]string{"X-Function": {"Test"}}},
+		{"/r/myapp/myroute", ``, "GET", http.StatusOK, map[string][]string{"X-Function": {"Test"}}},
+		{"/r/myapp/myerror", ``, "GET", http.StatusInternalServerError, map[string][]string{"X-Function": {"Test"}}},
 
 		// Added same tests again to check if time is reduced by the auth cache
-		{"/r/myapp/myroute", ``, http.StatusOK, map[string][]string{"X-Function": {"Test"}}},
-		{"/r/myapp/myerror", ``, http.StatusInternalServerError, map[string][]string{"X-Function": {"Test"}}},
+		{"/r/myapp/myroute", ``, "GET", http.StatusOK, map[string][]string{"X-Function": {"Test"}}},
+		{"/r/myapp/myerror", ``, "GET", http.StatusInternalServerError, map[string][]string{"X-Function": {"Test"}}},
 	} {
-		body := bytes.NewBuffer([]byte(test.body))
-		_, rec := routerRequest(t, router, "GET", test.path, body)
+		body := strings.NewReader(test.body)
+		_, rec := routerRequest(t, router, test.method, test.path, body)
 
 		if rec.Code != test.expectedCode {
 			t.Log(buf.String())
@@ -145,13 +146,65 @@ func TestRouteRunnerExecution(t *testing.T) {
 				i, test.expectedCode, rec.Code)
 		}
 
-		if test.expectedHeaders != nil {
-			for name, header := range test.expectedHeaders {
-				if header[0] != rec.Header().Get(name) {
-					t.Log(buf.String())
-					t.Errorf("Test %d: Expected header `%s` to be %s but was %s",
-						i, name, header[0], rec.Header().Get(name))
-				}
+		if test.expectedHeaders == nil {
+			continue
+		}
+		for name, header := range test.expectedHeaders {
+			if header[0] != rec.Header().Get(name) {
+				t.Log(buf.String())
+				t.Errorf("Test %d: Expected header `%s` to be %s but was %s",
+					i, name, header[0], rec.Header().Get(name))
+			}
+		}
+	}
+}
+
+func TestRouteRunnerTimeout(t *testing.T) {
+	t.Skip("doesn't work on old Ubuntu")
+	buf := setLogBuffer()
+
+	tasks := make(chan task.Request)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	go runner.StartWorkers(ctx, testRunner(t), tasks)
+
+	router := testRouter(&datastore.Mock{
+		Apps: []*models.App{
+			{Name: "myapp", Config: models.Config{}},
+		},
+		Routes: []*models.Route{
+			{Path: "/sleeper", AppName: "myapp", Image: "iron/sleeper", Timeout: 1},
+		},
+	}, &mqs.Mock{}, testRunner(t), tasks)
+
+	for i, test := range []struct {
+		path            string
+		body            string
+		method          string
+		expectedCode    int
+		expectedHeaders map[string][]string
+	}{
+		{"/r/myapp/sleeper", `{"sleep": 0}`, "POST", http.StatusOK, nil},
+		{"/r/myapp/sleeper", `{"sleep": 2}`, "POST", http.StatusGatewayTimeout, nil},
+	} {
+		body := strings.NewReader(test.body)
+		_, rec := routerRequest(t, router, test.method, test.path, body)
+
+		if rec.Code != test.expectedCode {
+			t.Log(buf.String())
+			t.Errorf("Test %d: Expected status code to be %d but was %d",
+				i, test.expectedCode, rec.Code)
+		}
+
+		if test.expectedHeaders == nil {
+			continue
+		}
+		for name, header := range test.expectedHeaders {
+			if header[0] != rec.Header().Get(name) {
+				t.Log(buf.String())
+				t.Errorf("Test %d: Expected header `%s` to be %s but was %s",
+					i, name, header[0], rec.Header().Get(name))
 			}
 		}
 	}