def ssm_command()

in api/PclusterApiHandler.py [0:0]

• 37 lines of code
• 7 McCabe index (conditional complexity)

def ssm_command(region, instance_id, user, run_command):
    # working_directory |= f"/home/{user}"
    start = time.time()

    if region:
        config = botocore.config.Config(region_name=region)
        ssm = boto3.client("ssm", config=config)
    else:
        ssm = boto3.client("ssm")

    command = f"runuser -l {user} -c '{run_command}'"

    ssm_resp = ssm.send_command(
        InstanceIds=[instance_id],
        DocumentName="AWS-RunShellScript",
        Comment=f"Run ssm command.",
        Parameters={"commands": [command]},
        CloudWatchOutputConfig={
            'CloudWatchLogGroupName': SSM_LOG_GROUP_NAME,
            'CloudWatchOutputEnabled': True
        },
    )

    command_id = ssm_resp["Command"]["CommandId"]

    logger.info(f"Submitted SSM command {command_id}")

    # Wait for command to complete
    time.sleep(0.75)
    while time.time() - start < 60:
        status = ssm.get_command_invocation(CommandId=command_id, InstanceId=instance_id)
        if status["Status"] != "InProgress":
            break
        time.sleep(0.75)

    if time.time() - start > 60:
        raise Exception("Timed out waiting for command to complete.")

    if status["Status"] != "Success":
        raise Exception(status["StandardErrorContent"])

    output = read_and_delete_ssm_output_from_cloudwatch(
        region=region,
        log_group_name=SSM_LOG_GROUP_NAME,
        command_id=command_id,
        instance_id=instance_id,
    )

    return output