We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Storage initializer container succeeded even with invalid storage uri
/kind bug

**What steps did you take and what happened:**
Set `storageUri` with an invalid uri `s3://platform-test/s3/moviesentiment` (s3 subdir does not exist)

The storage initializer container succeeded,
```
kubectl logs moviesentiment-predictor-default-xh8c8-deployment-5955bbccxfnfc -c storage-initializer
INFO:root:Initializing, args: src_uri [s3://platform-test/s3/moviesentiment] dest_path[ [/mnt/models]
INFO:root:Copying contents of s3://platform-test/s3/moviesentiment to local
INFO:root:Successfully copied s3://platform-test/s3/moviesentiment to /mnt/models
```
But main container failed
```
kubectl logs moviesentiment-predictor-default-xh8c8-deployment-5955bbccxfnfc -c user-container
INFO:root:Copying contents of /mnt/models to local
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "/usr/local/lib/python3.7/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/sklearnserver/sklearnserver/__main__.py", line 33, in <module>
    model.load()
  File "/sklearnserver/sklearnserver/model.py", line 32, in load
    self._joblib = joblib.load(model_file) #pylint:disable=attribute-defined-outside-init
  File "/usr/local/lib/python3.7/site-packages/joblib/numpy_pickle.py", line 597, in load
    with open(filename, 'rb') as f:
FileNotFoundError: [Errno 2] No such file or directory: '/mnt/models/model.joblib'
```

**What did you expect to happen:**
It should fail in storage initializer container instead.

**Anything else you would like to add:**
[Miscellaneous information that will assist in solving the issue.]


**Environment:**

