Orchestrations stuck in running state

[multipleofzero]

Description

I am using Fan-out/fan-in to do a two step calculation. The first step, "CalculateRouteProfile",
calls a third party API that returns data used in the second step "CalculateCo2FromRouteProfile".

The calculation is triggered by an HTTPStarter.

When running this for about 15'000 inputs, some of the calculations never seem to finish (most
of them do, though). The ones that do not finsh are stuck in the "Running" state.

Example of a finished calculation: c553159a6c994173bf6c4381376954dc. Examples of not finished
calculations follow below.

Expected behavior

All ~15'000 inputs are processed.

Actual behavior

Out of 754 orchestrators all but 14 eventually complete. The remaining 14 stay in "Running" state.

Relevant source code snippets

import logging
import azure.durable_functions as df
from azure.durable_functions.models.RetryOptions import RetryOptions


MAX_ROUTES_PER_POST_REQUEST = 100
FAN_OUT_CHUNK_SIZE = 25


def chunks(lst, n):
    """Yield successive n-sized chunks from list."""
    for i in range(0, len(lst), n):
        yield lst[i:i + n]


def orchestrator_function(context: df.DurableOrchestrationContext):
    """ """
    list_of_route_params = context.get_input()
    logging.info(
        f"client input in orchestrator function: {len(list_of_route_params)} items."
    )
    if len(list_of_route_params) > 0:
        logging.info(
            f"first item: {list_of_route_params[0]}."
        )
    custom_status = {
        "msg": "just started",
        "total_number_of_routes": len(list_of_route_params),
        "current_chunk": None,
        "routes_in_current_chunk": None
    }
    context.set_custom_status(custom_status)

    out = []  # this will contain our final output
    # We need to avoid that one call immediately try to calculate thousands of
    # routes without limits.  So we split the total list of input routes into
    # chunks, with FAN_OUT_CHUNK_SIZE as the limit per chunk.
    chunk_list = list(chunks(list_of_route_params, FAN_OUT_CHUNK_SIZE))
    logging.info(
        (f"I have {len(chunk_list)} chunks  with chunk size {FAN_OUT_CHUNK_SIZE} "
         f"for calculating {len(list_of_route_params)} route items")
    )
    custom_status["number_of_chunks"] = len(chunk_list)
    context.set_custom_status(custom_status)
    for idx, chunks_of_route_params in enumerate(chunk_list):
        # this is a chained "fan-in-fan-out" call: first get the profiles,
        # then calculate the profile and the emissions

        custom_status["msg"] = "processing route profiles"
        custom_status["routes_in_current_chunk"] = len(chunks_of_route_params)
        custom_status["current_chunk"] = idx
        context.set_custom_status(custom_status)

        parallel_tasks = []
        retry_options = RetryOptions(1000, 3)
        for route_params in chunks_of_route_params:
            # first calculate the routes including the profile
            parallel_tasks.append(
                context.call_activity_with_retry("CalculateRouteProfile",
                                                 retry_options, route_params)
            )
        chunk_profiles = yield context.task_all(parallel_tasks)

        custom_status["msg"] = "processing from route profiles"
        context.set_custom_status(custom_status)
        parallel_tasks = []
        for profile in chunk_profiles:
            # calculate emission based on the profile
            parallel_tasks.append(
                context.call_activity(
                    "CalculateCo2FromRouteProfile",
                    {
                        "profile": profile["serialized_profile"],
                        "route_params": profile["route_params"]
                    },
                )
            )
        out += yield context.task_all(parallel_tasks)

    custom_status["msg"] = "done with processing"
    custom_status["routes_in_current_chunk"] = len(chunk_list[-1])
    custom_status["current_chunk"] = len(chunk_list)
    context.set_custom_status(custom_status)

    return out


main = df.Orchestrator.create(orchestrator_function)

host.json:

{
    "version": "2.0",
    "logging": {
        "applicationInsights": {
            "samplingSettings": {
                "isEnabled": true,
                "excludedTypes": "Request"
            }
        }
    },
    "extensionBundle": {
        "id": "Microsoft.Azure.Functions.ExtensionBundle",
        "version": "[2.*, 3.0.0)"
    },
    "extensions": {
        "durableTask": {
            "maxConcurrentActivityFunctions": 3,
            "maxConcurrentOrchestratorFunctions": 1
        }
    }
}

Known workarounds

Using small batches of input (around 100 routes) seems to work reliably.

App Details

• Durable Functions extension version (e.g. v1.8.3): 1.3.0 (just called "Azure Functions" in VS Code)
• Azure Functions runtime version (1.0 or 2.0): 2.0
• Programming language used: Python

