From e4850c15e7e8458e9665bdf3f8a0572ee5b137b5 Mon Sep 17 00:00:00 2001
From: Oleg Avdeev <oleg.v.avdeev@gmail.com>
Date: Tue, 5 Jan 2021 19:39:58 -0800
Subject: [PATCH 1/5] minor fixes + scripts to run e2e tests in minikube

Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com>
---
 .dockerignore                                 |   1 +
 Makefile                                      |   3 +
 infra/docker/tests/Dockerfile                 |  24 ++++
 infra/scripts/k8s-common-functions.sh         | 109 ++++++++++++++++++
 infra/scripts/run-minikube-test.sh            |  38 ++++++
 infra/scripts/setup-e2e-local.sh              |  26 +++++
 infra/scripts/test-end-to-end-local.sh        |  69 +++++++++++
 infra/scripts/test_job.yaml                   |  35 ++++++
 sdk/python/feast/contrib/validation/ge.py     |   2 +-
 sdk/python/feast/pyspark/launcher.py          |   5 +-
 sdk/python/feast/pyspark/launchers/k8s/k8s.py |  12 +-
 tests/README.md                               |  90 +++++++++++++++
 tests/e2e/test_historical_features.py         |   9 +-
 13 files changed, 413 insertions(+), 10 deletions(-)
 create mode 100644 infra/docker/tests/Dockerfile
 create mode 100644 infra/scripts/k8s-common-functions.sh
 create mode 100755 infra/scripts/run-minikube-test.sh
 create mode 100644 infra/scripts/setup-e2e-local.sh
 create mode 100755 infra/scripts/test-end-to-end-local.sh
 create mode 100644 infra/scripts/test_job.yaml
 create mode 100644 tests/README.md

diff --git a/.dockerignore b/.dockerignore
index bef7cf70696..e28f081aff3 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -2,3 +2,4 @@ docs
 !docs/coverage
 charts
 env
+**/.terraform
diff --git a/Makefile b/Makefile
index 89e886ca8e0..c8dc7f0c3a6 100644
--- a/Makefile
+++ b/Makefile
@@ -156,6 +156,9 @@ build-ci-docker:
 build-jupyter-docker:
 	docker build -t $(REGISTRY)/feast-jupyter:$(VERSION) -f infra/docker/jupyter/Dockerfile .
 
+build-local-test-docker:
+	docker build -t feast:local -f infra/docker/tests/Dockerfile .
+
 # Documentation
 
 install-dependencies-proto-docs:
