Boto3 client in multiprocessing pool fails with "botocore.exceptions.NoCredentialsError: Unable to locate credentials"

user_10025504 0
Viewed 1306

I'm using boto3 to connect to s3, download objects and do some processing. I'm using a multiprocessing pool to do the above.

Following is a synopsis of the code I'm using:

session = None

def set_global_session():
    global session
    if not session:
        session = boto3.Session(region_name='us-east-1')

def function_to_be_sent_to_mp_pool():
    s3 = session.client('s3', region_name='us-east-1')
    list_of_b_n_o = list_of_buckets_and_objects
    for bucket, object in list_of_b_n_o:
        content = s3.get_object(Bucket=bucket, Key=key)
        data = json.loads(content['Body'].read().decode('utf-8'))
        write_processed_data_to_a_location()

def main():
    pool = mp.Pool(initializer=set_global_session, processes=40)
    pool.starmap(function_to_be_sent_to_mp_pool, list_of_b_n_o_i)

Now, when processes=40, everything works good. When processes = 64, still good.

However, when I increases to processes=128, I get the following error:

botocore.exceptions.NoCredentialsError: Unable to locate credentials

Our machine has the required IAM roles for accessing S3. Moreover, the weird thing that happens is that for some processes, it works fine, whereas for some others, it throws the credentials error. Why is this happening, and how to resolve this?

Another weird thing that happens is that I'm able to trigger two jobs in 2 separate terminal tabs (each of which has a separate ssh login shell to the machine). Each job spawns 64 processes, and that works fine as well, which means there are 128 processes running simultaneously. But 80 processes in one login shell fails.

Follow up:

I tried creating separate sessions for separate processes in one approach. In the other, I directly created s3-client using boto3.client. However, both of them throw the same error with 80 processes.

I also created separate clients with the following extra config:

Config(retries=dict(max_attempts=40), max_pool_connections=800)

This allowed me to use 80 processes at once, but anything > 80 fails with the same error.

Post follow up:

Can someone confirm if they've been able to use boto3 in multiprocessing with 128 processes?

boto3 python-3.x multiprocessing
3 Answers

This is actually a race condition on fetching the credentials. I'm not sure how fetching credentials under the hood works, but the I saw this question in Stack Overflow and this ticket in github.

I was able to resolve this by keeping a random wait time for each of the processes. The following is the updated code which works for me:

client_config = Config(retries=dict(max_attempts=400), max_pool_connections=800)
time.sleep(random.randint(0, num_processes*10)/1000) # random sleep time in milliseconds
s3 = boto3.client('s3', region_name='us-east-1', config=client_config)

I tried keeping the range for sleep time lesser than num_processes*10, but that failed again with the same issue.

@DenisDmitriev, since you are getting the credentials and storing them explicitly, I think that resolves the race condition and hence the issue is resolved.

PS: values for max_attempts and max_pool_connections don't have a logic. I was plugging several values until the race condition was figured out.

user_100255040

I suspect that AWS recently reduced throttling limits for metadata requests because I suddenly started running into the same issue. The solution that appears to work is to query credentials once before creating the pool and have the processes in the pool use them explicitly instead of making them query credentials again.

I am using fsspec with s3fs, and here's what my code for this looks like:

def get_aws_credentials():
    '''
    Retrieve current AWS credentials.
    '''
    import asyncio, s3fs
    fs = s3fs.S3FileSystem()

    # Try getting credentials
    num_attempts = 5
    for attempt in range(num_attempts):
        credentials = asyncio.run(fs.session.get_credentials())
        if credentials is not None:
            if attempt > 0:
                log.info('received credentials on attempt %s', 1 + attempt)
            return asyncio.run(credentials.get_frozen_credentials())

        time.sleep(15 * (random.random() + 0.5))

    raise RuntimeError('failed to request AWS credentials '
                       'after %d attempts' % num_attempts)


def process_parallel(fn_d, max_processes):
    # [...]
    c = get_aws_credentials()

    # Cache credentials
    import fsspec.config
    prev_s3_cfg = fsspec.config.conf.get('s3', {})
    try:
        fsspec.config.conf['s3'] = dict(prev_s3_cfg,
                                        key=c.access_key,
                                        secret=c.secret_key)

        num_processes = min(len(fn_d), max_processes)

        from concurrent.futures import ProcessPoolExecutor
        with ProcessPoolExecutor(max_workers=num_processes) as pool:
            for data in pool.map(process_file, fn_d, chunksize=10):
                yield data
    finally:
        fsspec.config.conf['s3'] = prev_s3_cfg

Raw boto3 code will look essentially the same, except instead of the whole fs.session and asyncio.run() song and dance, you'll work with boto3.Session itself and call its get_credentials() and get_frozen_credentials() methods directly.

user_27315160

I get the same problem with multi process situation. I guess there is a client init problem when you use multi process. So I suggest that you can use get function to get s3 client. It works for me.

g_s3_cli = None

def get_s3_client(refresh=False):
    global g_s3_cli
    if not g_s3_cli or refresh:
        g_s3_cli = boto3.client('s3')
    return g_s3_cli
user_23257780
Related