If deployed to Azure

• Timeframe issue observed: "createdTime": "2021-03-31T15:29:19Z"
• Function App name: co2-route-calculator-dev
• Function name(s): CalculateCo2FromRouteProfile, CalculateRouteProfile
• Azure region: Switzerland North
• Orchestration instance ID(s): a9ee8c72713644619b3e1d1eaeed7811, 1d986a195d1f4708be368e86c9e5cd06, adacc094cf934a68a5ae45f4562ec961, ...
• Azure storage account name: co2routecalculatordev

[pcopley]

I am also seeing this; for example, instance ID 64877078bdf54fde97cc8dea383d9a78. I have an orchestrator that saves a single record to CosmosDB (Mongo API) then publishes a corresponding Service Bus message.

[ConnorMcMahon]

@davidmrdavid, it's a bit hard to tell, but it appears that a failure in the fan-out call_actvity_with_retry created something weird with the Python implementation. We may want to see if we can repro this ourselves.

@pcopley, each of these stuck orchestration bugs can have fairly independent root causes, so I would recommend opening a separate ticket for your issue.

[ConnorMcMahon]

@multipleofzero

Just a couple of questions:

1. For the ~750 that didn't run into this isssue, did any of them have one of the initial call_activity_with_retry("CalculateRouteProfile") calls fail? If so, that orchestration instance id would be a great comparison point.
2. Would it be possible to get a reproduction of this issue where you compare the size of chunk_list and chunk_profiles? For some reason, we don't see as many CalculateCo2FromRouteProfile executions scheduled as CalculateRouteProfile calls.

Finally, as some unrelated advice, I am noticing that your activities are returning incredibly large outputs (and passing them as inputs to subsequent orchestrations). Some of these are on the order of MB. It is worth noting that when executing an orchestration function, we have to load the entire history into memory at once, which includes all orchestration, suborchestration, and activity inputs/outputs. If you have fan-out/fan-in operations with 100+ MB worth of history, this could easily cause you to run into the 1.5 GB limit of consumption function apps.

In the meantime, I am going to transfer this to a separate repo, as this bug seems specific to the Durable Python implementation.

[ConnorMcMahon]

I realize my previous instructions regarding potential memory issues didn't cover the workaround. Instead of passing huge inputs directly as activity inputs/outputs, you can have your activity store the processed data in a blob and then pass around a reference to that blob. This has huge benefits from both a perf standpoint and a storage IO cost perspective, as you only fetch the data when you actually need it.

The current downside is that it involves boiler plate code. We have an upcoming feature that will make this substantially easier.

[scherand]

I just realised that I used a different Github account for this and the following comment, please excuse!

1. For the ~750 that didn't run into this isssue, did any of them have one of the initial `call_activity_with_retry("CalculateRouteProfile")` calls fail? If so, that orchestration instance id would be a great comparison point.

I will try to figure this out and will update this issue with my findings.

2. Would it be possible to get a reproduction of this issue where you compare the size of `chunk_list` and `chunk_profiles`? For some reason, we don't see as many `CalculateCo2FromRouteProfile` executions scheduled as `CalculateRouteProfile` calls.

I will see what I can come up with; and again, update this issue with my findings.

Finally, as some unrelated advice, I am noticing that your activities are returning incredibly large outputs (and passing them as inputs to subsequent orchestrations). Some of these are on the order of MB. It is worth noting that when executing an orchestration function, we have to load the entire history into memory at once, which includes all orchestration, suborchestration, and activity inputs/outputs. If you have fan-out/fan-in operations with 100+ MB worth of history, this could easily cause you to run into the 1.5 GB limit of consumption function apps.

Thanks very much for the hint (as well as the workaround you provided later!). I think I understand where the problem lies. We did see "Python exited with code 137" errors some time (but I did not notice any in the execution under investigation here).

[scherand]

Unfortunately I do not seem to be able to collect the information you asked for in 1. I cannot find the cases where CalculateRouteProfile failed.

What I did is use the Monitoring menu of the CalculateRouteProfile function (cf 01_calc_profile_monitor.png) where 124 Errors are reported in the last 30 days. They do not show up in the list below, presumably because the list shows the most recent traces (?) and the errors are older.

I ran the query in Application Insights (cf 02_app_insight_success_true.png). Still only "Succes == True" traces show up, which is more or less expected.

I then changed the filter on the Success column to "not True" (cf 03_success_not_true.png).

