Repository: SeldonIO/seldon-core (language: python)

Issue:
Allow rclone init container for custom servers
Allow rclone init containers be configured when a user specifies a modelUri and this is not a prepackaed server.


Discussion / hints:

Fix patch:
diff --git a/notebooks/protocol_examples.ipynb b/notebooks/protocol_examples.ipynb
index 0ef1e69585..5851cfdbba 100644
--- a/notebooks/protocol_examples.ipynb
+++ b/notebooks/protocol_examples.ipynb
@@ -96,7 +96,7 @@
     {
      "data": {
       "text/plain": [
-       "'1.16.0-dev'"
+       "'1.17.0-dev'"
       ]
      },
      "execution_count": 5,
@@ -186,14 +186,16 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 9,
-   "metadata": {},
+   "execution_count": 10,
+   "metadata": {
+    "scrolled": true
+   },
    "outputs": [
     {
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "{'data': {'names': ['proba'], 'ndarray': [[0.43782349911420193]]}, 'meta': {'requestPath': {'classifier': 'seldonio/mock_classifier:1.16.0-dev'}}}\n"
+      "{'data': {'names': ['proba'], 'ndarray': [[0.43782349911420193]]}, 'meta': {'requestPath': {'classifier': 'seldonio/mock_classifier:1.17.0-dev'}}}\n"
      ]
     }
    ],
@@ -208,14 +210,14 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 10,
+   "execution_count": 11,
    "metadata": {},
    "outputs": [
     {
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "{'meta': {'requestPath': {'classifier': 'seldonio/mock_classifier:1.16.0-dev'}}, 'data': {'names': ['proba'], 'ndarray': [[0.43782349911420193]]}}\n"
+      "{'meta': {'requestPath': {'classifier': 'seldonio/mock_classifier:1.17.0-dev'}}, 'data': {'names': ['proba'], 'ndarray': [[0.43782349911420193]]}}\n"
      ]
     }
    ],
