Update eager: make sure client secret can be specified as env var

flytekit/bin/entrypoint.py

@@ -177,6 +177,13 @@ def _dispatch_execute(
     for k, v in output_file_dict.items():
         utils.write_proto_to_file(v.to_flyte_idl(), os.path.join(ctx.execution_state.engine_dir, k))
 
+    # make sure an event loop exists for data persistence step
+    try:
+        asyncio.get_event_loop()
+    except Exception:
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+
     ctx.file_access.put_data(ctx.execution_state.engine_dir, output_prefix, is_multipart=True)
     logger.info(f"Engine folder written successfully to the output prefix {output_prefix}")
 

flytekit/experimental/eager_function.py

@@ -1,5 +1,6 @@
 import asyncio
 import inspect
+import os
 import signal
 from contextlib import asynccontextmanager
 from datetime import datetime, timedelta, timezone
@@ -382,6 +383,7 @@
     timeout: Optional[timedelta] = None,
     poll_interval: Optional[timedelta] = None,
     local_entrypoint: bool = False,
+    client_secret_env_var: Optional[str] = None,
     **kwargs,
 ):
     """Eager workflow decorator.
@@ -396,6 +398,8 @@
     :param local_entrypoint: If True, the eager workflow will can be executed locally but use the provided
         :py:func:`~flytekit.remote.FlyteRemote` object to create task/workflow executions. This is useful for local
         testing against a remote Flyte cluster.
+    :param client_secret_env_var: if specified, binds the client secret to the specified environment variable for
+        remote authentication.
     :param kwargs: keyword-arguments forwarded to :py:func:`~flytekit.task`.
 
     This type of workflow will execute all flyte entities within it eagerly, meaning that all python constructs can be
@@ -488,7 +492,10 @@
             remote=remote,
             client_secret_group=client_secret_group,
             client_secret_key=client_secret_key,
+            timeout=timeout,
+            poll_interval=poll_interval,
             local_entrypoint=local_entrypoint,
+            client_secret_env_var=client_secret_env_var,
             **kwargs,
         )
 
@@ -510,7 +517,9 @@
             execution_id = exec_params.execution_id
 
         async_stack = AsyncStack(task_id, execution_id)
-        _remote = _prepare_remote(_remote, ctx, client_secret_group, client_secret_key, local_entrypoint)
+        _remote = _prepare_remote(
+            _remote, ctx, client_secret_group, client_secret_key, local_entrypoint, client_secret_env_var
+        )
 
         # make sure sub-nodes as cleaned up on termination signal
         loop = asyncio.get_event_loop()
@@ -533,8 +542,10 @@
                 await cleanup_fn()
 
     secret_requests = kwargs.pop("secret_requests", None) or []
-    if client_secret_group is not None and client_secret_key is not None:
+    try:
         secret_requests.append(Secret(group=client_secret_group, key=client_secret_key))
+    except ValueError:
+        pass
 
     return task(
         wrapper,
@@ -551,6 +562,7 @@
     client_secret_group: Optional[str] = None,
     client_secret_key: Optional[str] = None,
     local_entrypoint: bool = False,
+    client_secret_env_var: Optional[str] = None,
 ) -> Optional[FlyteRemote]:
     """Prepare FlyteRemote object for accessing Flyte cluster in a task running on the same cluster."""
 
@@ -576,7 +588,7 @@
     if remote.config.platform.endpoint.startswith("localhost"):
         # replace sandbox endpoints with internal dns, since localhost won't exist within the Flyte cluster
         return _internal_demo_remote(remote)
-    return _internal_remote(remote, client_secret_group, client_secret_key)
+    return _internal_remote(remote, client_secret_group, client_secret_key, client_secret_env_var)
 
 
 def _internal_demo_remote(remote: FlyteRemote) -> FlyteRemote:
@@ -605,16 +617,25 @@
 
 def _internal_remote(
     remote: FlyteRemote,
-    client_secret_group: str,
-    client_secret_key: str,
+    client_secret_group: Optional[str],
+    client_secret_key: Optional[str],
+    client_secret_env_var: Optional[str],
 ) -> FlyteRemote:
     """Derives a FlyteRemote object from a yaml configuration file, modifying parts to make it work internally."""
-    assert client_secret_group is not None, "secret_group must be defined when using a remote cluster"
-    assert client_secret_key is not None, "secret_key must be defined a remote cluster"
     secrets_manager = current_context().secrets
     client_secret = secrets_manager.get(client_secret_group, client_secret_key)
     # get the raw output prefix from the context that's set from the pyflyte-execute entrypoint
     # (see flytekit/bin/entrypoint.py)
+
+    if client_secret_env_var is not None:
+        # this creates a remote client where the env var client secret is sufficient for authentication
+        os.environ[client_secret_env_var] = client_secret
+        try:
+            remote_cls = type(remote)
+            return remote_cls()
+        except Exception as exc:
+            raise TypeError(f"Unable to authenticate remote class {remote_cls} with client secret") from exc
+
     ctx = FlyteContextManager.current_context()
     return FlyteRemote(
         config=remote.config.with_params(