Ensure error status codes are surfaced from client failures

executor/api/grpc/seldon/server_test.go

@@ -28,7 +28,7 @@ func TestPredict(t *testing.T) {
 		},
 	}
 	url, _ := url.Parse("http://localhost")
-	server := NewGrpcSeldonServer(&p, test.NewSeldonMessageTestClient(t, 0, nil, nil), url, "default")
+	server := NewGrpcSeldonServer(&p, test.SeldonMessageTestClient{}, url, "default")
 
 	var sm proto.SeldonMessage
 	var data = ` {"data":{"ndarray":[[1.1,2.0]]}}`
@@ -58,7 +58,7 @@ func TestFeedback(t *testing.T) {
 		},
 	}
 	url, _ := url.Parse("http://localhost")
-	server := NewGrpcSeldonServer(&p, test.NewSeldonMessageTestClient(t, 0, nil, nil), url, "default")
+	server := NewGrpcSeldonServer(&p, &test.SeldonMessageTestClient{}, url, "default")
 
 	var sm proto.Feedback
 	var data = ` {"request":{"data":{"ndarray":[[1.1,2.0]]}}}`

executor/api/rest/client.go

@@ -189,7 +189,7 @@ func (smc *JSONRestClient) doHttp(ctx context.Context, modelName string, method
 
 	if response.StatusCode != http.StatusOK {
 		smc.Log.Info("httpPost failed", "response code", response.StatusCode)
-		err = errors.Errorf("Internal service call from executor failed calling %s status code %d", url, response.StatusCode)
+		err = &httpStatusError{StatusCode: response.StatusCode, Url: url}
 	}
 
 	return b, contentType, err

executor/api/rest/errors.go

@@ -0,0 +1,15 @@
+package rest
+
+import (
+	"fmt"
+	"net/url"
+)
+
+type httpStatusError struct {
+	StatusCode int
+	Url        *url.URL
+}
+
+func (e *httpStatusError) Error() string {
+	return fmt.Sprintf("Internal service call from executor failed calling %s status code %d", e.Url, e.StatusCode)
+}

executor/api/rest/server.go

@@ -82,8 +82,13 @@ func (r *SeldonRestApi) respondWithSuccess(w http.ResponseWriter, code int, payl
 }
 
 func (r *SeldonRestApi) respondWithError(w http.ResponseWriter, payload payload.SeldonPayload, err error) {
-	w.Header().Set("Content-Type", "application/json")
-	w.WriteHeader(http.StatusInternalServerError)
+	w.Header().Set("Content-Type", payload.GetContentType())
+
+	if serr, ok := err.(*httpStatusError); ok {
+		w.WriteHeader(serr.StatusCode)
+	} else {
+		w.WriteHeader(http.StatusInternalServerError)
+	}
 
 	if payload != nil && payload.GetPayload() != nil {
 		err := r.Client.Marshall(w, payload)

executor/api/rest/server_test.go

@@ -55,7 +55,7 @@ func TestSimpleModel(t *testing.T) {
 	}
 
 	url, _ := url.Parse("http://localhost")
-	r := NewServerRestApi(&p, test.NewSeldonMessageTestClient(t, 0, nil, nil), false, url, "default", api.ProtocolSeldon, "test", "/metrics")
+	r := NewServerRestApi(&p, &test.SeldonMessageTestClient{}, false, url, "default", api.ProtocolSeldon, "test", "/metrics")
 	r.Initialise()
 	var data = ` {"data":{"ndarray":[1.1,2.0]}}`
 
@@ -84,7 +84,7 @@ func TestReponsePuuidIsSet(t *testing.T) {
 	}
 
 	url, _ := url.Parse("http://localhost")
-	r := NewServerRestApi(&p, test.NewSeldonMessageTestClient(t, 0, nil, nil), false, url, "default", api.ProtocolSeldon, "test", "/metrics")
+	r := NewServerRestApi(&p, &test.SeldonMessageTestClient{}, false, url, "default", api.ProtocolSeldon, "test", "/metrics")
 	r.Initialise()
 	var data = ` {"data":{"ndarray":[1.1,2.0]}}`
 
@@ -211,7 +211,7 @@ func TestServerMetrics(t *testing.T) {
 	}
 
 	url, _ := url.Parse("http://localhost")
-	r := NewServerRestApi(&p, test.NewSeldonMessageTestClient(t, 0, nil, nil), false, url, "default", api.ProtocolSeldon, "test", "/metrics")
+	r := NewServerRestApi(&p, &test.SeldonMessageTestClient{}, false, url, "default", api.ProtocolSeldon, "test", "/metrics")
 	r.Initialise()
 
 	var data = ` {"data":{"ndarray":[1.1,2.0]}}`
@@ -252,7 +252,7 @@ func TestTensorflowStatus(t *testing.T) {
 	}
 
 	url, _ := url.Parse("http://localhost")
-	r := NewServerRestApi(&p, test.NewSeldonMessageTestClient(t, 0, nil, nil), false, url, "default", api.ProtocolTensorflow, "test", "/metrics")
+	r := NewServerRestApi(&p, &test.SeldonMessageTestClient{}, false, url, "default", api.ProtocolTensorflow, "test", "/metrics")
 	r.Initialise()
 
 	req, _ := http.NewRequest("GET", "/v1/models/mymodel", nil)
@@ -281,7 +281,7 @@ func TestSeldonStatus(t *testing.T) {
 	}
 
 	url, _ := url.Parse("http://localhost")
-	r := NewServerRestApi(&p, test.NewSeldonMessageTestClient(t, 0, nil, nil), false, url, "default", api.ProtocolSeldon, "test", "/metrics")
+	r := NewServerRestApi(&p, &test.SeldonMessageTestClient{}, false, url, "default", api.ProtocolSeldon, "test", "/metrics")
 	r.Initialise()
 
 	req, _ := http.NewRequest("GET", "/api/v1.0/status/mymodel", nil)
@@ -310,7 +310,7 @@ func TestSeldonMetadata(t *testing.T) {
 	}
 
 	url, _ := url.Parse("http://localhost")
-	r := NewServerRestApi(&p, test.NewSeldonMessageTestClient(t, 0, nil, nil), false, url, "default", api.ProtocolSeldon, "test", "/metrics")
+	r := NewServerRestApi(&p, &test.SeldonMessageTestClient{}, false, url, "default", api.ProtocolSeldon, "test", "/metrics")
 	r.Initialise()
 
 	req, _ := http.NewRequest("GET", "/api/v1.0/metadata/mymodel", nil)
@@ -339,7 +339,7 @@ func TestTensorflowMetadata(t *testing.T) {
 	}
 
 	url, _ := url.Parse("http://localhost")
-	r := NewServerRestApi(&p, test.NewSeldonMessageTestClient(t, 0, nil, nil), false, url, "default", api.ProtocolTensorflow, "test", "/metrics")
+	r := NewServerRestApi(&p, &test.SeldonMessageTestClient{}, false, url, "default", api.ProtocolTensorflow, "test", "/metrics")
 	r.Initialise()
 
 	req, _ := http.NewRequest("GET", "/v1/models/mymodel/metadata", nil)
@@ -353,13 +353,14 @@ func TestPredictErrorWithServer(t *testing.T) {
 	t.Logf("Started")
 	g := NewGomegaWithT(t)
 	called := false
+	errorCode := http.StatusConflict
 
 	handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		_, err := ioutil.ReadAll(r.Body)
 		g.Expect(err).To(BeNil())
 		g.Expect(r.Header.Get(payload.SeldonPUIDHeader)).To(Equal(TestSeldonPuid))
 		called = true
-		w.WriteHeader(http.StatusInternalServerError)
+		w.WriteHeader(errorCode)
 		w.Write([]byte(errorPredictResponse))
 	})
 	server := httptest.NewServer(handler)
@@ -392,7 +393,8 @@ func TestPredictErrorWithServer(t *testing.T) {
 	req.Header = map[string][]string{"Content-Type": []string{"application/json"}, payload.SeldonPUIDHeader: []string{TestSeldonPuid}}
 	res := httptest.NewRecorder()
 	r.Router.ServeHTTP(res, req)
-	g.Expect(res.Code).To(Equal(http.StatusInternalServerError))
+	// check error code is the one returned by client
+	g.Expect(res.Code).To(Equal(errorCode))
 	g.Expect(called).To(Equal(true))
 	b, err := ioutil.ReadAll(res.Body)
 	g.Expect(err).Should(BeNil())

executor/api/test/seldonmessage_test_client.go

@@ -2,20 +2,18 @@ package test
 
 import (
 	"context"
-	"github.com/seldonio/seldon-core/executor/api/client"
 	"github.com/seldonio/seldon-core/executor/api/grpc/seldon/proto"
 	"github.com/seldonio/seldon-core/executor/api/payload"
 	"github.com/seldonio/seldon-core/operator/apis/machinelearning/v1"
 	"io"
 	"net/http"
-	"testing"
 )
 
 type SeldonMessageTestClient struct {
-	t           *testing.T
-	chosenRoute int
-	errMethod   *v1.PredictiveUnitMethod
-	err         error
+	ChosenRoute int
+	ErrMethod   *v1.PredictiveUnitMethod
+	Err         error
+	ErrPayload  payload.SeldonPayload
 }
 
 const (
@@ -52,51 +50,35 @@ func (s SeldonMessageTestClient) CreateErrorPayload(err error) payload.SeldonPay
 }
 
 func (s SeldonMessageTestClient) Predict(ctx context.Context, modelName string, host string, port int32, msg payload.SeldonPayload, meta map[string][]string) (payload.SeldonPayload, error) {
-	s.t.Logf("Predict %s %d", host, port)
-	if s.errMethod != nil && *s.errMethod == v1.TRANSFORM_INPUT {
-		return nil, s.err
+	if s.ErrMethod != nil && *s.ErrMethod == v1.TRANSFORM_INPUT {
+		return s.ErrPayload, s.Err
 	}
 	return msg, nil
 }
 
 func (s SeldonMessageTestClient) TransformInput(ctx context.Context, modelName string, host string, port int32, msg payload.SeldonPayload, meta map[string][]string) (payload.SeldonPayload, error) {
-	s.t.Logf("TransformInput %s %d", host, port)
-	if s.errMethod != nil && *s.errMethod == v1.TRANSFORM_INPUT {
-		return nil, s.err
+	if s.ErrMethod != nil && *s.ErrMethod == v1.TRANSFORM_INPUT {
+		return s.ErrPayload, s.Err
 	}
 	return msg, nil
 }
 
 func (s SeldonMessageTestClient) Route(ctx context.Context, modelName string, host string, port int32, msg payload.SeldonPayload, meta map[string][]string) (int, error) {
-	s.t.Logf("Route %s %d", host, port)
-	return s.chosenRoute, nil
+	return s.ChosenRoute, nil
 }
 
 func (s SeldonMessageTestClient) Combine(ctx context.Context, modelName string, host string, port int32, msgs []payload.SeldonPayload, meta map[string][]string) (payload.SeldonPayload, error) {
-	s.t.Logf("Combine %s %d", host, port)
 	return msgs[0], nil
 }
 
 func (s SeldonMessageTestClient) TransformOutput(ctx context.Context, modelName string, host string, port int32, msg payload.SeldonPayload, meta map[string][]string) (payload.SeldonPayload, error) {
-	s.t.Logf("TransformOutput %s %d", host, port)
 	return msg, nil
 }
 
 func (s SeldonMessageTestClient) Feedback(ctx context.Context, modelName string, host string, port int32, msg payload.SeldonPayload, meta map[string][]string) (payload.SeldonPayload, error) {
-	s.t.Logf("Feedback %s %d", host, port)
-	if s.errMethod != nil && *s.errMethod == v1.SEND_FEEDBACK {
-		return nil, s.err
+	if s.ErrMethod != nil && *s.ErrMethod == v1.SEND_FEEDBACK {
+		return nil, s.Err
 	}
 	resp := &payload.ProtoPayload{Msg: msg.GetPayload().(*proto.Feedback).Request}
 	return resp, nil
 }
-
-func NewSeldonMessageTestClient(t *testing.T, chosenRoute int, errMethod *v1.PredictiveUnitMethod, err error) client.SeldonApiClient {
-	client := SeldonMessageTestClient{
-		t:           t,
-		chosenRoute: chosenRoute,
-		errMethod:   errMethod,
-		err:         err,
-	}
-	return &client
-}

executor/predictor/predictor_process_test.go

@@ -32,21 +32,21 @@ const (
 func createPredictorProcess(t *testing.T) *PredictorProcess {
 	url, _ := url.Parse(testSourceUrl)
 	ctx := context.WithValue(context.TODO(), payload.SeldonPUIDHeader, testSeldonPuid)
-	pp := NewPredictorProcess(ctx, test.NewSeldonMessageTestClient(t, -1, nil, nil), logf.Log.WithName("SeldonMessageRestClient"), url, "default", map[string][]string{testCustomMetaKey: []string{testCustomMetaValue}})
+	pp := NewPredictorProcess(ctx, &test.SeldonMessageTestClient{}, logf.Log.WithName("SeldonMessageRestClient"), url, "default", map[string][]string{testCustomMetaKey: []string{testCustomMetaValue}})
 	return &pp
 }
 
 func createPredictorProcessWithRoute(t *testing.T, chosenRoute int) *PredictorProcess {
 	url, _ := url.Parse(testSourceUrl)
 	ctx := context.WithValue(context.TODO(), payload.SeldonPUIDHeader, testSeldonPuid)
-	pp := NewPredictorProcess(ctx, test.NewSeldonMessageTestClient(t, chosenRoute, nil, nil), logf.Log.WithName("SeldonMessageRestClient"), url, "default", map[string][]string{})
+	pp := NewPredictorProcess(ctx, &test.SeldonMessageTestClient{ChosenRoute: chosenRoute}, logf.Log.WithName("SeldonMessageRestClient"), url, "default", map[string][]string{})
 	return &pp
 }
 
 func createPredictorProcessWithError(t *testing.T, errMethod *v1.PredictiveUnitMethod, err error) *PredictorProcess {
 	url, _ := url.Parse(testSourceUrl)
 	ctx := context.WithValue(context.TODO(), payload.SeldonPUIDHeader, testSeldonPuid)
-	pp := NewPredictorProcess(ctx, test.NewSeldonMessageTestClient(t, -1, errMethod, err), logf.Log.WithName("SeldonMessageRestClient"), url, "default", map[string][]string{})
+	pp := NewPredictorProcess(ctx, &test.SeldonMessageTestClient{ErrMethod: errMethod, Err: err}, logf.Log.WithName("SeldonMessageRestClient"), url, "default", map[string][]string{})
 	return &pp
 }
 

testing/scripts/test_s2i_python.py

@@ -145,7 +145,7 @@ def test_model_rest_non200(self, namespace, s2i_python_version):
         r = rest_request_ambassador("mymodel", namespace, API_AMBASSADOR, data=arr)
         res = r.json()
         logging.warning(res)
-        assert r.status_code == 500
+        assert r.status_code == 400
         assert r.json()["status"]["code"] == 400
         assert r.json()["status"]["info"] == "exception caught"
         run(