feat: implement GRPC xDS on merlin standard transformer

api/cluster/resource/templater.go

@@ -463,6 +463,8 @@ func (t *InferenceServiceTemplater) createTransformerSpec(
 			MinReplicas: &(transformer.ResourceRequest.MinReplica),
 			MaxReplicas: transformer.ResourceRequest.MaxReplica,
 			Logger:      loggerSpec,
+			Labels:      t.deploymentConfig.StandardTransformer.DefaultLabels,
+			Annotations: t.deploymentConfig.StandardTransformer.DefaultAnnotations,
 		},
 	}
 

api/config/config.go

@@ -345,6 +345,8 @@ type StandardTransformerConfig struct {
 	Kafka              KafkaConfig           `validate:"required"`
 	// Simulator configs
 	SimulatorFeastClientMaxConcurrentRequests int `validate:"required" default:"100"`
+	DefaultLabels                             map[string]string
+	DefaultAnnotations                        map[string]string
 }
 
 // KafkaConfig configuration for publishing prediction log

api/config/config_test.go

@@ -527,7 +527,9 @@ func TestLoad(t *testing.T) {
 					IndexPath:  "index.html",
 				},
 				StandardTransformerConfig: StandardTransformerConfig{
-					FeastServingURLs: []FeastServingURL{},
+					DefaultLabels:      map[string]string{},
+					DefaultAnnotations: map[string]string{},
+					FeastServingURLs:   []FeastServingURL{},
 					FeastBigtableConfig: &FeastBigtableConfig{
 						IsUsingDirectStorage: true,
 						ServingURL:           "10.1.1.3",

api/go.mod

@@ -206,7 +206,7 @@ require (
 	github.com/sergi/go-diff v1.2.0 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/smartystreets/assertions v1.0.0 // indirect
-	github.com/spf13/afero v1.9.2 // indirect
+	github.com/spf13/afero v1.10.0 // indirect
 	github.com/spf13/cast v1.3.1 // indirect
 	github.com/spf13/cobra v1.8.0 // indirect
 	github.com/spf13/jwalterweatherman v1.1.0 // indirect
@@ -265,11 +265,14 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.13.6 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sts v1.16.19 // indirect
 	github.com/aws/smithy-go v1.20.4 // indirect
+	github.com/cncf/udpa/go v0.0.0-20220112060539-c52dc94e7fbe // indirect
+	github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa // indirect
+	github.com/envoyproxy/go-control-plane v0.12.0 // indirect
+	github.com/envoyproxy/protoc-gen-validate v1.0.4 // indirect
 	golang.org/x/time v0.5.0 // indirect
 	k8s.io/klog/v2 v2.120.1 // indirect
 	k8s.io/kube-openapi v0.0.0-20231113174909-778a5567bc1e // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
-
 )
 
 replace (

api/go.sum

@@ -234,10 +234,14 @@ github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGX
 github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI=
+github.com/cncf/udpa/go v0.0.0-20220112060539-c52dc94e7fbe h1:QQ3GSy+MqSHxm/d8nCtnAiZdYFd45cYZPs8vOOIYKfk=
+github.com/cncf/udpa/go v0.0.0-20220112060539-c52dc94e7fbe/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI=
 github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20211001041855-01bcc9b48dfe/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
 github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
+github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa h1:jQCWAUqqlij9Pgj2i/PB79y4KOPYVyFYdROxgaCwdTQ=
+github.com/cncf/xds/go v0.0.0-20231128003011-0fa0005c9caa/go.mod h1:x/1Gn8zydmfq8dk6e9PdstVsDgu9RuyIIJqAaF//0IM=
 github.com/cockroachdb/apd v1.1.0 h1:3LFP3629v+1aKXU5Q37mxmRxX/pIu1nijXydLShEq5I=
 github.com/cockroachdb/apd v1.1.0/go.mod h1:8Sl8LxpKi29FqWXR16WEFZRNSz3SoPzUzeMeY4+DwBQ=
 github.com/cockroachdb/cockroach-go v0.0.0-20190925194419-606b3d062051/go.mod h1:XGLbWH/ujMcbPbhZq52Nv6UrCghb1yGn//133kEsvDk=
@@ -313,7 +317,11 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.m
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ=
 github.com/envoyproxy/go-control-plane v0.10.2-0.20220325020618-49ff273808a1/go.mod h1:KJwIaB5Mv44NWtYuAOFCVOjcI94vtpEz2JU/D2v6IjE=
+github.com/envoyproxy/go-control-plane v0.12.0 h1:4X+VP1GHd1Mhj6IB5mMeGbLCleqxjletLK6K0rbxyZI=
+github.com/envoyproxy/go-control-plane v0.12.0/go.mod h1:ZBTaoJ23lqITozF0M6G4/IragXCQKCnYbmlmtHvwRG0=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
+github.com/envoyproxy/protoc-gen-validate v1.0.4 h1:gVPz/FMfvh57HdSJQyvBtF00j8JU4zdyUgIUNhlgg0A=
+github.com/envoyproxy/protoc-gen-validate v1.0.4/go.mod h1:qys6tmnRsYrQqIhm2bvKZH4Blx/1gTIZ2UKVY1M+Yew=
 github.com/evanphx/json-patch v5.7.0+incompatible h1:vgGkfT/9f8zE6tvSCe74nfpAVDQ2tG6yudJd8LBksgI=
 github.com/evanphx/json-patch v5.7.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
 github.com/evanphx/json-patch/v5 v5.9.0 h1:kcBlZQbplgElYIlo/n1hJbls2z/1awpXxpRi0/FOJfg=
@@ -1020,8 +1028,8 @@ github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0b
 github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
 github.com/spf13/afero v1.6.0/go.mod h1:Ai8FlHk4v/PARR026UzYexafAt9roJ7LcLMAmO6Z93I=
-github.com/spf13/afero v1.9.2 h1:j49Hj62F0n+DaZ1dDCvhABaPNSGNkt32oRFxI33IEMw=
-github.com/spf13/afero v1.9.2/go.mod h1:iUV7ddyEEZPO5gA3zD4fJt6iStLlL+Lg4m2cihcDf8Y=
+github.com/spf13/afero v1.10.0 h1:EaGW2JJh15aKOejeuJ+wpFSHnbd7GE6Wvp3TsNhb6LY=
+github.com/spf13/afero v1.10.0/go.mod h1:UBogFpq8E9Hx+xc5CNTTEpTnuHVmXDwZcZcE1eb/UhQ=
 github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
 github.com/spf13/cast v1.3.1 h1:nFm6S0SMdyzrzcmThSipiEubIDy8WEXKNZ0UOgiRpng=
 github.com/spf13/cast v1.3.1/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
@@ -1165,7 +1173,6 @@ golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm
 golang.org/x/crypto v0.0.0-20210616213533-5ff15b29337e/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20210711020723-a769d52b0f97/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/crypto v0.0.0-20211108221036-ceb1ce70b4fa/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
 golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.0.0-20220722155217-630584e8d5aa/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
 golang.org/x/crypto v0.21.0 h1:X31++rzVUdKhX5sWmSOFZxx8UW/ldWx55cbf08iNAMA=

api/pkg/transformer/feast/client.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"net"
 	"strconv"
+	"strings"
 
 	"github.com/caraml-dev/merlin/pkg/transformer/feast/bigtablestore"
 	"github.com/caraml-dev/merlin/pkg/transformer/feast/redis"
@@ -56,14 +57,21 @@ func createFeastServingClient(feastOptions Options, featureTableMetadata []*spec
 }
 
 func newFeastGrpcClient(url string, options Options) (*GrpcClient, error) {
-	host, port, err := net.SplitHostPort(url)
-	if err != nil {
-		return nil, errors.Errorf("Unable to parse Feast Serving host (%s): %s", url, err)
-	}
+	{
+		// validate
 
-	portInt, err := strconv.Atoi(port)
-	if err != nil {
-		return nil, errors.Errorf("Unable to parse Feast Serving port (%s): %s", url, err)
+		urlTmp := url
+		urlTmp = strings.TrimPrefix(urlTmp, "dns:///")
+		urlTmp = strings.TrimPrefix(urlTmp, "xds:///")
+		_, port, err := net.SplitHostPort(urlTmp)
+		if err != nil {
+			return nil, errors.Errorf("Unable to parse Feast Serving host (%s): %s", url, err)
+		}
+
+		_, err = strconv.Atoi(port)
+		if err != nil {
+			return nil, errors.Errorf("Unable to parse Feast Serving port (%s): %s", url, err)
+		}
 	}
 
 	dialOpts := []grpc.DialOption{}
@@ -74,7 +82,10 @@ func newFeastGrpcClient(url string, options Options) (*GrpcClient, error) {
 		})
 		dialOpts = append(dialOpts, keepAliveOpt)
 	}
-	client, err := newInsecureGRPCClientWithDialOptions(host, portInt, options.FeastGRPCConnCount, true, dialOpts...)
+	if strings.HasPrefix(url, "dns:///") {
+		dialOpts = append(dialOpts, grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`))
+	}
+	client, err := newInsecureGRPCClientWithDialOptions(url, options.FeastGRPCConnCount, true, dialOpts...)
 	if err != nil {
 		return nil, errors.Errorf("Unable to initialize a Feast gRPC client: %s", err)
 	}

api/pkg/transformer/feast/sdk.go

@@ -13,6 +13,7 @@ import (
 	"go.opencensus.io/plugin/ocgrpc"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	_ "google.golang.org/grpc/xds"
 )
 
 // GrpcClient as wrapper of feastsdk
@@ -23,9 +24,8 @@ type GrpcClient struct {
 	waitForReady bool
 }
 
-func newInsecureGRPCClientWithDialOptions(host string, port int, numConn int, waitForReady bool, opts ...grpc.DialOption) (*GrpcClient, error) {
+func newInsecureGRPCClientWithDialOptions(addr string, numConn int, waitForReady bool, opts ...grpc.DialOption) (*GrpcClient, error) {
 	feastCli := &GrpcClient{}
-	adr := fmt.Sprintf("%s:%d", host, port)
 
 	// Compile grpc dial options from security config.
 	options := append(opts, []grpc.DialOption{grpc.WithStatsHandler(&ocgrpc.ClientHandler{}), grpc.WithTransportCredentials(insecure.NewCredentials())}...)
@@ -36,7 +36,7 @@ func newInsecureGRPCClientWithDialOptions(host string, port int, numConn int, wa
 		otgrpc.OpenTracingClientInterceptor(opentracing.GlobalTracer()))
 	options = append(options, tracingInterceptor)
 
-	conn, err := grpcpool.DialContext(context.Background(), adr, uint(numConn), options...)
+	conn, err := grpcpool.DialContext(context.Background(), addr, uint(numConn), options...)
 	if err != nil {
 		return nil, err
 	}

python/sdk/Pipfile

@@ -8,6 +8,7 @@ ipykernel = "*"
 jupyter = "*"
 pytest = "*"
 pytest-xdist = "*"
+setuptools = "==81.0.0"
 
 [packages]
 merlin-sdk = {extras = ["test"],path = "."}

scripts/e2e/config/kserve/kustomization.yaml

@@ -2,3 +2,7 @@ resources:
 - kserve.yaml
 patchesStrategicMerge:
 - overlay.yaml
+images:
+  - name: gcr.io/kubebuilder/kube-rbac-proxy
+    newName: kubebuilder/kube-rbac-proxy
+    newTag: v0.13.1

scripts/e2e/setup-cluster.sh

@@ -70,7 +70,7 @@ install_istio() {
     kubectl rollout status deployment/istiod -w -n istio-system --timeout=${TIMEOUT}
     kubectl rollout status deployment/cluster-local-gateway -n istio-system -w --timeout=${TIMEOUT}
 
-    kubectl apply -f config/istio/ingress-class.yaml
+    kubectl apply --server-side -f config/istio/ingress-class.yaml
 
     sleep 30
 
@@ -79,7 +79,7 @@ install_istio() {
 
 set_ingress_host() {
     echo "::group::Set Ingress Host"
-    if [[ -z "${INGRESS_HOST}" ]]; then
+    if [[ -z "${INGRESS_HOST:-}" ]]; then
         export INGRESS_HOST="$(kubectl -n istio-system get service istio-ingressgateway -o jsonpath='{.status.loadBalancer.ingress[0].ip}').nip.io"
     fi
     echo "INGRESS_HOST=${INGRESS_HOST}"
@@ -90,11 +90,11 @@ install_knative() {
     echo "::group::Knative Deployment"
 
     # Install CRD
-    kubectl apply -f https://github.com/knative/serving/releases/download/knative-v${KNATIVE_VERSION}/serving-crds.yaml
+    kubectl apply --server-side -f https://github.com/knative/serving/releases/download/knative-v${KNATIVE_VERSION}/serving-crds.yaml
 
     # Install knative serving
     wget https://github.com/knative/serving/releases/download/knative-v${KNATIVE_VERSION}/serving-core.yaml -O config/knative/serving-core.yaml
-    kubectl apply -k config/knative
+    kubectl apply --server-side -k config/knative
 
     kubectl rollout status deployment/autoscaler -n knative-serving -w --timeout=${TIMEOUT}
     kubectl rollout status deployment/controller -n knative-serving -w --timeout=${TIMEOUT}
@@ -106,11 +106,11 @@ install_knative() {
     # Update knative config-deployment config map to allow resolving of the local e2e image repository
     kubectl get configmap -n knative-serving config-deployment -o yaml > temp.yaml
     yq -i '.data.registries-skipping-tag-resolving |= ("\(.),${DOCKER_REGISTRY}" | envsubst)' temp.yaml
-    kubectl apply -f temp.yaml
+    kubectl apply --server-side -f temp.yaml
     kubectl rollout restart deployment -n knative-serving controller
 
     # Install knative-istio
-    kubectl apply -f https://github.com/knative-sandbox/net-istio/releases/download/knative-v${KNATIVE_NET_ISTIO_VERSION}/net-istio.yaml
+    kubectl apply --server-side -f https://github.com/knative-sandbox/net-istio/releases/download/knative-v${KNATIVE_NET_ISTIO_VERSION}/net-istio.yaml
 
     kubectl rollout status deployment/net-istio-controller -n knative-serving -w --timeout=${TIMEOUT}
     kubectl rollout status deployment/net-istio-webhook -n knative-serving -w --timeout=${TIMEOUT}
@@ -121,7 +121,7 @@ install_knative() {
 install_cert_manager() {
     echo "::group::Cert Manager Deployment"
 
-    kubectl apply --filename=https://github.com/jetstack/cert-manager/releases/download/v${CERT_MANAGER_VERSION}/cert-manager.yaml
+    kubectl apply --server-side --filename=https://github.com/jetstack/cert-manager/releases/download/v${CERT_MANAGER_VERSION}/cert-manager.yaml
 
     kubectl rollout status deployment/cert-manager-webhook -n cert-manager -w --timeout=${TIMEOUT}
     kubectl rollout status deployment/cert-manager-cainjector -n cert-manager -w --timeout=${TIMEOUT}
@@ -147,9 +147,9 @@ install_kserve() {
     echo "::group::KServe Deployment"
 
     wget https://raw.githubusercontent.com/kserve/kserve/master/install/v${KSERVE_VERSION}/kserve.yaml -O config/kserve/kserve.yaml
-    kubectl apply -k config/kserve
+    kubectl apply --server-side -k config/kserve
     kubectl rollout status deployment/kserve-controller-manager -n kserve -w --timeout=${TIMEOUT}
-    kubectl apply -f https://raw.githubusercontent.com/kserve/kserve/master/install/v${KSERVE_VERSION}/kserve-runtimes.yaml
+    kubectl apply --server-side -f https://raw.githubusercontent.com/kserve/kserve/master/install/v${KSERVE_VERSION}/kserve-runtimes.yaml
 
     echo "::endgroup::"
 }