diff --git a/infra/docker/tests/Dockerfile b/infra/docker/tests/Dockerfile
new file mode 100644
index 00000000000..1f127b9bed0
--- /dev/null
+++ b/infra/docker/tests/Dockerfile
@@ -0,0 +1,24 @@
+ARG BASE_IMAGE=gcr.io/kf-feast/feast-ci:latest
+
+FROM ${BASE_IMAGE}
+
+RUN mkdir -p /src/sdk /src/spark/ingestion
+
+COPY sdk/python /src/sdk/python
+
+COPY README.md /src/README.md
+
+WORKDIR /src
+
+RUN pip install -r sdk/python/requirements-ci.txt
+
+RUN git init .
+RUN pip install -e sdk/python -U
+RUN pip install "s3fs" "boto3" "urllib3>=1.25.4"
+
+COPY tests /src/tests
+
+RUN pip install -r tests/requirements.txt
+
+COPY infra/scripts /src/infra/scripts
+COPY spark/ingestion/target /src/spark/ingestion/target
diff --git a/infra/scripts/k8s-common-functions.sh b/infra/scripts/k8s-common-functions.sh
new file mode 100644
index 00000000000..54ef64f5b64
--- /dev/null
+++ b/infra/scripts/k8s-common-functions.sh
@@ -0,0 +1,109 @@
+#!/bin/bash
+
+set -euo pipefail
+
+function k8s_cleanup {
+    local RELEASE=$1
+    local NAMESPACE=$2
+
+    # Create namespace if it doesn't exist.
+    kubectl create namespace "$NAMESPACE" || true
+
+    # Uninstall previous feast release if there is any.
+    helm uninstall "$RELEASE" -n "$NAMESPACE" || true
+
+    # `helm uninstall` doesn't remove PVCs, delete them manually.
+    time kubectl delete pvc --all -n "$NAMESPACE" || true
+
+    kubectl get service -n "$NAMESPACE"
+
+    # Set a new postgres password. Note that the postgres instance is not available outside
+    # the k8s cluster anyway so it doesn't have to be super secure.
+    echo "${STEP_BREADCRUMB:-} Setting PG password"
+
+    # use either shasum or md5sum, whichever exists
+    SUM=$(which md5sum shasum | grep -v "not found" | tail -n1 || true )
+
+    PG_PASSWORD=$(head -c 59 /dev/urandom | $SUM | head -c 16)
+    kubectl delete secret feast-postgresql -n "$NAMESPACE" || true
+    kubectl create secret generic feast-postgresql --from-literal=postgresql-password="$PG_PASSWORD" -n "$NAMESPACE"
+}
+
+function helm_install {
+    # helm install Feast into k8s cluster and display a nice error if it fails.
+    # Usage: helm_install $RELEASE $DOCKER_REPOSITORY $GIT_TAG ...
+    # Args:
+    #   $RELEASE is helm release name
+    #   $DOCKER_REPOSITORY is the docker repo containing feast images tagged with $GIT_TAG
+    #   ... you can pass additional args to this function that are passed on to helm install
+
+    local RELEASE=$1
+    local DOCKER_REPOSITORY=$2
+    local GIT_TAG=$3
+
+    shift 3
+
+    # We skip statsd exporter and other metrics stuff since we're not using it anyway, and it
+    # has some issues with unbound PVCs (that cause kubectl delete pvc to hang).
+    echo "${STEP_BREADCRUMB:-} Helm installing feast"
+
+    if ! time helm install --wait "$RELEASE" ./infra/charts/feast \
+        --timeout 5m \
+        --set "feast-jupyter.image.repository=${DOCKER_REPOSITORY}/feast-jupyter" \
+        --set "feast-jupyter.image.tag=${GIT_TAG}" \
+        --set "feast-online-serving.image.repository=${DOCKER_REPOSITORY}/feast-serving" \
+        --set "feast-online-serving.image.tag=${GIT_TAG}" \
+        --set "feast-jobservice.image.repository=${DOCKER_REPOSITORY}/feast-jobservice" \
+        --set "feast-jobservice.image.tag=${GIT_TAG}" \
+        --set "feast-core.image.repository=${DOCKER_REPOSITORY}/feast-core" \
+        --set "feast-core.image.tag=${GIT_TAG}" \
+        --set "prometheus-statsd-exporter.enabled=false" \
+        --set "prometheus.enabled=false" \
+        --set "grafana.enabled=false" \
+        --set "feast-jobservice.enabled=false" \
+        "$@" ; then
+
+        echo "Error during helm install. "
+        kubectl get pods
+
+        readarray -t CRASHED_PODS < <(kubectl get pods --no-headers=true | grep cicd | awk '{if ($2 == "0/1") { print $1 } }')
+        echo "Crashed pods: ${CRASHED_PODS[*]}"
+
+        for POD in "${CRASHED_PODS[@]}"; do
+            echo "Logs from pod error $POD:"
+            kubectl logs "$POD" --previous
+        done
+
+        exit 1
+    fi
+}
+
+function setup_sparkop_role {
+    # Set up permissions for the default user in sparkop namespace so that Feast SDK can manage
+    # sparkapplication resources from the test runner pod.
+
+    cat <<EOF | kubectl apply -f -
+kind: Role
+apiVersion: rbac.authorization.k8s.io/v1beta1
+metadata:
+  name: use-spark-operator
+  namespace: sparkop
+rules:
+- apiGroups: ["sparkoperator.k8s.io"]
+  resources: ["sparkapplications"]
+  verbs: ["create", "delete", "deletecollection", "get", "list", "update", "watch", "patch"]
+---
+apiVersion: rbac.authorization.k8s.io/v1beta1
+kind: RoleBinding
+metadata:
+  name: use-spark-operator
+  namespace: sparkop
+roleRef:
+  kind: Role
+  name: use-spark-operator
+  apiGroup: rbac.authorization.k8s.io
+subjects:
+  - kind: ServiceAccount
+    name: default
+EOF
+}
\ No newline at end of file
diff --git a/infra/scripts/run-minikube-test.sh b/infra/scripts/run-minikube-test.sh
new file mode 100755
index 00000000000..4c9a1344f85
--- /dev/null
+++ b/infra/scripts/run-minikube-test.sh
@@ -0,0 +1,38 @@
+#!/bin/bash
+
+set -euo pipefail
+
+NAMESPACE=sparkop
+JOB_NAME=test-runner
+
+# Delete all sparkapplication resources that may be left over from the previous test runs.
+kubectl delete sparkapplication --all -n sparkop || true
+
+JOB_SPEC=$(dirname $0)/test_job.yaml
+
+# Delete previous instance of the job if it exists
+kubectl delete -n ${NAMESPACE} "job/$JOB_NAME"  2>/dev/null || true
+
+# Create the job
+kubectl apply -n ${NAMESPACE} -f "$JOB_SPEC"
+
+# Wait for job to have a pod.
+for i in {1..10}
+do
+	POD=$(kubectl get pods -n ${NAMESPACE} --selector=job-name=$JOB_NAME --output=jsonpath='{.items[0].metadata.name}')
+	if [ ! -z "$POD" ]; then
+		break
+	else
+		sleep 1
+	fi
+done
+
+echo "Waiting for pod to be ready:"
+kubectl wait -n ${NAMESPACE} --for=condition=ContainersReady "pod/$POD" --timeout=60s || true
+
+echo "Job output:"
+kubectl logs -n ${NAMESPACE} -f "job/$JOB_NAME"
+
+# Can't wait for both conditions at once, so wait for complete first then wait for failure
+kubectl wait -n ${NAMESPACE} --for=condition=complete "job/$JOB_NAME" --timeout=60s && exit 0
+kubectl wait -n ${NAMESPACE} --for=condition=failure "job/$JOB_NAME" --timeout=60s && exit 1
diff --git a/infra/scripts/setup-e2e-local.sh b/infra/scripts/setup-e2e-local.sh
new file mode 100644
index 00000000000..720c418abe4
--- /dev/null
+++ b/infra/scripts/setup-e2e-local.sh
@@ -0,0 +1,26 @@
+#!/bin/bash
+set -euo pipefail
+
+STEP_BREADCRUMB='~~~~~~~~'
+
+pushd "$(dirname $0)"
+source k8s-common-functions.sh
+
+# spark k8s test - runs in sparkop namespace (so it doesn't interfere with a concurrently
+# running EMR test).
+NAMESPACE=sparkop
+RELEASE=sparkop
+
+# Clean up old release
+k8s_cleanup "$RELEASE" "$NAMESPACE"
+
+# Helm install everything in a namespace
+helm_install "$RELEASE" "${DOCKER_REPOSITORY}" "${GIT_TAG}" --namespace "$NAMESPACE" --create-namespace
+
+# Delete all sparkapplication resources that may be left over from the previous test runs.
+kubectl delete sparkapplication --all -n "$NAMESPACE" || true
+
+# Make sure the test pod has permissions to create sparkapplication resources
+setup_sparkop_role
+
+echo "DONE"
\ No newline at end of file
diff --git a/infra/scripts/test-end-to-end-local.sh b/infra/scripts/test-end-to-end-local.sh
new file mode 100755
index 00000000000..ec6fd0bff34
--- /dev/null
+++ b/infra/scripts/test-end-to-end-local.sh
@@ -0,0 +1,69 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+export DISABLE_FEAST_SERVICE_FIXTURES=1
+export DISABLE_SERVICE_FIXTURES=1
+
+export FEAST_SPARK_K8S_NAMESPACE=sparkop
+export FEAST_S3_ENDPOINT_URL=http://minio.minio.svc.cluster.local:9000
+
+# Used by tests
+export AWS_S3_ENDPOINT_URL=http://minio.minio.svc.cluster.local:9000
+
+cat << SPARK_CONF_END >/tmp/spark_conf.yml
+apiVersion: "sparkoperator.k8s.io/v1beta2"
+kind: SparkApplication
+metadata:
+  namespace: default
+spec:
+  type: Scala
+  mode: cluster
+  image: "gcr.io/kf-feast/spark-py:v3.0.1"
+  imagePullPolicy: Always
+  sparkVersion: "3.0.1"
+  timeToLiveSeconds: 3600
+  pythonVersion: "3"
+  sparkConf:
+    "spark.hadoop.fs.s3a.endpoint": http://minio.minio.svc.cluster.local:9000
+    "spark.hadoop.fs.s3a.path.style.access": "true"
+    "spark.hadoop.fs.s3a.access.key": ${AWS_ACCESS_KEY_ID}
+    "spark.hadoop.fs.s3a.secret.key": ${AWS_SECRET_ACCESS_KEY}
+  restartPolicy:
+    type: Never
+  volumes:
+    - name: "test-volume"
+      hostPath:
+        path: "/tmp"
+        type: Directory
+  driver:
+    cores: 1
+    coreLimit: "1200m"
+    memory: "512m"
+    labels:
+      version: 3.0.1
+    serviceAccount: spark
+    volumeMounts:
+      - name: "test-volume"
+        mountPath: "/tmp"
+  executor:
+    cores: 1
+    instances: 1
+    memory: "512m"
+    labels:
+      version: 3.0.1
+    volumeMounts:
+      - name: "test-volume"
+        mountPath: "/tmp"
+SPARK_CONF_END
+export FEAST_SPARK_K8S_JOB_TEMPLATE_PATH=/tmp/spark_conf.yml
+
+PYTHONPATH=sdk/python pytest tests/e2e/ \
+      --feast-version develop \
+      --core-url sparkop-feast-core:6565 \
+      --serving-url sparkop-feast-online-serving:6566 \
+      --env k8s \
+      --staging-path s3a://feast-staging \
+      --redis-url sparkop-redis-master.sparkop.svc.cluster.local:6379 \
+      --kafka-brokers sparkop-kafka.sparkop.svc.cluster.local:9092 \
+      -m "not bq"
\ No newline at end of file
diff --git a/infra/scripts/test_job.yaml b/infra/scripts/test_job.yaml
new file mode 100644
index 00000000000..4995b7d4f06
--- /dev/null
+++ b/infra/scripts/test_job.yaml
@@ -0,0 +1,35 @@
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: test-runner
+  namespace: sparkop
+spec:
+  backoffLimit: 1
+  template:
+    spec:
+      containers:
+      - name: ubuntu
+        image: feast:local
+        command: ["bash",  "-c", "./infra/scripts/test-end-to-end-local.sh"]
+        imagePullPolicy: Never
+        args:
+        - bash
+        stdin: true
+        stdinOnce: true
+        tty: true
+        env:
+          - name: AWS_ACCESS_KEY_ID
+            valueFrom:
+              secretKeyRef:
+                name: minio
+                key: accesskey
+          - name: AWS_SECRET_ACCESS_KEY
+            valueFrom:
+              secretKeyRef:
+                name: minio
+                key: secretkey
+          - name: AWS_DEFAULT_REGION
+            value: us-east-1
+          - name: AWS_S3_SIGNATURE_VERSION
+            value: s3v4
+      restartPolicy: Never
diff --git a/sdk/python/feast/contrib/validation/ge.py b/sdk/python/feast/contrib/validation/ge.py
index 2f617b86184..6259817cfa0 100644
--- a/sdk/python/feast/contrib/validation/ge.py
+++ b/sdk/python/feast/contrib/validation/ge.py
@@ -103,7 +103,7 @@ def apply_validation(
         "/"
     )
     staging_scheme = urlparse(staging_location).scheme
