Add turing_req_id in requests made to Turing Router components (#230)

* Add turing request id to requests sent to router components

* Replace deprecated ioutil.ReadAll function with io.ReadAll

* Fix http handler test

* Add mention of turing_req_id to docs

* Add sample of how turing_req_id is accessed in the pyfunc ensembler

* Rename turing_req_id in docs to Turing-Req-Id

Author: deadlycoconuts

docs/how-to/create-a-router/configure-enricher.md

@@ -10,7 +10,12 @@ The Turing enricher has the ability to perform arbitrary transformations on the
 Original request will be sent to configured routes without enrichment.
 
 ### Docker
-Turing will deploy specified Docker image as a pre-processor and will send original request to it for enrichment. To configure a Docker enricher, there are 3 sections to fill.  
+Turing will deploy specified Docker image as a pre-processor and will send original request to it for enrichment. 
+
+The request to the enricher is constructed from the request headers to the Router, and an identifier `Turing-Req-Id` 
+that is uniquely assigned to each request received by the Router.
+
+To configure a Docker enricher, there are 3 sections to fill.  
 
 Configure the Docker Container. There are 4 required inputs.
 

docs/how-to/create-a-router/configure-ensembler.md

@@ -26,7 +26,14 @@ It is not possible to select as the fallback response a route that has traffic r
 {% endhint %}
 
 ## Docker
-Turing will deploy the specified image as a post-processor and will send in the request payload the following, for ensembling - the original request, responses from all routes, and the treatment configuration (if an Experiment Engine is selected, in Configure Experiment Engine). The ensembler's request headers will contain the original request headers sent to Turing, merged with the enricher's response headers (if there are duplicates, the value in the enricher's response headers will take precedence). To configure a Docker ensembler, there are 3 sections to be filled.
+
+Turing will deploy the specified image as a post-processor and will send in the request payload the following, for 
+ensembling - the original request, responses from all routes, and the treatment configuration (if an Experiment 
+Engine is selected, in the Configure Experiment Engine step). The ensembler's request headers will contain the original 
+request headers sent to Turing, merged with the enricher's response headers (if there are duplicates, the value in 
+the enricher's response headers will take precedence), and an identifier `Turing-Req-Id` that is uniquely assigned to each request received by the Router. 
+
+To configure a Docker ensembler, there are 3 sections to be filled.
 
 Configure the Docker Container. There are 4 required inputs.
 
@@ -57,20 +64,26 @@ Configure the resources required for the ensembler. There are 3 required inputs,
 **Min/Max Replicas**: Min/max number of replicas for your ensembler. Scaling of the ensembler based on traffic volume will be automatically done for you.
 
 ## Pyfunc Ensembler
-Turing will deploy a previously registered pyfunc ensembler (refer to 
+Turing will deploy a previously registered Pyfunc ensembler (refer to 
 [the samples](https://github.com/caraml-dev/turing/tree/main/sdk/samples) in the SDK section for more information on how to 
 deploy one) as a containerised web service. 
 
 This allows you to simply define the logic required for the ensembling 
 step by implementing a Python `mlflow`-based interface, and rely on Turing API to containerise and package your 
 implementation as an entire web service automatically.
 
+Similar to requests sent to a Docker Ensembler, the request payload sent to a Pyfunc ensembler will contain the 
+original request, responses from all routes, and the treatment configuration (if an Experiment
+Engine is selected, in the Configure Experiment Engine step). The ensembler's request headers will contain the original
+request headers sent to Turing, merged with the enricher's response headers (if there are duplicates, the value in
+the enricher's response headers will take precedence), and an identifier `Turing-Req-Id` that is uniquely assigned to each request received by the Router. 
+
 To configure your router with a Pyfunc ensembler, simply select from the drop down list your desired ensembler, 
 registered in your current project. You'll also need to indicate your desired timeout value and resource request values:
 
 ![](../../.gitbook/assets/pyfunc_ensembler_config.png)
 
-**Pyfunc Ensembler**: The name of the pyfunc ensembler that has been deployed in your *current* project 
+**Pyfunc Ensembler**: The name of the Pyfunc ensembler that has been deployed in your *current* project 
 
 **Timeout**: Request timeout, which when exceeded, the request to the ensembler will be terminated
 
@@ -87,7 +100,8 @@ The router will send responses from all routes, together with treatment configur
 
 
 ## Ensembler Request Payload Format
-When the ensembler type is Docker/External, the ensembler will receive the following information in the request payload and the behaviour of the ensembler is up to the implementer.
+When the ensembler type is Docker/Pyfunc/External, the ensembler will receive the following information in the request 
+payload and the behaviour of the ensembler is up to the implementer.
 
 ```json
 {

docs/how-to/create-a-router/configure-experiment-engine.md

@@ -4,6 +4,10 @@
 This step is **optional** and the default behaviour will be to send the request to all of the configured routes, but the response from Turing will depend on the Ensembler configuration.
 {% endhint %}
 
+The request sent to the experiment engine is constructed from the request headers to the Router (including an
+identifier `Turing-Req-Id` that is uniquely assigned to each request received by the Router), and either the request 
+body to the Router (if there is no Enricher) or the response body from the Enricher (if enabled).
+
 If you intend to determine how to select/combine responses from the individual routes (typically in the Ensembler) to send back as the Turing response, configure the Experiment Engine as shown in Configure Experiment Engine. 
 
 ![](../../.gitbook/assets/configure_expriement_engine.png)

docs/how-to/create-a-router/configure-routes.md

@@ -2,6 +2,10 @@
 
 Routes are an essential part of your Turing Router setup. Each route is defined by its ID and the endpoint. A route can be your deployed ML model, exposed via HTTP interface, or an arbitrary non-ML web-service. Each router should be configured with at least one route and each route's ID should be unique among other routes of this router.
 
+The request sent to the each route is constructed from the request headers to the Router (including an
+identifier `Turing-Req-Id` that is uniquely assigned to each request received by the Router), and either the request
+body to the Router (if there is no Enricher) or the response body from the Enricher (if enabled).
+
 ![](../../.gitbook/assets/routes_panel.png)
 
 {% hint style="info" %}

engines/router/go.mod

@@ -5,11 +5,11 @@ go 1.18
 require (
 	bou.ke/monkey v1.0.2
 	cloud.google.com/go/bigquery v1.14.0
+	github.com/caraml-dev/turing/engines/experiment v0.0.0
 	github.com/fluent/fluent-logger-golang v1.5.0
 	github.com/go-playground/validator/v10 v10.3.0
 	github.com/gojek/fiber v0.0.0-20201008181849-4f0f8284dc84
 	github.com/gojek/mlp v1.4.7
-	github.com/caraml-dev/turing/engines/experiment v0.0.0
 	github.com/google/go-cmp v0.5.5
 	github.com/google/uuid v1.1.2
 	github.com/heptiolabs/healthcheck v0.0.0-20180807145615-6ff867650f40

engines/router/missionctl/handlers/batch_http_handler.go

@@ -2,7 +2,7 @@ package handlers
 
 import (
 	"encoding/json"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"strconv"
 	"sync"
@@ -62,7 +62,7 @@ func (h *batchHTTPHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request)
 	}
 
 	// Read the request body
-	requestBody, err := ioutil.ReadAll(req.Body)
+	requestBody, err := io.ReadAll(req.Body)
 	if err != nil {
 		h.error(ctx, rw, errors.NewHTTPError(err))
 		return

engines/router/missionctl/handlers/batch_http_handler_test.go

@@ -2,7 +2,7 @@ package handlers
 
 import (
 	"bytes"
-	"io/ioutil"
+	"io"
 	"net"
 	"net/http"
 	"net/http/httptest"
@@ -25,7 +25,7 @@ func TestNewBatchHTTPHandler(t *testing.T) {
 
 	//Create test routes endpoint. Route will write request body as response
 	testRouterHandler := http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
-		requestBody, err := ioutil.ReadAll(request.Body)
+		requestBody, err := io.ReadAll(request.Body)
 		assert.NoError(t, err)
 		_, err = writer.Write(requestBody)
 		assert.NoError(t, err)
@@ -101,7 +101,7 @@ func TestNewBatchHTTPHandler(t *testing.T) {
 			batchHTTPHandler.ServeHTTP(w, req)
 			res := w.Result()
 			defer res.Body.Close()
-			result, _ := ioutil.ReadAll(res.Body)
+			result, _ := io.ReadAll(res.Body)
 
 			assert.Equal(t, test.expectedStatusCode, res.StatusCode)
 			assert.Equal(t, test.expectedContentType, res.Header.Get("Content-Type"))

engines/router/missionctl/handlers/http_handler.go

@@ -3,7 +3,7 @@ package handlers
 import (
 	"context"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"time"
 
@@ -161,6 +161,10 @@ func (h *httpHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 	}
 	ctxLogger.Debugf("Received request for %v", turingReqID)
 
+	// Sets the turing request id in the header of the original request so that it gets sent to the enricher,
+	// the experiment engine, the model routes, and the ensembler
+	req.Header.Set(turingReqIDHeaderKey, turingReqID)
+
 	if tracing.Glob().IsEnabled() {
 		var sp opentracing.Span
 		ctx, sp = h.enableTracingSpan(ctx, req, httpHandlerID)
@@ -170,7 +174,7 @@ func (h *httpHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
 	}
 
 	// Read the request body
-	requestBody, err := ioutil.ReadAll(req.Body)
+	requestBody, err := io.ReadAll(req.Body)
 	if err != nil {
 		h.error(ctx, rw, errors.NewHTTPError(err))
 		return

engines/router/missionctl/handlers/http_handler_test.go

@@ -192,7 +192,8 @@ func TestHTTPService(t *testing.T) {
 
 	// Check that ensembler was called with the expected headers
 	mc.AssertCalled(t, "Ensemble",
-		http.Header{"Context-Type": []string{"application/json"}, "Enricher": []string{"value"}})
+		http.Header{"Context-Type": []string{"application/json"}, "Enricher": []string{"value"},
+			"Turing-Req-Id": []string{rr.Header().Get("Turing-Req-Id")}})
 }
 
 // TestHTTPServiceBadRequest tests for a HTTP InternalServerError on bad

engines/router/missionctl/http/response.go

@@ -1,7 +1,7 @@
 package http
 
 import (
-	"io/ioutil"
+	"io"
 	"net/http"
 )
 
@@ -32,7 +32,7 @@ func (r *CachedResponse) Body() []byte {
 // a HTTP response as input
 func NewCachedResponseFromHTTP(r *http.Response) (*CachedResponse, error) {
 	// Consume response body
-	data, err := ioutil.ReadAll(r.Body)
+	data, err := io.ReadAll(r.Body)
 	if err != nil {
 		return nil, err
 	}