Repository: fission/fission (go)

Issue:
Fission buffers server-sent event requests until they are complete before streaming the response
**Fission/Kubernetes version**

Fission 1.17.11
K8S 1.26
Kubernetes platform: docker-desktop

**Describe the bug**

When running a serverless function which pushes a response slowly over time (on purpose), fission appears to buffer the whole result, and only send it to the client once the response is complete. This slow sending is useful for [server-sent events](https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events) (SSE), a common way of streaming a slow computation to the client as it is ready on the server.

Notably, OpenAI's APIs use SSE to stream chat responses from LLMs, so any developers wanting to use LLMs in their fission-based apps will consume SSE, and likely want to provide a good user experience by returning SSE to the client.

**To Reproduce**

We're using a custom nodejs environment that controls the HTTP response directly instead of using the `fission/node-env` environment, so we're able to write raw buffers to the response. Within `fission/node-env`, it's kind of hard. Reading the node-env code, I think you can dodge it's response sending by just returning `{}` from your handler function, and then sending the response later, but that's kind of cheating. In any case, here's a low-level nodejs server with no dependencies which streams a response using server-sent events:

```javascript
// Import the built-in http module
const http = require('http');

// Create an HTTP server
const server = http.createServer((req, res) => {
    // Set the appropriate headers for server-sent events
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Connection', 'keep-alive');
    res.flushHeaders();

    let sentTimes = 0;

    // Send the current time every 500 ms
    const interval = setInterval(() => {
      // Increment the sentTimes counter
      sentTimes++;

      // Send the current time as an SSE
      res.write(`data: ${new Date().toISOString()}\n\n`);

      // If we've sent the time 10 times, close the connection and clear the interval
      if (sentTimes >= 10) {
        clearInterval(interval);
        res.end();
      }
    }, 500);
});

// Start the server on port 3000
server.listen(3000, () => {
  console.log('Server running on http://localhost:3000');
});
```

**Expected result**

Fission relays each chunk of the response to the caller as it arrives, without buffering. 

**Actual result**

Fission buffers the whole response, and only sends it once complete.

**Additional context**

My suspicion is that something inside the fission router accidentally buffers the response. I took a look at the `httputil` proxy code and nothing immediately stood out to me, but I am not a great golang programmer. Perhaps you folks know what this might be! SSE support would be really great to get!


Pull request:
fix: add flush to support chunked for the metric middleware
<!--  Thanks for sending a pull request! We request you provide detailed description as much as possible. -->

## Description
<!--- Describe your changes in detail. -->
<!-- Typically try to give details of what, why and how of the PR changes. -->
No Flush implement for the metric middleware cause not support chunked response for the router

## Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
Fixes #2767

## Testing
<!--- Please describe in detail how you tested your changes. -->

## Checklist:
<!-- Please tick following checkboxes as per your understanding. -->
- [x] I ran tests as well as code linting locally to verify my changes. 
- [x] I have done manual verification of my changes, changes working as expected.
- [x] I have added new tests to cover my changes.
- [x] My changes follow contributing guidelines of Fission.
- [x] I have signed all of my commits.


Patch:
diff --git a/pkg/utils/metrics/http_metrics.go b/pkg/utils/metrics/http_metrics.go
index 02549750..c1f393f0 100644
--- a/pkg/utils/metrics/http_metrics.go
+++ b/pkg/utils/metrics/http_metrics.go
@@ -67,6 +67,12 @@ func (rw *ResponseWriterWrapper) WriteHeader(statuscode int) {
 	rw.ResponseWriter.WriteHeader(statuscode)
 }
 
