Issue creating/calling Globus app multiple times

[Anthony Weaver]

I'm working on refining some Python code used for Globus compute to extract and ingest metadata.  Where I'm running into an issue is trying to perform multiple ingests in a loop.  If I ingest all my metadata at once it works fine.  I'm trying to break up the ingestion because there are some data sets that produce a very large number (10k-20k) metadata entries that I think would be better ingested in smaller chunks.

My current code (just the relevant portion) is as follows and it seems to ingest the first chunk of metadata but not the remaining chunks (currently I'm breaking into 4 chunks).  Any guidance on where I'm going wrong would be very much appreciated.

    # Info for interacting with the metadata server
    CLIENT_ID = os.environ.get("GLOBUS_COMPUTE_CLIENT_ID")
    SECRET = os.environ.get("GLOBUS_COMPUTE_CLIENT_SECRET")
    INDEX_ID = "XXXXXXXXXXXXXXXXXXXXXXXXXXXX"
   
    # There can be a very large number of entries so process in chunks
    ingest_chunks = split_entries_into_chunks(entries)
    for chunk in ingest_chunks:
        # Convert to JSON
        tmp_ingest = prettyprint_json({"ingest_type": "GMetaList", "ingest_data": {"gmeta": chunk}})
        ingest_data = json.loads(tmp_ingest)

        # Create the Globus App
        APP = globus_sdk.ClientApp("ingest-app", client_id=CLIENT_ID, client_secret=SECRET)
        SEARCH_CLIENT = globus_sdk.SearchClient(
                        app=APP,
                        app_scopes=[globus_sdk.Scope(globus_sdk.SearchClient.scopes.all)])

        ingest_res = SEARCH_CLIENT.ingest(INDEX_ID, ingest_data)
        task_id = ingest_res["task_id"]
        waited = 0
        max_wait = 300 # 5 minutes
        while True:
            res = SEARCH_CLIENT.get_task(task_id)
            if res["state"] == "SUCCESS":
                continue
            if res["state"] == "FAILED":
                raise ValueError(f"Ingest failed")
       
            # wait 1s and check for timeout
            waited += 1
            if waited >= max_wait:
                raise ValueError(f"Timeout occurred ingesting data")
       
            time.sleep(1)

[Stephen Rosen]

Hi Anthony,

I think I see the issue, and it's not related to the way you're constructing the app or client -- it's a problem with your task polling loop (the inner, `while True` loop).

I also have some follow-up notes on some things you can tidy up but which aren't as important to correctness.

Your loop reads:

    while True:

        res = SEARCH_CLIENT.get_task(task_id)

        if res["state"] == "SUCCESS":

            continue

        if res["state"] == "FAILED":

            raise ValueError(f"Ingest failed")

        # wait 1s and check for timeout

        waited += 1

        if waited >= max_wait:

            raise ValueError(f"Timeout occurred ingesting data")

        time.sleep(1)

The only way this loop can exit is with an error. One trivial fix is to replace `continue` with `break`.

However, I think we can make some other improvements. For example, counting 300 1 second sleeps is not 5 minutes -- given that `get_task()` calls take time, it's probably a good amount more! We can avoid that issue by measuring a time delta, instead of counting steps. I also think it will be less error-prone to make the loop condition express our actual stop criteria. Altogether, I'd express it like so:

    # poll in a "do-while" loop with a 5 minute time limit, until the task completes

    start_time = time.monotonic()

    res = SEARCH_CLIENT.get_task(task_id)

    status = res["state"]

    while (time.monotonic() - start_time) < 300 and status not in ("SUCCESS", "FAILED"):

        time.sleep(1)

        res = SEARCH_CLIENT.get_task(task_id)

        status = res["state"]

    if status == "FAILED":

        raise ValueError(f"Ingest failed for task '{task_id}'")

    # if the state was not failure (above) or success, something went wrong

    # so raise an error

    if res["state"] != "SUCCESS":

        raise ValueError(

            f"Ingest did not complete within time limit for task '{task_id}'"

        )

I would also be concerned that something is not correct in your error handling in the broader context where this code is called. Because the current loop doesn't exit on success, I expect that it is waiting 5+ minutes and then raising an exception. You may want to investigate to see where and how the exception is being handled (or not).

There are two other minor improvements I see, which I think will make your code more efficient but aren't essential:

1. Are you sure that you want to `prettyprint_json` and then `json.loads()` the result? That looks a little unusual. If possible, I would avoid converting the data to and from JSON.

2. Do you need to create a new client and a new app object with each run of the loop? In most scenarios, one app object and one client could be reused across loop iterations. Both of these objects have components that they will reuse across multiple requests, so reusing them makes them more efficient.

I'm pretty sure that fixing the loop will fix the main problem though, so let's start with that!

Let us know if it still doesn't work afterwards.

Cheers,

-Stephen

-- 
Written by a human.

[Anthony Weaver]

Stephen,

Thank you and I apologize.  I have no idea what I was thinking when I put a continue in that while loop.  That was a bonehead mistake.

Some of the other code there was there as a way to try and get the chunk processing working.  For example, I was recreating the ingest app

thinking that perhaps once the ingest finish, the app essentially "died" (for lack of a better term) and that I needed to 

reinstantiate it. I readjusted that loop and it's fine now

Tony

To unsubscribe from this group and stop receiving emails from it, send an email to discuss+u...@globus.org.