Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add Seldon PUID Header to gRPC context + tests #1790

Merged
merged 6 commits into from
May 19, 2020
Merged

Add Seldon PUID Header to gRPC context + tests #1790

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
1dccd80
Add Seldon PUID Header to gRPC context
May 1, 2020
33f0e82
Merge branch 'PUID' of git://github.com/glindsell/seldon-core into gl…
May 4, 2020
0b8dad6
Add tests for nil PUID error to predictor
May 4, 2020
60a02af
Change PUID to lowercase for gRPC
May 5, 2020
5e03400
Return grpc metadata type to use get and set methods
May 6, 2020
1d3ca47
PUID typo and error constant fix
May 18, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 17 additions & 5 deletions executor/api/grpc/seldon/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,14 +2,15 @@ package seldon

import (
"context"
"net/url"

"github.com/go-logr/logr"
"github.com/seldonio/seldon-core/executor/api/client"
"github.com/seldonio/seldon-core/executor/api/grpc"
"github.com/seldonio/seldon-core/executor/api/grpc/seldon/proto"
"github.com/seldonio/seldon-core/executor/api/payload"
"github.com/seldonio/seldon-core/executor/predictor"
"github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
"net/url"
v1 "github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
)

Expand All @@ -32,13 +33,16 @@ func NewGrpcSeldonServer(predictor *v1.PredictorSpec, client client.SeldonApiCli
}

func (g GrpcSeldonServer) Predict(ctx context.Context, req *proto.SeldonMessage) (*proto.SeldonMessage, error) {
seldonPredictorProcess := predictor.NewPredictorProcess(ctx, g.Client, logf.Log.WithName("SeldonMessageRestClient"), g.ServerUrl, g.Namespace, grpc.CollectMetadata(ctx))
md := grpc.CollectMetadata(ctx)
ctx = context.WithValue(ctx, payload.SeldonPUIDHeader, md.Get(payload.SeldonPUIDHeader)[0])
seldonPredictorProcess := predictor.NewPredictorProcess(ctx, g.Client, logf.Log.WithName("SeldonMessageRestClient"), g.ServerUrl, g.Namespace, md)
reqPayload := payload.ProtoPayload{Msg: req}
resPayload, err := seldonPredictorProcess.Predict(g.predictor.Graph, &reqPayload)
if err != nil {
g.Log.Error(err, "Failed to call predict")
return payloadToMessage(resPayload), err
}
return resPayload.GetPayload().(*proto.SeldonMessage), err
return payloadToMessage(resPayload), nil
}

func (g GrpcSeldonServer) SendFeedback(ctx context.Context, req *proto.Feedback) (*proto.SeldonMessage, error) {
Expand All @@ -47,6 +51,14 @@ func (g GrpcSeldonServer) SendFeedback(ctx context.Context, req *proto.Feedback)
resPayload, err := seldonPredictorProcess.Feedback(g.predictor.Graph, &reqPayload)
if err != nil {
g.Log.Error(err, "Failed to call feedback")
return payloadToMessage(resPayload), err
}
return payloadToMessage(resPayload), nil
}

func payloadToMessage(p payload.SeldonPayload) *proto.SeldonMessage {
if m, ok := p.GetPayload().(*proto.SeldonMessage); ok {
return m
}
return resPayload.GetPayload().(*proto.SeldonMessage), err
return nil
}
9 changes: 6 additions & 3 deletions executor/api/grpc/seldon/server_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,18 @@ package seldon

import (
"context"
"net/url"
"testing"

"github.com/golang/protobuf/jsonpb"
. "github.com/onsi/gomega"
"github.com/seldonio/seldon-core/executor/api/grpc/seldon/proto"
"github.com/seldonio/seldon-core/executor/api/test"
"github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
"net/url"
"testing"
v1 "github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
)

const testSeldonPuid = "1"

