Hi, can someone please help with this? Thanks.

Actually looking at this again, I do not think this is k8s doing this at all.. it seems prefect is doing graceful shutdown (SIGTERM) due to an exception in cancelling the flow run..

I suspect these are not full logs. It would be interesting to see why the cancellation was emitted in the first place.

Anyhow.. from what I can tell in the logs is that

prefect.exceptions.TerminationSignal

a SIGTERM was received.. possibly from k8s, killing the pod (please do check the pod metadata.. i.e. is the status

terminated

[if so, what is the reason] or

failed

) When the graceful shutdown was init, the code threw another exception

AssertionError

I do not know what this is..

I can reproduce this daily btw - long running flow, 3k tasks - randomly fails between 10 mins -> 2 hours and everything in between

no other pods on the node or in the cluster get terminated around the same time

has to be something with the prefect event loop

@jpuris I may have something - just got another flow to fail but the flow & agent logs are interesting, the flow appears to start again - notice the downloading flow code

the agent meanwhile reported 2023-08-04T081512.362509340Z 081338.857 | INFO | Flow run 'meek-limpet' - Created task run 'update_rooftop_metrics-3407' for task 'update_rooftop_metrics' 2023-08-04T081512.362511460Z 081338.859 | INFO | Flow run 'meek-limpet' - Executing 'update_rooftop_metrics-3407' immediately... 2023-08-04T081512.362513900Z rpc error: code = NotFound desc = an error occurred when try to find container "f782a10e7abe00b4484840abd243bd0eabdcfc5dcf0e7a242812035ffc31c4e1": not found Fri, Aug 4 2023 91512 am +pip install s3fs==2023.1.0 pydantic==1.10.6 datadog python-dotenv r7insight_python python-dotenv prefect-kv r7insight_python datadog psycopg2-binary clickhouse-connect

I assume the rpc error is k8s logging but cant be 100% - I cannot find that container id anywhere

I have narrowed this down - I have created a dedicated agent in a container in the same pod for this flow and it has run twice with no issues

@Ankit maybe something for you to try

found one that does

basically iterate a dataframe and call an api for each row

as simple as it gets

or so you would hope

This error happens to me from time to time too. I have a flow that launches many other subflows. Some of these subflows are falling. It is noteworthy that they always crash if run together, but if run separately, no errors occur. Most often, the flow with this code falls. I'll share the code in the thread, maybe it will help in finding the error.

@task()
def select(sql_path: str, params: BigQueryParamsMixin) -> pd.DataFrame:
    query = render(sql_path, params=params.render_params)
    job_config = bq.QueryJobConfig(query_parameters=params.bq_params)

    return bq_client.query(query, location='EU', job_config=job_config).to_dataframe()


def _split(params: BigQueryParamsMixin, chunks_size: int = 150_000):
    input_data = getattr(params, 'input_data', pd.DataFrame())
    chunks_number = len(input_data.index) // chunks_size

    if chunks_number >= 1:
        for chunk in np.array_split(input_data, chunks_number):
            params_copy = deepcopy(params)
            # the task which saves the result in GCS
            input_data_path = save_result(chunk) 
            setattr(params_copy, 'input_data_path', input_data_path)
            yield params_copy
    else:
        yield params


@flow(task_runner=SequentialTaskRunner())
def select_metric(sql: str, params: BigQueryParamsMixin, from_file: bool = True) -> pd.DataFrame:
    df = pd.DataFrame()

    for params in _split(params):
        #  BigQuery has limits on the size of a query. 
        #  Therefore, it is worth splitting the input 
        #  into chunks if it's needed.
        df_2 = select(sql, params)
        df = pd.concat([df, df_2])

    return df

And also traceback which is displayed when the error occurs:

Crash details:
Traceback (most recent call last):
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 1884, in capture_sigterm
    yield
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/api.py", line 243, in wait_for_call_in_loop_thread
    return call.result()
           ^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 283, in result
    return self.future.result(timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 169, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 346, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/client/utilities.py", line 51, in with_injected_client
    return await fn(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 681, in create_and_begin_subflow_run
    terminal_state = await orchestrate_flow_run(
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 830, in orchestrate_flow_run
    result = await flow_call.aresult()
             ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 292, in aresult
    return await asyncio.wrap_future(self.future)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 316, in _run_sync
    result = self.fn(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/trebufect/flows/select_metric.py", line 53, in select_metric
    df_2 = select(sql, params)
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/tasks.py", line 730, in submit
    return enter_task_run_engine(
           ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 1145, in enter_task_run_engine
    return from_sync.wait_for_call_in_loop_thread(begin_run)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/api.py", line 242, in wait_for_call_in_loop_thread
    waiter.wait()
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/waiters.py", line 152, in wait
    self._handle_waiting_callbacks()
  File "/usr/local/lib/python3.11/site-packages/prefect/_internal/concurrency/waiters.py", line 119, in _handle_waiting_callbacks
    callback: Call = self._queue.get()
                     ^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/queue.py", line 171, in get
    self.not_empty.wait()
  File "/usr/local/lib/python3.11/threading.py", line 320, in wait
    waiter.acquire()
  File "/usr/local/lib/python3.11/site-packages/prefect/engine.py", line 1874, in cancel_flow_run
    raise TerminationSignal(signal=signal.SIGTERM)
prefect.exceptions.TerminationSignal

Prefect Version:

2.11.3

Python Version:

3.11

I have one default agent. Each subflow runs its own Kubernetes Job.

Could someone take a look and help us please? 🙏

@Yuriy Trofimentsev I have fixed my issue btw - I changed my prefect tasks that query the db to just normal functions

wasted 4 days trying to work it out

@Yuriy Trofimentsev are yours long lasting jobs?