-    staging_client = get_staging_client(staging_scheme)
+    staging_client = get_staging_client(staging_scheme, client._config)
 
     pickled_code_fp = io.BytesIO(udf.pickled_code)
     remote_path = f"{staging_location}/udfs/{udf.name}.pickle"
diff --git a/sdk/python/feast/pyspark/launcher.py b/sdk/python/feast/pyspark/launcher.py
index 8b12d7e2fca..6d13edd89d6 100644
--- a/sdk/python/feast/pyspark/launcher.py
+++ b/sdk/python/feast/pyspark/launcher.py
@@ -67,12 +67,15 @@ def _get_optional(option):
 
 def _k8s_launcher(config: Config) -> JobLauncher:
     from feast.pyspark.launchers import k8s
+    staging_location = config.get(opt.SPARK_STAGING_LOCATION)
+    staging_uri = urlparse(staging_location)
 
     return k8s.KubernetesJobLauncher(
         namespace=config.get(opt.SPARK_K8S_NAMESPACE),
         resource_template_path=config.get(opt.SPARK_K8S_JOB_TEMPLATE_PATH, None),
-        staging_location=config.get(opt.SPARK_STAGING_LOCATION),
+        staging_location=staging_location,
         incluster=config.getboolean(opt.SPARK_K8S_USE_INCLUSTER_CONFIG),
+        staging_client=get_staging_client(staging_uri.scheme, config)
     )
 
 
