Offer unlimited log buffer size when scanning handler logs

Some users have complained that they cannot log lines longer
than 16KB. A 16KB log line seems excessive, but this has been
requested at least 2-3 times over the years.

To enable the feature, set log_buffer_size to -1. This is not
a default, because it is not as efficient as using a pre-
determined buffer size.

Tested by running curl with --data-binary and sending a 6.7M
Go binary into a function with the 16KB maximum log line size
set. That produced the error, when the size was set to -1, the
output was written, albeit quite slowly.

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

README.md

@@ -201,7 +201,7 @@ Environmental variables:
 | `jwt_auth`                       | For OpenFaaS for Enterprises customers only. When set to `true`, the watchdog will require a JWT token to be passed as a Bearer token in the Authorization header. This token can only be obtained through the OpenFaaS gateway using a token exchange using the `http://gateway.openfaas:8080` address as the authority. |
 | `jwt_auth_debug`                 | Print out debug messages from the JWT authentication process (OpenFaaS for Enterprises only). |
 | `jwt_auth_local`                 | When set to `true`, the watchdog will attempt to validate the JWT token using a port-forwarded or local gateway running at `http://127.0.0.1:8080` instead of attempting to reach it via an in-cluster service name  (OpenFaaS for Enterprises only). |
-| `log_buffer_size`                | The amount of bytes to read from stderr/stdout for log lines. When exceeded, the user will see an "bufio.Scanner: token too long" error. The default value is `bufio.MaxScanTokenSize`           |
+| `log_buffer_size`                | The amount of bytes to read from stderr/stdout for log lines. When exceeded, the user will see an "bufio.Scanner: token too long" error. The default value is `bufio.MaxScanTokenSize`. To turn off buffering for unlimited log line lengths, set this value to `-1` and `bufio.Reader` will be used which does not allocate a buffer. |
 | `log_call_id`                    | In HTTP mode, when printing a response code, content-length and timing, include the X-Call-Id header at the end of the line in brackets i.e. `[079d9ff9-d7b7-4e37-b195-5ad520e6f797]` or `[none]` when it's empty. Default: `false` |
 | `max_inflight`                   |  Limit the maximum number of requests in flight, and return a HTTP status 429 when exceeded           |
 | `mode`                           |  The mode which of-watchdog operates in, Default `streaming` [see doc](#3-streaming-fork-modestreaming---default). Options are [http](#1-http-modehttp), [serialising fork](#2-serializing-fork-modeserializing), [streaming fork](#3-streaming-fork-modestreaming---default), [static](#4-static-modestatic) |

executor/logging.go

@@ -13,11 +13,6 @@ import (
 func bindLoggingPipe(name string, pipe io.Reader, output io.Writer, logPrefix bool, maxBufferSize int) {
 	log.Printf("Started logging: %s from function.", name)
 
-	scanner := bufio.NewScanner(pipe)
-
-	buffer := make([]byte, maxBufferSize)
-	scanner.Buffer(buffer, maxBufferSize)
-
 	logFlags := log.Flags()
 	prefix := log.Prefix()
 	if logPrefix == false {
@@ -27,16 +22,47 @@ func bindLoggingPipe(name string, pipe io.Reader, output io.Writer, logPrefix bo
 
 	logger := log.New(output, prefix, logFlags)
 
-	go func() {
-		for scanner.Scan() {
-			if logPrefix {
-				logger.Printf("%s: %s", name, scanner.Text())
-			} else {
-				logger.Print(scanner.Text())
+	if maxBufferSize >= 0 {
+		go pipeBuffered(name, pipe, logger, logPrefix, maxBufferSize)
+	} else {
+		go pipeUnbuffered(name, pipe, logger, logPrefix)
+	}
+}
+
+func pipeBuffered(name string, pipe io.Reader, logger *log.Logger, logPrefix bool, maxBufferSize int) {
+	buf := make([]byte, maxBufferSize)
+	scanner := bufio.NewScanner(pipe)
+	scanner.Buffer(buf, maxBufferSize)
+
+	for scanner.Scan() {
+		if logPrefix {
+			logger.Printf("%s: %s", name, scanner.Text())
+		} else {
+			logger.Print(scanner.Text())
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		log.Printf("Error reading %s: %s", name, err)
+	}
+}
+
+func pipeUnbuffered(name string, pipe io.Reader, logger *log.Logger, logPrefix bool) {
+
+	r := bufio.NewReader(pipe)
+
+	for {
+		line, err := r.ReadString('\n')
+		if err != nil {
+			if err != io.EOF {
+				log.Printf("Error reading %s: %s", name, err)
 			}
+			break
 		}
-		if err := scanner.Err(); err != nil {
-			log.Printf("Error scanning %s: %s", name, err.Error())
+		if logPrefix {
+			logger.Printf("%s: %s", name, line)
+		} else {
+			logger.Print(line)
 		}
-	}()
+	}
+
 }