Updates for text streaming functions

Signed-off-by: Alex Ellis (OpenFaaS Ltd) <alexellis2@gmail.com>

vendor/github.com/openfaas/faas-provider/README.md

@@ -20,7 +20,15 @@ The following is used in OpenFaaS and recommended for those seeking to build the
 
 All the required HTTP routes are configured automatically including a HTTP server on port 8080. Your task is to implement the supplied HTTP handler functions.
 
-For an example see the [main.go](https://github.com/openfaas/faas-netes/blob/master/main.go) file in the [faas-netes](https://github.com/openfaas/faas-netes) Kubernetes backend.
+Examples:
+
+**OpenFaaS for Kubernetes**
+
+See the [main.go](https://github.com/openfaas/faas-netes/blob/master/main.go) file in the [faas-netes](https://github.com/openfaas/faas-netes) Kubernetes backend.
+
+**OpenFaaS for containerd (faasd)**
+
+See [provider.go](https://github.com/openfaas/faasd/blob/master/cmd/provider.go#L100) for the [faasd backend](https://github.com/openfaas/faasd/)
 
 I.e.:
 

vendor/github.com/openfaas/faas-provider/httputil/write_interceptor.go

@@ -7,12 +7,17 @@ import (
 )
 
 func NewHttpWriteInterceptor(w http.ResponseWriter) *HttpWriteInterceptor {
-	return &HttpWriteInterceptor{w, 0}
+	return &HttpWriteInterceptor{
+		ResponseWriter: w,
+		statusCode:     0,
+		bytesWritten:   0,
+	}
 }
 
 type HttpWriteInterceptor struct {
 	http.ResponseWriter
-	statusCode int
+	statusCode   int
+	bytesWritten int64
 }
 
 func (c *HttpWriteInterceptor) Status() int {
@@ -22,6 +27,10 @@ func (c *HttpWriteInterceptor) Status() int {
 	return c.statusCode
 }
 
+func (c *HttpWriteInterceptor) BytesWritten() int64 {
+	return c.bytesWritten
+}
+
 func (c *HttpWriteInterceptor) Header() http.Header {
 	return c.ResponseWriter.Header()
 }
@@ -30,6 +39,9 @@ func (c *HttpWriteInterceptor) Write(data []byte) (int, error) {
 	if c.statusCode == 0 {
 		c.WriteHeader(http.StatusOK)
 	}
+
+	c.bytesWritten += int64(len(data))
+
 	return c.ResponseWriter.Write(data)
 }
 

vendor/github.com/openfaas/faas-provider/proxy/proxy.go

@@ -26,11 +26,12 @@ import (
 	"log"
 	"net"
 	"net/http"
+	"net/http/httputil"
 	"net/url"
 	"time"
 
 	"github.com/gorilla/mux"
-	"github.com/openfaas/faas-provider/httputil"
+	fhttputil "github.com/openfaas/faas-provider/httputil"
 	"github.com/openfaas/faas-provider/types"
 )
 
@@ -68,6 +69,17 @@ func NewHandlerFunc(config types.FaaSConfig, resolver BaseURLResolver, verbose b
 
 	proxyClient := NewProxyClientFromConfig(config)
 
+	reverseProxy := httputil.ReverseProxy{}
+	reverseProxy.Director = func(req *http.Request) {
+		// At least an empty director is required to prevent runtime errors.
+		req.URL.Scheme = "http"
+	}
+	reverseProxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, err error) {
+	}
+
+	// Errors are common during disconnect of client, no need to log them.
+	reverseProxy.ErrorLog = log.New(io.Discard, "", 0)
+
 	return func(w http.ResponseWriter, r *http.Request) {
 		if r.Body != nil {
 			defer r.Body.Close()
@@ -81,7 +93,7 @@ func NewHandlerFunc(config types.FaaSConfig, resolver BaseURLResolver, verbose b
 			http.MethodGet,
 			http.MethodOptions,
 			http.MethodHead:
-			proxyRequest(w, r, proxyClient, resolver, verbose)
+			proxyRequest(w, r, proxyClient, resolver, &reverseProxy, verbose)
 
 		default:
 			w.WriteHeader(http.StatusMethodNotAllowed)
@@ -134,7 +146,7 @@ func NewProxyClient(timeout time.Duration, maxIdleConns int, maxIdleConnsPerHost
 }
 
 // proxyRequest handles the actual resolution of and then request to the function service.
-func proxyRequest(w http.ResponseWriter, originalReq *http.Request, proxyClient *http.Client, resolver BaseURLResolver, verbose bool) {
+func proxyRequest(w http.ResponseWriter, originalReq *http.Request, proxyClient *http.Client, resolver BaseURLResolver, reverseProxy *httputil.ReverseProxy, verbose bool) {
 	ctx := originalReq.Context()
 
 	pathVars := mux.Vars(originalReq)
@@ -142,7 +154,7 @@ func proxyRequest(w http.ResponseWriter, originalReq *http.Request, proxyClient
 	if functionName == "" {
 		w.Header().Add(openFaaSInternalHeader, "proxy")
 
-		httputil.Errorf(w, http.StatusBadRequest, "Provide function name in the request path")
+		fhttputil.Errorf(w, http.StatusBadRequest, "Provide function name in the request path")
 		return
 	}
 
@@ -152,7 +164,7 @@ func proxyRequest(w http.ResponseWriter, originalReq *http.Request, proxyClient
 
 		// TODO: Should record the 404/not found error in Prometheus.
 		log.Printf("resolver error: no endpoints for %s: %s\n", functionName, err.Error())
-		httputil.Errorf(w, http.StatusServiceUnavailable, "No endpoints available for: %s.", functionName)
+		fhttputil.Errorf(w, http.StatusServiceUnavailable, "No endpoints available for: %s.", functionName)
 		return
 	}
 
@@ -161,7 +173,7 @@ func proxyRequest(w http.ResponseWriter, originalReq *http.Request, proxyClient
 
 		w.Header().Add(openFaaSInternalHeader, "proxy")
 
-		httputil.Errorf(w, http.StatusInternalServerError, "Failed to resolve service: %s.", functionName)
+		fhttputil.Errorf(w, http.StatusInternalServerError, "Failed to resolve service: %s.", functionName)
 		return
 	}
 
@@ -169,16 +181,29 @@ func proxyRequest(w http.ResponseWriter, originalReq *http.Request, proxyClient
 		defer proxyReq.Body.Close()
 	}
 
-	start := time.Now()
+	if verbose {
+		start := time.Now()
+		defer func() {
+			seconds := time.Since(start)
+			log.Printf("%s took %f seconds\n", functionName, seconds.Seconds())
+		}()
+	}
+
+	if v := originalReq.Header.Get("Accept"); v == "text/event-stream" {
+		originalReq.URL = proxyReq.URL
+
+		reverseProxy.ServeHTTP(w, originalReq)
+		return
+	}
+
 	response, err := proxyClient.Do(proxyReq.WithContext(ctx))
-	seconds := time.Since(start)
 
 	if err != nil {
 		log.Printf("error with proxy request to: %s, %s\n", proxyReq.URL.String(), err.Error())
 
 		w.Header().Add(openFaaSInternalHeader, "proxy")
 
-		httputil.Errorf(w, http.StatusInternalServerError, "Can't reach service for: %s.", functionName)
+		fhttputil.Errorf(w, http.StatusInternalServerError, "Can't reach service for: %s.", functionName)
 		return
 	}
 
@@ -186,10 +211,6 @@ func proxyRequest(w http.ResponseWriter, originalReq *http.Request, proxyClient
 		defer response.Body.Close()
 	}
 
-	if verbose {
-		log.Printf("%s took %f seconds\n", functionName, seconds.Seconds())
-	}
-
 	clientHeader := w.Header()
 	copyHeaders(clientHeader, &response.Header)
 	w.Header().Set("Content-Type", getContentType(originalReq.Header, response.Header))

vendor/github.com/openfaas/faas-provider/types/system_events.go

@@ -0,0 +1,52 @@
+package types
+
+import "time"
+
+const (
+	TypeFunctionUsage = "function_usage"
+	TypeAPIAccess     = "api_access"
+)
+
+type Event interface {
+	EventType() string
+}
+
+type FunctionUsageEvent struct {
+	Namespace    string        `json:"namespace"`
+	FunctionName string        `json:"function_name"`
+	Started      time.Time     `json:"started"`
+	Duration     time.Duration `json:"duration"`
+	MemoryBytes  int64         `json:"memory_bytes"`
+}
+
+func (e FunctionUsageEvent) EventType() string {
+	return TypeFunctionUsage
+}
+
+type APIAccessEvent struct {
+	Actor         *Actor    `json:"actor,omitempty"`
+	Path          string    `json:"path"`
+	Method        string    `json:"method"`
+	Actions       []string  `json:"actions"`
+	ResponseCode  int       `json:"response_code"`
+	CustomMessage string    `json:"custom_message,omitempty"`
+	Namespace     string    `json:"namespace,omitempty"`
+	Time          time.Time `json:"time"`
+}
+
+func (e APIAccessEvent) EventType() string {
+	return TypeAPIAccess
+}
+
+// Actor is the user that triggered an event.
+// Get from OIDC claims, we can add any of the default OIDC profile or email claim fields if desired.
+type Actor struct {
+	// OIDC subject, a unique identifier of the user.
+	Sub string `json:"sub"`
+	// Full name of the subject, can be the name of a user of OpenFaaS component.
+	Name string `json:"name,omitempty"`
+	// OpenFaaS issuer
+	Issuer string `json:"issuer,omitempty"`
+	// Federated issuer
+	FedIssuer string `json:"fed_issuer,omitempty"`
+}