@ancient-wolf-19325, "asctime": "2025-01-14 101641,670", "name": "flytekit", "levelname": "ERROR", "message": "!! Begin Non-recoverable System Error Captured by Flyte !!"} {"asctime": "2025-01-14 101641,670", "name": "flytekit", "levelname": "ERROR", "message": "Trace:\n\n Traceback (most recent call last):\n File \"/opt/venv/lib/python3.10/site-packages/s3fs/core.py\", line 729, in _lsdir\n async for c in self._iterdir(\n File \"/opt/venv/lib/python3.10/site-packages/s3fs/core.py\", line 779, in _iterdir\n async for i in it:\n File \"/opt/venv/lib/python3.10/site-packages/aiobotocore/paginate.py\", line 30, in __anext__\n response = await self._make_request(current_kwargs)\n File \"/opt/venv/lib/python3.10/site-packages/aiobotocore/client.py\", line 412, in _make_api_call\n raise error_class(parsed_response, operation_name)\n botocore.exceptions.ClientError: An error occurred (SlowDown) when calling the ListObjectsV2 operation (reached max retries: 4): Please reduce your request rate.\n\n The above exception was the direct cause of the following exception:\n\n Traceback (most recent call last):\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/core/base_task.py\", line 800, in dispatch_execute\n literals_map, native_outputs_as_map = run_sync(\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/utils/asyn.py\", line 93, in run_sync\n return self._runner_map[name].run(coro)\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/utils/asyn.py\", line 72, in run\n res = fut.result(None)\n File \"/usr/local/lib/python3.10/concurrent/futures/_base.py\", line 458, in result\n return self.__get_result()\n File \"/usr/local/lib/python3.10/concurrent/futures/_base.py\", line 403, in __get_result\n raise self._exception\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/core/base_task.py\", line 655, in _output_to_literal_map\n raise e\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/core/type_engine.py\", line 1397, in async_to_literal\n lv = await transformer.async_to_literal(ctx, python_val, python_type, expected)\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/core/type_engine.py\", line 1677, in async_to_literal\n lit_list = await _run_coros_in_chunks(lit_list)\n File \"/opt/venv/lib/python3.10/site-packages/fsspec/asyn.py\", line 268, in _run_coros_in_chunks\n result, k = await done.pop()\n File \"/opt/venv/lib/python3.10/site-packages/fsspec/asyn.py\", line 245, in _run_coro\n return await asyncio.wait_for(coro, timeout=timeout), i\n File \"/usr/local/lib/python3.10/asyncio/tasks.py\", line 408, in wait_for\n return await fut\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/core/type_engine.py\", line 1397, in async_to_literal\n lv = await transformer.async_to_literal(ctx, python_val, python_type, expected)\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/core/type_engine.py\", line 2150, in async_to_literal\n await _run_coros_in_chunks([c for c in lit_map.values()])\n File \"/opt/venv/lib/python3.10/site-packages/fsspec/asyn.py\", line 268, in _run_coros_in_chunks\n result, k = await done.pop()\n File \"/opt/venv/lib/python3.10/site-packages/fsspec/asyn.py\", line 245, in _run_coro\n return await asyncio.wait_for(coro, timeout=timeout), i\n File \"/usr/local/lib/python3.10/asyncio/tasks.py\", line 408, in wait_for\n return await fut\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/core/type_engine.py\", line 1397, in async_to_literal\n lv = await transformer.async_to_literal(ctx, python_val, python_type, expected)\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/types/pickle/pickle.py\", line 94, in async_to_literal\n remote_path = await FlytePickle.to_pickle(ctx, python_val)\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/types/pickle/pickle.py\", line 50, in to_pickle\n return await ctx.file_access.async_put_raw_data(uri)\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/core/data_persistence.py\", line 423, in async_put_raw_data\n r = await self._put(from_path, to_path, **kwargs)\n File \"/opt/venv/lib/python3.10/site-packages/flytekit/core/data_persistence.py\", line 357, in _put\n dst = await file_system._put(from_path, to_path, recursive=recursive, **kwargs) # pylint: disable=W0212\n File \"/opt/venv/lib/python3.10/site-packages/fsspec/asyn.py\", line 565, in _put\n trailing_sep(rpath) or await self._isdir(rpath)\n File \"/opt/venv/lib/python3.10/site-packages/s3fs/core.py\", line 1498, in _isdir\n return bool(await self._lsdir(path))\n File \"/opt/venv/lib/python3.10/site-packages/s3fs/core.py\", line 742, in _lsdir\n raise translate_boto_error(e)\n OSError: [Errno 16] Please reduce your request rate.\n\nMessage:\n\n OSError: [Errno 16] Please reduce your request rate."} {"asctime": "2025-01-14 101641,671", "name": "flytekit", "levelname": "ERROR", "message": "!! End Error Captured by Flyte !!"} ERRORasyncioTask exception was never retrieved future: <Task finished name='Task-2307' coro=<_run_coros_in_chunks.<locals>._run_coro() done, defined at /opt/venv/lib/python3.10/site-packages/fsspec/asyn.py:243> exception=OSError(16, 'Please reduce your request rate.')> Traceback (most recent call last): File "/opt/venv/lib/python3.10/site-packages/s3fs/core.py", line 729, in _lsdir async for c in self._iterdir( File "/opt/venv/lib/python3.10/site-packages/s3fs/core.py", line 779, in _iterdir async for i in it: File "/opt/venv/lib/python3.10/site-packages/aiobotocore/paginate.py", line 30, in anext response = await self._make_request(current_kwargs) File "/opt/venv/lib/python3.10/site-packages/aiobotocore/client.py", line 412, in _make_api_call raise error_class(parsed_response, operation_name) botocore.exceptions.ClientError: An error occurred (SlowDown) when calling the ListObjectsV2 operation (reached max retries: 4): Please reduce your request rate. The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/opt/venv/lib/python3.10/site-packages/fsspec/asyn.py", line 245, in _run_coro return await asyncio.wait_for(coro, timeout=timeout), i File "/usr/local/lib/python3.10/asyncio/tasks.py", line 408, in wait_for return await fut File "/opt/venv/lib/python3.10/site-packages/flytekit/core/type_engine.py", line 1397, in async_to_literal lv = await transformer.async_to_literal(ctx, python_val, python_type, expected) File "/opt/venv/lib/python3.10/site-packages/flytekit/core/type_engine.py", line 2150, in async_to_literal await _run_coros_in_chunks([c for c in lit_map.values()]) File "/opt/venv/lib/python3.10/site-packages/fsspec/asyn.py", line 268, in _run_coros_in_chunks result, k = await done.pop() File "/opt/venv/lib/python3.10/site-packages/fsspec/asyn.py", line 245, in _run_coro return await asyncio.wait_for(coro, timeout=timeout), i File "/usr/local/lib/python3.10/asyncio/tasks.py", line 408, in wait_for return await fut File "/opt/venv/lib/python3.10/site-packages/flytekit/core/type_engine.py", line 1397, in async_to_literal lv = await transformer.async_to_literal(ctx, python_val, python_type, expected) File "/opt/venv/lib/python3.10/site-packages/flytekit/types/pickle/pickle.py", line 94, in async_to_literal remote_path = await FlytePickle.to_pickle(ctx, python_val) File "/opt/venv/lib/python3.10/site-packages/flytekit/types/pickle/pickle.py", line 50, in to_pickle return await ctx.file_access.async_put_raw_data(uri) File "/opt/venv/lib/python3.10/site-packages/flytekit/core/data_persistence.py", line 423, in async_put_raw_data r = await self._put(from_path, to_path, **kwargs) File "/opt/venv/lib/python3.10/site-packages/flytekit/core/data_persistence.py", line 357, in _put dst = await file_system._put(from_path, to_path, recursive=recursive, **kwargs) # pylint: disable=W0212 File "/opt/venv/lib/python3.10/site-packages/fsspec/asyn.py", line 565, in _put trailing_sep(rpath) or await self._isdir(rpath) File "/opt/venv/lib/python3.10/site-packages/s3fs/core.py", line 1498, in _isdir return bool(await self._lsdir(path)) File "/opt/venv/lib/python3.10/site-packages/s3fs/core.py", line 742, in _lsdir raise translate_boto_error(e) OSError: [Errno 16] Please reduce your request rate.

The error message indicates that your Flyte task is encountering a rate limit issue with AWS S3, specifically a "SlowDown" error when calling the

ListObjectsV2

operation. This suggests that the request rate to S3 is too high, and AWS is asking to reduce the request rate. To address this, consider implementing a rate limiter in your Flyte tasks to control the frequency of requests to S3. You can also review your S3 usage patterns to ensure that requests are distributed more evenly over time. Additionally, check if there are any configuration settings in Flyte or your AWS setup that can help manage the request rate. Would you like more details on how to implement a rate limiter or adjust your Flyte configuration? Sources: - GitHub Issue: Failed to launch workflows due to too many requests - GitHub Issue: webApi readRateLimiter settings are not respected 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325, sink events saved in s3 ?? admin_eventsink.go

It seems likely that you're asking about how Flyte handles event sinks, particularly in relation to S3. My understanding is that Flyte uses event sinks to record workflow events, and these can be configured to store data in various backends, including S3. The

admin_eventsink.go

file in Flyte's codebase is responsible for handling these events. However, the specific details of how events are saved to S3 or any other storage backend would depend on the configuration of your Flyte deployment. Could you clarify if you're looking for information on configuring Flyte to use S3 for event sinks, or are you interested in the internal workings of the

admin_eventsink.go

file? Sources: - https://discuss.flyte.org/t/16082768/hello-team-i-am-trying-to-pass-s3-client-object-to-a-functio/ 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 configuring Flyte to use S3 for event sinks

To configure Flyte to use S3 for event sinks, you need to set up the FlyteAdmin configuration to direct events to an S3 bucket. This involves specifying the S3 bucket details and ensuring that FlyteAdmin is configured to send events to this storage backend. 1. Configuration File: Update the FlyteAdmin configuration file to include the S3 bucket details. This typically involves setting the

external_events

configuration in the

values.yaml

file used by Flyte's Helm charts. 2. Enable External Events: Ensure that the

external_events

feature is enabled in your FlyteAdmin configuration. This can be done by setting the

enable

flag to

true

under the

external_events

key. 3. Specify S3 Details: Provide the necessary S3 details such as the bucket name, region, and any required authentication credentials. Would you like more detailed steps or examples on how to configure this? Sources: - https://discuss.flyte.org/t/9555705/hi-team-is-there-a-way-to-configure-the-s3-endpoint-url-we-r/ - https://discuss.flyte.org/t/2216158/hey-all-i-am-trying-to-access-aws-services-from-tasks-runnin/ 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 How many times flyte internal calls s3 put to save the map_task output in s3 buckets ?? and is there any limit/rate to control it ?

I'm not able to find any data sources to answer your question. You might want to specify the Flyte version or configuration details you're using. If this doesn't help, please reach out to the Flyte support team for further assistance. 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325, flyte saves task output to s3 ?

Yes, Flyte saves task outputs to S3. Flyte uses S3 as a storage backend to store intermediate data, task outputs, and other artifacts. When a task completes, its outputs are serialized and uploaded to the specified S3 bucket. This allows Flyte to manage data efficiently and ensure reproducibility across different task executions. Would you like more details on how this process works or how to configure it? Sources: - Flyte Data Management 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 more details on how this process works or how to configure it?

It seems likely that Flyte uses a pluggable data persistence layer to manage how task outputs are stored, including using S3 as a backend. Here's a breakdown of how this process typically works and how you can configure it: 1. Task Output Handling: When a Flyte task completes, its outputs are serialized into a format that can be stored in a blob storage system like S3. Flyte uses the FlyteFile abstraction to manage file-based outputs, which are automatically uploaded to the configured storage backend. 2. Configuration: To configure Flyte to use S3 for storing task outputs, you need to set up the storage configuration in your Flyte deployment. This involves specifying the S3 bucket details, access credentials, and any other necessary parameters in the Flyte configuration files. 3. FlyteKit and FlyteAdmin: FlyteKit, the Python SDK for Flyte, interacts with FlyteAdmin to register tasks and workflows. During execution, FlyteAdmin coordinates with FlytePropeller to manage data movement, including uploading task outputs to S3. 4. Data Movement: Flyte handles data movement between tasks by passing references to the data (e.g., S3 URLs) rather than the data itself. This ensures efficient data handling and reduces the need for data duplication. 5. Pluggable Storage: Flyte's storage layer is pluggable, allowing you to use different storage backends. By default, Flyte supports major blob stores like S3, and you can extend this by implementing custom storage plugins if needed. For more detailed configuration steps, you might want to refer to Flyte's documentation or specific configuration files like

flyteadmin_config.yaml

or

flytepropeller_config.yaml

. Would you like more specific examples or guidance on a particular aspect of this process? Sources: - FlyteKit Data Persistence 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325 specific examples or guidance on a particular aspect of this process?

To provide specific examples or guidance on configuring Flyte to use S3 for task outputs, let's focus on setting up the storage configuration in Flyte: 1. Flyte Configuration: You need to configure Flyte to use S3 by specifying the storage settings in the Flyte configuration files. This typically involves editing the

values.yaml

file if you're using Helm for deployment. 2. Storage Type and Bucket: Set the storage type to

s3

and specify the S3 bucket name where task outputs will be stored. You also need to provide the region and authentication details. Example configuration in `values.yaml`:

storage:
     type: s3
     bucketName: my-s3-bucket
     s3:
       region: us-east-1
       authType: iam  # or 'accesskey' if using access keys
       accessKey: "<your-access-key>"  # if using accesskey authType
       secretKey: "<your-secret-key>"  # if using accesskey authType

3. FlyteFile and FlyteDirectory: Use Flyte's abstractions like

FlyteFile

and

FlyteDirectory

in your tasks to handle file-based inputs and outputs. These abstractions automatically manage the upload and download of files to and from S3. Example task using `FlyteFile`:

from flytekit import task, FlyteFile

   @task
   def process_data(input_file: FlyteFile) -> FlyteFile:
       # Process the input file and generate an output file
       output_file_path = "processed_data.csv"
       # Save the processed data to the output file
       # Return the output file as a FlyteFile
       return FlyteFile(output_file_path)

4. Data Movement: Flyte automatically handles the movement of data between tasks by passing references (e.g., S3 URLs) to the data. This ensures efficient data handling and reduces the need for data duplication. If you need more detailed guidance on a specific part of this process, please let me know! Sources: - Flyte Data Management - FlyteKit Data Persistence 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325, right after creating a flyte project, if we run a pipeline it is failing wit below error. failed to create workflow in propeller namespaces "xxxxxxx" How to check whether we are ready to trigger the workflow/execution ?

The error message "failed to create workflow in propeller namespaces" suggests that the namespace required for the Flyte project and domain does not exist. This is a common issue when the necessary Kubernetes namespaces are not created after setting up a new Flyte project. To ensure that you are ready to trigger a workflow execution, follow these steps: 1. Check Namespace Creation: Verify that the namespace for your project and domain exists in your Kubernetes cluster. The default naming convention is

<project>-<domain>

. You can check this using:

kubectl get namespaces

2. Create Namespace Manually: If the namespace does not exist, you can create it manually:

kubectl create namespace <project>-<domain>

3. Sync Resources: Ensure that the Flyte cluster resource manager is running and configured to automatically create namespaces. If it's not running, you might need to manually trigger the resource synchronization:

kubectl create job --from=cronjob/syncresources <job-name> -n flyte

4. Check FlyteAdmin and FlytePropeller Logs: Look at the logs for FlyteAdmin and FlytePropeller to see if there are any errors or warnings related to namespace creation or resource management. 5. Configuration Check: Ensure that your Flyte deployment is configured to manage namespaces automatically. This is typically controlled by the cluster resource manager settings in your Flyte configuration. Would you like more detailed guidance on any of these steps? Sources: - Creating workflows in sandbox projects return 500 - Flyte Namespace and Project Configuration 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325, which module in flyte (i mean service) creates namespace as part of project creation request

The module responsible for creating namespaces as part of a project creation request in Flyte is the Cluster Resource Manager. This component interacts with FlyteAdmin to manage Kubernetes resources, including namespaces, based on the project and domain configurations. When a new project is created, the Cluster Resource Manager ensures that the necessary namespaces are provisioned in the Kubernetes cluster. This is typically configured to automatically create namespaces following the

<project>-<domain>

naming convention. Would you like more details on how to configure or troubleshoot the Cluster Resource Manager? Sources: - Flyte Namespace and Project Configuration - Cluster Resource Manager Documentation 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.

@ancient-wolf-19325, How do flyte executes aa_namespace.yaml where namespace creation is there

It seems likely that Flyte uses a component called the Cluster Resource Manager to handle the execution of resource templates like

aa_namespace.yaml

, which includes namespace creation. This component is responsible for synchronizing Kubernetes resources based on the configurations specified in Flyte's deployment setup. Here's how it typically works: 1. Configuration: The

aa_namespace.yaml

file would be part of the cluster resource templates configured in Flyte's deployment. This file specifies the namespace and other Kubernetes resources that need to be created. 2. Cluster Resource Manager: When a new project or domain is created, the Cluster Resource Manager reads the resource templates and applies them to the Kubernetes cluster. This includes creating namespaces as specified in the YAML files. 3. Automatic Synchronization: The Cluster Resource Manager is usually configured to automatically synchronize resources, ensuring that the necessary namespaces and other resources are created and updated as needed. 4. Manual Trigger: If automatic synchronization is not enabled, you might need to manually trigger the resource synchronization using a command like:

kubectl create job --from=cronjob/syncresources <job-name> -n flyte

Would you like more detailed guidance on configuring or troubleshooting this process? Sources: - Cluster Resource Manager Documentation - Flyte Namespace and Project Configuration 👍 0 button 👎 0 button 💡 Hint: Mention @runllm in the thread for followups.