- Istio Version: 1.1.7
- Knative Version: 0.8.0
- KFServing Version: 0.2.0
- Kubeflow version:
- Minikube version:
- Kubernetes version: (use `kubectl version`):
- OS (e.g. from `/etc/os-release`):


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/kfserving/kfserving/storage.py`
Content:
```
1 # Copyright 2019 kubeflow.org.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #    http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import glob
16 import logging
17 import tempfile
18 import os
19 import re
20 from urllib.parse import urlparse
21 from azure.common import AzureMissingResourceHttpError
22 from azure.storage.blob import BlockBlobService
23 from google.auth import exceptions
24 from google.cloud import storage
25 from minio import Minio
26 
27 _GCS_PREFIX = "gs://"
28 _S3_PREFIX = "s3://"
29 _BLOB_RE = "https://(.+?).blob.core.windows.net/(.+)"
30 _LOCAL_PREFIX = "file://"
31 
32 
33 class Storage(object): # pylint: disable=too-few-public-methods
34     @staticmethod
35     def download(uri: str, out_dir: str = None) -> str:
36         logging.info("Copying contents of %s to local", uri)
37 
38         is_local = False
39         if uri.startswith(_LOCAL_PREFIX) or os.path.exists(uri):
40             is_local = True
41 
42         if out_dir is None:
43             if is_local:
44                 # noop if out_dir is not set and the path is local
45                 return Storage._download_local(uri)
46             out_dir = tempfile.mkdtemp()
47 
48         if uri.startswith(_GCS_PREFIX):
49             Storage._download_gcs(uri, out_dir)
50         elif uri.startswith(_S3_PREFIX):
51             Storage._download_s3(uri, out_dir)
52         elif re.search(_BLOB_RE, uri):
53             Storage._download_blob(uri, out_dir)
54         elif is_local:
55             return Storage._download_local(uri, out_dir)
56         else:
57             raise Exception("Cannot recognize storage type for " + uri +
58                             "\n'%s', '%s', and '%s' are the current available storage type." %
59                             (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX))
60 
61         logging.info("Successfully copied %s to %s", uri, out_dir)
62         return out_dir
63 
64     @staticmethod
65     def _download_s3(uri, temp_dir: str):
66         client = Storage._create_minio_client()
67         bucket_args = uri.replace(_S3_PREFIX, "", 1).split("/", 1)
68         bucket_name = bucket_args[0]
69         bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
70         objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)
71         for obj in objects:
72             # Replace any prefix from the object key with temp_dir
73             subdir_object_key = obj.object_name.replace(bucket_path, "", 1).strip("/")
74             # fget_object handles directory creation if does not exist
75             if not obj.is_dir:
76                 if subdir_object_key == "":
77                     subdir_object_key = obj.object_name
78                 client.fget_object(bucket_name, obj.object_name,
79                                    os.path.join(temp_dir, subdir_object_key))
80 
81     @staticmethod
82     def _download_gcs(uri, temp_dir: str):
83         try:
84             storage_client = storage.Client()
85         except exceptions.DefaultCredentialsError:
86             storage_client = storage.Client.create_anonymous_client()
87         bucket_args = uri.replace(_GCS_PREFIX, "", 1).split("/", 1)
88         bucket_name = bucket_args[0]
89         bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
90         bucket = storage_client.bucket(bucket_name)
91         prefix = bucket_path
92         if not prefix.endswith("/"):
93             prefix = prefix + "/"
94         blobs = bucket.list_blobs(prefix=prefix)
95         for blob in blobs:
96             # Replace any prefix from the object key with temp_dir
97             subdir_object_key = blob.name.replace(bucket_path, "", 1).strip("/")
98 
99             # Create necessary subdirectory to store the object locally
100             if "/" in subdir_object_key:
101                 local_object_dir = os.path.join(temp_dir, subdir_object_key.rsplit("/", 1)[0])
102                 if not os.path.isdir(local_object_dir):
103                     os.makedirs(local_object_dir, exist_ok=True)
104             if subdir_object_key.strip() != "":
105                 dest_path = os.path.join(temp_dir, subdir_object_key)
106                 logging.info("Downloading: %s", dest_path)
107                 blob.download_to_filename(dest_path)
108 
109     @staticmethod
110     def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals
111         match = re.search(_BLOB_RE, uri)
112         account_name = match.group(1)
113         storage_url = match.group(2)
114         container_name, prefix = storage_url.split("/", 1)
115 
116         logging.info("Connecting to BLOB account: [%s], container: [%s], prefix: [%s]",
117                      account_name,
118                      container_name,
119                      prefix)
120         try:
121             block_blob_service = BlockBlobService(account_name=account_name)
122             blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
123         except AzureMissingResourceHttpError:
124             token = Storage._get_azure_storage_token()
125             if token is None:
126                 logging.warning("Azure credentials not found, retrying anonymous access")
127             block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)
128             blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
129 
130         for blob in blobs:
131             dest_path = os.path.join(out_dir, blob.name)
132             if "/" in blob.name:
133                 head, tail = os.path.split(blob.name)
134                 if prefix is not None:
135                     head = head[len(prefix):]
136                 if head.startswith('/'):
137                     head = head[1:]
138                 dir_path = os.path.join(out_dir, head)
139                 dest_path = os.path.join(dir_path, tail)
140                 if not os.path.isdir(dir_path):
141                     os.makedirs(dir_path)
142 
143             logging.info("Downloading: %s to %s", blob.name, dest_path)
144             block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)
145 
146     @staticmethod
147     def _get_azure_storage_token():
148         tenant_id = os.getenv("AZ_TENANT_ID", "")
149         client_id = os.getenv("AZ_CLIENT_ID", "")
150         client_secret = os.getenv("AZ_CLIENT_SECRET", "")
151         subscription_id = os.getenv("AZ_SUBSCRIPTION_ID", "")
152 
153         if tenant_id == "" or client_id == "" or client_secret == "" or subscription_id == "":
154             return None
155 
156         # note the SP must have "Storage Blob Data Owner" perms for this to work
157         import adal
158         from azure.storage.common import TokenCredential
159 
160         authority_url = "https://login.microsoftonline.com/" + tenant_id
161 
162         context = adal.AuthenticationContext(authority_url)
163 
164         token = context.acquire_token_with_client_credentials(
165             "https://storage.azure.com/",
166             client_id,
167             client_secret)
168 
169         token_credential = TokenCredential(token["accessToken"])
170 
171         logging.info("Retrieved SP token credential for client_id: %s", client_id)
172 
173         return token_credential
174 
175     @staticmethod
176     def _download_local(uri, out_dir=None):
177         local_path = uri.replace(_LOCAL_PREFIX, "", 1)
178         if not os.path.exists(local_path):
179             raise Exception("Local path %s does not exist." % (uri))
180 
181         if out_dir is None:
182             return local_path
183         elif not os.path.isdir(out_dir):
184             os.makedirs(out_dir)
185 
186         if os.path.isdir(local_path):
187             local_path = os.path.join(local_path, "*")
188 
189         for src in glob.glob(local_path):
190             _, tail = os.path.split(src)
191             dest_path = os.path.join(out_dir, tail)
192             logging.info("Linking: %s to %s", src, dest_path)
193             os.symlink(src, dest_path)
194         return out_dir
195 
196     @staticmethod
197     def _create_minio_client():
198         # Remove possible http scheme for Minio
199         url = urlparse(os.getenv("AWS_ENDPOINT_URL", ""))
200         use_ssl = url.scheme == 'https' if url.scheme else bool(os.getenv("S3_USE_HTTPS", "true"))
201         return Minio(url.netloc,
202                      access_key=os.getenv("AWS_ACCESS_KEY_ID", ""),
203                      secret_key=os.getenv("AWS_SECRET_ACCESS_KEY", ""),
204                      secure=use_ssl)
205
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python/kfserving/kfserving/storage.py b/python/kfserving/kfserving/storage.py
--- a/python/kfserving/kfserving/storage.py
+++ b/python/kfserving/kfserving/storage.py
@@ -68,6 +68,7 @@
         bucket_name = bucket_args[0]
         bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
         objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)
