Pipeline Experiments (#360)

* initial commit

* lint

* Allow sticky sessions for pipelines

* lint and missing artifacts

* lint

* add docs

* review fixes

* ensure headers are returned by models but some are not transferred between topics

* update to custom mlserver image

* review fixes

apis/mlops/scheduler/scheduler.proto

@@ -197,26 +197,32 @@ message StartExperimentRequest {
   Experiment experiment = 1;
 }
 
+enum ResourceType {
+  MODEL = 0;
+  PIPELINE = 1;
+}
+
 message Experiment {
   string name = 1;
-  optional string defaultModel = 2;
+  optional string default = 2;
   repeated ExperimentCandidate candidates = 3;
   optional ExperimentMirror mirror = 4;
   optional ExperimentConfig config = 5;
   optional KubernetesMeta kubernetesMeta = 6;
+  ResourceType resourceType = 7;
 }
 
 message ExperimentConfig {
   bool stickySessions = 1;
 }
 
 message ExperimentCandidate {
-  string modelName = 1;
+  string name = 1;
   uint32 weight = 2;
 }
 
 message ExperimentMirror {
-  string modelName = 1;
+  string name = 1;
   uint32 percent = 2;
 }
 

docs/source/contents/examples/index.md

@@ -3,6 +3,7 @@
 This section will provide some examples to allow operations with Seldon to be tested so you can run your own models, experiments, pipelines and explainers.
 
  * [Local examples](local-examples.md)
+ * [Local experiments](local-experiments.md)
  * [Kubernetes examples](k8s-examples.md)
  * [Pipeline examples](pipeline-examples.md)
  * [Explainer examples](explainer-examples.md)  
@@ -16,6 +17,7 @@ This section will provide some examples to allow operations with Seldon to be te
 :hidden:
 
 local-examples.md
+local-experiments.md
 k8s-examples.md
 pipeline-examples.md
 cifar10.md

docs/source/contents/examples/local-experiments.md

@@ -0,0 +1,8 @@
+# Local Experiments
+
+Run these examples from the `samples` folder.
+
+
+```{include} ../../../../samples/local-experiments.md
+```
+

docs/source/contents/kubernetes/resources/experiment/index.md

@@ -1,6 +1,6 @@
 # Experiment
 
-An Experiment defines a traffic split between Models. This allows new versions of models to be tested and experiments between models to be run.
+An Experiment defines a traffic split between Models or Pipelines. This allows new versions of models and pipelines to be tested.
 
 An experiment spec has three sections:
 
@@ -32,15 +32,26 @@ curl http://${MESH_IP}/v2/models/experiment-iris/infer \
    -d '{"inputs": [{"name": "predict", "shape": [1, 4], "datatype": "FP32", "data": [[1, 2, 3, 4]]}]}'
 ```
 
+For examples see the [local experiments notebook](../../../examples/local-experiments.md).
+
+## Pipeline Experiments
+
+Running an experiment between some pipelines is very similar. The difference is `resourceType: pipeline` needs to be defined and in this case the candidates or mirrors will refer to pipelines. An example is shown below:
+
+```{literalinclude} ../../../../../../samples/experiments/addmul10.yaml 
+:language: yaml
+```
+For an example see the [local experiments notebook](../../../examples/local-experiments.md).
+
 ## Sticky Sessions
 
-To allow cohorts to get consistent views in an experiment each inference request passes back a response header `seldon-route` which can be passed in future requests to an experiment to bypass the random traffic splits and get a prediction from the same model used in the initial request.
+To allow cohorts to get consistent views in an experiment each inference request passes back a response header `x-seldon-route` which can be passed in future requests to an experiment to bypass the random traffic splits and get a prediction from the sequence of models and pipelines used in the initial request.
 
-This is illustrated in the [local examples notebook](../../../examples/local-examples.md).
+This is illustrated in the [local experiments notebook](../../../examples/local-experiments.md).
 
 Caveats:
 
-  * Note this is the same model but not necessarily the same replica. This means at present this will not work for stateful models that need to go to the same model replica.
+  * Note the models used will be the same but not necessarily the same replica instances. This means at present this will not work for stateful models that need to go to the same model replica instance.
 
 ## Service Meshes
 

k8s/Makefile

@@ -12,7 +12,7 @@ PIPELINEGATEWAY_IMG ?= ${DOCKERHUB_USERNAME}/seldon-pipelinegateway:${CUSTOM_IMA
 RCLONE_IMG ?= ${DOCKERHUB_USERNAME}/seldon-rclone:${CUSTOM_IMAGE_TAG}
 SCHEDULER_IMG ?= ${DOCKERHUB_USERNAME}/seldon-scheduler:${CUSTOM_IMAGE_TAG}
 
-MLSERVER_IMG ?= cliveseldon/mlserver:1.1.0.explain
+MLSERVER_IMG ?= cliveseldon/mlserver:1.2.0.dev1
 TRITON_IMG ?= nvcr.io/nvidia/tritonserver:22.05-py3
 
 .PHONY: create

k8s/helm-charts/seldon-core-v2-setup/values.yaml

@@ -104,7 +104,7 @@ serverConfig:
       pullPolicy: IfNotPresent
       registry: docker.io
       repository: cliveseldon/mlserver
-      tag: 1.1.0.explain
+      tag: 1.2.0.dev1
     serverCapabilities: "mlserver,alibi-detect,lightgbm,mlflow,python,sklearn,spark-mlib,xgboost"
     overcommitPercentage: "10"
     modelVolumeStorage: 1Gi

k8s/helm-charts/seldon-core-v2-setup/values.yaml.template

@@ -104,8 +104,8 @@ serverConfig:
     image:
       pullPolicy: IfNotPresent
       registry: docker.io
-      repository: seldonio/mlserver
-      tag: 1.1.0
+      repository: cliveseldon/mlserver
+      tag: 1.2.0.dev1
     serverCapabilities: "mlserver,alibi-detect,lightgbm,mlflow,python,sklearn,spark-mlib,xgboost"
     overcommitPercentage: "10"
     modelVolumeStorage: 1Gi

k8s/yaml/seldon-v2-components.yaml

@@ -1043,7 +1043,7 @@ spec:
         value: "1"
       - name: MLSERVER_LOAD_MODELS_AT_STARTUP
         value: "false"
-      image: cliveseldon/mlserver:1.1.0.explain
+      image: cliveseldon/mlserver:1.2.0.dev1
       imagePullPolicy: IfNotPresent
       lifecycle:
         preStop:

operator/apis/mlops/v1alpha1/experiment_types.go

@@ -27,21 +27,29 @@ import (
 // EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
 // NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.
 
+type ResourceType string
+
+const (
+	ModelResourceType    ResourceType = "model"
+	PipelineResourceType ResourceType = "pipeline"
+)
+
 // ExperimentSpec defines the desired state of Experiment
 type ExperimentSpec struct {
-	DefaultModel *string               `json:"defaultModel,omitempty"`
+	Default      *string               `json:"default,omitempty"`
 	Candidates   []ExperimentCandidate `json:"candidates"`
 	Mirror       *ExperimentMirror     `json:"mirror,omitempty"`
+	ResourceType ResourceType          `json:"resourceType,omitempty"`
 }
 
 type ExperimentCandidate struct {
-	ModelName string `json:"modelName"`
-	Weight    uint32 `json:"weight"`
+	Name   string `json:"modelName"`
+	Weight uint32 `json:"weight"`
 }
 
 type ExperimentMirror struct {
-	ModelName string `json:"model_name"`
-	Percent   uint32 `json:"percent"`
+	Name    string `json:"model_name"`
+	Percent uint32 `json:"percent"`
 }
 
 // ExperimentStatus defines the observed state of Experiment
@@ -77,8 +85,8 @@ func init() {
 
 func asSchedulerCandidate(candidate *ExperimentCandidate) *scheduler.ExperimentCandidate {
 	return &scheduler.ExperimentCandidate{
-		ModelName: candidate.ModelName,
-		Weight:    candidate.Weight,
+		Name:   candidate.Name,
+		Weight: candidate.Weight,
 	}
 }
 
@@ -90,19 +98,27 @@ func (e *Experiment) AsSchedulerExperimentRequest() *scheduler.Experiment {
 	}
 	if e.Spec.Mirror != nil {
 		mirror = &scheduler.ExperimentMirror{
-			ModelName: e.Spec.Mirror.ModelName,
-			Percent:   e.Spec.Mirror.Percent,
+			Name:    e.Spec.Mirror.Name,
+			Percent: e.Spec.Mirror.Percent,
 		}
 	}
+	var resourceType scheduler.ResourceType
+	switch e.Spec.ResourceType {
+	case PipelineResourceType:
+		resourceType = scheduler.ResourceType_PIPELINE
+	default:
+		resourceType = scheduler.ResourceType_MODEL
+	}
 	return &scheduler.Experiment{
-		Name:         e.Name,
-		DefaultModel: e.Spec.DefaultModel,
-		Candidates:   candidates,
-		Mirror:       mirror,
+		Name:       e.Name,
+		Default:    e.Spec.Default,
+		Candidates: candidates,
+		Mirror:     mirror,
 		KubernetesMeta: &scheduler.KubernetesMeta{
 			Namespace:  e.Namespace,
 			Generation: e.Generation,
 		},
+		ResourceType: resourceType,
 	}
 }
 

operator/apis/mlops/v1alpha1/experiment_types_test.go

@@ -19,47 +19,99 @@ func TestAsSchedulerExperimentRequest(t *testing.T) {
 	getStrPtr := func(val string) *string { return &val }
 	tests := []test{
 		{
-			name: "basic",
+			name: "model",
 			experiment: &Experiment{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:       "foo",
 					Namespace:  "default",
 					Generation: 1,
 				},
 				Spec: ExperimentSpec{
-					DefaultModel: getStrPtr("model1"),
+					Default: getStrPtr("model1"),
 					Candidates: []ExperimentCandidate{
 						{
-							ModelName: "model1",
-							Weight:    20,
+							Name:   "model1",
+							Weight: 20,
 						},
 						{
-							ModelName: "model2",
-							Weight:    30,
+							Name:   "model2",
+							Weight: 30,
 						},
 					},
 					Mirror: &ExperimentMirror{
-						ModelName: "model4",
-						Percent:   40,
+						Name:    "model4",
+						Percent: 40,
 					},
 				},
 			},
 			proto: &scheduler.Experiment{
 				Name:         "foo",
-				DefaultModel: getStrPtr("model1"),
+				Default:      getStrPtr("model1"),
+				ResourceType: scheduler.ResourceType_MODEL,
 				Candidates: []*scheduler.ExperimentCandidate{
 					{
-						ModelName: "model1",
-						Weight:    20,
+						Name:   "model1",
+						Weight: 20,
 					},
 					{
-						ModelName: "model2",
-						Weight:    30,
+						Name:   "model2",
+						Weight: 30,
 					},
 				},
 				Mirror: &scheduler.ExperimentMirror{
-					ModelName: "model4",
-					Percent:   40,
+					Name:    "model4",
+					Percent: 40,
+				},
+				KubernetesMeta: &scheduler.KubernetesMeta{
+					Namespace:  "default",
+					Generation: 1,
+				},
+			},
+		},
+		{
+			name: "pipeline",
+			experiment: &Experiment{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:       "foo",
+					Namespace:  "default",
+					Generation: 1,
+				},
+				Spec: ExperimentSpec{
+					Default:      getStrPtr("pipeline1"),
+					ResourceType: PipelineResourceType,
+					Candidates: []ExperimentCandidate{
+						{
+							Name:   "pipeline1",
+							Weight: 20,
+						},
+						{
+							Name:   "pipeline2",
+							Weight: 30,
+						},
+					},
+					Mirror: &ExperimentMirror{
+						Name:    "pipeline4",
+						Percent: 40,
+					},
+				},
+			},
+			proto: &scheduler.Experiment{
+				Name:         "foo",
+				Default:      getStrPtr("pipeline1"),
+				ResourceType: scheduler.ResourceType_PIPELINE,
+				Candidates: []*scheduler.ExperimentCandidate{
+					{
+						Name:   "pipeline1",
+						Weight: 20,
+					},
+					{
+						Name:   "pipeline2",
+						Weight: 30,
+					},
+				},
+				Mirror: &scheduler.ExperimentMirror{
+					Name:    "pipeline4",
+					Percent: 40,
 				},
 				KubernetesMeta: &scheduler.KubernetesMeta{
 					Namespace:  "default",

operator/cmd/seldon/cli/pipeline_infer.go

@@ -28,6 +28,10 @@ func createPipelineInfer() *cobra.Command {
 			if err != nil {
 				return err
 			}
+			stickySesion, err := cmd.Flags().GetBool(stickySessionFlag)
+			if err != nil {
+				return err
+			}
 			showResponse, err := cmd.Flags().GetBool(showResponseFlag)
 			if err != nil {
 				return err
@@ -60,11 +64,12 @@ func createPipelineInfer() *cobra.Command {
 				return fmt.Errorf("required inline data or from file with -f <file-path>")
 			}
 
-			err = inferenceClient.Infer(pipelineName, inferMode, data, showRequest, showResponse, iterations, cli.InferPipeline, showHeaders, headers, false)
+			err = inferenceClient.Infer(pipelineName, inferMode, data, showRequest, showResponse, iterations, cli.InferPipeline, showHeaders, headers, stickySesion)
 			return err
 		},
 	}
 	cmdPipelineInfer.Flags().StringP(fileFlag, "f", "", "inference payload file")
+	cmdPipelineInfer.Flags().BoolP(stickySessionFlag, "s", false, "use sticky session from last infer (only works with inference to experiments)")
 	cmdPipelineInfer.Flags().String(inferenceHostFlag, env.GetString(EnvInfer, DefaultInferHost), "seldon inference host")
 	cmdPipelineInfer.Flags().String(inferenceModeFlag, "rest", "inference mode rest or grpc")
 	cmdPipelineInfer.Flags().IntP(inferenceIterationsFlag, "i", 1, "inference iterations")

operator/config/serverconfigs/kustomization.yaml

@@ -15,7 +15,7 @@ images:
   newTag: latest
 - name: mlserver
   newName: cliveseldon/mlserver
-  newTag: 1.1.0.explain
+  newTag: 1.2.0.dev1
 - name: rclone
   newName: seldonio/seldon-rclone
   newTag: latest