Better Azure blob storage support

[Tom-Newton]

TL;DR

Support Azure blob storage for storing metadata and raw data including structured datasets, without interrupting other standard Azure authentication. Also ensure storage_options are set consistently for all uses of fsspec.

Type

Its debatable. The original github issue was a bug for fixing Azure authentication but it could be argued that Azure blob storage support is a new feature.

• Bug Fix
• Feature
• Plugin

Are all requirements met?

• Code completed
• Smoke tested - Not really sure what is expected but I have used these change to deploy real workflows.
• Unit tests added - I added a bit of new coverage, e.g. initialising the AzureBlobFilesystem.
• Code documentation added - There are code comments, but I think ideally there would be a page in the documentation for Azure.
• Any pending items have an associated Issue - There aren't any

Complete description

Added a new component to DataConfig for Azure similar to the ones for S3 and GCS. This allows configuring the flytekit's storage account without disrupting other Azure auth that may be used by flyte task implementations. This is done by setting the environment variables FLYTE_AZURE_STORAGE_ACCOUNT_NAME, FLYTE_AZURE_STORAGE_ACCOUNT_KEY, FLYTE_AZURE_TENANT_ID, FLYTE_AZURE_CLIENT_ID, FLYTE_AZURE_CLIENT_SECRET. You can set these using a pod template so that authentication to your flyte storage account works across your whole flyte deployment, inrespective of what authentication each task uses.

Use DataConfig to construct fsspec storage options in data_persistence.py. We reuse the same function for fetchign storage_options elsewhere inlcuding for encode and decode of structured datasets. This ensures local, S3, GCS and Azure work wherever fsspec is used.

You should specify configmap.core.propeller.rawoutput-prefix to be in the format abfs://<container-name>/<path-within-container>.

Tracking Issue

Related discussion was on flyteorg/flyte#3962 but technically that issues was about workload identity which was mostly fixed by #1813

Follow-up issue

NA

[welcome]

Thank you for opening this pull request! 🙌

These tips will help get your PR across the finish line:

• Most of the repos have a PR template; if not, fill it out to the best of your knowledge.
• Sign off your commits (Reference: DCO Guide).

[kumare3]

cc @fiedlerNr9

[codecov]

Codecov Report

Attention: 27 lines in your changes are missing coverage. Please review.

Comparison is base (1737048) 18.73% compared to head (72a440e) 54.74%.
Report is 27 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff             @@
##           master    #1842       +/-   ##
===========================================
+ Coverage   18.73%   54.74%   +36.01%     
===========================================
  Files         332      293       -39     
  Lines       31398    22092     -9306     
  Branches     3082     2167      -915     
===========================================
+ Hits         5882    12094     +6212     
+ Misses      25432     9849    -15583     
- Partials       84      149       +65     

Files	Coverage Δ
flytekit/configuration/__init__.py	73.42% <100.00%> (+1.91%)	⬆️
flytekit/configuration/internal.py	80.26% <100.00%> (-0.26%)	⬇️
...t-polars/flytekitplugins/polars/sd_transformers.py	100.00% <100.00%> (ø)
flytekit/types/structured/basic_dfs.py	37.96% <25.00%> (+37.96%)	⬆️
flytekit/core/data_persistence.py	38.88% <30.00%> (+4.29%)	⬆️

... and 359 files with indirect coverage changes

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[Tom-Newton]

From the type hint on get_filesystem_for_path it looks like we assume get_filesystem never returns None so I thought probably best to delete this. If anyone thinks its important I'm happy to add it back.

[devictr]

@wild-endeavor I think you wanted to delete this already? So this seems like a good idea

[Tom-Newton]

Hmm... it looks like a couple of my new tests are failing on the windows builds.

E           Expected: to_parquet('abfs://container/parquet_df/00000', coerce_timestamps=<ANY>, allow_truncated_timestamps=<ANY>, storage_options={'account_name': 'accountname_from_storage_options'})
E           Actual: to_parquet('abfs://container/parquet_df\\00000', coerce_timestamps='us', allow_truncated_timestamps=False, storage_options={'account_name': 'accountname_from_storage_options'})

Asserting on the path was not really the primary purpose of this test so I will probably make the assertion just on the storage_options. I suspect this would work because blob storage blob names can be arbitrary strings but normally forward slashes / are used, and various tools will expect / to be the directory seporator.

[fiedlerNr9]

tested on a live cluster and works like expected - also with workload identities 👍

[Tom-Newton]

There are 3 test suites that timeout out after 6 hours. It looks like they all failed in the Cache pip stage so I don't think these are related to my changes.

[wild-endeavor]

just some nits

[welcome]

Congrats on merging your first pull request! 🎉