Repository: odahu/odahu-flow (language: python)

Issue:
Handle request-id header on service mesh layer
Handle request-id header on service mesh layer. Currently it is automatically set on ingress for request, but ODAHU ML Server (Flask service) has a logic to also put it to response. It is required for proper feedback flow. Since we want to support third-party Inference Servers we cannot rely on this behaviour and have to move it to service mesh layer.

Currently ML server uses X-REQUEST-ID and pastes it to request ID.


Discussion / hints:

Fix patch:
diff --git a/helms/odahu-flow-core/templates/edge-ingress.yaml b/helms/odahu-flow-core/templates/edge-ingress.yaml
index e0a403d4a..6f2f302c1 100644
--- a/helms/odahu-flow-core/templates/edge-ingress.yaml
+++ b/helms/odahu-flow-core/templates/edge-ingress.yaml
@@ -41,30 +41,47 @@ spec:
       {{- end }}
 {{ end }}
 ---
-# This filter is to preserve a request ID of incoming external request on Ingress.
-# The motivation is that to adopt Knative routing mechanism we use "double" routing
-# on istio ingress. The first routing (that we provide) bases on the request path and
-# forwards the request to the same ingress, but rewrites Host header for the second routing.
-# The second one (automatically added by Knative Service) bases on the Host header and forwards
-# the request to the corresponding service. Without preserve_external_request_id these 2 phases of routing
-# both override the X-Request-ID header, while we want it to stay the same. Otherwise feedback mechanism breaks.
+# This filter saves a request ID (taken from "request-id" or "x-request-id" header) to a Dynamic Metadata,
+# a shared cache for all the filters of current HTTP stream. On forwarding response it retrieves the request-id value
+# and sets the corresponding header. This ensures that client always receives a request ID in response headers,
+# no matter what inference server is used in Model Deployment. The request ID can then be used to provide a feedback
+# for a corresponding prediction.
+# TODO: used syntax is for Istio 1.4 and is deprecated. Don't forget to update it after upgrading Istio
 apiVersion: networking.istio.io/v1alpha3
 kind: EnvoyFilter
 metadata:
-  name: preserve-request-id-filter
+  name: request-id-filter
   namespace: {{ .Values.feedback.istio_namespace }}
 spec:
-  configPatches:
-    - applyTo: NETWORK_FILTER
-      match:
-        context: GATEWAY
-        listener:
-          filterChain:
-            filter:
-              name: envoy.http_connection_manager
-      patch:
-        operation: MERGE
-        value:
-          typed_config:
-            '@type': type.googleapis.com/envoy.config.filter.network.http_connection_manager.v2.HttpConnectionManager
-            preserve_external_request_id: true
+  workloadSelector:
+    labels:
+      istio: ingressgateway
+  filters:
+    - filterName: envoy.lua
+      filterType: HTTP
+      listenerMatch:
+        listenerType: GATEWAY
+        listenerProtocol: HTTP
+      filterConfig:
+        inlineCode: |
+          function envoy_on_request(request_handle)
+            local headers = request_handle:headers()
+            local x_request_id = headers:get("x-request-id")
+            local request_id = headers:get("request-id")
+
+            if request_id == nil then
+              request_id = x_request_id
+              request_handle:logInfo("request-id header is nil, using x-request-id")
+              headers:add("request-id", request_id)
+            end
+
+            request_handle:logInfo("request-id from headers: "..request_id)
+            request_handle:streamInfo():dynamicMetadata():set("odahu", "request-id", request_id)
+          end
+          function envoy_on_response(response_handle)
+            local headers = response_handle:headers()
+            local request_id = response_handle:streamInfo():dynamicMetadata():get("odahu")["request-id"]
+            headers:replace("request-id", request_id)
+
+            response_handle:logInfo("request-id from dynamic metadata: "..request_id)
+          end
diff --git a/packages/feedback/pkg/collector/handlers.go b/packages/feedback/pkg/collector/handlers.go
index e4376b18b..7f3fe2af7 100644
--- a/packages/feedback/pkg/collector/handlers.go
+++ b/packages/feedback/pkg/collector/handlers.go
@@ -20,8 +20,8 @@ import (
 	"fmt"
 	"github.com/odahu/odahu-flow/packages/feedback/pkg/feedback"
 	"net/http"
-	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	commons_feedback "odahu-commons/feedback"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
 	"github.com/gin-gonic/gin"
 )
