Issue 561 pipeline err (#569)

scheduler/pkg/kafka/pipeline/grpcserver.go

@@ -113,6 +113,9 @@ func (g *GatewayGrpcServer) ModelInfer(ctx context.Context, r *v2.ModelInferRequ
 	if err != nil {
 		go g.metrics.AddPipelineInferMetrics(resourceName, metrics.MethodTypeGrpc, elapsedTime, codes.FailedPrecondition.String())
 		return nil, status.Errorf(codes.FailedPrecondition, err.Error())
+	} else if kafkaRequest.isError {
+		go g.metrics.AddPipelineInferMetrics(resourceName, metrics.MethodTypeGrpc, elapsedTime, codes.Unknown.String())
+		return nil, status.Errorf(codes.Unknown, string(kafkaRequest.response))
 	}
 	meta := convertKafkaHeadersToGrpcMetadata(kafkaRequest.headers)
 	meta[util.RequestIdHeader] = []string{kafkaRequest.key}

scheduler/pkg/kafka/pipeline/httpserver.go

@@ -127,15 +127,24 @@ func (g *GatewayHttpServer) infer(w http.ResponseWriter, req *http.Request, reso
 
 	kafkaRequest, err := g.gateway.Infer(req.Context(), resourceName, isModel, dataProto, convertHttpHeadersToKafkaHeaders(req.Header), g.getRequestId(req))
 	elapsedTime := time.Since(startTime).Seconds()
-	for k, vals := range convertKafkaHeadersToHttpHeaders(kafkaRequest.headers) {
-		for _, val := range vals {
-			w.Header().Add(k, val)
+	if kafkaRequest != nil {
+		for k, vals := range convertKafkaHeadersToHttpHeaders(kafkaRequest.headers) {
+			for _, val := range vals {
+				w.Header().Add(k, val)
+			}
 		}
 	}
 	w.Header().Set(util.RequestIdHeader, kafkaRequest.key)
 	if err != nil {
 		logger.WithError(err).Error("Failed to call infer")
 		w.WriteHeader(http.StatusInternalServerError)
+	} else if kafkaRequest.isError {
+		logger.Error(string(kafkaRequest.response))
+		w.WriteHeader(http.StatusBadRequest)
+		_, err = w.Write(kafkaRequest.response)
+		if err != nil {
+			logger.WithError(err).Error("Failed to write error payload")
+		}
 	} else {
 		resJson, err := ConvertV2ResponseBytesToJson(kafkaRequest.response)
 		if err != nil {

scheduler/pkg/kafka/pipeline/httpserver_test.go

@@ -23,16 +23,17 @@ import (
 )
 
 type fakePipelineInferer struct {
-	err  error
-	data []byte
-	key  string
+	err          error
+	data         []byte
+	key          string
+	isPayloadErr bool
 }
 
 func (f *fakePipelineInferer) Infer(ctx context.Context, resourceName string, isModel bool, data []byte, headers []kafka.Header, requestId string) (*Request, error) {
 	if f.err != nil {
 		return nil, f.err
 	} else {
-		return &Request{key: f.key, response: f.data}, nil
+		return &Request{key: f.key, response: f.data, isError: f.isPayloadErr}, nil
 	}
 }
 
@@ -72,12 +73,13 @@ func TestHttpServer(t *testing.T) {
 	g := NewGomegaWithT(t)
 
 	type test struct {
-		name       string
-		path       string
-		header     string
-		req        string
-		res        *v2.ModelInferResponse
-		statusCode int
+		name         string
+		path         string
+		header       string
+		req          string
+		res          *v2.ModelInferResponse
+		isPayloadErr bool
+		statusCode   int
 	}
 	tests := []test{
 		{
@@ -102,6 +104,15 @@ func TestHttpServer(t *testing.T) {
 			},
 			statusCode: http.StatusOK,
 		},
+		{
+			name:         "payload error",
+			path:         "/v2/models/foo/infer",
+			header:       "foo",
+			req:          `{"inputs":[{"name":"input1","datatype":"BOOL","shape":[500],"data":[true,false,true,false,true]}]}`,
+			res:          &v2.ModelInferResponse{},
+			isPayloadErr: true,
+			statusCode:   http.StatusBadRequest,
+		},
 		{
 			name:       "wrong path",
 			path:       "/foo",
@@ -144,9 +155,10 @@ func TestHttpServer(t *testing.T) {
 			b, err := proto.Marshal(test.res)
 			g.Expect(err).To(BeNil())
 			mockInferer := &fakePipelineInferer{
-				err:  nil,
-				data: b,
-				key:  testRequestId,
+				err:          nil,
+				data:         b,
+				key:          testRequestId,
+				isPayloadErr: test.isPayloadErr,
 			}
 			httpServer.gateway = mockInferer
 			inferV2Path := test.path

scheduler/pkg/kafka/pipeline/kafkamanager.go

@@ -208,9 +208,6 @@ func (km *KafkaManager) Infer(ctx context.Context, resourceName string, isModel
 	logger.Debugf("Waiting for response for key %s", key)
 	request.wg.Wait()
 	logger.Debugf("Got response for key %s", key)
-	if request.isError {
-		return nil, fmt.Errorf("%s", string(request.response))
-	}
 	return request, nil
 }