@@ -231,7 +233,143 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 11,
+   "execution_count": 12,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "seldondeployment.machinelearning.seldon.io \"example-seldon\" deleted\r\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl delete -f resources/model_seldon.yaml"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Seldon protocol Model with ModelUri with two custom models"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 26,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "%%writetemplate resources/model_seldon.yaml\n",
+    "apiVersion: machinelearning.seldon.io/v1\n",
+    "kind: SeldonDeployment\n",
+    "metadata:\n",
+    "  name: example-seldon\n",
+    "spec:\n",
+    "  protocol: seldon\n",
+    "  predictors:\n",
+    "  - componentSpecs:\n",
+    "    - spec:\n",
+    "        containers:\n",
+    "        - image: seldonio/mock_classifier:{VERSION}\n",
+    "          name: classifier\n",
+    "        - image: seldonio/mock_classifier:{VERSION}\n",
+    "          name: classifier2\n",
+    "    graph:\n",
+    "      name: classifier\n",
+    "      type: MODEL\n",
+    "      modelUri: gs://seldon-models/v1.17.0-dev/sklearn/iris\n",
+    "      children:\n",
+    "      - name: classifier2\n",
+    "        type: MODEL\n",
+    "        modelUri: gs://seldon-models/v1.17.0-dev/sklearn/iris\n",
+    "    name: model\n",
+    "    replicas: 1"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 27,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "seldondeployment.machinelearning.seldon.io/example-seldon unchanged\r\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl apply -f resources/model_seldon.yaml"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 28,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "seldondeployment.machinelearning.seldon.io/example-seldon condition met\r\n"
+     ]
+    }
+   ],
+   "source": [
+    "!kubectl wait --for condition=ready --timeout=300s sdep --all -n seldon"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 30,
+   "metadata": {
+    "scrolled": true
+   },
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "{'data': {'names': ['proba'], 'ndarray': [[0.07735472603574542]]}, 'meta': {'requestPath': {'classifier': 'seldonio/mock_classifier:1.17.0-dev', 'classifier2': 'seldonio/mock_classifier:1.17.0-dev'}}}\n"
+     ]
+    }
+   ],
+   "source": [
+    "X=!curl -s -d '{\"data\": {\"ndarray\":[[1.0, 2.0, 5.0]]}}' \\\n",
+    "   -X POST http://localhost:8003/seldon/seldon/example-seldon/api/v1.0/predictions \\\n",
+    "   -H \"Content-Type: application/json\"\n",
+    "d=json.loads(X[0])\n",
+    "print(d)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 31,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "{'meta': {'requestPath': {'classifier': 'seldonio/mock_classifier:1.17.0-dev', 'classifier2': 'seldonio/mock_classifier:1.17.0-dev'}}, 'data': {'names': ['proba'], 'ndarray': [[0.07735472603574542]]}}\n"
+     ]
+    }
+   ],
+   "source": [
+    "X=!cd ../executor/proto && grpcurl -d '{\"data\":{\"ndarray\":[[1.0,2.0,5.0]]}}' \\\n",
+    "         -rpc-header seldon:example-seldon -rpc-header namespace:seldon \\\n",
+    "         -plaintext \\\n",
+    "         -proto ./prediction.proto  0.0.0.0:8003 seldon.protos.Seldon/Predict\n",
+    "d=json.loads(\"\".join(X))\n",
+    "print(d)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 32,
    "metadata": {},
    "outputs": [
     {
diff --git a/operator/controllers/seldondeployment_controller.go b/operator/controllers/seldondeployment_controller.go
index a1bf4f1b15..852e1dd212 100644
--- a/operator/controllers/seldondeployment_controller.go
+++ b/operator/controllers/seldondeployment_controller.go
@@ -584,7 +584,7 @@ func (r *SeldonDeploymentReconciler) createComponents(ctx context.Context, mlDep
 		}
 
 		pi := NewPrePackedInitializer(ctx, r.ClientSet)
-		err = pi.createStandaloneModelServers(mlDep, &p, &c, &p.Graph, securityContext)
+		err = pi.addModelServersAndInitContainers(mlDep, &p, &c, &p.Graph, securityContext, log)
 		if err != nil {
 			return nil, err
 		}
diff --git a/operator/controllers/seldondeployment_prepackaged_servers.go b/operator/controllers/seldondeployment_prepackaged_servers.go
index cadb885b85..6d98abd047 100644
--- a/operator/controllers/seldondeployment_prepackaged_servers.go
+++ b/operator/controllers/seldondeployment_prepackaged_servers.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/go-logr/logr"
 	"strconv"
 	"strings"
 
@@ -401,25 +402,33 @@ func SetUriParamsForTFServingProxyContainer(pu *machinelearningv1.PredictiveUnit
 	}
 }
 
-func (pi *PrePackedInitialiser) createStandaloneModelServers(mlDep *machinelearningv1.SeldonDeployment, p *machinelearningv1.PredictorSpec, c *components, pu *machinelearningv1.PredictiveUnit, podSecurityContext *v1.PodSecurityContext) error {
+func (pi *PrePackedInitialiser) findDeployment(c *components, depName string) (*appsv1.Deployment, bool, error) {
+	for i := 0; i < len(c.deployments); i++ {
+		d := c.deployments[i]
+		if strings.Compare(d.Name, depName) == 0 {
+			return d, true, nil
+		}
+	}
+	return nil, false, nil
+}
 
+func (pi *PrePackedInitialiser) addModelServersAndInitContainers(mlDep *machinelearningv1.SeldonDeployment,
+	p *machinelearningv1.PredictorSpec,
+	c *components,
+	pu *machinelearningv1.PredictiveUnit,
+	podSecurityContext *v1.PodSecurityContext,
+	log logr.Logger) error {
+
+	sPodSpec, idx := utils.GetSeldonPodSpecForPredictiveUnit(p, pu.Name)
+	if sPodSpec == nil {
+		return fmt.Errorf("Failed to find PodSpec for Prepackaged server PreditiveUnit named %s", pu.Name)
+	}
+	depName := machinelearningv1.GetDeploymentName(mlDep, *p, sPodSpec, idx)
 	if machinelearningv1.IsPrepack(pu) {
-		sPodSpec, idx := utils.GetSeldonPodSpecForPredictiveUnit(p, pu.Name)
-		if sPodSpec == nil {
-			return fmt.Errorf("Failed to find PodSpec for Prepackaged server PreditiveUnit named %s", pu.Name)
-		}
-		depName := machinelearningv1.GetDeploymentName(mlDep, *p, sPodSpec, idx)
-		seldonId := machinelearningv1.GetSeldonDeploymentName(mlDep)
-
-		var deploy *appsv1.Deployment
-		existing := false
-		for i := 0; i < len(c.deployments); i++ {
-			d := c.deployments[i]
-			if strings.Compare(d.Name, depName) == 0 {
-				deploy = d
-				existing = true
-				break
-			}
+
+		deploy, existing, err := pi.findDeployment(c, depName)
+		if err != nil {
+			return err
 		}
 
 		// might not be a Deployment yet - if so we have to create one
@@ -462,7 +471,7 @@ func (pi *PrePackedInitialiser) createStandaloneModelServers(mlDep *machinelearn
 		}
 
 		if !existing {
-
+			seldonId := machinelearningv1.GetSeldonDeploymentName(mlDep)
 			// this is a new deployment so its containers won't have a containerService
 			for k := 0; k < len(deploy.Spec.Template.Spec.Containers); k++ {
 				con := &deploy.Spec.Template.Spec.Containers[k]
@@ -478,10 +487,32 @@ func (pi *PrePackedInitialiser) createStandaloneModelServers(mlDep *machinelearn
 				c.deployments = append(c.deployments, deploy)
 			}
 		}
+	} else {
+		// add model uri initializer for non server components
+		if pu.ModelURI != "" {
+			log.Info("Add rclone init container for predictive unit", "predictive unit", pu.Name)
+			deploy, existing, err := pi.findDeployment(c, depName)
+			if err != nil {
+				return err
+			}
+			if !existing {
+				return fmt.Errorf("Expected to find a deployment for predictive unit %s", pu.Name)
+			}
+			mi := NewModelInitializer(pi.ctx, pi.clientset)
+			c := utils.GetContainerForDeployment(deploy, pu.Name)
+			if c == nil {
+				return fmt.Errorf("Expected to find container for predictive unit %s", pu.Name)
+			}
+			envSecretRefName := extractEnvSecretRefName(pu)
+			_, err = mi.InjectModelInitializer(deploy, c.Name, pu.ModelURI, pu.ServiceAccountName, envSecretRefName, pu.StorageInitializerImage)
+			if err != nil {
+				return err
+			}
+		}
 	}
 
 	for i := 0; i < len(pu.Children); i++ {
-		if err := pi.createStandaloneModelServers(mlDep, p, c, &pu.Children[i], podSecurityContext); err != nil {
+		if err := pi.addModelServersAndInitContainers(mlDep, p, c, &pu.Children[i], podSecurityContext, log); err != nil {
 			return err
 		}
 	}


Test patch:
diff --git a/operator/controllers/seldondeployment_prepackaged_servers_test.go b/operator/controllers/seldondeployment_prepackaged_servers_test.go
index 896f72c91e..24ea4f2271 100644
--- a/operator/controllers/seldondeployment_prepackaged_servers_test.go
+++ b/operator/controllers/seldondeployment_prepackaged_servers_test.go
@@ -2,8 +2,11 @@ package controllers
 
 import (
 	"context"
+	"github.com/go-logr/logr/testr"
+	"k8s.io/client-go/kubernetes/fake"
 	"strconv"
 	"strings"
+	"testing"
 	"time"
 
 	. "github.com/onsi/ginkgo"
@@ -16,6 +19,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
+	crfake "sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 var _ = Describe("Create a prepacked sklearn server.", func() {
@@ -1213,3 +1217,177 @@ var _ = Describe("Create a prepacked triton server with seldon.io/no-storage-ini
 	})
 
 })
+
+func createCustomModelWithUri() (*machinelearningv1.SeldonDeployment,
+	*machinelearningv1.PredictorSpec,
+	*components,
+	*machinelearningv1.PredictiveUnit) {
+	impl := machinelearningv1.SIMPLE_MODEL
+	sdep := &machinelearningv1.SeldonDeployment{
+		Spec: machinelearningv1.SeldonDeploymentSpec{
+			Predictors: []machinelearningv1.PredictorSpec{
+				{
+					Name: "p1",
+					ComponentSpecs: []*machinelearningv1.SeldonPodSpec{
+						{
+							Spec: corev1.PodSpec{
+								Containers: []corev1.Container{
+									{
+										Name: "model1",
+									},
+								},
+							},
+						},
+					},
+					Graph: machinelearningv1.PredictiveUnit{
+						Name:           "model1",
+						ModelURI:       "gs://abc",
+						Implementation: &impl,
+					},
+				},
+			},
+		},
+	}
+	depName := machinelearningv1.GetDeploymentName(sdep, sdep.Spec.Predictors[0], sdep.Spec.Predictors[0].ComponentSpecs[0], 0)
+	c := &components{
+		deployments: []*appsv1.Deployment{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: depName,
+				},
+				Spec: appsv1.DeploymentSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name: "model1",
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	return sdep, &sdep.Spec.Predictors[0], c, &sdep.Spec.Predictors[0].Graph
+}
+
+func createPrePackedServer() (*machinelearningv1.SeldonDeployment,
+	*machinelearningv1.PredictorSpec,
+	*components,
+	*machinelearningv1.PredictiveUnit) {
+	impl := machinelearningv1.PredictiveUnitImplementation("SKLEARN_SERVER")
+	sdep := &machinelearningv1.SeldonDeployment{
+		Spec: machinelearningv1.SeldonDeploymentSpec{
+			Predictors: []machinelearningv1.PredictorSpec{
+				{
+					Name: "p1",
+					ComponentSpecs: []*machinelearningv1.SeldonPodSpec{
+						{
+							Spec: corev1.PodSpec{
+								Containers: []corev1.Container{
+									{
+										Name: "model1",
+									},
+								},
+							},
+						},
+					},
+					Graph: machinelearningv1.PredictiveUnit{
+						Name:           "model1",
+						ModelURI:       "gs://abc",
+						Implementation: &impl,
+					},
+				},
+			},
+		},
+	}
+	depName := machinelearningv1.GetDeploymentName(sdep, sdep.Spec.Predictors[0], sdep.Spec.Predictors[0].ComponentSpecs[0], 0)
+	c := &components{
+		deployments: []*appsv1.Deployment{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: depName,
+				},
+				Spec: appsv1.DeploymentSpec{
+					Template: corev1.PodTemplateSpec{
+						Spec: corev1.PodSpec{
+							Containers: []corev1.Container{
+								{
+									Name: "model1",
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+	return sdep, &sdep.Spec.Predictors[0], c, &sdep.Spec.Predictors[0].Graph
+}
+
+func getNumInitContainers(c *components) int {
+	tot := 0
+	for _, d := range c.deployments {
+		tot = tot + len(d.Spec.Template.Spec.InitContainers)
+	}
+	return tot
+}
+
+func setupTestConfigMap() error {
+	scheme := createScheme()
+	machinelearningv1.C = crfake.NewFakeClientWithScheme(scheme)
+	testConfigMap1 := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      ControllerConfigMapName,
+			Namespace: ControllerNamespace,
+		},
+		Data: configs,
+	}
+	return machinelearningv1.C.Create(context.TODO(), testConfigMap1)
+}
+
+func TestAddModelServersAndInitContainers(t *testing.T) {
+	g := NewGomegaWithT(t)
+	err := setupTestConfigMap()
+	g.Expect(err).To(BeNil())
+	type test struct {
+		name      string
+		error     bool
+		generator func() (
+			*machinelearningv1.SeldonDeployment,
+			*machinelearningv1.PredictorSpec,
+			*components,
+			*machinelearningv1.PredictiveUnit)
+		expectedNumInitContainers int
+	}
+
+	tests := []test{
+		{
+			name:                      "model uri in custom model",
+			generator:                 createCustomModelWithUri,
+			expectedNumInitContainers: 1,
+		},
+		{
+			name:                      "model uri in prepackaged server",
+			generator:                 createPrePackedServer,
+			expectedNumInitContainers: 1,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			sdep, p, c, pu := test.generator()
+			cs := fake.NewSimpleClientset(configMap)
+			pi := NewPrePackedInitializer(context.TODO(), cs)
+			logger := testr.New(t)
+			err := pi.addModelServersAndInitContainers(sdep, p, c, pu, nil, logger)
+			if test.error {
+				g.Expect(err).ToNot(BeNil())
+			} else {
+				g.Expect(err).To(BeNil())
+				g.Expect(getNumInitContainers(c)).To(Equal(test.expectedNumInitContainers))
+			}
+		})
+	}
+}