Hi team, I've been successfully using Prefect with Kubernetes for a while now. But over the past two weeks, I've encountered intermittent errors during flow deployments. Sometimes everything works fine, and other times, all tasks fail to deploy. The error message typically looks like the one below. It seems to be an error with a connection refused to warden-mutating.common-webhooks.networking.gke.io. Could it be caused by an overload with too many tasks being scheduled? I’d love to hear your thoughts if you have any insights into what might be causing these errors and how to solve it

Failed to submit flow run '6bddeb25-71d8-4add-92de-41d200887583' to infrastructure.
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect_kubernetes/worker.py", line 632, in _create_job
    job = batch_client.create_namespaced_job(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api/batch_v1_api.py", line 210, in create_namespaced_job
    return self.create_namespaced_job_with_http_info(namespace, body, **kwargs)  # noqa: E501
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api/batch_v1_api.py", line 309, in create_namespaced_job_with_http_info
    return self.api_client.call_api(
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api_client.py", line 348, in call_api
    return self.__call_api(resource_path, method,
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api_client.py", line 180, in __call_api
    response_data = self.request(
                    ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/api_client.py", line 391, in request
    return <http://self.rest_client.POST|self.rest_client.POST>(url,
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/rest.py", line 276, in POST
    return self.request("POST", url,
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/kubernetes/client/rest.py", line 235, in request
    raise ApiException(http_resp=r)
kubernetes.client.exceptions.ApiException: (500)
Reason: Internal Server Error
HTTP response headers: HTTPHeaderDict({'Audit-Id': 'b3aa49b8-60fd-4da9-b6c5-55d47dbc1f6d', 'Cache-Control': 'no-cache, private', 'Content-Type': 'application/json', 'Warning': '299 - "unknown field \\"spec.template.spec.completions\\"", 299 - "unknown field \\"spec.template.spec.parallelism\\""', 'X-Kubernetes-Pf-Flowschema-Uid': '312d4f4a-edd6-4ea6-aee4-cb1ea8654799', 'X-Kubernetes-Pf-Prioritylevel-Uid': '3f4f1eb8-f614-433c-af7b-3844f105a5a8', 'Date': 'Thu, 11 Jul 2024 15:00:23 GMT', 'Content-Length': '619'})
HTTP response body: {"kind":"Status","apiVersion":"v1","metadata":{},"status":"Failure","message":"Internal error occurred: failed calling webhook \"<http://warden-mutating.common-webhooks.networking.gke.io|warden-mutating.common-webhooks.networking.gke.io>\": failed to call webhook: Post \"<https://localhost:5443/webhook/warden-mutating?timeout=10s>\": dial tcp [::1]:5443: connect: connection refused","reason":"InternalError","details":{"causes":[{"message":"failed calling webhook \"<http://warden-mutating.common-webhooks.networking.gke.io|warden-mutating.common-webhooks.networking.gke.io>\": failed to call webhook: Post \"<https://localhost:5443/webhook/warden-mutating?timeout=10s>\": dial tcp [::1]:5443: connect: connection refused"}]},"code":500}


During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/workers/base.py", line 834, in _submit_run_and_capture_errors
    result = await self.run(
             ^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect_kubernetes/worker.py", line 510, in run
    job = await run_sync_in_worker_thread(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/utilities/asyncutils.py", line 91, in run_sync_in_worker_thread
    return await anyio.to_thread.run_sync(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/anyio/to_thread.py", line 33, in run_sync
    return await get_asynclib().run_sync_in_worker_thread(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 877, in run_sync_in_worker_thread
    return await future
           ^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/anyio/_backends/_asyncio.py", line 807, in run
    result = context.run(func, *args)
             ^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect_kubernetes/worker.py", line 641, in _create_job
    message += ": " + exc.body["message"]
                      ~~~~~~~~^^^^^^^^^^^
TypeError: string indices must be integers, not 'str'

@Marvin how can I query worker or work-pool health status with an API call to the server or via the prefect-cli?

@Marvin if there a way that I can use:

import logging

logger = logging.getLogger(__name__)

instead of

from prefect import  get_run_logger
logger = get_run_logger()

I want to write my code in standard way and not to be coupled with prefect, for example when I am writing libraries

@Marvin what does this mean? Task parameter introspection took 406.297 seconds , exceeding

PREFECT_TASK_INTROSPECTION_WARN_THRESHOLD

of 10.0. Try wrapping large task parameters with

prefect.utilities.annotations.quote

for increased performance, e.g.

my_task(quote(param))

. To disable this message set

PREFECT_TASK_INTROSPECTION_WARN_THRESHOLD=0

.

Hi, I have been working on a what I expect is a memory leak and have been led to use memory_profiler. The trouble I'm having is that the output of memory_profiler does not go to the Prefect logger. I have tried to run the deployment from the command line, which I can do, but I don't get any console output, only what is shown below: How can I force the output to show when I run the deployment from the CLI?

@Marvin how do I trigger a deployment in a python file

Hi All, I get this error when running say large amount of flows at a time. I am using ECS task to run the flow from docker. Error: when calling the RegisterTaskDefinition operation: Too many concurrent attempts to create a new revision of the specified family. I have not passed Task Definition, so letting Prefect Generate it. However I see prefect always generates/registers new type definition with no changes except incremeting task definition version and registered at, by changes I already have these envirnment variables on Worker Pool / Agent

'AWS_RETRY_MODE': 'adaptive',
  'AWS_MAX_ATTEMPTS': '100'

How to ensure Prefect does NOT generate new task definition for each flow?

I am wondering if there is a pattern that exists in prefect 2.0 to say that a Flow had nothing to do. My use case is that I have a Flow checking every 5min for files in the s3 bucket of a partner and when they are files I process them with a series of tasks. The dashboard is showing a lot of successful tasks. Would using a subflow be a good pattern there? One flow "checking" and one flow "processing" ? Maybe having a sub flow status "success nothing to do" would work. I also have another use case, in the same flow I am processing those files and 0.1% of the records are not "valid" I am logging them as warnings but wondering if there is a way to report on those "invalid" records, seems to be a pretty common use case

is there a way to customize slack notifications via prefect? for example, if i wanted to tag people on slack in the alert message based on which flow is running. can i pass in a custom message/alert?

Hi - we have the “mono-repo” layout - we intend to keep all our flows in a package and leverage shared stuff in a common package. We are able to register flows with our AWS hosted instance, but only a very simple “stand-alone” flow executes, we cannot get ones that reference common code to run, details follow in thread… Any help much appreciated!