diff --git a/sdk/python/feast/pyspark/launchers/k8s/k8s.py b/sdk/python/feast/pyspark/launchers/k8s/k8s.py
index 6132c046a91..dd79eaa1074 100644
--- a/sdk/python/feast/pyspark/launchers/k8s/k8s.py
+++ b/sdk/python/feast/pyspark/launchers/k8s/k8s.py
@@ -22,7 +22,7 @@
     StreamIngestionJob,
     StreamIngestionJobParameters,
 )
-from feast.staging.storage_client import get_staging_client
+from feast.staging.storage_client import AbstractStagingClient
 
 from .k8s_utils import (
     DEFAULT_JOB_TEMPLATE,
@@ -146,10 +146,12 @@ def __init__(
         incluster: bool,
         staging_location: str,
         resource_template_path: Optional[Path],
+        staging_client: AbstractStagingClient,
     ):
         self._namespace = namespace
         self._api = _get_api(incluster=incluster)
         self._staging_location = staging_location
+        self._staging_client = staging_client
         if resource_template_path is not None:
             self._resource_template = _load_resource_template(resource_template_path)
         else:
@@ -181,10 +183,6 @@ def _job_from_job_info(self, job_info: JobInfo) -> SparkJob:
             # We should never get here
             raise ValueError(f"Unknown job type {job_info.job_type}")
 
-    def _get_staging_client(self):
-        uri = urlparse(self._staging_location)
-        return get_staging_client(uri.scheme)
-
     def historical_feature_retrieval(
         self, job_params: RetrievalJobParameters
     ) -> RetrievalJob:
@@ -203,7 +201,7 @@ def historical_feature_retrieval(
             pyspark_script = f.read()
 
         pyspark_script_path = urlunparse(
-            self._get_staging_client().upload_fileobj(
+            self._staging_client.upload_fileobj(
                 BytesIO(pyspark_script.encode("utf8")),
                 local_path="historical_retrieval.py",
                 remote_path_prefix=self._staging_location,
@@ -241,7 +239,7 @@ def _upload_jar(self, jar_path: str) -> str:
             local_jar_path = jar_path
         with open(local_jar_path, "rb") as f:
             return urlunparse(
-                self._get_staging_client().upload_fileobj(
+                self._staging_client.upload_fileobj(
                     f,
                     local_jar_path,
                     remote_path_prefix=self._staging_location,
diff --git a/tests/README.md b/tests/README.md
new file mode 100644
index 00000000000..ea4be2b873c
--- /dev/null
+++ b/tests/README.md
@@ -0,0 +1,90 @@
+# Running full e2e test suite in Minikube
+
+This doc describes how to to run the entire suite of e2e tests locally, with no external (cloud) dependencies. It will use minikube, spark k8s operator and minio for storage.
+
+The tests will be run against your local copy of the Python SDK and ingestion jar. For other components like core and serving this setup will use docker images from the public GCR repo, built from the latest master.
+
+## Prerequisites:
+* Docker (highly recommend increasing default disk image size on OSX in docker settings)
+* awscli
+* kubectl
+* minikube (tested using docker driver on OSX)
+* helm 3
+* bash 5.0+ (you'll need to brew install it on OSX)
+* Java 11 toolchain with maven
+* make
+
+## Steps
+
+1. Start minikube. We'll need more memory and disk than default.
+```bash
+minikube start --disk-size='40000mb' --memory 4096
+```
+
+2. Install minio. 
+```bash
+helm repo add minio https://helm.min.io/
+helm install --namespace minio --create-namespace minio minio/minio --set resources.requests.memory=2Gi
+```
+
+3. Create k8s namespace to run tests in
+```bash
+kubectl create namespace sparkop
+```
+
+4. Install spark operator.
+```bash
+helm repo add spark-operator https://googlecloudplatform.github.io/spark-on-k8s-operator
+helm install spark-op spark-operator/spark-operator \
+	--namespace spark-operator \
+	--create-namespace \
+	--set "image.tag=v1beta2-1.1.2-2.4.5" \
+	--set "sparkJobNamespace=sparkop" \
+	--set "serviceAccounts.spark.name=spark"
+```
+
+5. Copy secret from minio into new sparkop namespace.
+```bash
+kubectl get secret minio --namespace=minio -oyaml | grep -v '^\s*namespace:\s' | kubectl apply --namespace=sparkop -f -
+```
+
+6. Install Feast. That may fail due to timeout, rerun again in that case
+```bash
+#
+# NB!! make sure to use bash 5.0+ for the next step. You'd need to brew install it on OSX
+DOCKER_REPOSITORY=gcr.io/kf-feast GIT_TAG=develop bash ./infra/scripts/setuplocal.sh
+```
+
+7. Build the ingestion jar locally
+```bash
+make build-java-no-tests REVISION=develop
+```
+
+8. Create staging bucket using awscli. First, create a port-forward for minio:
+```bash
+# port forward minio in a separate terminal
+export POD_NAME=$(kubectl get pods --namespace minio -l "release=minio" -o jsonpath="{.items[0].metadata.name}")
+
+kubectl port-forward $POD_NAME 9000 --namespace minio
+```
+Then create the bucket
+```bash
+
+export AWS_ACCESS_KEY_ID=$(kubectl get secret minio -o jsonpath="{.data.accesskey}" -n minio | base64 --decode)
+export AWS_SECRET_ACCESS_KEY=$(kubectl get secret minio -o jsonpath="{.data.secretkey}" -n minio | base64 --decode)
+export AWS_DEFAULT_REGION=us-east-1
+
+# Finally, create staging bucket
+aws --endpoint-url http://localhost:9000 s3 mb s3://feast-staging
+```
+
+9. Use minikube docker to build a docker image with tests from your working copy.
+```bash
+eval $(minikube docker-env)
+make build-local-test-docker
+```
+
+10. Finally, run tests:
+```bash
+./infra/scripts/run-minikube-test.sh
+```
diff --git a/tests/e2e/test_historical_features.py b/tests/e2e/test_historical_features.py
index 5cf2449582c..4a45b2e38f1 100644
--- a/tests/e2e/test_historical_features.py
+++ b/tests/e2e/test_historical_features.py
@@ -2,6 +2,7 @@
 from typing import Union
 from urllib.parse import urlparse, urlunparse
 
+import os
 import gcsfs
 import numpy as np
 import pandas as pd
@@ -30,7 +31,13 @@ def read_parquet(uri):
 
         import s3fs
 
-        fs = s3fs.S3FileSystem()
+        # AWS_S3_ENDPOINT_URL needs to be set when using minio
+        if 'AWS_S3_ENDPOINT_URL' in os.environ:
+            fs = s3fs.S3FileSystem(client_kwargs={
+                'endpoint_url': os.getenv('AWS_S3_ENDPOINT_URL')
+            })
+        else:
+            fs = s3fs.S3FileSystem()
         files = ["s3://" + path for path in fs.glob(s3uri + "/part-*")]
         ds = parquet.ParquetDataset(files, filesystem=fs)
         return ds.read().to_pandas()

From c90198332282c2b919c9970f247eb02e01f1862e Mon Sep 17 00:00:00 2001
From: Oleg Avdeev <oleg.v.avdeev@gmail.com>
Date: Tue, 5 Jan 2021 21:31:30 -0800
Subject: [PATCH 2/5] script rename

Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com>
---
 tests/README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/README.md b/tests/README.md
index ea4be2b873c..fcbbc2f02d2 100644
--- a/tests/README.md
+++ b/tests/README.md
@@ -52,7 +52,7 @@ kubectl get secret minio --namespace=minio -oyaml | grep -v '^\s*namespace:\s' |
 ```bash
 #
 # NB!! make sure to use bash 5.0+ for the next step. You'd need to brew install it on OSX
-DOCKER_REPOSITORY=gcr.io/kf-feast GIT_TAG=develop bash ./infra/scripts/setuplocal.sh
+DOCKER_REPOSITORY=gcr.io/kf-feast GIT_TAG=develop bash ./infra/scripts/setup-e2e-local.sh
 ```
 
 7. Build the ingestion jar locally

From 14ec1e8a8ea10edc4bb871993eaa174f3cb6e2a6 Mon Sep 17 00:00:00 2001
From: Oleg Avdeev <oleg.v.avdeev@gmail.com>
Date: Thu, 7 Jan 2021 16:34:48 -0800
Subject: [PATCH 3/5] refactor bash function library

Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com>
---
 .gitignore                            |   5 ++
 infra/scripts/azure-runner.sh         |   8 +-
 infra/scripts/k8s-common-functions.sh |  18 ++++-
 infra/scripts/runner-helper.sh        | 108 --------------------------
 infra/scripts/setup-e2e-local.sh      |   2 +-
 5 files changed, 26 insertions(+), 115 deletions(-)
 delete mode 100755 infra/scripts/runner-helper.sh

diff --git a/.gitignore b/.gitignore
index b2c3f77f8c3..546aff03830 100644
--- a/.gitignore
+++ b/.gitignore
@@ -184,3 +184,8 @@ sdk/python/docs/html
 *_pb2.py
 *_pb2.pyi
 *_pb2_grpc.py
+
+# VSCode
+.bloop
+.metals
+*.code-workspace
diff --git a/infra/scripts/azure-runner.sh b/infra/scripts/azure-runner.sh
index a8e731d400e..dde64e9e521 100755
--- a/infra/scripts/azure-runner.sh
+++ b/infra/scripts/azure-runner.sh
@@ -12,7 +12,7 @@ GIT_REMOTE_URL=https://github.com/feast-dev/feast.git
 echo "########## Starting e2e tests for ${GIT_REMOTE_URL} ${GIT_TAG} ###########"
 
 # Note requires running in root feast directory
-source infra/scripts/runner-helper.sh
+source infra/scripts/k8s-common-functions.sh
 
 # Workaround for COPY command in core docker image that pulls local maven repo into the image
 # itself.
@@ -34,8 +34,12 @@ RELEASE=sparkop
 # Delete old helm release and PVCs
 k8s_cleanup "$RELEASE" "$NAMESPACE"
 
+wait_for_images "${DOCKER_REPOSITORY}" "${GIT_TAG}"
+
 # Helm install everything in a namespace
-helm_install "$RELEASE" "${DOCKER_REPOSITORY}" "${GIT_TAG}" --namespace "$NAMESPACE"
+helm_install "$RELEASE" "${DOCKER_REPOSITORY}" "${GIT_TAG}" "$NAMESPACE" \
+        --set "feast-jobservice.envOverrides.FEAST_AZURE_BLOB_ACCOUNT_NAME=${AZURE_BLOB_ACCOUNT_NAME}" \
+        --set "feast-jobservice.envOverrides.FEAST_AZURE_BLOB_ACCOUNT_ACCESS_KEY=${AZURE_BLOB_ACCOUNT_ACCESS_KEY}"
 
 # Delete old test running pod if it exists
 kubectl delete pod -n "$NAMESPACE" ci-test-runner 2>/dev/null || true
diff --git a/infra/scripts/k8s-common-functions.sh b/infra/scripts/k8s-common-functions.sh
index 54ef64f5b64..2864c252dcc 100644
--- a/infra/scripts/k8s-common-functions.sh
+++ b/infra/scripts/k8s-common-functions.sh
@@ -2,6 +2,13 @@
 
 set -euo pipefail
 
+function wait_for_images {
+    local DOCKER_REPOSITORY=$1
+    local GIT_TAG=$2
+    # Wait for images to be available in the docker repository; ci is the last image built
+    timeout 15m bash -c "while ! gcloud container images list-tags ${DOCKER_REPOSITORY}/feast-ci --format=json | jq -e \".[] | select(.tags[] | contains (\\\"${GIT_TAG}\\\"))\" > /dev/null; do sleep 10s; done"
+}
+
 function k8s_cleanup {
     local RELEASE=$1
     local NAMESPACE=$2
@@ -36,12 +43,14 @@ function helm_install {
     #   $RELEASE is helm release name
     #   $DOCKER_REPOSITORY is the docker repo containing feast images tagged with $GIT_TAG
     #   ... you can pass additional args to this function that are passed on to helm install
+    #   $NAMESPACE is the namespace name
 
     local RELEASE=$1
     local DOCKER_REPOSITORY=$2
     local GIT_TAG=$3
+    local NAMESPACE=$4
 
-    shift 3
+    shift 4
 
     # We skip statsd exporter and other metrics stuff since we're not using it anyway, and it
     # has some issues with unbound PVCs (that cause kubectl delete pvc to hang).
@@ -61,17 +70,18 @@ function helm_install {
         --set "prometheus.enabled=false" \
         --set "grafana.enabled=false" \
         --set "feast-jobservice.enabled=false" \
+        --namespace "$NAMESPACE" \
         "$@" ; then
 
         echo "Error during helm install. "
-        kubectl get pods
+        kubectl -n "$NAMESPACE" get pods
 
-        readarray -t CRASHED_PODS < <(kubectl get pods --no-headers=true | grep cicd | awk '{if ($2 == "0/1") { print $1 } }')
+        readarray -t CRASHED_PODS < <(kubectl -n "$NAMESPACE" get pods --no-headers=true | grep "$RELEASE" | awk '{if ($2 == "0/1") { print $1 } }')
         echo "Crashed pods: ${CRASHED_PODS[*]}"
 
         for POD in "${CRASHED_PODS[@]}"; do
             echo "Logs from pod error $POD:"
-            kubectl logs "$POD" --previous
+            kubectl -n "$NAMESPACE" logs "$POD" --previous
         done
 
         exit 1
diff --git a/infra/scripts/runner-helper.sh b/infra/scripts/runner-helper.sh
deleted file mode 100755
index c03e29b9a33..00000000000
--- a/infra/scripts/runner-helper.sh
+++ /dev/null
@@ -1,108 +0,0 @@
-#!/usr/bin/env bash
-
-function k8s_cleanup {
-    local RELEASE=$1
-    local NAMESPACE=$2
-
-    # Create namespace if it doesn't exist.
-    kubectl create namespace "$NAMESPACE" || true
-
-    # Uninstall previous feast release if there is any.
-    helm uninstall "$RELEASE" -n "$NAMESPACE" || true
-
-    # `helm uninstall` doesn't remove PVCs, delete them manually.
-    time kubectl delete pvc --all -n "$NAMESPACE" || true
-
-    kubectl get service -n "$NAMESPACE"
-
-    # Set a new postgres password. Note that the postgres instance is not available outside
-    # the k8s cluster anyway so it doesn't have to be super secure.
-    echo "${STEP_BREADCRUMB} Setting PG password"
-    PG_PASSWORD=$(head -c 59 /dev/urandom | md5sum | head -c 16)
-    kubectl delete secret feast-postgresql -n "$NAMESPACE" || true
-    kubectl create secret generic feast-postgresql --from-literal=postgresql-password="$PG_PASSWORD" -n "$NAMESPACE"
-}
-
-function helm_install {
-    # helm install Feast into k8s cluster and display a nice error if it fails.
-    # Usage: helm_install $RELEASE $DOCKER_REPOSITORY $GIT_TAG ...
-    # Args:
-    #   $RELEASE is helm release name
-    #   $DOCKER_REPOSITORY is the docker repo containing feast images tagged with $GIT_TAG
-    #   ... you can pass additional args to this function that are passed on to helm install
-
-    local RELEASE=$1
-    local DOCKER_REPOSITORY=$2
-    local GIT_TAG=$3
-
-    shift 3
-
-    # Wait for images to be available in the docker repository; ci is the last image built
-    timeout 15m bash -c "while ! gcloud container images list-tags ${DOCKER_REPOSITORY}/feast-ci --format=json | jq -e \".[] | select(.tags[] | contains (\\\"${GIT_TAG}\\\"))\" > /dev/null; do sleep 10s; done"
-
-    # We skip statsd exporter and other metrics stuff since we're not using it anyway, and it
-    # has some issues with unbound PVCs (that cause kubectl delete pvc to hang).
-    echo "${STEP_BREADCRUMB} Helm installing feast"
-
-    if ! time helm install --wait "$RELEASE" infra/charts/feast \
-        --timeout 15m \
-        --set "feast-jupyter.image.repository=${DOCKER_REPOSITORY}/feast-jupyter" \
-        --set "feast-jupyter.image.tag=${GIT_TAG}" \
-        --set "feast-online-serving.image.repository=${DOCKER_REPOSITORY}/feast-serving" \
-        --set "feast-online-serving.image.tag=${GIT_TAG}" \
-        --set "feast-jobservice.image.repository=${DOCKER_REPOSITORY}/feast-jobservice" \
-        --set "feast-jobservice.image.tag=${GIT_TAG}" \
-        --set "feast-jobservice.envOverrides.FEAST_AZURE_BLOB_ACCOUNT_NAME=${AZURE_BLOB_ACCOUNT_NAME}" \
-        --set "feast-jobservice.envOverrides.FEAST_AZURE_BLOB_ACCOUNT_ACCESS_KEY=${AZURE_BLOB_ACCOUNT_ACCESS_KEY}" \
-        --set "feast-core.image.repository=${DOCKER_REPOSITORY}/feast-core" \
-        --set "feast-core.image.tag=${GIT_TAG}" \
-        --set "prometheus-statsd-exporter.enabled=false" \
-        --set "prometheus.enabled=false" \
-        --set "grafana.enabled=false" \
-        --set "feast-jobservice.enabled=false" \
-        "$@" ; then
-
-        echo "Error during helm install. "
-        kubectl -n "$NAMESPACE" get pods
-
-        readarray -t CRASHED_PODS < <(kubectl -n "$NAMESPACE" get pods --no-headers=true | grep "$RELEASE" | awk '{if ($2 == "0/1") { print $1 } }')
-        echo "Crashed pods: ${CRASHED_PODS[*]}"
-
-        for POD in "${CRASHED_PODS[@]}"; do
-            echo "Logs from pod error $POD:"
-            kubectl -n "$NAMESPACE" logs "$POD" --previous
-        done
-
-        exit 1
-    fi
-}
-
-function setup_sparkop_role {
-    # Set up permissions for the default user in sparkop namespace so that Feast SDK can manage
-    # sparkapplication resources from the test runner pod.
-
-    cat <<EOF | kubectl apply -f -
-kind: Role
-apiVersion: rbac.authorization.k8s.io/v1beta1
-metadata:
-  name: use-spark-operator
-  namespace: sparkop
-rules:
-- apiGroups: ["sparkoperator.k8s.io"]
-  resources: ["sparkapplications"]
-  verbs: ["create", "delete", "deletecollection", "get", "list", "update", "watch", "patch"]
----
-apiVersion: rbac.authorization.k8s.io/v1beta1
-kind: RoleBinding
-metadata:
-  name: use-spark-operator
-  namespace: sparkop
-roleRef:
-  kind: Role
-  name: use-spark-operator
-  apiGroup: rbac.authorization.k8s.io
-subjects:
-  - kind: ServiceAccount
-    name: default
-EOF
-}
diff --git a/infra/scripts/setup-e2e-local.sh b/infra/scripts/setup-e2e-local.sh
index 720c418abe4..3432673d6dc 100644
--- a/infra/scripts/setup-e2e-local.sh
+++ b/infra/scripts/setup-e2e-local.sh
@@ -15,7 +15,7 @@ RELEASE=sparkop
 k8s_cleanup "$RELEASE" "$NAMESPACE"
 
 # Helm install everything in a namespace
-helm_install "$RELEASE" "${DOCKER_REPOSITORY}" "${GIT_TAG}" --namespace "$NAMESPACE" --create-namespace
+helm_install "$RELEASE" "${DOCKER_REPOSITORY}" "${GIT_TAG}" "$NAMESPACE" --create-namespace
 
 # Delete all sparkapplication resources that may be left over from the previous test runs.
 kubectl delete sparkapplication --all -n "$NAMESPACE" || true

From c76c6022f0c068c9d4321f7eb7910963d6f49ef4 Mon Sep 17 00:00:00 2001
From: Oleg Avdeev <oleg.v.avdeev@gmail.com>
Date: Thu, 7 Jan 2021 18:49:20 -0800
Subject: [PATCH 4/5] fix linter warnings

Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com>
---
 sdk/python/feast/pyspark/launcher.py          |  3 ++-
 sdk/python/feast/pyspark/launchers/k8s/k8s.py |  1 -
 sdk/python/feast/staging/storage_client.py    |  1 +
 tests/e2e/test_historical_features.py         | 10 +++++-----
 4 files changed, 8 insertions(+), 7 deletions(-)

diff --git a/sdk/python/feast/pyspark/launcher.py b/sdk/python/feast/pyspark/launcher.py
index 6d13edd89d6..9b372db654c 100644
--- a/sdk/python/feast/pyspark/launcher.py
+++ b/sdk/python/feast/pyspark/launcher.py
@@ -67,6 +67,7 @@ def _get_optional(option):
 
 def _k8s_launcher(config: Config) -> JobLauncher:
     from feast.pyspark.launchers import k8s
+
     staging_location = config.get(opt.SPARK_STAGING_LOCATION)
     staging_uri = urlparse(staging_location)
 
@@ -75,7 +76,7 @@ def _k8s_launcher(config: Config) -> JobLauncher:
         resource_template_path=config.get(opt.SPARK_K8S_JOB_TEMPLATE_PATH, None),
         staging_location=staging_location,
         incluster=config.getboolean(opt.SPARK_K8S_USE_INCLUSTER_CONFIG),
-        staging_client=get_staging_client(staging_uri.scheme, config)
+        staging_client=get_staging_client(staging_uri.scheme, config),
     )
 
 
diff --git a/sdk/python/feast/pyspark/launchers/k8s/k8s.py b/sdk/python/feast/pyspark/launchers/k8s/k8s.py
index 2b3440c15f2..dc555f7b78c 100644
--- a/sdk/python/feast/pyspark/launchers/k8s/k8s.py
+++ b/sdk/python/feast/pyspark/launchers/k8s/k8s.py
@@ -9,7 +9,6 @@
 import yaml
 from kubernetes.client.api import CustomObjectsApi
 
-from feast.config import Config
 from feast.constants import ConfigOptions as opt
 from feast.pyspark.abc import (
     BQ_SPARK_PACKAGE,
diff --git a/sdk/python/feast/staging/storage_client.py b/sdk/python/feast/staging/storage_client.py
index cdec19a8b12..07027447d65 100644
--- a/sdk/python/feast/staging/storage_client.py
+++ b/sdk/python/feast/staging/storage_client.py
@@ -378,6 +378,7 @@ def list_files(self, uri: ParseResult) -> List[str]:
 
     def _uri_to_bucket_key(self, uri: ParseResult) -> Tuple[str, str]:
         assert uri.hostname == f"{self.account_name}.blob.core.windows.net"
+        assert uri.username
         bucket = uri.username
         key = uri.path.lstrip("/")
         return bucket, key
diff --git a/tests/e2e/test_historical_features.py b/tests/e2e/test_historical_features.py
index 4a45b2e38f1..595bb066f23 100644
--- a/tests/e2e/test_historical_features.py
+++ b/tests/e2e/test_historical_features.py
@@ -1,8 +1,8 @@
+import os
 from datetime import datetime, timedelta
 from typing import Union
 from urllib.parse import urlparse, urlunparse
 
-import os
 import gcsfs
 import numpy as np
 import pandas as pd
@@ -32,10 +32,10 @@ def read_parquet(uri):
         import s3fs
 
         # AWS_S3_ENDPOINT_URL needs to be set when using minio
-        if 'AWS_S3_ENDPOINT_URL' in os.environ:
-            fs = s3fs.S3FileSystem(client_kwargs={
-                'endpoint_url': os.getenv('AWS_S3_ENDPOINT_URL')
-            })
+        if "AWS_S3_ENDPOINT_URL" in os.environ:
+            fs = s3fs.S3FileSystem(
+                client_kwargs={"endpoint_url": os.getenv("AWS_S3_ENDPOINT_URL")}
+            )
         else:
             fs = s3fs.S3FileSystem()
         files = ["s3://" + path for path in fs.glob(s3uri + "/part-*")]

From 1c755a0825dc7d8b4b36207d3ce2742d1710aa0c Mon Sep 17 00:00:00 2001
From: Oleg Avdeev <oleg.v.avdeev@gmail.com>
Date: Thu, 7 Jan 2021 20:31:03 -0800
Subject: [PATCH 5/5] fix the timeout

Signed-off-by: Oleg Avdeev <oleg.v.avdeev@gmail.com>
---
 infra/scripts/k8s-common-functions.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/infra/scripts/k8s-common-functions.sh b/infra/scripts/k8s-common-functions.sh
index 2864c252dcc..520dbd48177 100644
--- a/infra/scripts/k8s-common-functions.sh
+++ b/infra/scripts/k8s-common-functions.sh
@@ -57,7 +57,7 @@ function helm_install {
     echo "${STEP_BREADCRUMB:-} Helm installing feast"
 
     if ! time helm install --wait "$RELEASE" ./infra/charts/feast \
-        --timeout 5m \
+        --timeout 15m \
         --set "feast-jupyter.image.repository=${DOCKER_REPOSITORY}/feast-jupyter" \
         --set "feast-jupyter.image.tag=${GIT_TAG}" \
         --set "feast-online-serving.image.repository=${DOCKER_REPOSITORY}/feast-serving" \