Confusingly that does not produce any result at all (cf 04_successs_not_true_no_result.png). This surprises me because the Monitor reported Errors in the last 30 days (which is also the search time frame in App Insights).

What is the best/most straight forward way to find the answer to your first question?

[multipleofzero]

I have switched to using blob storage and references to blobs to pass the results from one activity function to the next now. As you assumed, this alone does not solve the issue though.

I could not find any orchestration instance that fulfills your request in 1., i.e. that did succeed but had a failed call to CalculateRouteProfile. And I see the difference in number of calls to CalculateRouteProfile vs CalculateCo2FromRouteProfile!

Currently it seems to me as if the _with_retry part in call_activity_with_retry is causing the issue. When looking at the history table of an orchestration instance, it seems that the triggers that are scheduled for the retry do not (properly) fire? Because I suspected the 1 second backoff time might be too short for the system to properly register the timer, I changed the wait time before the retry to 10 seconds (which is correctly reflected in the difference between Timestamp and FireAt in the screenshot below).

I am not sure about the Timestamp vs _Timestamp column though. This screenshot is from Azure Storage Explorer. (Note: this orchestration instance ran in a different storage account: co2routecalculatortest)

My next test will be a run without retry. I will report back as soon as I am done with the test.

[multipleofzero]

I now have an orchestration id of a run where there does not seem to be any error in CalculateRouteProfile but still it does not terminate: 9c7a8fceb0dd4ddf9f94119cc4262184. This is from a run without _with_retry which seems to make the "perpetual orchestrations" (almost) go away (the one I have listed is the only one in 1000+). However, I have to deal with failed orchestrations now (rewinding them).

I also have two orchestrations that seem to fail right after they have been rewound: 3a453dc5bcb24ca388a2f1bac253aa4e and 647205c287ee4a1a98c2468c9ae0eb0d.

These are all in the storage account named co2routecalculatortest.

[davidmrdavid]

Hi @multipleofzero

Thank you for the detailed information, this helps us a lot. I want to take a step back and ask about this sentence you made two comments ago: "And I see the difference in number of calls to CalculateRouteProfile vs CalculateCo2FromRouteProfile!"

When you say that, do you mean that you see the difference in orchestration history log, or that you see the difference via a log statement that counts the length of elements in chunk_profiles versus chunk_list? Having the latter would be ideal for us and it would help pinpoint if this issue is occurring at the level of the Python SDK, or deeper in the framework.

cc @ConnorMcMahon.

[multipleofzero]

@davidmrdavid

ask about this sentence you made two comments ago: "And I see the difference in number of calls to CalculateRouteProfile vs CalculateCo2FromRouteProfile!"
When you say that, do you mean that you see the difference in orchestration history log, or that you see the difference via a log statement that counts the length of elements in chunk_profiles versus chunk_list?

Unfortunately I was referring to the first: I see a difference in the orchestration history log table. Somehow I am not able to find the logs for my orchestration instances (other than when live streaming?). There is a log statement that should show the counts; if only I was able to get at the logs...

[davidmrdavid]

@multipleofzero, we should be able to help with that. Do you mind showing us how you're logging and how you're trying to retrieve those logs? and does your application have Application Insights enabled?

[scherand]

@davidmrdavid

we should be able to help with that

Awesome!
And yes, Application Insights is (are :P) enabled.

I have tried two approaches, one via logging.info() and one via setting custom_status. For example, there is a custom state set when the orchestrator_function starts:

if not context.is_replaying:
    msg = (f"orchestrator has {len(chunk_list)} chunks  with chunk size {FAN_OUT_CHUNK_SIZE} "
           f"for calculating {len(list_of_route_params)} route items")
    logging.info(msg)

Then, before calling CalculateRouteProfile a custom status is set:

custom_status["msg"] = "processing route profiles"
custom_status["routes_in_current_chunk"] = len(chunks_of_route_params)
custom_status["current_chunk"] = idx
context.set_custom_status(custom_status)

And before calling CalculateCo2FromRouteProfile the custom status is updated:

custom_status["msg"] = f"processing co2 from {len(chunk_profiles)} route profiles"
context.set_custom_status(custom_status)

And I have a new Activity Function to clean up the referenced input data from the blob, CalculationCleanup:

custom_status["msg"] = (f"cleaning up after co2 calculation of {len(chunk_profiles)} "
                        "route profiles")
context.set_custom_status(custom_status)

This should all be visible for the orchestration 9c7a8fceb0dd4ddf9f94119cc4262184 in the storage account named co2routecalculatortest.

Note that I will be away for a week nowish, but I will come back to this when I am back at the latest.