Repository: Netflix/metaflow
Pull request: Adding support for Azure Blob Storage as a datastore
The primary change is implementing `AzureStorage` (analogous to existing `S3Storage`, `LocalStorage`).  We are consciously deferring the decision of having first class "data tools" support for Azure.

There are some necessary changes to ensure full Azure support on all Metaflow surfaces:
* includefile
* conda
* cards
* mflog
* kubernetes
* argo

We take care to ensure there is no cross disruption to users not using Azure.  More specifically:
* Users need to setup AWS dependencies (boto3, config params), iff they are using AWS.
* Users need to setup Azure dependencies (azure SDK libs, config params), iff they are using Azure.

We aggressively use local imports to achieve this.

Much effort was also spent to ensure good performance of Metaflow's usage of Azure Blob Storage.   See context docs for more details.

Some docs for context:
* Overall [Metaflow-on-Azure](https://docs.google.com/document/d/1_W7zDiImLtxY0Xq47fohGvtEPrd3KWV4ULfOxYNjlhs/edit#heading=h.6xrup8gk65q4) effort.
* AzureStorage [implementation](https://docs.google.com/document/d/18_WiFdagdaV0ATfHcK7CfG6wMTpap691hAoMxlfCMEg/edit#heading=h.ikim125j24da) notes (may not match up 100% to code in PR in minor ways, but the bones should be consistent).
  * [Section on performant implementation.](https://docs.google.com/document/d/18_WiFdagdaV0ATfHcK7CfG6wMTpap691hAoMxlfCMEg/edit#heading=h.x77ca5o2wmep)
* [Performance discovery](https://docs.google.com/document/d/1PkR_0Fz9B3OJukyIKwD-QiUC4jMeAGx9Snh_cUQyfNg/edit#heading=h.z4v8dso6utye) (pre-implementation).


Problem statement:
Support for another public cloud - Microsoft Azure
Currently, Metaflow is set up to work with AWS as the default public cloud. The architecture of Metaflow allows for additional public clouds to be supported.

Adding support for Microsoft Azure might broaden the potential user base, which could increase the adaption rate. This, in turn, could lead to increased community attention.

Hints from issue discussion:
I can dedicate a few hours here and there for Azure support, but I don't have time to take the reins on this one. If someone goes through the trouble of designing and proposing a solution and could use some extra hands for the implementation, loop me in.
@gerryhernandez: I have asked my contacts at Microsoft (Norwegian HQ) whether they would be willing to pitch in with funding and/or time from their engineers.
Yes, any update from Microsoft? If they don't have plan to do so. Can we fork a branch and add Azure enhancements on our own?
Hello! I'm in discussion with my colleagues from Microsoft Norway about this project. @jwang01 do you want to help with implementing?
What's the main challenge you can see now? Converting the AWS Cloudformation templates to ARM templates? 
That might be a good start :)
I wonder if Kubernetes/Helm would be a better option than ARM? The result would then potentially be cloud-agnostic.
Any chances of this getting traction
> Any chances of this getting traction

Ditto.  Any luck?
Also curious about this feature. Any updates, @webmaxru or @gerryhernandez?

Commit message: remove signature

Patch to review:
diff --git a/metaflow/datastore/__init__.py b/metaflow/datastore/__init__.py
index 794ea23ccef..8f0f5f97717 100644
--- a/metaflow/datastore/__init__.py
+++ b/metaflow/datastore/__init__.py
@@ -5,5 +5,6 @@
 
 from .local_storage import LocalStorage
 from .s3_storage import S3Storage
+from .azure_storage import AzureStorage
 
-DATASTORES = {"local": LocalStorage, "s3": S3Storage}
+DATASTORES = {"local": LocalStorage, "s3": S3Storage, "azure": AzureStorage}
diff --git a/metaflow/datastore/azure_storage.py b/metaflow/datastore/azure_storage.py
new file mode 100644
index 00000000000..fd00626a99c
--- /dev/null
+++ b/metaflow/datastore/azure_storage.py
@@ -0,0 +1,550 @@
+import math
+import platform
+
+import multiprocessing
+import json
+import os
+import shutil
+import uuid
+import sys
+import time
+from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, as_completed
+from tempfile import mkdtemp
+
+from metaflow.datastore.datastore_storage import DataStoreStorage, CloseAfterUse
+from metaflow.exception import MetaflowInternalError, MetaflowException
+from metaflow.metaflow_config import (
+    DATASTORE_SYSROOT_AZURE,
+    ARTIFACT_LOCALROOT,
+    AZURE_STORAGE_WORKLOAD_TYPE,
+)
+from metaflow.plugins.azure.azure_utils import (
+    check_azure_deps,
+    process_exception,
+    handle_exceptions,
+    create_static_token_credential,
+    parse_azure_full_path,
+)
+
+from metaflow.plugins.azure.blob_service_client_factory import (
+    get_azure_blob_service_client,
+)
+
+if sys.version_info[:2] < (3, 7):
+    # in 3.6, Only BrokenProcessPool exists (there is no BrokenThreadPool)
+    from concurrent.futures.process import BrokenProcessPool as BrokenExecutor
+else:
+    # in 3.7 and newer, BrokenExecutor is a base class that parents BrokenProcessPool AND BrokenThreadPool
+    from concurrent.futures import BrokenExecutor
+
+# TODO keyboard interrupt crazy tracebacks in process pool
+
+
+def _determine_effective_cpu_limit():
+    """Calculate CPU limit (in number of cores) based on:
+
+    - /sys/fs/cgroup/cpu/cpu.max (if available, cgroup 2)
+    OR
+    - /sys/fs/cgroup/cpu/cpu.cfs_quota_us
+    - /sys/fs/cgroup/cpu/cpu.cfs_period_us
+
+    Returns:
+        > 0 if limit was successfully calculated
+        = 0 if we determined that there is no limit
+        -1 if we failed to determine the limit
+    """
+    try:
+        if platform.system() == "Darwin":
+            # On MacOS, and not in a container
+            # We are assuming it is extremely out-of-the-way to run a Darwin container
+            return 0
+        elif platform.system() == "Linux":
+            # Bare metal Linux, or a Linux container running on either Linux or MacOS system
+            # Linux containers on MacOS have this cpu.max file
+            with open("/sys/fs/cgroup/cpu.max", "rb") as f:
+                parts = f.read().decode("utf-8").split(" ")
+                if len(parts) == 2:
+                    if parts[0] == "max":
+                        return 0
+                    return int(parts[0]) / int(parts[1])
+
+            with open("/sys/fs/cgroup/cpu/cpu.cfs_quota_us", "rb") as f:
+                quota = int(f.read())
+                # this file shows -1 for no limit
+                if quota == -1:
+                    return 0
+                # some other negative number - this is weird...return "undetermined"
+                if quota < 0:
+                    return -1
+            with open("/sys/fs/cgroup/cpu/cpu.cfs_period_us", "rb") as f:
+                period = int(f.read())
+                # should be positive. we don't want div by zero errors.
+                if period <= 0:
+                    return -1
+                return quota / period
+        else:
+            # Not Linux or MacOS...give up and return "undetermined"
+            return -1
+    except Exception:
+        return -1
+
+
+def _noop_for_executor_warm_up():
+    pass
+
+
+def _compute_executor_max_workers():
+    # For processes, let's start conservative. Let's restrict to 4-18 always. Can be configurable one day.
+    min_processes = 4
+    max_processes = 18
+    # make an effort to get cgroup cpu limits
+    effective_cpu_limit = _determine_effective_cpu_limit()
+
+    def _bracket(min_v, v, max_v):
+        assert min_v <= max_v
+        if v < min_v:
+            return min_v
+        if v > max_v:
+            return max_v
+        return v
+
+    if effective_cpu_limit < 0:
+        # We don't know the limit, stick to min
+        processpool_max_workers = min_processes
+    elif effective_cpu_limit == 0:
+        # There is (probably) no limit, use physical core count
+        processpool_max_workers = _bracket(
+            min_processes, os.cpu_count() or 1, max_processes
+        )
+    else:
+        # There is a limit, so let's bracket it within min / max
+        processpool_max_workers = _bracket(
+            min_processes, math.ceil(effective_cpu_limit), max_processes
+        )
+    # Threads a lighter than processes... so just tack on a little more
+    threadpool_max_workers = processpool_max_workers + 4
+    return processpool_max_workers, threadpool_max_workers
+
+
+class AzureStorageExecutor(object):
+    """Thin wrapper around a ProcessPoolExecutor, or a ThreadPoolExecutor where
+    the former may be unsafe.
+    """
+
+    def __init__(self, use_processes=False):
+        (
+            processpool_max_workers,
+            threadpool_max_workers,
+        ) = _compute_executor_max_workers()
+        if use_processes:
+            mp_start_method = multiprocessing.get_start_method(allow_none=True)
+            if mp_start_method == "spawn":
+                self._executor = ProcessPoolExecutor(
+                    max_workers=processpool_max_workers
+                )
+            elif sys.version_info[:2] >= (3, 7):
+                self._executor = ProcessPoolExecutor(
+                    mp_context=multiprocessing.get_context("spawn"),
+                    max_workers=processpool_max_workers,
+                )
+            else:
+                raise MetaflowException(
+                    msg="Cannot use ProcessPoolExecutor because Python version is older than 3.7 and multiprocess start method has been set to something other than 'spawn'"
+                )
+        else:
+            self._executor = ThreadPoolExecutor(max_workers=threadpool_max_workers)
+
+    def warm_up(self):
+        # warm up at least one process or thread in the pool.
+        # we don't await future... just let it complete in background
+        self._executor.submit(_noop_for_executor_warm_up)
+
+    def submit(self, *args, **kwargs):
+        return self._executor.submit(*args, **kwargs)
+
+
+# How many threads / connections to use per upload or download operation
+AZURE_STORAGE_DOWNLOAD_MAX_CONCURRENCY = 4
+AZURE_STORAGE_UPLOAD_MAX_CONCURRENCY = 16
+
+BYTES_IN_MB = 1024 * 1024
+
+AZURE_STORAGE_DEFAULT_SCOPE = "https://storage.azure.com/.default"
+
+
+class _AzureRootClient(object):
+    """
+    This exists independently from AzureBlobStorage as a wrapper around SDK clients.
+    It carries around parameters needed to construct Azure SDK clients on demand.
+
+    _AzureRootClient objects will be passed from main to worker threads or processes. They
+    must be picklable. We delay constructing Azure SDK objects because they are not picklable.
+
+    For example, azure.core.TokenCredential objects are not picklable. Therefore we pass around an
+    AccessToken (token) instead, and construct TokenCredential on demand in the target thread (or process).
+    Note that we do this to amortize credential retrieval cost across threads (or processes). Depending on
+    the credential methods available to DefaultAzureCredential, credential retrieval can be expensive.
+    E.g. Azure CLI based credential may take 500-1000ms.
+
+    _AzureRootClient  also carries around with it blob methods that operate relative to
+    datastore_root.
+    """
+
+    def __init__(self, datastore_root=None, token=None, shared_access_signature=None):
+        if datastore_root is None:
+            raise MetaflowInternalError("datastore_root must be set")
+        if token is None and shared_access_signature is None:
+            raise MetaflowInternalError(
+                "either shared_access_signature or token must be set"
+            )
+        if token and shared_access_signature:
+            raise MetaflowInternalError(
+                "cannot set both shared_access_signature and token"
+            )
+        self._datastore_root = datastore_root
+        self._token = token
+        self._shared_access_signature = shared_access_signature
+
+    def get_datastore_root(self):
+        return self._datastore_root
+
+    def get_blob_container_client(self):
+        if self._shared_access_signature:
+            credential = self._shared_access_signature
+            credential_is_cacheable = True
+        else:
+            credential = create_static_token_credential(self._token)
+            credential_is_cacheable = True
+        service = get_azure_blob_service_client(
+            credential=credential,
+            credential_is_cacheable=credential_is_cacheable,
+        )
+        # datastore_root is <container_name>/<blob_prefix>
+        container_name, _ = parse_azure_full_path(self._datastore_root)
+        return service.get_container_client(container_name)
+
+    def get_blob_client(self, path):
+        container = self.get_blob_container_client()
+        blob_full_path = self.get_blob_full_path(path)
+        return container.get_blob_client(blob_full_path)
+
+    def get_blob_full_path(self, path):
+        """
+        Full path means <blob_prefix>/<path> where:
+        datastore_root is <container_name>/<blob_prefix>
+        """
+        _, blob_prefix = parse_azure_full_path(self._datastore_root)
+        if blob_prefix is None:
+            return path
+        path = path.lstrip("/")
+        return "/".join([blob_prefix, path])
+
+    # Azure blob operations. These are meant to be single units of work
+    # to be performed by thread or process pool workers.
+    def is_file_single(self, path):
+        """Drives AzureStorage.is_file()"""
+        try:
+            blob = self.get_blob_client(path)
+            return blob.exists()
+        except Exception as e:
+            process_exception(e)
+
+    def save_bytes_single(
+        self,
+        path_tmpfile_metadata_triple,
+        overwrite=False,
+    ):
+        """Drives AzureStorage.save_bytes()"""
+        try:
+            path, tmpfile, metadata = path_tmpfile_metadata_triple
+
+            metadata_to_upload = None
+            if metadata:
+                metadata_to_upload = {
+                    # Azure metadata rules:
+                    # https://docs.microsoft.com/en-us/azure/storage/blobs/storage-blob-properties-metadata#set-and-retrieve-metadata
+                    # https://docs.microsoft.com/en-us/rest/api/storageservices/setting-and-retrieving-properties-and-metadata-for-blob-resources#Subheading1
+                    "metaflow_user_attributes": json.dumps(metadata),
+                }
+            blob = self.get_blob_client(path)
+            from azure.core.exceptions import ResourceExistsError
+
+            with open(tmpfile, "rb") as byte_stream:
+                try:
+                    # This is a racy existence check worth doing.
+                    # It is good enough 99.9% of the time.
+                    # Depending on ResourceExistsError is more costly, though
+                    # we are still going to handle it right.
+                    if overwrite or not blob.exists():
+                        blob.upload_blob(
+                            byte_stream,
+                            overwrite=overwrite,
+                            metadata=metadata_to_upload,
+                            max_concurrency=AZURE_STORAGE_UPLOAD_MAX_CONCURRENCY,
+                        )
+                except ResourceExistsError:
+                    if overwrite:
+                        # this is an unexpected condition - operation should not complain about
+                        # resource exists if we already said it's fine to overwrite
+                        raise
+                    else:
+                        # we did not want to overwrite. We swallow the exception because the behavior we
+                        # want is "try to upload, but just no-op if already exists".
+                        # this is consistent with S3 and Local implementations.
+                        #
+                        # Note: In other implementations, we may do a pre-upload object existence check.
+                        # Race conditions are possible in those implementations - and that appears to
+                        # be tolerated by our datastore usage patterns.
+                        #
+                        # For Azure, we let azure-storage-blob and underlying REST API handle this. It looks
+                        # race free (as of 6/28/2022)
+                        pass
+        except Exception as e:
+            process_exception(e)
+
+    def load_bytes_single(self, tmpdir, key):
+        """Drives AzureStorage.load_bytes()"""
+        from azure.core.exceptions import ResourceNotFoundError
+
+        try:
+            blob = self.get_blob_client(key)
+            try:
+                blob_properties = blob.get_blob_properties()
+            except ResourceNotFoundError:
+                # load_bytes() needs to return None for keys that don't exist
+                return key, None, None
+            tmp_filename = os.path.join(tmpdir, str(uuid.uuid4()))
+            try:
+                with open(tmp_filename, "wb") as f:
+                    blob.download_blob(
+                        max_concurrency=AZURE_STORAGE_DOWNLOAD_MAX_CONCURRENCY
+                    ).readinto(f)
+                metaflow_user_attributes = None
+                if (
+                    blob_properties.metadata
+                    and "metaflow_user_attributes" in blob_properties.metadata
+                ):
+                    metaflow_user_attributes = json.loads(
+                        blob_properties.metadata["metaflow_user_attributes"]
+                    )
+            except Exception:
+                # clean up the tmp file for the one specific failed load
+                if os.path.exists(tmp_filename):
+                    os.unlink(tmp_filename)
+                raise
+            return key, tmp_filename, metaflow_user_attributes
+        except Exception as e:
+            process_exception(e)
+
+    def list_content_single(self, path):
+        """Drives AzureStorage.list_content()"""
+        try:
+            result = []
+            # all query paths are assumed to be folders. This replicates S3 behavior.
+            path = path.rstrip("/") + "/"
+            full_path = self.get_blob_full_path(path)
+            container = self.get_blob_container_client()
+            # "file" blobs show up as BlobProperties. We assume we always have "blob_type" key
+            # "directories" show up as BlobPrefix. We assume we never have "blob_type" key
+            for blob_properties_or_prefix in container.walk_blobs(
+                name_starts_with=full_path
+            ):
+                name = blob_properties_or_prefix.name
+                # there are other ways. Like checking the returned name ends with slash.
+                # But checking blob_type is more robust
+                is_file = blob_properties_or_prefix.has_key("blob_type")
+                if not is_file:
+                    # for directories we don't want trailing slashes in results
+                    name = name.rstrip("/")
+
+                # Now massage the resulting blob paths - we need to strip off the common blob prefix
+                _, top_level_blob_prefix = parse_azure_full_path(
+                    self.get_datastore_root()
+                )
+                if (
+                    top_level_blob_prefix is not None
+                    and name[: len(top_level_blob_prefix)] == top_level_blob_prefix
+                ):
+                    name = name[len(top_level_blob_prefix) + 1 :]
+
+                # DataStorage.list_content_result is not pickle-able, because it is defined
+                # inline as a class member. So let's just return a regular tuple. list_content()
+                # can pack it up later.
+                # TODO(jackie) Why is it defined as a class member at all? Probably should not be.
+                result.append((name, is_file))
+            return result
+        except Exception as e:
+            process_exception(e)
+
+
+def handle_executor_exceptions(func):
+    """
+    Decorator for handling errors that come from an Executor. This decorator should
+    only be used on functions where executor errors are possible. I.e. the function
+    uses AzureStorageExecutor.
+    """
+
+    def inner_function(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except BrokenExecutor:
+            # This is fatal. So we bail ASAP.
+            # We also don't want to log, because KeyboardInterrupt on worker processes
+            # also take us here, so it's going to be "normal" user operation most of the
+            # time.
+            # BrokenExecutor parents both BrokenThreadPool and BrokenProcessPool.
+            sys.exit(1)
+
+    return inner_function
+
+
+class AzureStorage(DataStoreStorage):
+    TYPE = "azure"
+
+    def __init__(self, root=None):
+        # cannot decorate __init__... invoke it with dummy decoratee
+        check_azure_deps(lambda: 0)
+        super(AzureStorage, self).__init__(root)
+        self._tmproot = ARTIFACT_LOCALROOT
+        self._default_scope_token = None
+        self._root_client = None
+
+        self._use_processes = AZURE_STORAGE_WORKLOAD_TYPE == "high_throughput"
+        self._executor = AzureStorageExecutor(use_processes=self._use_processes)
+        self._executor.warm_up()
+
+    @handle_exceptions
+    def _get_default_token(self):
+        # either we never got a default token, or the one we got is expiring in 5 min
+        if not self._default_scope_token or (
+            self._default_scope_token.expires_on - time.time() < 300
+        ):
+            from azure.identity import DefaultAzureCredential
+
+            with DefaultAzureCredential() as credential:
+                self._default_scope_token = credential.get_token(
+                    AZURE_STORAGE_DEFAULT_SCOPE
+                )
+        return self._default_scope_token
+
+    @property
+    def root_client(self):
+        """Note this is for optimization only - it allows slow-initialization credentials to be
+        reused across multiple threads and processes.
+
+        Speed up applies mainly to the "no access key" path.
+        """
+        if self._root_client is None:
+            self._root_client = _AzureRootClient(
+                datastore_root=self.datastore_root,
+                token=self._get_default_token(),
+            )
+        return self._root_client
+
+    @classmethod
+    def get_datastore_root_from_config(cls, echo, create_on_absent=True):
+        # create_on_absent doesn't do anything.  This matches S3Storage
+        return DATASTORE_SYSROOT_AZURE
+
+    @handle_executor_exceptions
+    def is_file(self, paths):
+        # preserving order is important...
+        futures = [
+            self._executor.submit(
+                self.root_client.is_file_single,
+                path,
+            )
+            for path in paths
+        ]
+        # preserving order is important...
+        return [future.result() for future in futures]
+
+    def info_file(self, path):
+        # not used anywhere... we should consider killing this on all data storage implementations
+        raise NotImplementedError()
+
+    def size_file(self, path):
+        from azure.core.exceptions import ResourceNotFoundError
+
+        try:
+            return self.root_client.get_blob_client(path).get_blob_properties().size
+        except ResourceNotFoundError:
+            return None
+        except Exception as e:
+            process_exception(e)
+
+    @handle_executor_exceptions
+    def list_content(self, paths):
+        futures = [
+            self._executor.submit(self.root_client.list_content_single, path)
+            for path in paths
+        ]
+        result = []
+        for future in as_completed(futures):
+            result.extend(self.list_content_result(*x) for x in future.result())
+        return result
+
+    @handle_executor_exceptions
+    def save_bytes(self, path_and_bytes_iter, overwrite=False, len_hint=0):
+        tmpdir = None
+        try:
+            tmpdir = mkdtemp(
+                dir=ARTIFACT_LOCALROOT, prefix="metaflow.azure.save_bytes."
+            )
+            futures = []
+            for path, byte_stream in path_and_bytes_iter:
+                metadata = None
+                # bytes_stream could actually be (bytes_stream, metadata) instead.
+                # Read the small print on DatastoreStorage.save_bytes()
+                if isinstance(byte_stream, tuple):
+                    byte_stream, metadata = byte_stream
+                tmp_filename = os.path.join(tmpdir, str(uuid.uuid4()))
+                with open(tmp_filename, "wb") as f:
+                    f.write(byte_stream.read())
+                # Fully finish writing the file, before submitting work. Careful with indentation.
+
+                futures.append(
+                    self._executor.submit(
+                        self.root_client.save_bytes_single,
+                        (path, tmp_filename, metadata),
+                        overwrite=overwrite,
+                    )
+                )
+            for future in as_completed(futures):
+                future.result()
+        finally:
+            # *Future* improvement: We could clean up individual tmp files as each future completes
+            if tmpdir and os.path.exists(tmpdir):
+                shutil.rmtree(tmpdir)
+
+    @handle_executor_exceptions
+    def load_bytes(self, keys):
+
+        tmpdir = mkdtemp(dir=self._tmproot, prefix="metaflow.azure.load_bytes.")
+        try:
+            futures = [
+                self._executor.submit(
+                    self.root_client.load_bytes_single,
+                    tmpdir,
+                    key,
+                )
+                for key in keys
+            ]
+
+            # Let's detect any failures fast. Stop and cleanup ASAP.
+            # Note that messing up return order is beneficial re: Hyrum's law too.
+            items = [future.result() for future in as_completed(futures)]
+        except Exception:
+            # Other BaseExceptions will skip clean up here.
+            # We let this go - smooth exit in those circumstances is more important than cleaning up a few tmp files
+            if os.path.exists(tmpdir):
+                shutil.rmtree(tmpdir)
+            raise
+
+        class _Closer(object):
+            @staticmethod
+            def close():
+                if os.path.isdir(tmpdir):
+                    shutil.rmtree(tmpdir)
+
+        return CloseAfterUse(iter(items), closer=_Closer)
diff --git a/metaflow/datastore/datastore_storage.py b/metaflow/datastore/datastore_storage.py
index dfbbd0ef2c0..ebdacc45793 100644
--- a/metaflow/datastore/datastore_storage.py
+++ b/metaflow/datastore/datastore_storage.py
@@ -229,6 +229,8 @@ def save_bytes(self, path_and_bytes_iter, overwrite=False, len_hint=0):
             BufferedIOBase.
         overwrite : bool
             True if the objects can be overwritten. Defaults to False.
+            Even when False, it is NOT an error condition to see an existing object.
+            Simply do not perform the upload operation.
         len_hint : int
             Estimated number of items produced by the iterator
 
diff --git a/metaflow/includefile.py b/metaflow/includefile.py
index 9131c012ed6..879c9109071 100644
--- a/metaflow/includefile.py
+++ b/metaflow/includefile.py
@@ -3,14 +3,17 @@
 import io
 import json
 import os
+import shutil
+import uuid
 
 from hashlib import sha1
+from tempfile import mkdtemp
 
 from metaflow._vendor import click
 
 from . import parameters
 from .current import current
-from .exception import MetaflowException
+from .exception import MetaflowException, MetaflowInternalError
 from .metaflow_config import DATATOOLS_LOCALROOT, DATATOOLS_SUFFIX
 from .parameters import DeployTimeField, Parameter
 from .util import to_unicode
@@ -80,6 +83,93 @@ def path(self):
         return self._path
 
 
+class Azure(object):
+    @classmethod
+    def get_root_from_config(cls, echo, create_on_absent=True):
+        from metaflow.metaflow_config import DATATOOLS_AZUREROOT
+
+        return DATATOOLS_AZUREROOT
+
+    def __init__(self):
+        # This local directory is used to house any downloaded blobs, for lifetime of
+        # this object as a context manager.
+        self._tmpdir = None
+
+    def _get_storage_backend(self, key):
+        """
+        Return an AzureDatastore, rooted at the container level, no prefix.
+        Key MUST be a fully qualified path. e.g. <container_name>/b/l/o/b/n/a/m/e
+        """
+        from .plugins.azure.azure_utils import parse_azure_full_path
+
+        # we parse out the container name only, and use that to root our storage implementation
+        container_name, _ = parse_azure_full_path(key)
+        # Import DATASTORES dynamically... otherwise, circular import
+        from .datastore import DATASTORES
+
+        storage_impl = DATASTORES["azure"]
+        return storage_impl(container_name)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        if self._tmpdir and os.path.exists(self._tmpdir):
+            shutil.rmtree(self._tmpdir)
+
+    def get(self, key=None, return_missing=False):
+        """Key MUST be a fully qualified path.  <container_name>/b/l/o/b/n/a/m/e"""
+        if not return_missing:
+            raise MetaflowException("Azure object supports only return_missing=True")
+        # We fabricate a uri scheme to fit into existing includefile code (just like local://)
+        if not key.startswith("azure://"):
+            raise MetaflowInternalError(
+                msg="Expected Azure object key to start with 'azure://'"
+            )
+        uri_style_key = key
+        short_key = key[8:]
+        storage = self._get_storage_backend(short_key)
+        azure_object = None
+        with storage.load_bytes([short_key]) as load_result:
+            for _, tmpfile, _ in load_result:
+                if tmpfile is None:
+                    azure_object = AzureObject(uri_style_key, None, False)
+                else:
+                    if not self._tmpdir:
+                        self._tmpdir = mkdtemp(prefix="metaflow.includefile.azure.")
+                    output_file_path = os.path.join(self._tmpdir, str(uuid.uuid4()))
+                    shutil.move(tmpfile, output_file_path)
+                    azure_object = AzureObject(uri_style_key, output_file_path, True)
+                break
+        return azure_object
+
+    def put(self, key, obj, overwrite=True):
+        """Key MUST be a fully qualified path.  <container_name>/b/l/o/b/n/a/m/e"""
+        storage = self._get_storage_backend(key)
+        storage.save_bytes([(key, io.BytesIO(obj))], overwrite=overwrite)
+        # We fabricate a uri scheme to fit into existing includefile code (just like local://)
+        return "azure://%s" % key
+
+
+class AzureObject(object):
+    def __init__(self, url, path, exists):
+        self._path = path
+        self._url = url
+        self._exists = exists
+
+    @property
+    def path(self):
+        return self._path
+
+    @property
+    def url(self):
+        return self._url
+
+    @property
+    def exists(self):
+        return self._exists
+
+
 class Local(object):
     """
     This class allows you to access the local filesystem in a way similar to the S3 datatools
@@ -165,7 +255,7 @@ def put(self, key, obj, overwrite=True):
 # From here on out, this is the IncludeFile implementation.
 from .datatools import S3
 
-DATACLIENTS = {"local": Local, "s3": S3}
+DATACLIENTS = {"local": Local, "s3": S3, "azure": Azure}
 
 
 class LocalFile:
diff --git a/metaflow/main_cli.py b/metaflow/main_cli.py
index 85e8b28e4f2..7b3f542f0d5 100644
--- a/metaflow/main_cli.py
+++ b/metaflow/main_cli.py
@@ -502,6 +502,29 @@ def configure_s3_datastore(existing_env):
     return env
 
 
+def configure_azure_datastore(existing_env):
+    env = {}
+    # Set Azure Blob Storage as default datastore.
+    env["METAFLOW_DEFAULT_DATASTORE"] = "azure"
+    # Set Azure Blob Storage folder for datastore.
+    # TODO rename this Blob Endpoint!
+    env["METAFLOW_AZURE_STORAGE_BLOB_SERVICE_ENDPOINT"] = click.prompt(
+        cyan("[METAFLOW_AZURE_STORAGE_BLOB_SERVICE_ENDPOINT]")
+        + " Azure Storage Account URL, for the account holding the Blob container to be used. "
+        + "(E.g. https://<storage_account>.blob.core.windows.net/)",
+        default=existing_env.get("METAFLOW_AZURE_STORAGE_BLOB_SERVICE_ENDPOINT"),
+        show_default=True,
+    )
+    env["METAFLOW_DATASTORE_SYSROOT_AZURE"] = click.prompt(
+        cyan("[METAFLOW_DATASTORE_SYSROOT_AZURE]")
+        + " Azure Blob Storage folder for Metaflow artifact storage "
+        + "(Format: <container_name>/<prefix>)",
+        default=existing_env.get("METAFLOW_DATASTORE_SYSROOT_AZURE"),
+        show_default=True,
+    )
+    return env
+
+
 def configure_metadata_service(existing_env):
     empty_profile = False
     if not existing_env:
@@ -521,7 +544,7 @@ def configure_metadata_service(existing_env):
         cyan("[METAFLOW_SERVICE_INTERNAL_URL]")
         + yellow(" (optional)")
         + " URL for Metaflow Service "
-        + "(Accessible only within VPC).",
+        + "(Accessible only within VPC [AWS] or a Kubernetes cluster [if the service runs in one]).",
         default=existing_env.get(
             "METAFLOW_SERVICE_INTERNAL_URL", env["METAFLOW_SERVICE_URL"]
         ),
@@ -538,7 +561,45 @@ def configure_metadata_service(existing_env):
     return env
 
 
-def configure_datastore_and_metadata(existing_env):
+def configure_azure_datastore_and_metadata(existing_env):
+    empty_profile = False
+    if not existing_env:
+        empty_profile = True
+    env = {}
+
+    # Configure Azure Blob Storage as the datastore.
+    use_azure_as_datastore = click.confirm(
+        "\nMetaflow can use "
+        + yellow("Azure Blob Storage as the storage backend")
+        + " for all code and data artifacts on "
+        + "Azure.\nAzure Blob Storage is a strict requirement if you "
+        + "intend to execute your flows on a Kubernetes cluster on Azure (AKS or self-managed)"
+        + ".\nWould you like to configure Azure Blob Storage "
+        + "as the default storage backend?",
+        default=empty_profile
+        or existing_env.get("METAFLOW_DEFAULT_DATASTORE", "") == "azure",
+        abort=False,
+    )
+    if use_azure_as_datastore:
+        env.update(configure_azure_datastore(existing_env))
+
+    # Configure Metadata service for tracking.
+    if click.confirm(
+        "\nMetaflow can use a "
+        + yellow("remote Metadata Service to track")
+        + " and persist flow execution metadata.\nConfiguring the "
+        "service is a requirement if you intend to schedule your "
+        "flows with Kubernetes on Azure (AKS or self-managed).\nWould you like to "
+        "configure the Metadata Service?",
+        default=empty_profile
+        or existing_env.get("METAFLOW_DEFAULT_METADATA", "") == "service",
+        abort=False,
+    ):
+        env.update(configure_metadata_service(existing_env))
+    return env
+
+
+def configure_aws_datastore_and_metadata(existing_env):
     empty_profile = False
     if not existing_env:
         empty_profile = True
@@ -747,6 +808,15 @@ def configure_kubernetes(existing_env):
         default=existing_env.get("METAFLOW_KUBERNETES_CONTAINER_IMAGE", ""),
         show_default=True,
     )
+    # Set default Kubernetes secrets to source into pod envs
+    env["METAFLOW_KUBERNETES_SECRETS"] = click.prompt(
+        cyan("[METAFLOW_KUBERNETES_SECRETS]")
+        + yellow(" (optional)")
+        + " Comma-delimited list of secret names. Jobs will"
+        " gain environment variables from these secrets. ",
+        default=existing_env.get("METAFLOW_KUBERNETES_SECRETS", ""),
+        show_default=True,
+    )
 
     return env
 
@@ -775,6 +845,77 @@ def verify_aws_credentials(ctx):
         ctx.abort()
 
 
+def verify_azure_credentials(ctx):
+    # Verify that the user has configured AWS credentials on their computer.
+    if not click.confirm(
+        "\nMetaflow relies on "
+        + yellow("Azure access credentials")
+        + " present on your computer to access resources on Azure."
+        "\nBefore proceeding further, please confirm that you "
+        "have already configured these access credentials on "
+        "this computer.",
+        default=True,
+    ):
+        echo(
+            "There are many ways to setup your Azure access credentials. You "
+            "can get started by getting familiar with the following: ",
+            nl=False,
+            fg="yellow",
+        )
+        echo("")
+        echo(
+            "- https://docs.microsoft.com/en-us/cli/azure/authenticate-azure-cli",
+            fg="cyan",
+        )
+        echo(
+            "- https://docs.microsoft.com/en-us/cli/azure/azure-cli-configuration",
+            fg="cyan",
+        )
+        ctx.abort()
+
+
+@configure.command(help="Configure metaflow to access Azure Blob Storage.")
+@click.option(
+    "--profile",
+    "-p",
+    default="",
+    help="Configure a named profile. Activate the profile by setting "
+    "`METAFLOW_PROFILE` environment variable.",
+)
+@click.pass_context
+def azure(ctx, profile):
+
+    # Greet the user!
+    echo(
+        "Welcome to Metaflow! Follow the prompts to configure your installation.\n",
+        bold=True,
+    )
+
+    # Check for existing configuration.
+    if not overwrite_config(profile):
+        ctx.abort()
+
+    verify_azure_credentials(ctx)
+
+    existing_env = get_env(profile)
+
+    env = {}
+    env.update(configure_azure_datastore_and_metadata(existing_env))
+
+    persist_env({k: v for k, v in env.items() if v}, profile)
+
+    # Prompt user to also configure Kubernetes for compute if using azure
+    if env.get("METAFLOW_DEFAULT_DATASTORE") == "azure":
+        click.echo(
+            "\nFinal note! Metaflow can scale your flows by "
+            + yellow("executing your steps on Kubernetes.")
+            + "\nYou may use Azure Kubernetes Service (AKS)"
+            " or a self-managed Kubernetes cluster on Azure VMs."
+            + " If/when your Kubernetes cluster is ready for use,"
+            " please run 'metaflow configure kubernetes'.",
+        )
+
+
 @configure.command(help="Configure metaflow to access self-managed AWS resources.")
 @click.option(
     "--profile",
@@ -804,7 +945,7 @@ def aws(ctx, profile):
         empty_profile = True
 
     env = {}
-    env.update(configure_datastore_and_metadata(existing_env))
+    env.update(configure_aws_datastore_and_metadata(existing_env))
 
     # Configure AWS Batch for compute if using S3
     if env.get("METAFLOW_DEFAULT_DATASTORE") == "s3":
@@ -848,27 +989,20 @@ def kubernetes(ctx, profile):
     if not overwrite_config(profile):
         ctx.abort()
 
-    verify_aws_credentials(ctx)
-
     existing_env = get_env(profile)
 
     env = existing_env.copy()
 
-    if existing_env.get("METAFLOW_DEFAULT_DATASTORE") == "s3":
-        # Skip S3 configuration if it is already configured
-        pass
-    elif not existing_env.get("METAFLOW_DEFAULT_DATASTORE"):
-        env.update(configure_s3_datastore(existing_env))
-    else:
-        # If configured to use something else, offer to switch to S3
-        click.confirm(
-            "\nMetaflow on Kubernetes needs to use Amazon S3 as a datastore, "
-            + "but your existing configuration is not using Amazon S3. "
-            + "Would you like to reconfigure it to use Amazon S3?",
-            default=True,
-            abort=True,
+    # We used to push user straight to S3 configuration inline.
+    # Now that we support >1 cloud, it gets too complicated.
+    # Therefore, we instruct the user to configure datastore first, by
+    # a separate command.
+    if existing_env.get("METAFLOW_DEFAULT_DATASTORE") == "local":
+        click.echo(
+            "\nCannot run Kubernetes with local datastore. Please run"
+            " 'metaflow configure aws' or 'metaflow configure azure'."
         )
-        env.update(configure_s3_datastore(existing_env))
+        click.abort()
 
     # Configure remote metadata.
     if existing_env.get("METAFLOW_DEFAULT_METADATA") == "service":
diff --git a/metaflow/metaflow_config.py b/metaflow/metaflow_config.py
index b40b855b26a..3711715bda3 100644
--- a/metaflow/metaflow_config.py
+++ b/metaflow/metaflow_config.py
@@ -36,8 +36,50 @@ def init_config():
 METAFLOW_CONFIG = init_config()
 
 
-def from_conf(name, default=None):
-    return os.environ.get(name, METAFLOW_CONFIG.get(name, default))
+def from_conf(name, default=None, from_env_only=False, validate_fn=None):
+    """
+    First try to pull value from environment, then from metaflow config JSON.
+
+    If from_env_only is True, we never use the value in metaflow config JSON.
+    We will raise an Error on seeing the value set in metaflow config JSON.
+
+    Prior to a value being returned, we will validate using validate_fn (if provided).
+    Only non-None values are validated.
+
+    validate_fn should accept (name, value).
+    If the value validates, return None, else raise an MetaflowException.
+    """
+    value_from_env = os.environ.get(name, None)
+    value_from_config = METAFLOW_CONFIG.get(name, default)
+    if from_env_only:
+        if value_from_config is not None:
+            raise MetaflowException(
+                "%s may only be set from environment variable, NOT from metaflow JSON configs."
+                % name
+            )
+        value = value_from_env
+    else:
+        if value_from_env is not None:
+            value = value_from_env
+        else:
+            value = value_from_config
+    if validate_fn and value is not None:
+        validate_fn(name, value)
+    return value
+
+
+def _get_validate_choice_fn(choices):
+    """Returns a validate_fn for use with from_conf().
+    The validate_fn will check a value against a list of allowed choices.
+    """
+
+    def _validate_choice(name, value):
+        if value not in choices:
+            raise MetaflowException(
+                "%s must be set to one of %s. Got '%s'." % (name, choices, value)
+            )
+
+    return _validate_choice
 
 
 ###
@@ -60,6 +102,8 @@ def from_conf(name, default=None):
 DATASTORE_SYSROOT_LOCAL = from_conf("METAFLOW_DATASTORE_SYSROOT_LOCAL")
 # S3 bucket and prefix to store artifacts for 's3' datastore.
 DATASTORE_SYSROOT_S3 = from_conf("METAFLOW_DATASTORE_SYSROOT_S3")
+# Azure Blob Storage container and blob prefix
+DATASTORE_SYSROOT_AZURE = from_conf("METAFLOW_DATASTORE_SYSROOT_AZURE")
 
 
 ###
@@ -131,6 +175,15 @@ def from_conf(name, default=None):
         % DATATOOLS_DEFAULT_SESSION_VARS
     )
 
+# Azure datatools root location
+# Note: we do not expose an actual datatools library for Azure (like we do for S3)
+# Similar to DATATOOLS_LOCALROOT, this is used ONLY by the IncludeFile's internal implementation.
+DATATOOLS_AZUREROOT = from_conf(
+    "METAFLOW_DATATOOLS_AZUREROOT",
+    os.path.join(from_conf("METAFLOW_DATASTORE_SYSROOT_AZURE"), DATATOOLS_SUFFIX)
+    if from_conf("METAFLOW_DATASTORE_SYSROOT_AZURE")
+    else None,
+)
 # Local datatools root location
 DATATOOLS_LOCALROOT = from_conf(
     "METAFLOW_DATATOOLS_LOCALROOT",
@@ -139,7 +192,7 @@ def from_conf(name, default=None):
     else None,
 )
 
-# The root directory to save artifact pulls in, when using S3
+# The root directory to save artifact pulls in, when using S3 or Azure
 ARTIFACT_LOCALROOT = from_conf("METAFLOW_ARTIFACT_LOCALROOT", os.getcwd())
 
 # Cards related config variables
@@ -151,8 +204,27 @@ def from_conf(name, default=None):
     if from_conf("METAFLOW_DATASTORE_SYSROOT_S3")
     else None,
 )
+DATASTORE_CARD_AZUREROOT = from_conf(
+    "METAFLOW_CARD_AZUREROOT",
+    os.path.join(from_conf("METAFLOW_DATASTORE_SYSROOT_AZURE"), DATASTORE_CARD_SUFFIX)
+    if from_conf("METAFLOW_DATASTORE_SYSROOT_AZURE")
+    else None,
+)
 CARD_NO_WARNING = from_conf("METAFLOW_CARD_NO_WARNING", False)
 
+# Azure storage account URL
+AZURE_STORAGE_BLOB_SERVICE_ENDPOINT = from_conf(
+    "METAFLOW_AZURE_STORAGE_BLOB_SERVICE_ENDPOINT"
+)
+
+# Azure storage can use process-based parallelism instead of threads.
+# Processes perform better for high throughput workloads (e.g. many huge artifacts)
+AZURE_STORAGE_WORKLOAD_TYPE = from_conf(
+    "METAFLOW_AZURE_STORAGE_WORKLOAD_TYPE",
+    "general",
+    validate_fn=_get_validate_choice_fn(["general", "high_throughput"]),
+)
+
 
 ###
 # Metadata configuration
@@ -246,10 +318,9 @@ def from_conf(name, default=None):
 # Conda configuration
 ###
 # Conda package root location on S3
-CONDA_PACKAGE_S3ROOT = from_conf(
-    "METAFLOW_CONDA_PACKAGE_S3ROOT",
-    "%s/conda" % from_conf("METAFLOW_DATASTORE_SYSROOT_S3"),
-)
+CONDA_PACKAGE_S3ROOT = from_conf("METAFLOW_CONDA_PACKAGE_S3ROOT")
+# Conda package root location on Azure
+CONDA_PACKAGE_AZUREROOT = from_conf("METAFLOW_CONDA_PACKAGE_AZUREROOT")
 
 # Use an alternate dependency resolver for conda packages instead of conda
 # Mamba promises faster package dependency resolution times, which
@@ -325,11 +396,22 @@ def get_version(pkg):
 
 # PINNED_CONDA_LIBS are the libraries that metaflow depends on for execution
 # and are needed within a conda environment
-def get_pinned_conda_libs(python_version):
-    return {
+def get_pinned_conda_libs(python_version, datastore_type):
+    pins = {
         "requests": ">=2.21.0",
-        "boto3": ">=1.14.0",
     }
+    if datastore_type == "s3":
+        pins["boto3"] = ">=1.14.0"
+    elif datastore_type == "azure":
+        pins["azure-identity"] = ">=1.10.0"
+        pins["azure-storage-blob"] = ">=12.12.0"
+    elif datastore_type == "local":
+        pass
+    else:
+        raise MetaflowException(
+            msg="conda lib pins for datastore %s are undefined" % (datastore_type,)
+        )
+    return pins
 
 
 # Check if there are extensions to Metaflow to load and override everything
diff --git a/metaflow/metaflow_environment.py b/metaflow/metaflow_environment.py
index 9a0d7bc182b..b45b958ad68 100644
--- a/metaflow/metaflow_environment.py
+++ b/metaflow/metaflow_environment.py
@@ -42,7 +42,7 @@ def decospecs(self):
         """
         return ()
 
-    def bootstrap_commands(self, step_name):
+    def bootstrap_commands(self, step_name, datastore_type):
         """
         A list of shell commands to bootstrap this environment in a remote runtime.
         """
@@ -79,20 +79,67 @@ def get_client_info(cls, flow_name, metadata):
         """
         return "Local environment"
 
-    def get_package_commands(self, code_package_url):
+    def _get_download_code_package_cmd(self, code_package_url, datastore_type):
+        """Return a command that downloads the code package from the datastore. We use various
+        cloud storage CLI tools because we don't have access to Metaflow codebase (which we
+        are about to download in the command).
+
+        The command should download the package to "job.tar" in the current directory.
+
+        It should work silently if everything goes well.
+        """
+        if datastore_type == "s3":
+            return (
+                '%s -m awscli ${METAFLOW_S3_ENDPOINT_URL:+--endpoint-url=\\"${METAFLOW_S3_ENDPOINT_URL}\\"} '
+                + "s3 cp %s job.tar >/dev/null"
+            ) % (self._python(), code_package_url)
+        elif datastore_type == "azure":
+            from .plugins.azure.azure_utils import parse_azure_full_path
+
+            container_name, blob = parse_azure_full_path(code_package_url)
+            # remove a trailing slash, if present
+            blob_endpoint = "${METAFLOW_AZURE_STORAGE_BLOB_SERVICE_ENDPOINT%/}"
+            return "download-azure-blob --blob-endpoint={blob_endpoint} --container={container} --blob={blob} --output-file=job.tar".format(
+                blob_endpoint=blob_endpoint,
+                blob=blob,
+                container=container_name,
+            )
+        else:
+            raise NotImplementedError(
+                "We don't know how to generate a download code package cmd for datastore %s"
+                % datastore_type
+            )
+
+    def _get_install_dependencies_cmd(self, datastore_type):
+        cmds = ["%s -m pip install requests -qqq" % self._python()]
+        if datastore_type == "s3":
+            cmds.append("%s -m pip install awscli boto3 -qqq" % self._python())
+        elif datastore_type == "azure":
+            cmds.append(
+                "%s -m pip install azure-identity azure-storage-blob simple-azure-blob-downloader -qqq"
+                % self._python()
+            )
+        else:
+            raise NotImplementedError(
+                "We don't know how to generate an install dependencies cmd for datastore %s"
+                % datastore_type
+            )
+        return " && ".join(cmds)
+
+    def get_package_commands(self, code_package_url, datastore_type):
         cmds = [
             BASH_MFLOG,
             "mflog 'Setting up task environment.'",
-            "%s -m pip install awscli requests boto3 -qqq" % self._python(),
+            self._get_install_dependencies_cmd(datastore_type),
             "mkdir metaflow",
             "cd metaflow",
             "mkdir .metaflow",  # mute local datastore creation log
             "i=0; while [ $i -le 5 ]; do "
             "mflog 'Downloading code package...'; "
-            '%s -m awscli ${METAFLOW_S3_ENDPOINT_URL:+--endpoint-url=\\"${METAFLOW_S3_ENDPOINT_URL}\\"} '
-            "s3 cp %s job.tar >/dev/null && mflog 'Code package downloaded.' && break; "
+            + self._get_download_code_package_cmd(code_package_url, datastore_type)
+            + " && mflog 'Code package downloaded.' && break; "
             "sleep 10; i=$((i+1)); "
-            "done" % (self._python(), code_package_url),
+            "done",
             "if [ $i -gt 5 ]; then "
             "mflog 'Failed to download code package from %s "
             "after 6 tries. Exiting...' && exit 1; "
diff --git a/metaflow/mflog/__init__.py b/metaflow/mflog/__init__.py
index 68777ff9790..1bcc05e8b47 100644
--- a/metaflow/mflog/__init__.py
+++ b/metaflow/mflog/__init__.py
@@ -4,6 +4,7 @@
 from .mflog import refine, set_should_persist
 
 from metaflow.util import to_unicode
+from metaflow.exception import MetaflowInternalError
 
 # Log source indicates the system that *minted the timestamp*
 # for the logline. This means that for a single task we can
@@ -142,3 +143,19 @@ def _available_logs(tail, stream, echo, should_persist=False):
     # tailed.
     _available_logs(stdout_tail, "stdout", echo)
     _available_logs(stderr_tail, "stderr", echo)
+
+
+def get_log_tailer(log_url, datastore_type):
+    if datastore_type == "s3":
+        from metaflow.datatools.s3tail import S3Tail
+
+        return S3Tail(log_url)
+    elif datastore_type == "azure":
+        from metaflow.plugins.azure.azure_tail import AzureTail
+
+        return AzureTail(log_url)
+    else:
+        raise MetaflowInternalError(
+            "Log tailing implementation missing for datastore type %s"
+            % (datastore_type,)
+        )
diff --git a/metaflow/plugins/argo/argo_workflows.py b/metaflow/plugins/argo/argo_workflows.py
index 9204ff9ec82..fbee3d2d240 100644
--- a/metaflow/plugins/argo/argo_workflows.py
+++ b/metaflow/plugins/argo/argo_workflows.py
@@ -6,12 +6,11 @@
 
 from metaflow import current
 from metaflow.decorators import flow_decorators
-from metaflow.exception import MetaflowException, MetaflowInternalError
+from metaflow.exception import MetaflowException
 from metaflow.metaflow_config import (
     BATCH_METADATA_SERVICE_HEADERS,
     BATCH_METADATA_SERVICE_URL,
     DATASTORE_CARD_S3ROOT,
-    DATASTORE_LOCAL_DIR,
     DATASTORE_SYSROOT_S3,
     DATATOOLS_S3ROOT,
     DEFAULT_METADATA,
@@ -20,6 +19,9 @@
     KUBERNETES_SANDBOX_INIT_SCRIPT,
     KUBERNETES_SECRETS,
     S3_ENDPOINT_URL,
+    AZURE_STORAGE_BLOB_SERVICE_ENDPOINT,
+    DATASTORE_SYSROOT_AZURE,
+    DATASTORE_CARD_AZUREROOT,
 )
 from metaflow.mflog import BASH_SAVE_LOGS, bash_capture_logs, export_mflog_env_vars
 from metaflow.parameters import deploy_time_eval
@@ -606,9 +608,13 @@ def _container_templates(self):
                     task_id_expr,
                     mflog_expr,
                 ]
-                + self.environment.get_package_commands(self.code_package_url)
+                + self.environment.get_package_commands(
+                    self.code_package_url, self.flow_datastore.TYPE
+                )
+            )
+            step_cmds = self.environment.bootstrap_commands(
+                node.name, self.flow_datastore.TYPE
             )
-            step_cmds = self.environment.bootstrap_commands(node.name)
 
             input_paths = "{{inputs.parameters.input-paths}}"
 
@@ -778,12 +784,17 @@ def _container_templates(self):
                 }
             )
             # add METAFLOW_S3_ENDPOINT_URL
-            if S3_ENDPOINT_URL is not None:
-                env["METAFLOW_S3_ENDPOINT_URL"] = S3_ENDPOINT_URL
+            env["METAFLOW_S3_ENDPOINT_URL"] = S3_ENDPOINT_URL
 
             # support Metaflow sandboxes
-            if KUBERNETES_SANDBOX_INIT_SCRIPT is not None:
-                env["METAFLOW_INIT_SCRIPT"] = KUBERNETES_SANDBOX_INIT_SCRIPT
+            env["METAFLOW_INIT_SCRIPT"] = KUBERNETES_SANDBOX_INIT_SCRIPT
+
+            # Azure stuff
+            env[
+                "METAFLOW_AZURE_STORAGE_BLOB_SERVICE_ENDPOINT"
+            ] = AZURE_STORAGE_BLOB_SERVICE_ENDPOINT
+            env["METAFLOW_DATASTORE_SYSROOT_AZURE"] = DATASTORE_SYSROOT_AZURE
+            env["METAFLOW_DATASTORE_CARD_AZUREROOT"] = DATASTORE_CARD_AZUREROOT
 
             metaflow_version = self.environment.get_environment_info()
             metaflow_version["flow_name"] = self.graph.name
@@ -815,6 +826,10 @@ def _container_templates(self):
                     Parameter("num-splits").valueFrom({"path": "/mnt/out/splits"})
                 )
 
+            # It makes no sense to set env vars to None (shows up as "None" string)
+            env_without_none_values = {k: v for k, v in env.items() if v is not None}
+            del env
+
             # Create a ContainerTemplate for this node. Ideally, we would have
             # liked to inline this ContainerTemplate and avoid scanning the workflow
             # twice, but due to issues with variable substitution, we will have to
@@ -876,7 +891,7 @@ def _container_templates(self):
                             command=cmds,
                             env=[
                                 kubernetes_sdk.V1EnvVar(name=k, value=str(v))
-                                for k, v in env.items()
+                                for k, v in env_without_none_values.items()
                             ]
                             # Add environment variables for book-keeping.
                             # https://argoproj.github.io/argo-workflows/fields/#fields_155
diff --git a/metaflow/plugins/argo/argo_workflows_cli.py b/metaflow/plugins/argo/argo_workflows_cli.py
index 8737112ff79..35e2092af4e 100644
--- a/metaflow/plugins/argo/argo_workflows_cli.py
+++ b/metaflow/plugins/argo/argo_workflows_cli.py
@@ -328,8 +328,10 @@ def make_flow(
 ):
     # TODO: Make this check less specific to Amazon S3 as we introduce
     #       support for more cloud object stores.
-    if obj.flow_datastore.TYPE != "s3":
-        raise MetaflowException("Argo Workflows requires --datastore=s3.")
+    if obj.flow_datastore.TYPE not in ("azure", "s3"):
+        raise MetaflowException(
+            "Argo Workflows requires --datastore=s3 or --datastore=azure"
+        )
 
     # Attach @kubernetes and @environment decorator to the flow to
     # ensure that the related decorator hooks are invoked.
diff --git a/metaflow/plugins/aws/batch/batch.py b/metaflow/plugins/aws/batch/batch.py
index 0af3d56e55d..f505c2936d7 100644
--- a/metaflow/plugins/aws/batch/batch.py
+++ b/metaflow/plugins/aws/batch/batch.py
@@ -8,11 +8,10 @@
 
 from metaflow import util
 from metaflow.datatools.s3tail import S3Tail
-from metaflow.exception import MetaflowException, MetaflowInternalError
+from metaflow.exception import MetaflowException
 from metaflow.metaflow_config import (
     BATCH_METADATA_SERVICE_URL,
     DATATOOLS_S3ROOT,
-    DATASTORE_LOCAL_DIR,
     DATASTORE_SYSROOT_S3,
     DEFAULT_METADATA,
     BATCH_METADATA_SERVICE_HEADERS,
@@ -20,7 +19,6 @@
     DATASTORE_CARD_S3ROOT,
     S3_ENDPOINT_URL,
 )
-from metaflow.mflog.mflog import refine, set_should_persist
 from metaflow.mflog import (
     export_mflog_env_vars,
     bash_capture_logs,
@@ -60,10 +58,10 @@ def _command(self, environment, code_package_url, step_name, step_cmds, task_spe
             stderr_path=STDERR_PATH,
             **task_spec
         )
-        init_cmds = environment.get_package_commands(code_package_url)
+        init_cmds = environment.get_package_commands(code_package_url, "s3")
         init_expr = " && ".join(init_cmds)
         step_expr = bash_capture_logs(
-            " && ".join(environment.bootstrap_commands(step_name) + step_cmds)
+            " && ".join(environment.bootstrap_commands(step_name, "s3") + step_cmds)
         )
 
         # construct an entry point that
diff --git a/metaflow/plugins/aws/batch/batch_client.py b/metaflow/plugins/aws/batch/batch_client.py
index 078e437d54c..d5cd594e5c1 100644
--- a/metaflow/plugins/aws/batch/batch_client.py
+++ b/metaflow/plugins/aws/batch/batch_client.py
@@ -417,6 +417,8 @@ def gpu(self, gpu):
         return self
 
     def environment_variable(self, name, value):
+        if value is None:
+            return self
         if "environment" not in self.payload["containerOverrides"]:
             self.payload["containerOverrides"]["environment"] = []
         value = str(value)
diff --git a/metaflow/plugins/aws/step_functions/step_functions.py b/metaflow/plugins/aws/step_functions/step_functions.py
index 343daf428dc..482fa2a9dde 100644
--- a/metaflow/plugins/aws/step_functions/step_functions.py
+++ b/metaflow/plugins/aws/step_functions/step_functions.py
@@ -629,6 +629,10 @@ def _batch(self, node):
                 )
             env["METAFLOW_SFN_DYNAMO_DB_TABLE"] = SFN_DYNAMO_DB_TABLE
 
+        # It makes no sense to set env vars to None (shows up as "None" string)
+        env_without_none_values = {k: v for k, v in env.items() if v is not None}
+        del env
+
         # Resolve AWS Batch resource requirements.
         batch_deco = [deco for deco in node.decorators if deco.name == "batch"][0]
         resources = {}
@@ -675,7 +679,7 @@ def _batch(self, node):
                 shared_memory=resources["shared_memory"],
                 max_swap=resources["max_swap"],
                 swappiness=resources["swappiness"],
-                env=env,
+                env=env_without_none_values,
                 attrs=attrs,
                 host_volumes=resources["host_volumes"],
             )
diff --git a/metaflow/plugins/azure/__init__.py b/metaflow/plugins/azure/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/metaflow/plugins/azure/azure_exceptions.py b/metaflow/plugins/azure/azure_exceptions.py
new file mode 100644
index 00000000000..f4f500b1268
--- /dev/null
+++ b/metaflow/plugins/azure/azure_exceptions.py
@@ -0,0 +1,13 @@
+from metaflow.exception import MetaflowException
+
+
+class MetaflowAzureAuthenticationError(MetaflowException):
+    headline = "Failed to authenticate with Azure"
+
+
+class MetaflowAzureResourceError(MetaflowException):
+    headline = "Failed to access Azure resource"
+
+
+class MetaflowAzurePackageError(MetaflowException):
+    headline = "Missing required packages azure-identity and azure-storage-blob"
diff --git a/metaflow/plugins/azure/azure_tail.py b/metaflow/plugins/azure/azure_tail.py
new file mode 100644
index 00000000000..d07fe08e94a
--- /dev/null
+++ b/metaflow/plugins/azure/azure_tail.py
@@ -0,0 +1,94 @@
+from io import BytesIO
+
+from azure.core.exceptions import ResourceNotFoundError, HttpResponseError
+
+from metaflow.exception import MetaflowException
+
+from metaflow.plugins.azure.blob_service_client_factory import (
+    get_azure_blob_service_client,
+)
+from metaflow.plugins.azure.azure_utils import (
+    parse_azure_full_path,
+)
+
+
+class AzureTail(object):
+    def __init__(self, blob_full_uri):
+        """Location should be something like <container_name>/blob"""
+        container_name, blob_name = parse_azure_full_path(blob_full_uri)
+        if not blob_name:
+            raise MetaflowException(
+                msg="Failed to parse blob_full_uri into <container_name>/<blob_name> (got %s)"
+                % blob_full_uri
+            )
+        service = get_azure_blob_service_client()
+        container = service.get_container_client(container_name)
+        self._blob_client = container.get_blob_client(blob_name)
+        self._pos = 0
+        self._tail = b""
+
+    def __iter__(self):
+        buf = self._fill_buf()
+        if buf is not None:
+            # If there are no line breaks in the entries
+            # file, then we will yield nothing, ever.
+            #
+            # This apes S3 tail. We can fix it here and in S3
+            # if/when this becomes an issue. It boils down to
+            # knowing when to give up waiting on partial lines
+            # to become full lines (tricky, need more info).
+            #
+            # Likely this has been OK because we control the
+            # line-break presence in the objects we tail.
+            for line in buf:
+                if line.endswith(b"\n"):
+                    yield line
+                else:
+                    self._tail = line
+                    break
+
+    def _make_range_request(self):
+        try:
+            # Yes we read to the end... memory blow up is possible. We can improve by specifying length param
+            return self._blob_client.download_blob(offset=self._pos).readall()
+        except ResourceNotFoundError:
+            # Maybe the log hasn't been uploaded yet, but will be soon.
+            return None
+        except HttpResponseError as e:
+            # be silent on range errors - it means log did not advance
+            if e.status_code != 416:
+                print(
+                    "Failed to tail log from step (status code = %d)" % (e.status_code,)
+                )
+            return None
+        except Exception as e:
+            print("Failed to tail log from step (%s)" % type(e))
+            return None
+
+    def _fill_buf(self):
+        data = self._make_range_request()
+        if data is None:
+            return None
+        if data:
+            buf = BytesIO(data)
+            self._pos += len(data)
+            self._tail = b""
+            return buf
+        else:
+            return None
+
+
+if __name__ == "__main__":
+    # This main program is for debugging and testing purposes
+    import argparse
+
+    parser = argparse.ArgumentParser(
+        description="Tail an Azure Blob. Must specify METAFLOW_AZURE_STORAGE_BLOB_SERVICE_ENDPOINT in environment."
+    )
+    parser.add_argument(
+        "blob_full_uri", help="The blob to tail. Format is <container_name>/<blob>"
+    )
+    args = parser.parse_args()
+    az_tail = AzureTail(args.blob_full_uri)
+    for line in az_tail:
+        print(line.strip().decode("utf-8"))
diff --git a/metaflow/plugins/azure/azure_utils.py b/metaflow/plugins/azure/azure_utils.py
new file mode 100644
index 00000000000..208a0d76f60
--- /dev/null
+++ b/metaflow/plugins/azure/azure_utils.py
@@ -0,0 +1,218 @@
+import sys
+import time
+
+from metaflow.plugins.azure.azure_exceptions import (
+    MetaflowAzureAuthenticationError,
+    MetaflowAzureResourceError,
+    MetaflowAzurePackageError,
+)
+from metaflow.exception import MetaflowInternalError, MetaflowException
+
+
+def _check_and_init_azure_deps():
+    try:
+        # Python 3.6 would print lots of warnings about deprecated cryptography usage when importing Azure modules
+        import warnings
+
+        warnings.filterwarnings("ignore")
+
+        import azure.storage.blob
+        import azure.identity
+
+        # cut down on crazy logging from azure.identity.
+        # TODO but what if folks want to debug on occasion?
+        import logging
+
+        logging.getLogger("azure.identity").setLevel(logging.ERROR)
+        logging.getLogger("msrest.serialization").setLevel(logging.ERROR)
+    except ImportError:
+        raise MetaflowAzurePackageError()
+
+    if sys.version_info[:2] < (3, 6):
+        raise MetaflowException(
+            msg="Metaflow may only use Azure Blob Storage with Python 3.6 or newer"
+        )
+
+
+def check_azure_deps(func):
+    """The decorated function checks Azure dependencies (as needed for Azure storage backend). This includes
+    various Azure SDK packages, as well as a Python version of >3.6
+
+    We also tune some warning and logging configurations to reduce excessive log lines from Azure SDK.
+    """
+
+    def _inner_func(*args, **kwargs):
+        _check_and_init_azure_deps()
+        return func(*args, **kwargs)
+
+    return _inner_func
+
+
+def parse_azure_full_path(blob_full_uri):
+    """
+    Parse an Azure Blob Storage path str into a tuple (container_name, blob).
+
+    Expected format is: <container_name>/<blob>
+
+    This is sometimes used to parse an Azure sys root, in which case:
+
+    - <container_name> is the Azure Blob Storage container name
+    - <blob> is effectively a blob_prefix, a subpath within the container in which blobs will live
+
+    Blob may be None, if input looks like <container_name>. I.e. no slashes present.
+
+    We take a strict validation approach, doing no implicit string manipulations on
+    the user's behalf.  Path manipulations by themselves are complicated enough without
+    adding magic.
+
+    We provide clear error messages so the user knows exactly how to fix any validation error.
+    """
+    if blob_full_uri.endswith("/"):
+        raise ValueError("sysroot may not end with slash (got %s)" % blob_full_uri)
+    if blob_full_uri.startswith("/"):
+        raise ValueError("sysroot may not start with slash (got %s)" % blob_full_uri)
+    if "//" in blob_full_uri:
+        raise ValueError(
+            "sysroot may not contain any consecutive slashes (got %s)" % blob_full_uri
+        )
+    parts = blob_full_uri.split("/", 1)
+    container_name = parts[0]
+    if container_name == "":
+        raise ValueError(
+            "Container name part of sysroot may not be empty (tried to parse %s)"
+            % (blob_full_uri,)
+        )
+    if len(parts) == 1:
+        blob_name = None
+    else:
+        blob_name = parts[1]
+
+    return container_name, blob_name
+
+
+@check_azure_deps
+def process_exception(e):
+    """
+    Translate errors to Metaflow errors for standardized messaging. The intent is that all
+    Azure Blob Storage integration logic should send errors to this function for
+    translation.
+
+    We explicitly EXCLUDE executor related errors here.  See handle_executor_exceptions
+    """
+    if isinstance(e, MetaflowException):
+        # If it's already a MetaflowException... no translation needed
+        raise
+    if isinstance(e, ImportError):
+        # Surprise ImportError here... (expected to see this handled and wrapped as MetaflowAzurePackagingError)
+        # Reraise it raw for visibility, it's a bug and is catastrophic anyway.
+        raise
+
+    from azure.core.exceptions import (
+        ClientAuthenticationError,
+        ResourceNotFoundError,
+        ResourceExistsError,
+        AzureError,
+    )
+
+    if isinstance(e, ClientAuthenticationError):
+        # Final line shows the TLDR
+        # Note we assume the str(e) is never empty string (otherwise, IndexError)
+        raise MetaflowAzureAuthenticationError(msg=str(e).splitlines()[-1])
+    elif isinstance(e, (ResourceNotFoundError, ResourceExistsError)):
+        raise MetaflowAzureResourceError(msg=str(e))
+    elif isinstance(e, AzureError):  # this is the base class for all Azure SDK errors
+        raise MetaflowInternalError(msg="Azure error: %s" % (str(e)))
+    else:
+        raise MetaflowInternalError(msg=str(e))
+
+
+def handle_exceptions(func):
+    """This is a decorator leveraging the logic from process_exception()"""
+
+    def _inner_func(*args, **kwargs):
+        try:
+            return func(*args, **kwargs)
+        except Exception as e:
+            process_exception(e)
+
+    return _inner_func
+
+
+@check_azure_deps
+def create_cacheable_default_azure_credentials(*args, **kwargs):
+    """azure.identity.DefaultAzureCredential is not readily cacheable in a dictionary
+    because it does not have a content based hash and equality implementations.
+
+    We implement a subclass CacheableDefaultAzureCredential to add them.
+
+    We need this because credentials will be part of the cache key in _ClientCache.
+    """
+    from azure.identity import DefaultAzureCredential
+
+    class CacheableDefaultAzureCredential(DefaultAzureCredential):
+        def __init__(self, *args, **kwargs):
+            super(CacheableDefaultAzureCredential, self).__init__(*args, **kwargs)
+            # Just hashing all the kwargs works because they are all individually
+            # hashable as of 7/15/2022.
+            #
+            # What if Azure adds unhashable things to kwargs?
+            # - We will have CI to catch this (it will always install latest Azure SDKs)
+            # - In Metaflow usage today we never specify any kwargs anyway. (see last line
+            #   of the outer function.
+            self._hash_code = hash((args, tuple(sorted(kwargs.items()))))
+
+        def __hash__(self):
+            return self._hash_code
+
+        def __eq__(self, other):
+            return hash(self) == hash(other)
+
+    return CacheableDefaultAzureCredential(*args, **kwargs)
+
+
+@check_azure_deps
+def create_static_token_credential(token_):
+    from azure.core.credentials import TokenCredential
+
+    class StaticTokenCredential(TokenCredential):
+        # We initialize with a fixed token (_cached_token).
+        #
+        # In most cases, we take a fast path - we always just return that fixed token.
+        # I.e. we generate the token once somewhere; subsequent operations use that same token.
+        #
+        # The fixed token can expire (defaults to several hours, but can be configured by an Azure admin)
+        #
+        # If we detect token expiration, we delegate all future token needs back to DefaultAzureCredential,
+        # which similarly supports token caching and regeneration of expired tokens.
+        #
+        # The net result is that we only generate new tokens when absolutely necessary.
+        #
+        # This dance is needed because DefaultAzureCredential is picklable and therefore cannot be shared
+        # across thread or process pool workers. Simply using a new DefaultAzureCredential in each thread
+        # imposes a one time penalty per object. That penalty comes from token generation may be large.
+        # e.g. Azure CLI is particularly slow (500ms - 1000ms).
+        #
+        # https://docs.microsoft.com/en-us/python/api/azure-identity/azure.identity.defaultazurecredential?view=azure-python
+        def __init__(self, token):
+            self._cached_token = token
+            self._credential = None
+
+        def get_token(self, *_scopes, **_kwargs):
+
+            if (self._cached_token.expires_on - time.time()) < 300:
+                from azure.identity import DefaultAzureCredential
+
+                self._credential = DefaultAzureCredential()
+            if self._credential:
+                return self._credential.get_token(*_scopes, **_kwargs)
+            return self._cached_token
+
+        # This object will be stored within a BlobServiceClient object.
+        # Implement __hash__ and __eq__ so this and the containing service objects become cacheable.
+        def __hash__(self):
+            return hash(self._cached_token)
+
+        def __eq__(self, other):
+            return self._cached_token == other._cached_token
+
+    return StaticTokenCredential(token_)
diff --git a/metaflow/plugins/azure/blob_service_client_factory.py b/metaflow/plugins/azure/blob_service_client_factory.py
new file mode 100644
index 00000000000..390cf6fa712
--- /dev/null
+++ b/metaflow/plugins/azure/blob_service_client_factory.py
@@ -0,0 +1,171 @@
+from metaflow.exception import MetaflowException
+from metaflow.metaflow_config import AZURE_STORAGE_BLOB_SERVICE_ENDPOINT
+from metaflow.plugins.azure.azure_utils import (
+    create_cacheable_default_azure_credentials,
+    check_azure_deps,
+)
+
+import os
+import threading
+
+
+class _ClientCache(object):
+    """
+    azure.storage.blob.BlobServiceClient objects internally cache HTTPS connections.
+    In order to reuse HTTPS connections, we need to reuse BlobServiceClient objects.
+
+    The effect we are going for is for each process or thread in the executor to
+    EACH REUSE ITS OWN long-lived BlobServiceClient object.
+    """
+
+    _cache = dict()
+
+    def __init__(self):
+        raise RuntimeError("_ClientCache may not be instantiated!")
+
+    @staticmethod
+    def get(
+        blob_service_endpoint,
+        credential=None,
+        max_single_put_size=None,
+        max_single_get_size=None,
+        max_chunk_get_size=None,
+        connection_data_block_size=None,
+    ):
+        """
+        Each spawned process will get its own fresh cache.
+
+        With each process's cache, each thread gets its own distinct service object.
+
+        The cache key includes all BlobServiceClient creation params PLUS process ID and thread ID.
+
+        This means that no more than one thread of control will ever access a cache key. This, and
+        the fact that dict() operations are thread-safe means that no additional synchronization
+        required here.
+        """
+        cache_key = (
+            os.getpid(),
+            threading.get_ident(),
+            credential,
+            blob_service_endpoint,
+            max_single_put_size,
+            max_single_get_size,
+            max_chunk_get_size,
+            connection_data_block_size,
+        )
+        service_just_created = None
+        if cache_key not in _ClientCache._cache:
+            service_just_created = _create_blob_service_client(
+                blob_service_endpoint,
+                credential=credential,
+                max_single_put_size=max_single_put_size,
+                max_single_get_size=max_single_get_size,
+                max_chunk_get_size=max_chunk_get_size,
+                connection_data_block_size=connection_data_block_size,
+                # BlobServiceClient accepts many more kwargs. We can add passthroughs as/when needed.
+            )
+            _ClientCache._cache[cache_key] = service_just_created
+        service_to_return = _ClientCache._cache[cache_key]
+        if (
+            service_just_created is not None
+            and service_just_created != service_to_return
+        ):
+            # This condition may not be fatal, but highly unexpected.
+            # Each thread of execution gets its own entry, so there *should* be no concurrent insertions
+            # on the same key.
+            #
+            # The Metaflow team REALLY wants to know if this ever happens, but we stop short of raising a
+            # fatal exception to not kill user jobs.
+            print(
+                "METAFLOW WARNING: Azure _ClientCache had the same cache key updated more than once"
+            )
+        return service_to_return
+
+
+# AZURE SDK tunables (AZURE_CLIENT_*)
+# ===================================
+# We try pick some sensible defaults here.
+#
+# Block size on underlying down TLS transport - Azure SDK defaults to 4096.
+# This larger block size dramatically improves aggregate download throughput.
+# 256KB (within order of 2x) seemed optimal for specific benchmark setup.
+# In future, exposing this as a tunable to users is a possibility if necessary.
+AZURE_CLIENT_CONNECTION_DATA_BLOCK_SIZE = 262144
+
+# When to use more than a single thread / connection for a single GET or PUT.
+AZURE_CLIENT_MAX_SINGLE_GET_SIZE_MB = 32
+AZURE_CLIENT_MAX_SINGLE_PUT_SIZE_MB = 64
+
+# Maximum chunk size when splitting a single blob GET into chunks for concurrent processing
+AZURE_CLIENT_MAX_CHUNK_GET_SIZE_MB = 16
+
+BYTES_IN_MB = 1024 * 1024
+
+
+def get_azure_blob_service_client(
+    credential=None,
+    credential_is_cacheable=False,
+    max_single_get_size=AZURE_CLIENT_MAX_SINGLE_GET_SIZE_MB * BYTES_IN_MB,
+    max_single_put_size=AZURE_CLIENT_MAX_SINGLE_PUT_SIZE_MB * BYTES_IN_MB,
+    max_chunk_get_size=AZURE_CLIENT_MAX_CHUNK_GET_SIZE_MB * BYTES_IN_MB,
+    connection_data_block_size=AZURE_CLIENT_CONNECTION_DATA_BLOCK_SIZE,
+):
+    """Returns a azure.storage.blob.BlobServiceClient.
+
+    The value adds are:
+    - connection caching (see _ClientCache)
+    - auto storage account URL detection
+    - auto credential handling (pull SAS token from environment, OR DefaultAzureCredential)
+    - sensible default values for Azure SDK tunables
+    """
+    if not AZURE_STORAGE_BLOB_SERVICE_ENDPOINT:
+        raise MetaflowException(
+            msg="Must configure METAFLOW_AZURE_STORAGE_BLOB_SERVICE_ENDPOINT"
+        )
+    blob_service_endpoint = AZURE_STORAGE_BLOB_SERVICE_ENDPOINT
+
+    if not credential:
+        credential = create_cacheable_default_azure_credentials()
+        credential_is_cacheable = True
+
+    if not credential_is_cacheable:
+        return _create_blob_service_client(
+            blob_service_endpoint,
+            credential=credential,
+            max_single_put_size=max_single_put_size,
+            max_single_get_size=max_single_get_size,
+            max_chunk_get_size=max_chunk_get_size,
+            connection_data_block_size=connection_data_block_size,
+            # BlobServiceClient accepts many more kwargs. We can add passthroughs as/when needed.
+        )
+
+    return _ClientCache.get(
+        blob_service_endpoint,
+        credential=credential,
+        max_single_put_size=max_single_put_size,
+        max_single_get_size=max_single_get_size,
+        max_chunk_get_size=max_chunk_get_size,
+        connection_data_block_size=connection_data_block_size,
+    )
+
+
+@check_azure_deps
+def _create_blob_service_client(
+    blob_service_endpoint,
+    credential=None,
+    max_single_put_size=None,
+    max_single_get_size=None,
+    max_chunk_get_size=None,
+    connection_data_block_size=None,
+):
+    from azure.storage.blob import BlobServiceClient
+
+    return BlobServiceClient(
+        blob_service_endpoint,
+        credential=credential,
+        max_single_put_size=max_single_put_size,
+        max_single_get_size=max_single_get_size,
+        max_chunk_get_size=max_chunk_get_size,
+        connection_data_block_size=connection_data_block_size,
+        # BlobServiceClient accepts many more kwargs. We can add passthroughs as/when needed.
+    )
diff --git a/metaflow/plugins/cards/card_datastore.py b/metaflow/plugins/cards/card_datastore.py
index 5065d4a7523..597b69b016e 100644
--- a/metaflow/plugins/cards/card_datastore.py
+++ b/metaflow/plugins/cards/card_datastore.py
@@ -14,6 +14,7 @@
     DATASTORE_CARD_LOCALROOT,
     DATASTORE_LOCAL_DIR,
     DATASTORE_CARD_SUFFIX,
+    DATASTORE_CARD_AZUREROOT,
 )
 
 from .exception import CardNotPresentException
@@ -45,7 +46,9 @@ class CardDatastore(object):
     def get_storage_root(cls, storage_type):
         if storage_type == "s3":
             return DATASTORE_CARD_S3ROOT
-        else:
+        elif storage_type == "azure":
+            return DATASTORE_CARD_AZUREROOT
+        elif storage_type == "local":
             # Borrowing some of the logic from LocalStorage.get_storage_root
             result = DATASTORE_CARD_LOCALROOT
             if result is None:
@@ -66,6 +69,11 @@ def get_storage_root(cls, storage_type):
                 result = orig_path
 
             return result
+        else:
+            # Let's make it obvious we need to update this block for each new datastore backend...
+            raise NotImplementedError(
+                "Card datastore does not support backend %s" % (storage_type,)
+            )
 
     def __init__(self, flow_datastore, pathspec=None):
         self._backend = flow_datastore._storage_impl
diff --git a/metaflow/plugins/conda/__init__.py b/metaflow/plugins/conda/__init__.py
index ec0321b4e8e..55dac5a41bd 100644
--- a/metaflow/plugins/conda/__init__.py
+++ b/metaflow/plugins/conda/__init__.py
@@ -3,6 +3,14 @@
 import json
 import fcntl
 
+from metaflow.exception import MetaflowException, MetaflowInternalError
+from metaflow.metaflow_config import (
+    CONDA_PACKAGE_S3ROOT,
+    DATASTORE_SYSROOT_S3,
+    CONDA_PACKAGE_AZUREROOT,
+    DATASTORE_SYSROOT_AZURE,
+)
+
 CONDA_MAGIC_FILE = "conda.dependencies"
 
 
@@ -42,3 +50,30 @@ def write_to_conda_manifest(ds_root, flow_name, key, value):
                 raise
         finally:
             fcntl.flock(f, fcntl.LOCK_UN)
+
+
+def get_conda_package_root(datastore_type):
+    # Yes, code duplication. But easier to read this way, at N=2
+    if datastore_type == "s3":
+        if CONDA_PACKAGE_S3ROOT is None:
+            if DATASTORE_SYSROOT_S3 is None:
+                raise MetaflowException(
+                    msg="METAFLOW_DATASTORE_SYSROOT_S3 must be set!"
+                )
+            return "%s/conda" % DATASTORE_SYSROOT_S3
+        else:
+            return CONDA_PACKAGE_S3ROOT
+    elif datastore_type == "azure":
+        if CONDA_PACKAGE_AZUREROOT is None:
+            if DATASTORE_SYSROOT_AZURE is None:
+                raise MetaflowException(
+                    msg="METAFLOW_DATASTORE_SYSROOT_AZURE must be set!"
+                )
+            return "%s/conda" % DATASTORE_SYSROOT_AZURE
+        else:
+            return CONDA_PACKAGE_AZUREROOT
+    else:
+        raise MetaflowInternalError(
+            msg="Unsupported storage backend '%s' for working with Conda"
+            % (datastore_type,)
+        )
diff --git a/metaflow/plugins/conda/batch_bootstrap.py b/metaflow/plugins/conda/batch_bootstrap.py
index 1217c72c39f..5b79a4b7c19 100644
--- a/metaflow/plugins/conda/batch_bootstrap.py
+++ b/metaflow/plugins/conda/batch_bootstrap.py
@@ -1,23 +1,19 @@
-import functools
 import json
-from multiprocessing import Pool
 import os
-import tarfile
 import shutil
-import subprocess
 import sys
 
-from metaflow.datatools import S3
+from metaflow.exception import MetaflowException
 from metaflow.metaflow_config import DATASTORE_LOCAL_DIR
 
 from ..env_escape import generate_trampolines, ENV_ESCAPE_PY
 
-from . import CONDA_MAGIC_FILE
+from . import CONDA_MAGIC_FILE, get_conda_package_root
 
 
-def bootstrap_environment(flow_name, env_id):
+def bootstrap_environment(flow_name, env_id, datastore_type):
     setup_conda_manifest(flow_name)
-    packages = download_conda_packages(flow_name, env_id)
+    packages = download_conda_packages(flow_name, env_id, datastore_type)
     install_conda_environment(env_id, packages)
 
 
@@ -31,18 +27,42 @@ def setup_conda_manifest(flow_name):
     )
 
 
-def download_conda_packages(flow_name, env_id):
+def download_conda_packages(flow_name, env_id, datastore_type):
     pkgs_folder = os.path.join(os.getcwd(), "pkgs")
     if not os.path.exists(pkgs_folder):
         os.makedirs(pkgs_folder)
+    # NOTE: if two runs use the same DATASTORE_LOCAL_DIR but different cloud based
+    # datastore roots (e.g. a different DATASTORE_SYSROOT_AZURE), then this breaks.
+    # This shared local manifest CONDA_MAGIC_FILE will say that cache_urls exist,
+    # but those URLs will not actually point to any real objects in the datastore
+    # for the second run.
     manifest_folder = os.path.join(os.getcwd(), DATASTORE_LOCAL_DIR, flow_name)
     with open(os.path.join(manifest_folder, CONDA_MAGIC_FILE)) as f:
         env = json.load(f)[env_id]
-        with S3() as s3:
-            for pkg in s3.get_many(env["cache_urls"]):
-                shutil.move(
-                    pkg.path, os.path.join(pkgs_folder, os.path.basename(pkg.key))
-                )
+        if datastore_type == "s3":
+            from metaflow.datatools import S3
+
+            with S3() as s3:
+                for pkg in s3.get_many(env["cache_urls"]):
+                    shutil.move(
+                        pkg.path, os.path.join(pkgs_folder, os.path.basename(pkg.key))
+                    )
+        elif datastore_type == "azure":
+            # Import DATASTORES dynamically... otherwise, circular import
+            from metaflow.datastore import DATASTORES
+
+            conda_package_root = get_conda_package_root("azure")
+            storage = DATASTORES["azure"](conda_package_root)
+            with storage.load_bytes(env["cache_urls"]) as load_result:
+                for key, tmpfile, _ in load_result:
+                    shutil.move(
+                        tmpfile, os.path.join(pkgs_folder, os.path.basename(key))
+                    )
+        else:
+            raise MetaflowException(
+                msg="Downloading conda code packages from datastore backend %s is unimplemented!"
+                % datastore_type
+            )
         return env["order"]
 
 
@@ -68,4 +88,4 @@ def install_conda_environment(env_id, packages):
 
 
 if __name__ == "__main__":
-    bootstrap_environment(sys.argv[1], sys.argv[2])
+    bootstrap_environment(sys.argv[1], sys.argv[2], sys.argv[3])
diff --git a/metaflow/plugins/conda/conda_environment.py b/metaflow/plugins/conda/conda_environment.py
index 7ca49e16385..713da1a337f 100644
--- a/metaflow/plugins/conda/conda_environment.py
+++ b/metaflow/plugins/conda/conda_environment.py
@@ -70,14 +70,14 @@ def _get_executable(self, step_name):
     def set_local_root(self, ds_root):
         self.local_root = ds_root
 
-    def bootstrap_commands(self, step_name):
+    def bootstrap_commands(self, step_name, datastore_type):
         # Bootstrap conda and execution environment for step
         env_id = self._get_env_id(step_name)
         if env_id is not None:
             return [
                 "echo 'Bootstrapping environment...'",
-                'python -m metaflow.plugins.conda.batch_bootstrap "%s" %s'
-                % (self.flow.name, env_id),
+                'python -m metaflow.plugins.conda.batch_bootstrap "%s" %s "%s"'
+                % (self.flow.name, env_id, datastore_type),
                 "echo 'Environment bootstrapped.'",
             ]
         return []
@@ -129,8 +129,8 @@ def get_client_info(cls, flow_name, metadata):
         }
         return new_info
 
-    def get_package_commands(self, code_package_url):
-        return self.base_env.get_package_commands(code_package_url)
+    def get_package_commands(self, code_package_url, datastore_type):
+        return self.base_env.get_package_commands(code_package_url, datastore_type)
 
     def get_environment_info(self):
         return self.base_env.get_environment_info()
diff --git a/metaflow/plugins/conda/conda_step_decorator.py b/metaflow/plugins/conda/conda_step_decorator.py
index 4e0dad78a01..b931957865d 100644
--- a/metaflow/plugins/conda/conda_step_decorator.py
+++ b/metaflow/plugins/conda/conda_step_decorator.py
@@ -18,14 +18,14 @@
 from metaflow.extension_support import EXT_PKG
 from metaflow.metaflow_environment import InvalidEnvironmentException
 from metaflow.metadata import MetaDatum
-from metaflow.metaflow_config import get_pinned_conda_libs, CONDA_PACKAGE_S3ROOT
+from metaflow.metaflow_config import (
+    get_pinned_conda_libs,
+)
 from metaflow.util import get_metaflow_root
-from metaflow.datastore import LocalStorage
-from metaflow.datatools import S3
-from metaflow.unbounded_foreach import UBF_CONTROL
+from metaflow.datastore import DATASTORES, LocalStorage
 
 from ..env_escape import generate_trampolines
-from . import read_conda_manifest, write_to_conda_manifest
+from . import read_conda_manifest, write_to_conda_manifest, get_conda_package_root
 from .conda import Conda
 
 try:
@@ -94,7 +94,7 @@ def is_enabled(self, ubf_context=None):
         )
 
     def _lib_deps(self):
-        deps = get_pinned_conda_libs(self._python_version())
+        deps = get_pinned_conda_libs(self._python_version(), self.flow_datastore.TYPE)
 
         base_deps = self.base_attributes["libraries"]
         deps.update(base_deps)
@@ -155,7 +155,11 @@ def _resolve_step_environment(self, ds_root, force=False):
                 }
             else:
                 payload = cached_deps[env_id]
-            if self.flow_datastore.TYPE == "s3" and "cache_urls" not in payload:
+
+            if (
+                self.flow_datastore.TYPE in ("s3", "azure")
+                and "cache_urls" not in payload
+            ):
                 payload["cache_urls"] = self._cache_env()
             write_to_conda_manifest(ds_root, self.flow.name, env_id, payload)
             CondaStepDecorator.environments = CondaStepDecorator.conda.environments(
@@ -176,7 +180,6 @@ def _download(entry):
         for package_info in self.conda.package_info(env_id):
             url = urlparse(package_info["url"])
             path = os.path.join(
-                CONDA_PACKAGE_S3ROOT,
                 url.netloc,
                 url.path.lstrip("/"),
                 package_info["md5"],
@@ -195,8 +198,19 @@ def _download(entry):
             files.append((path, tarball_path))
         if to_download:
             Pool(8).map(_download, to_download)
-        with S3() as s3:
-            s3.put_files(files, overwrite=False)
+
+        list_of_path_and_filehandle = [
+            (path, open(tarball_path, "rb")) for path, tarball_path in files
+        ]
+
+        # We need our own storage backend so that we can customize datastore_root on it
+        # in a clearly safe way, without the existing backend owned by FlowDatastore
+        storage_impl = DATASTORES[self.flow_datastore.TYPE]
+        storage = storage_impl(get_conda_package_root(self.flow_datastore.TYPE))
+        storage.save_bytes(
+            list_of_path_and_filehandle, len_hint=len(list_of_path_and_filehandle)
+        )
+
         return [files[0] for files in files]
 
     def _prepare_step_environment(self, step_name, ds_root):
diff --git a/metaflow/plugins/kubernetes/kubernetes.py b/metaflow/plugins/kubernetes/kubernetes.py
index 053761df2d4..1155133b0b3 100644
--- a/metaflow/plugins/kubernetes/kubernetes.py
+++ b/metaflow/plugins/kubernetes/kubernetes.py
@@ -1,34 +1,32 @@
-import hashlib
 import json
 import math
 import os
-import re
-import select
 import shlex
 import time
 
 from metaflow import current, util
-from metaflow.datatools.s3tail import S3Tail
-from metaflow.exception import MetaflowException, MetaflowInternalError
+from metaflow.exception import MetaflowException
 from metaflow.metaflow_config import (
     BATCH_METADATA_SERVICE_HEADERS,
     BATCH_METADATA_SERVICE_URL,
     DATASTORE_CARD_S3ROOT,
-    DATASTORE_LOCAL_DIR,
     DATASTORE_SYSROOT_S3,
     DATATOOLS_S3ROOT,
     DEFAULT_AWS_CLIENT_PROVIDER,
     DEFAULT_METADATA,
     KUBERNETES_SANDBOX_INIT_SCRIPT,
     S3_ENDPOINT_URL,
+    AZURE_STORAGE_BLOB_SERVICE_ENDPOINT,
+    DATASTORE_SYSROOT_AZURE,
+    DATASTORE_CARD_AZUREROOT,
 )
 from metaflow.mflog import (
     BASH_SAVE_LOGS,
     bash_capture_logs,
     export_mflog_env_vars,
     tail_logs,
+    get_log_tailer,
 )
-from metaflow.mflog.mflog import refine, set_should_persist
 
 from .kubernetes_client import KubernetesClient
 
@@ -79,10 +77,15 @@ def _command(
             stdout_path=STDOUT_PATH,
             stderr_path=STDERR_PATH,
         )
-        init_cmds = self._environment.get_package_commands(code_package_url)
+        init_cmds = self._environment.get_package_commands(
+            code_package_url, self._datastore.TYPE
+        )
         init_expr = " && ".join(init_cmds)
         step_expr = bash_capture_logs(
-            " && ".join(self._environment.bootstrap_commands(step_name) + step_cmds)
+            " && ".join(
+                self._environment.bootstrap_commands(step_name, self._datastore.TYPE)
+                + step_cmds
+            )
         )
 
         # Construct an entry point that
@@ -184,7 +187,7 @@ def create_job(
             )
             .environment_variable("METAFLOW_DATASTORE_SYSROOT_S3", DATASTORE_SYSROOT_S3)
             .environment_variable("METAFLOW_DATATOOLS_S3ROOT", DATATOOLS_S3ROOT)
-            .environment_variable("METAFLOW_DEFAULT_DATASTORE", "s3")
+            .environment_variable("METAFLOW_DEFAULT_DATASTORE", self._datastore.TYPE)
             .environment_variable("METAFLOW_DEFAULT_METADATA", DEFAULT_METADATA)
             .environment_variable("METAFLOW_KUBERNETES_WORKLOAD", 1)
             .environment_variable("METAFLOW_RUNTIME_ENVIRONMENT", "kubernetes")
@@ -192,6 +195,19 @@ def create_job(
             .environment_variable(
                 "METAFLOW_DEFAULT_AWS_CLIENT_PROVIDER", DEFAULT_AWS_CLIENT_PROVIDER
             )
+            .environment_variable("METAFLOW_S3_ENDPOINT_URL", S3_ENDPOINT_URL)
+            .environment_variable(
+                "METAFLOW_AZURE_STORAGE_BLOB_SERVICE_ENDPOINT",
+                AZURE_STORAGE_BLOB_SERVICE_ENDPOINT,
+            )
+            .environment_variable(
+                "METAFLOW_DATASTORE_SYSROOT_AZURE", DATASTORE_SYSROOT_AZURE
+            )
+            .environment_variable("METAFLOW_CARD_AZUREROOT", DATASTORE_CARD_AZUREROOT)
+            # support Metaflow sandboxes
+            .environment_variable(
+                "METAFLOW_INIT_SCRIPT", KUBERNETES_SANDBOX_INIT_SCRIPT
+            )
             # Skip setting METAFLOW_DATASTORE_SYSROOT_LOCAL because metadata sync
             # between the local user instance and the remote Kubernetes pod
             # assumes metadata is stored in DATASTORE_LOCAL_DIR on the Kubernetes
@@ -199,16 +215,6 @@ def create_job(
             # see get_datastore_root_from_config in datastore/local.py).
         )
 
-        # add METAFLOW_S3_ENDPOINT_URL
-        if S3_ENDPOINT_URL is not None:
-            job.environment_variable("METAFLOW_S3_ENDPOINT_URL", S3_ENDPOINT_URL)
-
-        # support Metaflow sandboxes
-        if KUBERNETES_SANDBOX_INIT_SCRIPT is not None:
-            job.environment_variable(
-                "METAFLOW_INIT_SCRIPT", KUBERNETES_SANDBOX_INIT_SCRIPT
-            )
-
         for name, value in env.items():
             job.environment_variable(name, value)
 
@@ -272,8 +278,9 @@ def wait_for_launch(job):
                 time.sleep(update_delay(time.time() - start_time))
 
         prefix = b"[%s] " % util.to_bytes(self._job.id)
-        stdout_tail = S3Tail(stdout_location)
-        stderr_tail = S3Tail(stderr_location)
+
+        stdout_tail = get_log_tailer(stdout_location, self._datastore.TYPE)
+        stderr_tail = get_log_tailer(stderr_location, self._datastore.TYPE)
 
         # 1) Loop until the job has started
         wait_for_launch(self._job)
@@ -286,7 +293,6 @@ def wait_for_launch(job):
             echo=echo,
             has_log_updates=lambda: self._job.is_running,
         )
-
         # 3) Fetch remaining logs
         #
         # It is possible that we exit the loop above before all logs have been
diff --git a/metaflow/plugins/kubernetes/kubernetes_decorator.py b/metaflow/plugins/kubernetes/kubernetes_decorator.py
index 974dbdfdd30..5f6336c3c6d 100644
--- a/metaflow/plugins/kubernetes/kubernetes_decorator.py
+++ b/metaflow/plugins/kubernetes/kubernetes_decorator.py
@@ -123,9 +123,9 @@ def __init__(self, attributes=None, statically_defined=False):
     # Refer https://github.com/Netflix/metaflow/blob/master/docs/lifecycle.png
     def step_init(self, flow, graph, step, decos, environment, flow_datastore, logger):
         # Executing Kubernetes jobs requires a non-local datastore.
-        if flow_datastore.TYPE != "s3":
+        if flow_datastore.TYPE not in ("s3", "azure"):
             raise KubernetesException(
-                "The *@kubernetes* decorator requires --datastore=s3 at the moment."
+                "The *@kubernetes* decorator requires --datastore=s3 or --datastore=azure at the moment."
             )
 
         # Set internal state.
diff --git a/metaflow/plugins/kubernetes/kubernetes_job.py b/metaflow/plugins/kubernetes/kubernetes_job.py
index a1f4d30425e..9a788b65e46 100644
--- a/metaflow/plugins/kubernetes/kubernetes_job.py
+++ b/metaflow/plugins/kubernetes/kubernetes_job.py
@@ -254,6 +254,9 @@ def memory(self, mem):
         return self
 
     def environment_variable(self, name, value):
+        # Never set to None
+        if value is None:
+            return self
         self._kwargs["environment_variables"] = dict(
             self._kwargs.get("environment_variables", {}), **{name: value}
         )
diff --git a/test/core/contexts.json b/test/core/contexts.json
index f469169cbdc..f8eac9d7468 100644
--- a/test/core/contexts.json
+++ b/test/core/contexts.json
@@ -30,6 +30,36 @@
                 "S3FailureTest"
             ]
         },
+        {
+            "name": "python3-all-local-azure-storage",
+            "disabled": true,
+            "env": {
+                "METAFLOW_USER": "tester",
+                "METAFLOW_RUN_BOOL_PARAM": "False",
+                "METAFLOW_RUN_NO_DEFAULT_PARAM": "test_str",
+                "METAFLOW_DEFAULT_METADATA": "local"
+            },
+            "python": "python3",
+            "top_options": [
+                "--metadata=local",
+                "--datastore=azure",
+                "--environment=local",
+                "--event-logger=nullSidecarLogger",
+                "--no-pylint",
+                "--quiet"
+            ],
+            "run_options": [
+                "--max-workers", "50",
+                "--max-num-splits", "10000",
+                "--tag", "\u523a\u8eab means sashimi",
+                "--tag", "multiple tags should be ok"
+            ],
+            "checks": [ "python3-cli", "python3-metadata"],
+            "disabled_tests": [
+                "LargeArtifactTest",
+                "S3FailureTest"
+            ]
+        },
         {
             "name": "dev-local",
             "disabled": true,


Code review comments:
diff_hunk: @@ -36,8 +36,50 @@ def init_config():
 METAFLOW_CONFIG = init_config()
 
 
-def from_conf(name, default=None):
-    return os.environ.get(name, METAFLOW_CONFIG.get(name, default))
+def from_conf(name, default=None, from_env_only=False, validate_fn=None):
path: metaflow/metaflow_config.py
text: @user1:
from_env_only does not seem to be used here (or I missed it), what is the use case?

@author:
I'll remove this. It was used for a security token value  ("shared access signature") that should not live in the config JSON - but we decided not to support signature at all.

diff_hunk: @@ -80,6 +83,93 @@ def path(self):
         return self._path
 
 
+class Azure(object):
path: metaflow/includefile.py
text: @user1:
nit: maybe this whole this can move over to the azure folder to keep this a little cleaner (kind of like S3 doesn't show up here except in the dictionary?

@author:
The intent is that this should stay logically private for includefile usage ONLY.  I think we can do something like `metaflow/plugins/azure/includefile_support.py` which would be sufficiently explicit about it.