Merge pull request #12 from triggermesh/apigateway

Apigateway

Author: tzununbekov

Gopkg.lock

@@ -0,0 +1,3 @@
+[[constraint]]
+  name = "github.com/aws/aws-lambda-go"
+  version = "1.6.0"

Gopkg.toml

@@ -1,4 +1,4 @@
-[![CircleCI](https://circleci.com/gh/triggermesh/aws-custom-runtime.svg?style=shield)](https://circleci.com/gh/triggermesh/aws-custom-runtime)
+[![Go Report Card](https://goreportcard.com/badge/github.com/triggermesh/aws-custom-runtime)](https://goreportcard.com/report/github.com/triggermesh/aws-custom-runtime) [![CircleCI](https://circleci.com/gh/triggermesh/aws-custom-runtime.svg?style=shield)](https://circleci.com/gh/triggermesh/aws-custom-runtime)
 
 ## Running AWS Lambda Custom Runtime in Knative
 

README.md

@@ -25,12 +25,8 @@ import (
 	"strings"
 	"sync"
 	"time"
-)
 
-const (
-	numberOfinvokers = 8    // Number of bootstrap processes
-	requestSizeLimit = 1e+7 // Request bosy size limit, 10Mb
-	functionTTL      = 3e+9 // Funtions deadline, 3 seconds
+	"github.com/triggermesh/aws-custom-runtime/pkg/events/apiGateway"
 )
 
 type message struct {
@@ -39,7 +35,16 @@ type message struct {
 	data     []byte
 }
 
+type responseWrapper struct {
+	http.ResponseWriter
+	Body []byte
+}
+
 var (
+	numberOfinvokers       = 8  // Number of bootstrap processes
+	requestSizeLimit int64 = 5  // Request body size limit, Mb
+	functionTTL      int64 = 10 // Funtions deadline, seconds
+
 	tasks   chan message
 	results map[string]chan message
 
@@ -60,6 +65,11 @@ var (
 	}
 )
 
+func (rw *responseWrapper) Write(data []byte) (int, error) {
+	rw.Body = data
+	return len(data), nil
+}
+
 func setupEnv() error {
 	environment["_HANDLER"], _ = os.LookupEnv("_HANDLER")
 	environment["LAMBDA_TASK_ROOT"], _ = os.LookupEnv("LAMBDA_TASK_ROOT")
@@ -73,7 +83,9 @@ func setupEnv() error {
 }
 
 func newTask(w http.ResponseWriter, r *http.Request) {
-	body, err := ioutil.ReadAll(http.MaxBytesReader(w, r.Body, requestSizeLimit))
+	requestSizeLimitInBytes := requestSizeLimit * 1e+6
+	functionTTLInNanoSeconds := functionTTL * 1e+9
+	body, err := ioutil.ReadAll(http.MaxBytesReader(w, r.Body, requestSizeLimitInBytes))
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
@@ -83,7 +95,7 @@ func newTask(w http.ResponseWriter, r *http.Request) {
 	now := time.Now().UnixNano()
 	task := message{
 		id:       fmt.Sprintf("%d", now),
-		deadline: now + functionTTL,
+		deadline: now + functionTTLInNanoSeconds,
 		data:     body,
 	}
 	fmt.Printf("<- %s %s\n", task.id, task.data)
@@ -97,7 +109,7 @@ func newTask(w http.ResponseWriter, r *http.Request) {
 	tasks <- task
 
 	select {
-	case <-time.After(time.Duration(functionTTL)):
+	case <-time.After(time.Duration(functionTTLInNanoSeconds)):
 		fmt.Printf("-> ! %s Deadline is reached\n", task.id)
 		w.WriteHeader(http.StatusGone)
 		w.Write([]byte(fmt.Sprintf("Deadline is reached, data %s", task.data)))
@@ -187,6 +199,45 @@ func responseHandler(w http.ResponseWriter, r *http.Request) {
 	return
 }
 
+func mapEvent(h http.Handler) http.Handler {
+	eventType, _ := os.LookupEnv("EVENT")
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		rw := responseWrapper{w, []byte{}}
+		switch eventType {
+		case "API_GATEWAY":
+			apiGateway.Request(r)
+			h.ServeHTTP(&rw, r)
+			apiGateway.Response(w, rw.Body)
+		default:
+			h.ServeHTTP(w, r)
+		}
+	})
+}
+
+func setLimits() {
+	if v, ok := os.LookupEnv("INVOKER_COUNT"); ok {
+		if vv, err := strconv.Atoi(v); err != nil {
+			fmt.Printf("can't set invokers limit, using default value %d\n", numberOfinvokers)
+		} else {
+			numberOfinvokers = vv
+		}
+	}
+	if v, ok := os.LookupEnv("REQUEST_SIZE_LIMIT"); ok {
+		if vv, err := strconv.Atoi(v); err != nil {
+			fmt.Printf("can't set request size limit, using default value %d\n", requestSizeLimit)
+		} else {
+			requestSizeLimit = int64(vv)
+		}
+	}
+	if v, ok := os.LookupEnv("FUNCTION_TTL"); ok {
+		if vv, err := strconv.Atoi(v); err != nil {
+			fmt.Printf("can't set function ttl, using default value %d\n", functionTTL)
+		} else {
+			functionTTL = int64(vv)
+		}
+	}
+}
+
 func api() error {
 	apiRouter := http.NewServeMux()
 	apiRouter.HandleFunc(awsEndpoint+"/init/error", initError)
@@ -200,6 +251,9 @@ func main() {
 	results = make(map[string]chan message)
 	defer close(tasks)
 
+	fmt.Println("Setting limits")
+	setLimits()
+
 	fmt.Println("Setup env")
 	if err := setupEnv(); err != nil {
 		log.Fatalln(err)
@@ -210,8 +264,8 @@ func main() {
 		log.Fatalln(api())
 	}()
 
-	fmt.Println("Starting invokers")
 	for i := 0; i < numberOfinvokers; i++ {
+		fmt.Println("Starting bootstrap", i+1)
 		go func() {
 			if err := exec.Command("sh", "-c", environment["LAMBDA_TASK_ROOT"]+"/bootstrap").Run(); err != nil {
 				log.Fatalln(err)
@@ -220,7 +274,8 @@ func main() {
 	}
 
 	taskRouter := http.NewServeMux()
-	taskRouter.HandleFunc("/", newTask)
+	taskHandler := http.HandlerFunc(newTask)
+	taskRouter.Handle("/", mapEvent(taskHandler))
 	fmt.Println("Listening...")
 	log.Fatalln(http.ListenAndServe(":8080", taskRouter))
 }

main.go

@@ -0,0 +1,53 @@
+package apiGateway
+
+import (
+	"bytes"
+	"encoding/json"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"strings"
+
+	"github.com/aws/aws-lambda-go/events"
+)
+
+func Request(r *http.Request) {
+	event := events.APIGatewayProxyRequest{}
+	body, err := ioutil.ReadAll(r.Body)
+	if err != nil {
+		log.Fatalln(err)
+	}
+
+	event.Body = string(body)
+	event.Headers = make(map[string]string)
+	for k, v := range r.Header {
+		event.Headers[k] = strings.Join(v, ",")
+	}
+	event.HTTPMethod = r.Method
+	event.Path = r.URL.Path
+	event.QueryStringParameters = make(map[string]string)
+	for k, v := range r.URL.Query() {
+		event.QueryStringParameters[k] = strings.Join(v, ",")
+	}
+	event.RequestContext = events.APIGatewayProxyRequestContext{}
+	// event.Resource = ""
+	// event.QueryStringParameters = make(map[string]string)
+	// event.IsBase64Encoded = false
+	js, err := json.Marshal(event)
+	if err != nil {
+		log.Fatalln(err)
+	}
+	r.Body = ioutil.NopCloser(bytes.NewBuffer(js))
+}
+
+func Response(w http.ResponseWriter, data []byte) (int, error) {
+	var js events.APIGatewayProxyResponse
+	if err := json.Unmarshal(data, &js); err != nil {
+		return 0, err
+	}
+	for k, v := range js.Headers {
+		w.Header().Set(k, v)
+	}
+	w.WriteHeader(js.StatusCode)
+	return w.Write([]byte(js.Body))
+}