func TestPredict(t *testing.T) {
t.Logf("Started")
g := NewGomegaWithT(t)
Expand Down
21 changes: 10 additions & 11 deletions executor/api/grpc/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,19 +2,20 @@ package grpc

import (
"context"
"math"
"strconv"

"github.com/go-logr/logr"
guuid "github.com/google/uuid"
grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
"github.com/grpc-ecosystem/go-grpc-middleware/tracing/opentracing"
grpc_opentracing "github.com/grpc-ecosystem/go-grpc-middleware/tracing/opentracing"
"github.com/opentracing/opentracing-go"
"github.com/seldonio/seldon-core/executor/api/metric"
"github.com/seldonio/seldon-core/executor/api/payload"
"github.com/seldonio/seldon-core/executor/k8s"
v1 "github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
"math"
"strconv"
)

const (
Expand Down Expand Up @@ -63,15 +64,13 @@ func CreateGrpcServer(spec *v1.PredictorSpec, deploymentName string, annotations
return grpcServer, nil
}

func CollectMetadata(ctx context.Context) map[string][]string {
md, ok := metadata.FromIncomingContext(ctx)
if ok {
val := md.Get(payload.SeldonPUIDHeader)
func CollectMetadata(ctx context.Context) metadata.MD {
if mdFromIncoming, ok := metadata.FromIncomingContext(ctx); ok {
val := mdFromIncoming.Get(payload.SeldonPUIDHeader)
if len(val) == 0 {
md.Set(payload.SeldonPUIDHeader, guuid.New().String())
mdFromIncoming.Set(payload.SeldonPUIDHeader, guuid.New().String())
}
return md
} else {
return map[string][]string{payload.SeldonPUIDHeader: []string{guuid.New().String()}}
return mdFromIncoming
}
return metadata.New(map[string]string{payload.SeldonPUIDHeader: guuid.New().String()})
}
14 changes: 7 additions & 7 deletions executor/api/grpc/server_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,11 @@ package grpc

import (
"context"
"testing"

. "github.com/onsi/gomega"
"github.com/seldonio/seldon-core/executor/api/payload"
"google.golang.org/grpc/metadata"
"strings"
"testing"
)

func TestAddPuid(t *testing.T) {
Expand All @@ -15,16 +15,16 @@ func TestAddPuid(t *testing.T) {
ctx := context.Background()
meta := CollectMetadata(ctx)

g.Expect(meta[payload.SeldonPUIDHeader]).NotTo(BeNil())
g.Expect(meta.Get(payload.SeldonPUIDHeader)).NotTo(BeNil())
}

func TestExistingPuid(t *testing.T) {
g := NewGomegaWithT(t)
guid := "1"
puid := "1"

ctx := metadata.NewIncomingContext(context.TODO(), metadata.New(map[string]string{payload.SeldonPUIDHeader: guid}))
ctx := metadata.NewIncomingContext(context.TODO(), metadata.New(map[string]string{payload.SeldonPUIDHeader: puid}))
meta := CollectMetadata(ctx)

g.Expect(meta[strings.ToLower(payload.SeldonPUIDHeader)]).NotTo(BeNil())
g.Expect(meta[strings.ToLower(payload.SeldonPUIDHeader)][0]).To(Equal(guid))
g.Expect(meta.Get(payload.SeldonPUIDHeader)).NotTo(BeNil())
g.Expect(meta.Get(payload.SeldonPUIDHeader)[0]).To(Equal(puid))
}
9 changes: 6 additions & 3 deletions executor/api/grpc/tensorflow/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@ package tensorflow

import (
"context"
"net/url"

"github.com/go-logr/logr"
"github.com/golang/protobuf/proto"
"github.com/pkg/errors"
Expand All @@ -10,8 +12,7 @@ import (
"github.com/seldonio/seldon-core/executor/api/payload"
"github.com/seldonio/seldon-core/executor/predictor"
"github.com/seldonio/seldon-core/executor/proto/tensorflow/serving"
"github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
"net/url"
v1 "github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
)

Expand All @@ -34,7 +35,9 @@ func NewGrpcTensorflowServer(predictor *v1.PredictorSpec, client client.SeldonAp
}

func (g *GrpcTensorflowServer) execute(ctx context.Context, req proto.Message, method string) (payload.SeldonPayload, error) {
seldonPredictorProcess := predictor.NewPredictorProcess(ctx, g.Client, logf.Log.WithName(method), g.ServerUrl, g.Namespace, grpc.CollectMetadata(ctx))
md := grpc.CollectMetadata(ctx)
ctx = context.WithValue(ctx, payload.SeldonPUIDHeader, md.Get(payload.SeldonPUIDHeader)[0])
seldonPredictorProcess := predictor.NewPredictorProcess(ctx, g.Client, logf.Log.WithName(method), g.ServerUrl, g.Namespace, md)
reqPayload := payload.ProtoPayload{Msg: req}
return seldonPredictorProcess.Predict(g.predictor.Graph, &reqPayload)
}
Expand Down
9 changes: 5 additions & 4 deletions executor/api/rest/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,10 @@ package rest
import (
"context"
"fmt"
"io/ioutil"
"net/http"
"net/url"

"github.com/go-logr/logr"
guuid "github.com/google/uuid"
"github.com/gorilla/mux"
Expand All @@ -15,10 +19,7 @@ import (
"github.com/seldonio/seldon-core/executor/api/metric"
"github.com/seldonio/seldon-core/executor/api/payload"
"github.com/seldonio/seldon-core/executor/predictor"
"github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
"io/ioutil"
"net/http"
"net/url"
v1 "github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
)

Expand Down
16 changes: 8 additions & 8 deletions executor/api/rest/server_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,14 @@
package rest

import (
"io/ioutil"
"net/http"
"net/http/httptest"
"net/url"
"strconv"
"strings"
"testing"

guuid "github.com/google/uuid"
. "github.com/onsi/gomega"
"github.com/prometheus/common/expfmt"
Expand All @@ -9,13 +17,6 @@ import (
"github.com/seldonio/seldon-core/executor/api/payload"
"github.com/seldonio/seldon-core/executor/api/test"
v1 "github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
"io/ioutil"
"net/http"
"net/http/httptest"
"net/url"
"strconv"
"strings"
"testing"
)

const (
Expand Down Expand Up @@ -266,7 +267,6 @@ func TestModelWithServer(t *testing.T) {
r.Router.ServeHTTP(res, req)
g.Expect(res.Code).To(Equal(200))
g.Expect(called).To(Equal(true))

}

func TestServerMetrics(t *testing.T) {
Expand Down
36 changes: 28 additions & 8 deletions executor/predictor/predictor_process.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,16 +3,19 @@ package predictor
import (
"context"
"fmt"
"net/url"
"sync"

"github.com/go-logr/logr"
guuid "github.com/google/uuid"
"github.com/seldonio/seldon-core/executor/api/client"
"github.com/seldonio/seldon-core/executor/api/payload"
payloadLogger "github.com/seldonio/seldon-core/executor/logger"
"github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
"net/url"
"sync"
v1 "github.com/seldonio/seldon-core/operator/apis/machinelearning.seldon.io/v1"
)

const NilPUIDError = "context value for Seldon PUID Header is nil"

type PredictorProcess struct {
Ctx context.Context
Client client.SeldonApiClient
Expand Down Expand Up @@ -242,7 +245,7 @@ func (p *PredictorProcess) getLogUrl(logger *v1.Logger) (*url.URL, error) {
}
}

func (p *PredictorProcess) logPayload(nodeName string, logger *v1.Logger, reqType payloadLogger.LogRequestType, msg payload.SeldonPayload) error {
func (p *PredictorProcess) logPayload(nodeName string, logger *v1.Logger, reqType payloadLogger.LogRequestType, msg payload.SeldonPayload, puid string) error {
data, err := msg.GetBytes()
if err != nil {
return err
Expand All @@ -251,7 +254,6 @@ func (p *PredictorProcess) logPayload(nodeName string, logger *v1.Logger, reqTyp
if err != nil {
return err
}

payloadLogger.QueueLogRequest(payloadLogger.LogRequest{
Url: logUrl,
Bytes: &data,
Expand All @@ -260,15 +262,30 @@ func (p *PredictorProcess) logPayload(nodeName string, logger *v1.Logger, reqTyp
Id: guuid.New().String(),
SourceUri: p.ServerUrl,
ModelId: nodeName,
RequestId: p.Ctx.Value(payload.SeldonPUIDHeader).(string),
RequestId: puid,
})
return nil
}

func (p *PredictorProcess) getPUIDHeader() (string, error) {
// Check request ID is not nil
if puid, ok := p.Ctx.Value(payload.SeldonPUIDHeader).(string); ok {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Will this not still panic if Value return nil? Should we not change for its existence before returning the string?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Interestingly not...

There's a similar question on this here.

Using the "comma-ok" form:
The value of ok is true if the assertion holds. Otherwise it is false and the value of n is the zero value for type T. No run-time panic occurs in this case.

return puid, nil
}
return "", fmt.Errorf(NilPUIDError)
}

func (p *PredictorProcess) Predict(node *v1.PredictiveUnit, msg payload.SeldonPayload) (payload.SeldonPayload, error) {
puid, err := p.getPUIDHeader()
if err != nil {
return nil, err
}
//Log Request
if node.Logger != nil && (node.Logger.Mode == v1.LogRequest || node.Logger.Mode == v1.LogAll) {
p.logPayload(node.Name, node.Logger, payloadLogger.InferenceRequest, msg)
err := p.logPayload(node.Name, node.Logger, payloadLogger.InferenceRequest, msg, puid)
if err != nil {
return nil, err
}
}
tmsg, err := p.transformInput(node, msg)
if err != nil {
Expand All @@ -281,7 +298,10 @@ func (p *PredictorProcess) Predict(node *v1.PredictiveUnit, msg payload.SeldonPa
response, err := p.transformOutput(node, cmsg)
// Log Response
if err == nil && node.Logger != nil && (node.Logger.Mode == v1.LogResponse || node.Logger.Mode == v1.LogAll) {
p.logPayload(node.Name, node.Logger, payloadLogger.InferenceResponse, response)
err := p.logPayload(node.Name, node.Logger, payloadLogger.InferenceResponse, response, puid)
if err != nil {
return nil, err
}
}
return response, err
}
Expand Down
Loading