+func (rw *ResponseWriterWrapper) Flush() {
+	if f, ok := rw.ResponseWriter.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
 func HTTPMetricMiddleware(next http.Handler) http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if util.IsWebsocketRequest(r) {
diff --git a/pkg/utils/namespace.go b/pkg/utils/namespace.go
index 1e552e8d..836bdde5 100644
--- a/pkg/utils/namespace.go
+++ b/pkg/utils/namespace.go
@@ -20,7 +20,7 @@ const (
 type (
 	NamespaceResolver struct {
 		FunctionNamespace string
-		BuiderNamespace   string
+		BuilderNamespace  string
 		DefaultNamespace  string
 		FissionResourceNS map[string]string
 		Logger            *zap.Logger
@@ -40,14 +40,14 @@ var nsResolver *NamespaceResolver
 func init() {
 	nsResolver = &NamespaceResolver{
 		FunctionNamespace: os.Getenv(ENV_FUNCTION_NAMESPACE),
-		BuiderNamespace:   os.Getenv(ENV_BUILDER_NAMESPACE),
+		BuilderNamespace:  os.Getenv(ENV_BUILDER_NAMESPACE),
 		DefaultNamespace:  os.Getenv(ENV_DEFAULT_NAMESPACE),
 		FissionResourceNS: GetNamespaces(),
 		Logger:            loggerfactory.GetLogger(),
 	}
 
 	nsResolver.Logger.Debug("namespaces", zap.String("function_namespace", nsResolver.FunctionNamespace),
-		zap.String("builder_namespace", nsResolver.BuiderNamespace),
+		zap.String("builder_namespace", nsResolver.BuilderNamespace),
 		zap.String("default_namespace", nsResolver.DefaultNamespace),
 		zap.Any("fission_resource_namespace", listNamespaces(nsResolver.FissionResourceNS)))
 }
@@ -96,8 +96,8 @@ func (nsr *NamespaceResolver) FissionNSWithOptions(option ...option) map[string]
 	if options.functionNS && nsr.FunctionNamespace != "" {
 		fissionResourceNS[nsr.FunctionNamespace] = nsr.FunctionNamespace
 	}
-	if options.builderNS && nsr.BuiderNamespace != "" {
-		fissionResourceNS[nsr.BuiderNamespace] = nsr.BuiderNamespace
+	if options.builderNS && nsr.BuilderNamespace != "" {
+		fissionResourceNS[nsr.BuilderNamespace] = nsr.BuilderNamespace
 	}
 	if options.defaultNs && nsr.DefaultNamespace != "" {
 		fissionResourceNS[nsr.DefaultNamespace] = nsr.DefaultNamespace
@@ -118,7 +118,7 @@ func GetNamespaces() map[string]string {
 	if len(envValue) > 0 {
 		lstNamespaces := strings.Split(envValue, ",")
 		for _, namespace := range lstNamespaces {
-			//check to handle string with additional comma at the end of string. eg- ns1,ns2,
+			// check to handle string with additional comma at the end of string. eg- ns1,ns2,
 			if namespace != "" {
 				namespaces[namespace] = namespace
 			}
@@ -132,14 +132,14 @@ func GetNamespaces() map[string]string {
 }
 
 func (nsr *NamespaceResolver) GetBuilderNS(namespace string) string {
-	if nsr.BuiderNamespace == "" {
+	if nsr.BuilderNamespace == "" {
 		return namespace
 	}
 
 	if namespace != metav1.NamespaceDefault {
 		return namespace
 	}
-	return nsr.BuiderNamespace
+	return nsr.BuilderNamespace
 }
 
 func (nsr *NamespaceResolver) GetFunctionNS(namespace string) string {
@@ -154,7 +154,7 @@ func (nsr *NamespaceResolver) GetFunctionNS(namespace string) string {
 }
 
 func (nsr *NamespaceResolver) ResolveNamespace(namespace string) string {
-	if nsr.FunctionNamespace == "" || nsr.BuiderNamespace == "" {
+	if nsr.FunctionNamespace == "" || nsr.BuilderNamespace == "" {
 		return nsr.DefaultNamespace
 	}
 	return namespace


Test patch:
diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr_test.go b/pkg/executor/executortype/newdeploy/newdeploymgr_test.go
index 3d2eea05..5cc9b288 100644
--- a/pkg/executor/executortype/newdeploy/newdeploymgr_test.go
+++ b/pkg/executor/executortype/newdeploy/newdeploymgr_test.go
@@ -66,7 +66,7 @@ func TestRefreshFuncPods(t *testing.T) {
 
 	nsResolver := utils.NamespaceResolver{
 		FunctionNamespace: functionNamespace,
-		BuiderNamespace:   builderNamespace,
+		BuilderNamespace:  builderNamespace,
 		DefaultNamespace:  defaultNamespace,
 	}
 	ndm.nsResolver = &nsResolver
diff --git a/pkg/utils/metrics/http_metrics_test.go b/pkg/utils/metrics/http_metrics_test.go
new file mode 100644
index 00000000..61ede88f
--- /dev/null
+++ b/pkg/utils/metrics/http_metrics_test.go
@@ -0,0 +1,98 @@
+package metrics
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/gorilla/mux"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var dataRow = []byte("I'm the data Row\n")
+
+func chunkedHandler(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := context.WithCancel(r.Context())
+	ticker := time.NewTicker(time.Second) // We may set it to 10 secs
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-ticker.C:
+				_, _ = w.Write(dataRow)
+				if f, ok := w.(http.Flusher); ok {
+					f.Flush()
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	// Emulate some work
+	time.Sleep(5 * time.Second)
+
+	// Telling the loop that keeps the connection alive to end
+	cancel()
+
+	// Waiting until the loop ends
+	wg.Wait()
+}
+
+func TestChunked(t *testing.T) {
+	mr := mux.NewRouter()
+	mr.Use(HTTPMetricMiddleware)
+	mr.Handle("/", http.HandlerFunc(chunkedHandler))
+	s := httptest.NewServer(mr)
+	defer s.Close()
+
+	resp, err := http.Get(s.URL)
+	require.NoError(t, err)
+	assert.Contains(t, resp.TransferEncoding, "chunked")
+	defer resp.Body.Close()
+
+	r := bufio.NewReader(resp.Body)
+	for {
+		line, err := readChunkedResponseLine(r)
+		if err != nil {
+			if err == io.EOF {
+				return
+			}
+			log.Fatal(err.Error())
+		}
+		if len(line) == 0 {
+			log.Println("Alive!")
+			continue
+		}
+
+		fmt.Println(string(line)) // we got the final response
+		assert.Equal(t, dataRow, append(line, '\n'))
+	}
+}
+
+func readChunkedResponseLine(r *bufio.Reader) ([]byte, error) {
+	line, isPrefix, err := r.ReadLine()
+	if err != nil {
+		return nil, err
+	}
+
+	if isPrefix {
+		rest, err := readChunkedResponseLine(r)
+		if err != nil {
+			return nil, err
+		}
+		line = append(line, rest...)
+	}
+
+	return line, nil
+}
diff --git a/pkg/utils/namespace_test.go b/pkg/utils/namespace_test.go
index 9620f2d0..ea722f86 100644
--- a/pkg/utils/namespace_test.go
+++ b/pkg/utils/namespace_test.go
@@ -226,7 +226,7 @@ func TestNamespaceResolver(t *testing.T) {
 func getFissionNamespaces(builderNS, functionNS, defaultNS string) *NamespaceResolver {
 	return &NamespaceResolver{
 		FunctionNamespace: functionNS,
-		BuiderNamespace:   builderNS,
+		BuilderNamespace:  builderNS,
 		DefaultNamespace:  defaultNS,
 	}
 }