@@ -43,11 +43,9 @@ func handleFeedbackEndpoint(c *gin.Context) {
 
 	message.RequestID = c.GetHeader(feedback.OdahuFlowRequestIdHeaderKey)
 	if len(message.RequestID) == 0 {
-		message.RequestID = c.GetHeader(feedback.RequestIdHeaderKey)
-		if len(message.RequestID) == 0 {
-			c.JSON(http.StatusBadRequest, gin.H{"error": fmt.Sprintf("%s header is missed", feedback.RequestIdHeaderKey)})
-			return
-		}
+		c.JSON(http.StatusBadRequest,
+			gin.H{"error": fmt.Sprintf("%s header is missed", feedback.OdahuFlowRequestIdHeaderKey)})
+		return
 	}
 
 	if len(modelName) == 0 || len(modelVersion) == 0 {
diff --git a/packages/feedback/pkg/feedback/types.go b/packages/feedback/pkg/feedback/types.go
index 2342e09b1..b3db17325 100644
--- a/packages/feedback/pkg/feedback/types.go
+++ b/packages/feedback/pkg/feedback/types.go
@@ -4,7 +4,6 @@ const (
 	ModelNameHeaderKey          = "model-name"
 	ModelVersionHeaderKey       = "model-version"
 	ModelEndpointHeaderKey      = "model-endpoint"
-	RequestIdHeaderKey          = "x-request-id"
 	OdahuFlowRequestIdHeaderKey = "request-id"
 	HttpMethodHeaderKey         = ":method"
 	OriginalUriHeaderKey        = "x-original-uri"
diff --git a/packages/feedback/pkg/tapping/collector.go b/packages/feedback/pkg/tapping/collector.go
index 60fef146f..6d392c5bd 100644
--- a/packages/feedback/pkg/tapping/collector.go
+++ b/packages/feedback/pkg/tapping/collector.go
@@ -103,12 +103,6 @@ func (rc *RequestCollector) convertToFeedback(message *Message) (*commons_feedba
 		case feedback.ForwardedHostHeaderKey:
 			requestResponse.RequestHost = header.Value
 
-		case feedback.RequestIdHeaderKey:
-			if len(responseBody.RequestID) == 0 || len(requestResponse.RequestID) == 0 {
-				responseBody.RequestID = header.Value
-				requestResponse.RequestID = header.Value
-			}
-
 		case feedback.OdahuFlowRequestIdHeaderKey:
 			responseBody.RequestID = header.Value
 			requestResponse.RequestID = header.Value


Test patch:
diff --git a/packages/feedback/pkg/collector/http_test.go b/packages/feedback/pkg/collector/http_test.go
index aa68bcd69..8865b8503 100644
--- a/packages/feedback/pkg/collector/http_test.go
+++ b/packages/feedback/pkg/collector/http_test.go
@@ -19,10 +19,10 @@ package collector
 import (
 	"encoding/json"
 	"fmt"
-	feedback_commons "odahu-commons/feedback"
 	"github.com/odahu/odahu-flow/packages/feedback/pkg/feedback"
 	"net/http"
 	"net/http/httptest"
+	feedback_commons "odahu-commons/feedback"
 	"testing"
 
 	"github.com/gin-gonic/gin"
@@ -69,7 +69,7 @@ func TestSendFeedbackWithoutHeader(t *testing.T) {
 	router.ServeHTTP(w, req)
 
 	assert.Equal(t, 400, w.Code)
-	assert.Equal(t, "{\"error\":\"x-request-id header is missed\"}", w.Body.String())
+	assert.Equal(t, "{\"error\":\"request-id header is missed\"}", w.Body.String())
 }
 
 func buildMessage(modelName, modelVersion, requestID string, payload map[string]interface{}) feedback_commons.ModelFeedback {
@@ -102,7 +102,7 @@ func TestSendFeedbackWithOnlyHeader(t *testing.T) {
 
 	w := httptest.NewRecorder()
 	req, _ := http.NewRequest("POST", testFeedbackUrl, nil)
-	req.Header.Set(feedback.RequestIdHeaderKey, requestID)
+	req.Header.Set(feedback.OdahuFlowRequestIdHeaderKey, requestID)
 	req.Header.Set(feedback.ModelNameHeaderKey, modelName)
 	req.Header.Set(feedback.ModelVersionHeaderKey, modelVersion)
 	router.ServeHTTP(w, req)