+        count = 0
         for obj in objects:
             # Replace any prefix from the object key with temp_dir
             subdir_object_key = obj.object_name.replace(bucket_path, "", 1).strip("/")
@@ -77,6 +78,10 @@
                     subdir_object_key = obj.object_name
                 client.fget_object(bucket_name, obj.object_name,
                                    os.path.join(temp_dir, subdir_object_key))
+            count = count + 1
+        if count == 0:
+            raise RuntimeError("Failed to fetch model. \
+The path or model %s does not exist." % (uri))
 
     @staticmethod
     def _download_gcs(uri, temp_dir: str):
@@ -92,6 +97,7 @@
         if not prefix.endswith("/"):
             prefix = prefix + "/"
         blobs = bucket.list_blobs(prefix=prefix)
+        count = 0
         for blob in blobs:
             # Replace any prefix from the object key with temp_dir
             subdir_object_key = blob.name.replace(bucket_path, "", 1).strip("/")
@@ -105,6 +111,10 @@
                 dest_path = os.path.join(temp_dir, subdir_object_key)
                 logging.info("Downloading: %s", dest_path)
                 blob.download_to_filename(dest_path)
+            count = count + 1
+        if count == 0:
+            raise RuntimeError("Failed to fetch model. \
+The path or model %s does not exist." % (uri))
 
     @staticmethod
     def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals
@@ -126,7 +136,7 @@
                 logging.warning("Azure credentials not found, retrying anonymous access")
             block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)
             blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
-
+        count = 0
         for blob in blobs:
             dest_path = os.path.join(out_dir, blob.name)
             if "/" in blob.name:
@@ -142,6 +152,10 @@
 
             logging.info("Downloading: %s to %s", blob.name, dest_path)
             block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)
+            count = count + 1
+        if count == 0:
+            raise RuntimeError("Failed to fetch model. \
+The path or model %s does not exist." % (uri))
 
     @staticmethod
     def _get_azure_storage_token():
@@ -176,7 +190,7 @@
     def _download_local(uri, out_dir=None):
         local_path = uri.replace(_LOCAL_PREFIX, "", 1)
         if not os.path.exists(local_path):
-            raise Exception("Local path %s does not exist." % (uri))
+            raise RuntimeError("Local path %s does not exist." % (uri))
 
         if out_dir is None:
             return local_path

