feast-dev · jklegar · Jan 6, 2021 · Jan 6, 2021 · Jan 6, 2021 · Jan 6, 2021
diff --git a/.prow/config.yaml b/.prow/config.yaml
@@ -312,6 +312,16 @@ presubmits:
               secretKeyRef:
                 name: feast-az-creds
                 key: AZ_SERVICE_PRINCIPAL_TENANT_ID
+          - name: AZURE_BLOB_ACCOUNT_NAME
+            valueFrom:
+              secretKeyRef:
+                name: feast-az-creds
+                key: AZURE_BLOB_ACCOUNT_NAME
+          - name: AZURE_BLOB_ACCOUNT_ACCESS_KEY
+            valueFrom:
+              secretKeyRef:
+                name: feast-az-creds
+                key: AZURE_BLOB_ACCOUNT_ACCESS_KEY
         volumeMounts:
           - mountPath: /etc/gcloud/service-account.json
             name: service-account

@@ -53,6 +53,8 @@ time kubectl run -n "$NAMESPACE" -i ci-test-runner  \
     --restart=Never \
     --image="${DOCKER_REPOSITORY}/feast-ci:${GIT_TAG}" \
     --env="STAGING_PATH=${STAGING_PATH}" \
+    --env="FEAST_AZURE_BLOB_ACCOUNT_NAME=${AZURE_BLOB_ACCOUNT_NAME}" \
+    --env="FEAST_AZURE_BLOB_ACCOUNT_ACCESS_KEY=${AZURE_BLOB_ACCOUNT_ACCESS_KEY}" \
     --  \
     bash -c "mkdir src && cd src && git clone ${GIT_REMOTE_URL} && cd feast && git config remote.origin.fetch '+refs/pull/*:refs/remotes/origin/pull/*' && git fetch -q && git checkout ${GIT_TAG} && ./infra/scripts/setup-e2e-env-sparkop.sh && ./infra/scripts/test-end-to-end-sparkop.sh"
 

@@ -52,6 +52,8 @@ function helm_install {
         --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" \

@@ -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"

@@ -68,12 +68,7 @@ def _get_optional(option):
 def _k8s_launcher(config: Config) -> JobLauncher:
     from feast.pyspark.launchers import k8s
 
-    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),
-        incluster=config.getboolean(opt.SPARK_K8S_USE_INCLUSTER_CONFIG),
-    )
+    return k8s.KubernetesJobLauncher(config=config)
 
 
 _launchers = {

@@ -9,6 +9,8 @@
 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,
     BatchIngestionJob,
@@ -140,16 +142,13 @@ class KubernetesJobLauncher(JobLauncher):
     Submits spark jobs to a spark cluster. Currently supports only historical feature retrieval jobs.
     """
 
-    def __init__(
-        self,
-        namespace: str,
-        incluster: bool,
-        staging_location: str,
-        resource_template_path: Optional[Path],
-    ):
-        self._namespace = namespace
+    def __init__(self, config: Config):
+        self._config = config
+        self._namespace = config.get(opt.SPARK_K8S_NAMESPACE)
+        incluster = config.getboolean(opt.SPARK_K8S_USE_INCLUSTER_CONFIG)
         self._api = _get_api(incluster=incluster)
-        self._staging_location = staging_location
+        self._staging_location = config.get(opt.SPARK_STAGING_LOCATION)
+        resource_template_path = config.get(opt.SPARK_K8S_JOB_TEMPLATE_PATH, None)
         if resource_template_path is not None:
             self._resource_template = _load_resource_template(resource_template_path)
         else:
@@ -183,7 +182,7 @@ def _job_from_job_info(self, job_info: JobInfo) -> SparkJob:
 
     def _get_staging_client(self):
         uri = urlparse(self._staging_location)
-        return get_staging_client(uri.scheme)
+        return get_staging_client(uri.scheme, self._config)
 
     def historical_feature_retrieval(
         self, job_params: RetrievalJobParameters

@@ -41,3 +41,4 @@ moto
 pyspark==3.0.1
 pyspark-stubs==3.0.0.post1
 kubernetes==12.0.*
+azure-storage-blob==12.6.0