catch Asset Definition Exception

[yamhirotoCS]

Hello,
Our Dagster loads instance from several tools, such as Airbyte, Fivetran, and a table containing dashboard names.
Right now, our dagster instance throws an error and becomes unusable if :

• our Airbyte instance is down
• Fivetran is down
• someone renames a dashboard which is part of the asset key.

If some assets become unavailable for the reason above, I still want my Dasgter instance to stand and load the other 'healthy' jobs.

How can we catch Exception related to AssetSelection when jobs are being loaded? :
_DagsterInvalidSubsetError: AssetKey(s) {AssetKey(['prefix', 'key'])} were selected, but no AssetsDefinition objects supply these keys_
I would then simply replace it with a dummy asset. until the root cause is fixed

[OwenKephart]

Hi @yamhirotoCS -- two general categories of ideas:

1. Don't use explicit asset keys when creating jobs:

# this call may return an empty list if fivetran is down
fivetran_assets = get_fivetran_assets(...)

fivetran_job = define_asset_job(asset_selection=fivetran_assets)

This solution would allow for empty jobs to be created in these situations.

2. Brute force:

try:
    defs = Definitions(assets=assets, ...)
except:
    dummy_assets = ... 
    defs = Definitions(assets=assets + dummy_assets, ...)

Without knowing more about your setup, it might be difficult to get the actual set of required dummy assets, or maybe it's just something along the lines of "any asset key explicitly mentioned in a job's AssetSelection"

[yamhirotoCS]

Hey @OwenKephart ,
thanks for coming back to me.

1. Some 'standard' jobs are handled that way in our code, but we have other jobs where assets need to be explicitly defined : job with a specific target asset that needs to be refreshed, with the required dependencies.
   For instance, we pull daily the list of dashboard made by analysts in a table, and we dynamically create an asset for each of them. Only a few dashboards need to have their own job, which we define in a config file with the specific dashboard asset and relevant dependencies. But say if we're not notified of a dashboard renaming, the asset gets a new name which doesn't match the assetKey in the job config . Now that I think about it, perhaps I can simply create an alias when defining those dashboard assets

I guess my issue is in two parts :

• If one service is down, its assets should be ignored but other assets should still be loaded.
• If there are jobs related to assets not available at a given time, they should be ignored.

As a test for part 1:
I replaced the actual credentials for Airbyte with dummies to raise the Exception that I want to catch :
from dagster._core.definitions.events import Failure
raise Failure(f"Max retries ({self.request_max_retries}) exceeded with url: {url}.")
Full stacktrace at the end

1. I played with try/catch at the asset loading for both Airbyte/ Fivetran, tested with/without precising Failure in the catch step

try:
    fivetran_instance = FivetranResource(
        api_key="dummy",
        api_secret=EnvVar("FIVETRAN_API_SECRET"),
    )

    fivetran_assets = load_assets_from_fivetran_instance(
        fivetran_instance,
        connector_to_group_fn=get_translator("group"),
    )
except:  / except Failure:
    fivetran_assets = my_dummy_asset

2. Tested the suggested brute force (for Fivetran let's say)

try:
    defs = Definitions(
        jobs=all_jobs,
        assets=load_assets_from_modules([airbyte, fivetran, other]),
        [...]
    )
except:
    defs = Definitions(
        jobs=all_jobs,
        assets=load_assets_from_modules([airbyte, other]),
        [...]
    )

Alas without success. Am I trying to catch the exception at the wrong place ? started digging into dagster/_grpc/server.py as well.

dagster._core.definitions.events.Failure: Max retries (3) exceeded with url: https://api.fivetran.com/v1/groups.
  File "[...]/.venv/lib/python3.11/site-packages/dagster/_grpc/server.py", line 295, in __init__
    self._loaded_repositories: Optional[LoadedRepositories] = LoadedRepositories(
                                                              ^^^^^^^^^^^^^^^^^^^
  File "[...]/.venv/lib/python3.11/site-packages/dagster/_grpc/server.py", line 154, in __init__
    repo_def = recon_repo.get_definition()
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "[...]/.venv/lib/python3.11/site-packages/dagster/_core/definitions/reconstruct.py", line 119, in get_definition
    return repository_def_from_pointer(self.pointer, self.repository_load_data)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "[...]/.venv/lib/python3.11/site-packages/dagster/_core/definitions/reconstruct.py", line 742, in repository_def_from_pointer
    repo_def = repository_def_from_target_def(target, repository_load_data)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "[...]/.venv/lib/python3.11/site-packages/dagster/_core/definitions/reconstruct.py", line 731, in repository_def_from_target_def
    return target.compute_repository_definition()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "[...]/.venv/lib/python3.11/site-packages/dagster/_core/definitions/repository_definition/repository_definition.py", line 491, in compute_repository_definition
    repository_load_data = self._compute_repository_load_data()
                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "[...]/.venv/lib/python3.11/site-packages/dagster/_core/definitions/repository_definition/repository_definition.py", line 436, in _compute_repository_load_data
    cached_data_by_key={
                       ^
  File "[...]/.venv/lib/python3.11/site-packages/dagster/_core/definitions/repository_definition/repository_definition.py", line 437, in <dictcomp>
    defn.unique_id: defn.compute_cacheable_data()
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "[...]/.venv/lib/python3.11/site-packages/dagster/_core/definitions/cacheable_assets.py", line 216, in compute_cacheable_data
    return self._wrapped.compute_cacheable_data()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "[...]/.venv/lib/python3.11/site-packages/dagster_fivetran/asset_defs.py", line 398, in compute_cacheable_data
    for connector in self._get_connectors():
                     ^^^^^^^^^^^^^^^^^^^^^^
  File "[...]/.venv/lib/python3.11/site-packages/dagster_fivetran/asset_defs.py", line 362, in _get_connectors
    groups = self._fivetran_instance.make_request("GET", "groups")["items"]
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "[...]/.venv/lib/python3.11/site-packages/dagster_fivetran/resources.py", line 126, in make_request
    raise Failure(f"Max retries ({self.request_max_retries}) exceeded with url: {url}.")

[yamhirotoCS]

I am unable to catch any exception, I was able to add some breakpoints() and navigated with the debugger to see how the flow was going : it somehow goes to dagster/_grpc/server.py where the fail occurs.

            self._loaded_repositories: Optional[LoadedRepositories] = LoadedRepositories(
                loadable_target_origin,
                self._entry_point,
                self._container_image,
            )

I was able to use the debugger with : dagster api grpc --module-name USER_CODE_REPO --host 0.0.0.0 --port 4266

I guess I need to review the dagster architecture