{"golden_diff": "diff --git a/python/kfserving/kfserving/storage.py b/python/kfserving/kfserving/storage.py\n--- a/python/kfserving/kfserving/storage.py\n+++ b/python/kfserving/kfserving/storage.py\n@@ -68,6 +68,7 @@\n         bucket_name = bucket_args[0]\n         bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n         objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)\n+        count = 0\n         for obj in objects:\n             # Replace any prefix from the object key with temp_dir\n             subdir_object_key = obj.object_name.replace(bucket_path, \"\", 1).strip(\"/\")\n@@ -77,6 +78,10 @@\n                     subdir_object_key = obj.object_name\n                 client.fget_object(bucket_name, obj.object_name,\n                                    os.path.join(temp_dir, subdir_object_key))\n+            count = count + 1\n+        if count == 0:\n+            raise RuntimeError(\"Failed to fetch model. \\\n+The path or model %s does not exist.\" % (uri))\n \n     @staticmethod\n     def _download_gcs(uri, temp_dir: str):\n@@ -92,6 +97,7 @@\n         if not prefix.endswith(\"/\"):\n             prefix = prefix + \"/\"\n         blobs = bucket.list_blobs(prefix=prefix)\n+        count = 0\n         for blob in blobs:\n             # Replace any prefix from the object key with temp_dir\n             subdir_object_key = blob.name.replace(bucket_path, \"\", 1).strip(\"/\")\n@@ -105,6 +111,10 @@\n                 dest_path = os.path.join(temp_dir, subdir_object_key)\n                 logging.info(\"Downloading: %s\", dest_path)\n                 blob.download_to_filename(dest_path)\n+            count = count + 1\n+        if count == 0:\n+            raise RuntimeError(\"Failed to fetch model. \\\n+The path or model %s does not exist.\" % (uri))\n \n     @staticmethod\n     def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals\n@@ -126,7 +136,7 @@\n                 logging.warning(\"Azure credentials not found, retrying anonymous access\")\n             block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)\n             blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n-\n+        count = 0\n         for blob in blobs:\n             dest_path = os.path.join(out_dir, blob.name)\n             if \"/\" in blob.name:\n@@ -142,6 +152,10 @@\n \n             logging.info(\"Downloading: %s to %s\", blob.name, dest_path)\n             block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)\n+            count = count + 1\n+        if count == 0:\n+            raise RuntimeError(\"Failed to fetch model. \\\n+The path or model %s does not exist.\" % (uri))\n \n     @staticmethod\n     def _get_azure_storage_token():\n@@ -176,7 +190,7 @@\n     def _download_local(uri, out_dir=None):\n         local_path = uri.replace(_LOCAL_PREFIX, \"\", 1)\n         if not os.path.exists(local_path):\n-            raise Exception(\"Local path %s does not exist.\" % (uri))\n+            raise RuntimeError(\"Local path %s does not exist.\" % (uri))\n \n         if out_dir is None:\n             return local_path\n", "issue": "Storage initializer container succeeded even with invalid storage uri\n/kind bug\r\n\r\n**What steps did you take and what happened:**\r\nSet `storageUri` with an invalid uri `s3://platform-test/s3/moviesentiment` (s3 subdir does not exist)\r\n\r\nThe storage initializer container succeeded,\r\n```\r\nkubectl logs moviesentiment-predictor-default-xh8c8-deployment-5955bbccxfnfc -c storage-initializer\r\nINFO:root:Initializing, args: src_uri [s3://platform-test/s3/moviesentiment] dest_path[ [/mnt/models]\r\nINFO:root:Copying contents of s3://platform-test/s3/moviesentiment to local\r\nINFO:root:Successfully copied s3://platform-test/s3/moviesentiment to /mnt/models\r\n```\r\nBut main container failed\r\n```\r\nkubectl logs moviesentiment-predictor-default-xh8c8-deployment-5955bbccxfnfc -c user-container\r\nINFO:root:Copying contents of /mnt/models to local\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.7/runpy.py\", line 193, in _run_module_as_main\r\n    \"__main__\", mod_spec)\r\n  File \"/usr/local/lib/python3.7/runpy.py\", line 85, in _run_code\r\n    exec(code, run_globals)\r\n  File \"/sklearnserver/sklearnserver/__main__.py\", line 33, in <module>\r\n    model.load()\r\n  File \"/sklearnserver/sklearnserver/model.py\", line 32, in load\r\n    self._joblib = joblib.load(model_file) #pylint:disable=attribute-defined-outside-init\r\n  File \"/usr/local/lib/python3.7/site-packages/joblib/numpy_pickle.py\", line 597, in load\r\n    with open(filename, 'rb') as f:\r\nFileNotFoundError: [Errno 2] No such file or directory: '/mnt/models/model.joblib'\r\n```\r\n\r\n**What did you expect to happen:**\r\nIt should fail in storage initializer container instead.\r\n\r\n**Anything else you would like to add:**\r\n[Miscellaneous information that will assist in solving the issue.]\r\n\r\n\r\n**Environment:**\r\n\r\n- Istio Version: 1.1.7\r\n- Knative Version: 0.8.0\r\n- KFServing Version: 0.2.0\r\n- Kubeflow version:\r\n- Minikube version:\r\n- Kubernetes version: (use `kubectl version`):\r\n- OS (e.g. from `/etc/os-release`):\r\n\n", "before_files": [{"content": "# Copyright 2019 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport glob\nimport logging\nimport tempfile\nimport os\nimport re\nfrom urllib.parse import urlparse\nfrom azure.common import AzureMissingResourceHttpError\nfrom azure.storage.blob import BlockBlobService\nfrom google.auth import exceptions\nfrom google.cloud import storage\nfrom minio import Minio\n\n_GCS_PREFIX = \"gs://\"\n_S3_PREFIX = \"s3://\"\n_BLOB_RE = \"https://(.+?).blob.core.windows.net/(.+)\"\n_LOCAL_PREFIX = \"file://\"\n\n\nclass Storage(object): # pylint: disable=too-few-public-methods\n    @staticmethod\n    def download(uri: str, out_dir: str = None) -> str:\n        logging.info(\"Copying contents of %s to local\", uri)\n\n        is_local = False\n        if uri.startswith(_LOCAL_PREFIX) or os.path.exists(uri):\n            is_local = True\n\n        if out_dir is None:\n            if is_local:\n                # noop if out_dir is not set and the path is local\n                return Storage._download_local(uri)\n            out_dir = tempfile.mkdtemp()\n\n        if uri.startswith(_GCS_PREFIX):\n            Storage._download_gcs(uri, out_dir)\n        elif uri.startswith(_S3_PREFIX):\n            Storage._download_s3(uri, out_dir)\n        elif re.search(_BLOB_RE, uri):\n            Storage._download_blob(uri, out_dir)\n        elif is_local:\n            return Storage._download_local(uri, out_dir)\n        else:\n            raise Exception(\"Cannot recognize storage type for \" + uri +\n                            \"\\n'%s', '%s', and '%s' are the current available storage type.\" %\n                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX))\n\n        logging.info(\"Successfully copied %s to %s\", uri, out_dir)\n        return out_dir\n\n    @staticmethod\n    def _download_s3(uri, temp_dir: str):\n        client = Storage._create_minio_client()\n        bucket_args = uri.replace(_S3_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)\n        for obj in objects:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = obj.object_name.replace(bucket_path, \"\", 1).strip(\"/\")\n            # fget_object handles directory creation if does not exist\n            if not obj.is_dir:\n                if subdir_object_key == \"\":\n                    subdir_object_key = obj.object_name\n                client.fget_object(bucket_name, obj.object_name,\n                                   os.path.join(temp_dir, subdir_object_key))\n\n    @staticmethod\n    def _download_gcs(uri, temp_dir: str):\n        try:\n            storage_client = storage.Client()\n        except exceptions.DefaultCredentialsError:\n            storage_client = storage.Client.create_anonymous_client()\n        bucket_args = uri.replace(_GCS_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        bucket = storage_client.bucket(bucket_name)\n        prefix = bucket_path\n        if not prefix.endswith(\"/\"):\n            prefix = prefix + \"/\"\n        blobs = bucket.list_blobs(prefix=prefix)\n        for blob in blobs:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = blob.name.replace(bucket_path, \"\", 1).strip(\"/\")\n\n            # Create necessary subdirectory to store the object locally\n            if \"/\" in subdir_object_key:\n                local_object_dir = os.path.join(temp_dir, subdir_object_key.rsplit(\"/\", 1)[0])\n                if not os.path.isdir(local_object_dir):\n                    os.makedirs(local_object_dir, exist_ok=True)\n            if subdir_object_key.strip() != \"\":\n                dest_path = os.path.join(temp_dir, subdir_object_key)\n                logging.info(\"Downloading: %s\", dest_path)\n                blob.download_to_filename(dest_path)\n\n    @staticmethod\n    def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals\n        match = re.search(_BLOB_RE, uri)\n        account_name = match.group(1)\n        storage_url = match.group(2)\n        container_name, prefix = storage_url.split(\"/\", 1)\n\n        logging.info(\"Connecting to BLOB account: [%s], container: [%s], prefix: [%s]\",\n                     account_name,\n                     container_name,\n                     prefix)\n        try:\n            block_blob_service = BlockBlobService(account_name=account_name)\n            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n        except AzureMissingResourceHttpError:\n            token = Storage._get_azure_storage_token()\n            if token is None:\n                logging.warning(\"Azure credentials not found, retrying anonymous access\")\n            block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)\n            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n\n        for blob in blobs:\n            dest_path = os.path.join(out_dir, blob.name)\n            if \"/\" in blob.name:\n                head, tail = os.path.split(blob.name)\n                if prefix is not None:\n                    head = head[len(prefix):]\n                if head.startswith('/'):\n                    head = head[1:]\n                dir_path = os.path.join(out_dir, head)\n                dest_path = os.path.join(dir_path, tail)\n                if not os.path.isdir(dir_path):\n                    os.makedirs(dir_path)\n\n            logging.info(\"Downloading: %s to %s\", blob.name, dest_path)\n            block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)\n\n    @staticmethod\n    def _get_azure_storage_token():\n        tenant_id = os.getenv(\"AZ_TENANT_ID\", \"\")\n        client_id = os.getenv(\"AZ_CLIENT_ID\", \"\")\n        client_secret = os.getenv(\"AZ_CLIENT_SECRET\", \"\")\n        subscription_id = os.getenv(\"AZ_SUBSCRIPTION_ID\", \"\")\n\n        if tenant_id == \"\" or client_id == \"\" or client_secret == \"\" or subscription_id == \"\":\n            return None\n\n        # note the SP must have \"Storage Blob Data Owner\" perms for this to work\n        import adal\n        from azure.storage.common import TokenCredential\n\n        authority_url = \"https://login.microsoftonline.com/\" + tenant_id\n\n        context = adal.AuthenticationContext(authority_url)\n\n        token = context.acquire_token_with_client_credentials(\n            \"https://storage.azure.com/\",\n            client_id,\n            client_secret)\n\n        token_credential = TokenCredential(token[\"accessToken\"])\n\n        logging.info(\"Retrieved SP token credential for client_id: %s\", client_id)\n\n        return token_credential\n\n    @staticmethod\n    def _download_local(uri, out_dir=None):\n        local_path = uri.replace(_LOCAL_PREFIX, \"\", 1)\n        if not os.path.exists(local_path):\n            raise Exception(\"Local path %s does not exist.\" % (uri))\n\n        if out_dir is None:\n            return local_path\n        elif not os.path.isdir(out_dir):\n            os.makedirs(out_dir)\n\n        if os.path.isdir(local_path):\n            local_path = os.path.join(local_path, \"*\")\n\n        for src in glob.glob(local_path):\n            _, tail = os.path.split(src)\n            dest_path = os.path.join(out_dir, tail)\n            logging.info(\"Linking: %s to %s\", src, dest_path)\n            os.symlink(src, dest_path)\n        return out_dir\n\n    @staticmethod\n    def _create_minio_client():\n        # Remove possible http scheme for Minio\n        url = urlparse(os.getenv(\"AWS_ENDPOINT_URL\", \"\"))\n        use_ssl = url.scheme == 'https' if url.scheme else bool(os.getenv(\"S3_USE_HTTPS\", \"true\"))\n        return Minio(url.netloc,\n                     access_key=os.getenv(\"AWS_ACCESS_KEY_ID\", \"\"),\n                     secret_key=os.getenv(\"AWS_SECRET_ACCESS_KEY\", \"\"),\n                     secure=use_ssl)\n", "path": "python/kfserving/kfserving/storage.py"}], "after_files": [{"content": "# Copyright 2019 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport glob\nimport logging\nimport tempfile\nimport os\nimport re\nfrom urllib.parse import urlparse\nfrom azure.common import AzureMissingResourceHttpError\nfrom azure.storage.blob import BlockBlobService\nfrom google.auth import exceptions\nfrom google.cloud import storage\nfrom minio import Minio\n\n_GCS_PREFIX = \"gs://\"\n_S3_PREFIX = \"s3://\"\n_BLOB_RE = \"https://(.+?).blob.core.windows.net/(.+)\"\n_LOCAL_PREFIX = \"file://\"\n\n\nclass Storage(object): # pylint: disable=too-few-public-methods\n    @staticmethod\n    def download(uri: str, out_dir: str = None) -> str:\n        logging.info(\"Copying contents of %s to local\", uri)\n\n        is_local = False\n        if uri.startswith(_LOCAL_PREFIX) or os.path.exists(uri):\n            is_local = True\n\n        if out_dir is None:\n            if is_local:\n                # noop if out_dir is not set and the path is local\n                return Storage._download_local(uri)\n            out_dir = tempfile.mkdtemp()\n\n        if uri.startswith(_GCS_PREFIX):\n            Storage._download_gcs(uri, out_dir)\n        elif uri.startswith(_S3_PREFIX):\n            Storage._download_s3(uri, out_dir)\n        elif re.search(_BLOB_RE, uri):\n            Storage._download_blob(uri, out_dir)\n        elif is_local:\n            return Storage._download_local(uri, out_dir)\n        else:\n            raise Exception(\"Cannot recognize storage type for \" + uri +\n                            \"\\n'%s', '%s', and '%s' are the current available storage type.\" %\n                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX))\n\n        logging.info(\"Successfully copied %s to %s\", uri, out_dir)\n        return out_dir\n\n    @staticmethod\n    def _download_s3(uri, temp_dir: str):\n        client = Storage._create_minio_client()\n        bucket_args = uri.replace(_S3_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)\n        count = 0\n        for obj in objects:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = obj.object_name.replace(bucket_path, \"\", 1).strip(\"/\")\n            # fget_object handles directory creation if does not exist\n            if not obj.is_dir:\n                if subdir_object_key == \"\":\n                    subdir_object_key = obj.object_name\n                client.fget_object(bucket_name, obj.object_name,\n                                   os.path.join(temp_dir, subdir_object_key))\n            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\n\n    @staticmethod\n    def _download_gcs(uri, temp_dir: str):\n        try:\n            storage_client = storage.Client()\n        except exceptions.DefaultCredentialsError:\n            storage_client = storage.Client.create_anonymous_client()\n        bucket_args = uri.replace(_GCS_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        bucket = storage_client.bucket(bucket_name)\n        prefix = bucket_path\n        if not prefix.endswith(\"/\"):\n            prefix = prefix + \"/\"\n        blobs = bucket.list_blobs(prefix=prefix)\n        count = 0\n        for blob in blobs:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = blob.name.replace(bucket_path, \"\", 1).strip(\"/\")\n\n            # Create necessary subdirectory to store the object locally\n            if \"/\" in subdir_object_key:\n                local_object_dir = os.path.join(temp_dir, subdir_object_key.rsplit(\"/\", 1)[0])\n                if not os.path.isdir(local_object_dir):\n                    os.makedirs(local_object_dir, exist_ok=True)\n            if subdir_object_key.strip() != \"\":\n                dest_path = os.path.join(temp_dir, subdir_object_key)\n                logging.info(\"Downloading: %s\", dest_path)\n                blob.download_to_filename(dest_path)\n            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\n\n    @staticmethod\n    def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals\n        match = re.search(_BLOB_RE, uri)\n        account_name = match.group(1)\n        storage_url = match.group(2)\n        container_name, prefix = storage_url.split(\"/\", 1)\n\n        logging.info(\"Connecting to BLOB account: [%s], container: [%s], prefix: [%s]\",\n                     account_name,\n                     container_name,\n                     prefix)\n        try:\n            block_blob_service = BlockBlobService(account_name=account_name)\n            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n        except AzureMissingResourceHttpError:\n            token = Storage._get_azure_storage_token()\n            if token is None:\n                logging.warning(\"Azure credentials not found, retrying anonymous access\")\n            block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)\n            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n        count = 0\n        for blob in blobs:\n            dest_path = os.path.join(out_dir, blob.name)\n            if \"/\" in blob.name:\n                head, tail = os.path.split(blob.name)\n                if prefix is not None:\n                    head = head[len(prefix):]\n                if head.startswith('/'):\n                    head = head[1:]\n                dir_path = os.path.join(out_dir, head)\n                dest_path = os.path.join(dir_path, tail)\n                if not os.path.isdir(dir_path):\n                    os.makedirs(dir_path)\n\n            logging.info(\"Downloading: %s to %s\", blob.name, dest_path)\n            block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)\n            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\n\n    @staticmethod\n    def _get_azure_storage_token():\n        tenant_id = os.getenv(\"AZ_TENANT_ID\", \"\")\n        client_id = os.getenv(\"AZ_CLIENT_ID\", \"\")\n        client_secret = os.getenv(\"AZ_CLIENT_SECRET\", \"\")\n        subscription_id = os.getenv(\"AZ_SUBSCRIPTION_ID\", \"\")\n\n        if tenant_id == \"\" or client_id == \"\" or client_secret == \"\" or subscription_id == \"\":\n            return None\n\n        # note the SP must have \"Storage Blob Data Owner\" perms for this to work\n        import adal\n        from azure.storage.common import TokenCredential\n\n        authority_url = \"https://login.microsoftonline.com/\" + tenant_id\n\n        context = adal.AuthenticationContext(authority_url)\n\n        token = context.acquire_token_with_client_credentials(\n            \"https://storage.azure.com/\",\n            client_id,\n            client_secret)\n\n        token_credential = TokenCredential(token[\"accessToken\"])\n\n        logging.info(\"Retrieved SP token credential for client_id: %s\", client_id)\n\n        return token_credential\n\n    @staticmethod\n    def _download_local(uri, out_dir=None):\n        local_path = uri.replace(_LOCAL_PREFIX, \"\", 1)\n        if not os.path.exists(local_path):\n            raise RuntimeError(\"Local path %s does not exist.\" % (uri))\n\n        if out_dir is None:\n            return local_path\n        elif not os.path.isdir(out_dir):\n            os.makedirs(out_dir)\n\n        if os.path.isdir(local_path):\n            local_path = os.path.join(local_path, \"*\")\n\n        for src in glob.glob(local_path):\n            _, tail = os.path.split(src)\n            dest_path = os.path.join(out_dir, tail)\n            logging.info(\"Linking: %s to %s\", src, dest_path)\n            os.symlink(src, dest_path)\n        return out_dir\n\n    @staticmethod\n    def _create_minio_client():\n        # Remove possible http scheme for Minio\n        url = urlparse(os.getenv(\"AWS_ENDPOINT_URL\", \"\"))\n        use_ssl = url.scheme == 'https' if url.scheme else bool(os.getenv(\"S3_USE_HTTPS\", \"true\"))\n        return Minio(url.netloc,\n                     access_key=os.getenv(\"AWS_ACCESS_KEY_ID\", \"\"),\n                     secret_key=os.getenv(\"AWS_SECRET_ACCESS_KEY\", \"\"),\n                     secure=use_ssl)\n", "path": "python/kfserving/kfserving/storage.py"}]}