executor: decompress gzip paloyads before logging them

executor/api/rest/kfserving.go

@@ -1,14 +1,37 @@
 package rest
 
 import (
+	"compress/gzip"
 	"encoding/json"
+	"io/ioutil"
+
 	"github.com/pkg/errors"
 	"github.com/seldonio/seldon-core/executor/api/payload"
+
+	"bytes"
 )
 
 func ChainKFserving(msg payload.SeldonPayload) (payload.SeldonPayload, error) {
+	data, err := msg.GetBytes()
+	if err != nil {
+		return nil, err
+	}
+
+	if msg.GetContentEncoding() == "gzip" {
+		bytesReader := bytes.NewReader(data)
+		gzipReader, err := gzip.NewReader(bytesReader)
+		if err != nil {
+			return nil, err
+		}
+		output, err := ioutil.ReadAll(gzipReader)
+		if err != nil {
+			return nil, err
+		}
+		data = output
+	}
+
 	var f interface{}
-	err := json.Unmarshal(msg.GetPayload().([]byte), &f)
+	err = json.Unmarshal(data, &f)
 	if err != nil {
 		return nil, err
 	}
@@ -21,10 +44,9 @@ func ChainKFserving(msg payload.SeldonPayload) (payload.SeldonPayload, error) {
 		b, err := json.Marshal(m)
 		if err != nil {
 			return nil, err
-		} else {
-			p := payload.BytesPayload{Msg: b}
-			return &p, nil
 		}
+		p := payload.BytesPayload{Msg: b, ContentType: msg.GetContentType()}
+		return &p, nil
 	} else {
 		return nil, errors.Errorf("Failed to convert kfserving response so it could be chained to new input")
 	}

executor/logger/types.go

@@ -13,12 +13,13 @@ const (
 )
 
 type LogRequest struct {
-	Url         *url.URL
-	Bytes       *[]byte
-	ContentType string
-	ReqType     LogRequestType
-	Id          string
-	SourceUri   *url.URL
-	ModelId     string
-	RequestId   string
+	Url             *url.URL
+	Bytes           *[]byte
+	ContentType     string
+	ContentEncoding string
+	ReqType         LogRequestType
+	Id              string
+	SourceUri       *url.URL
+	ModelId         string
+	RequestId       string
 }

executor/logger/worker.go

@@ -1,15 +1,19 @@
 package logger
 
 import (
+	"bytes"
+	"compress/gzip"
 	"context"
 	"errors"
 	"fmt"
+	"io/ioutil"
+	"net/http"
+	"time"
+
 	cloudevents "github.com/cloudevents/sdk-go"
 	"github.com/cloudevents/sdk-go/pkg/cloudevents/transport"
 	"github.com/confluentinc/confluent-kafka-go/kafka"
 	"github.com/go-logr/logr"
-	"net/http"
-	"time"
 )
 
 const (
@@ -122,6 +126,26 @@ func (w *Worker) sendKafkaEvent(logReq LogRequest) error {
 }
 
 func (w *Worker) sendCloudEvent(logReq LogRequest) error {
+	var data []byte
+
+	// This temporary fix related to the fact that Triton server responses
+	// are now gzipped compressed. Until we introduce support for gzip
+	// compressed payloads in the logger / adserver and include content-encoding
+	// header in the CloudEvent messages this can serve as temporary solution.
+	if logReq.ContentEncoding == "gzip" {
+		bytesReader := bytes.NewReader(*logReq.Bytes)
+		gzipReader, err := gzip.NewReader(bytesReader)
+		if err != nil {
+			return err
+		}
+		output, err := ioutil.ReadAll(gzipReader)
+		if err != nil {
+			return err
+		}
+		data = output
+	} else {
+		data = *logReq.Bytes
+	}
 
 	t, err := cloudevents.NewHTTPTransport(
 		cloudevents.WithTarget(logReq.Url.String()),
@@ -154,7 +178,7 @@ func (w *Worker) sendCloudEvent(logReq LogRequest) error {
 
 	event.SetSource(logReq.SourceUri.String())
 	event.SetDataContentType(logReq.ContentType)
-	if err := event.SetData(*logReq.Bytes); err != nil {
+	if err := event.SetData(data); err != nil {
 		return fmt.Errorf("while setting cloudevents data: %s", err)
 	}
 

executor/predictor/predictor_process.go

@@ -333,17 +333,17 @@ func (p *PredictorProcess) logPayload(nodeName string, logger *v1.Logger, reqTyp
 	if err != nil {
 		return err
 	}
-
 	go func() {
 		err := payloadLogger.QueueLogRequest(payloadLogger.LogRequest{
-			Url:         logUrl,
-			Bytes:       &data,
-			ContentType: msg.GetContentType(),
-			ReqType:     reqType,
-			Id:          guuid.New().String(),
-			SourceUri:   p.ServerUrl,
-			ModelId:     nodeName,
-			RequestId:   puid,
+			Url:             logUrl,
+			Bytes:           &data,
+			ContentType:     msg.GetContentType(),
+			ContentEncoding: msg.GetContentEncoding(),
+			ReqType:         reqType,
+			Id:              guuid.New().String(),
+			SourceUri:       p.ServerUrl,
+			ModelId:         nodeName,
+			RequestId:       puid,
 		})
 		if err != nil {
 			p.Log.Error(err, "failed to log request")