From a10c8409cd6989aa0b2de0202ab76bdec5ef6dbe Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Tue, 23 Mar 2021 11:29:44 -0500 Subject: [PATCH 01/69] First workflow to deploy layer --- .github/workflows/deploy_lambdas.yml | 49 ++++++++++++++++++++++++++++ aws/requirements.txt | 4 +++ 2 files changed, 53 insertions(+) create mode 100644 .github/workflows/deploy_lambdas.yml create mode 100644 aws/requirements.txt diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml new file mode 100644 index 0000000..17c1aa5 --- /dev/null +++ b/.github/workflows/deploy_lambdas.yml @@ -0,0 +1,49 @@ +name: CI + +# Controls when the action will run. Triggers the workflow on push or pull request +# events but only for the main branch +on: + #when there is a push to the main + push: + branches: [ serverless ] + #when there is a pull to the main + pull_request: + branches: [ serverless ] + +jobs: + build: + runs-on: ubuntu-latest + steps: + + # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it + - uses: actions/checkout@v2 + + #installs a version of python, but I need this if deploying to a severless Python Lambda? + - name: Set up Python + uses: actions/setup-python@v2 + with: + python-version: '3.7' + + #credentials to connect to AWS + - name: Configure AWS credentials from Production account + uses: aws-actions/configure-aws-credentials@v1 + with: + aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }} + aws-region: us-east-1 + profile_name: default + project_name: MDFConnect + runtime: python3.7 + + - name: Create dependency layer + run: | + python -m pip install --upgrade pip + mkdir python + #install all dependencies as defined by requirements.txt in the aws directory + pip3 install --use-deprecated=legacy-resolver -r aws/requirements.txt -t ./python + + #zip files into current directory + zip -r funcxLayer.zip ./python + aws lambda publish-layer-version --layer-name FuncxLayer --zip-file fileb://./funcxLayer.zip --compatible-runtimes python3.6 python3.7 > layer_info.json + cat layer_info.json + LAYER_VERSION=$(jq ".Version" layer_info.json) diff --git a/aws/requirements.txt b/aws/requirements.txt new file mode 100644 index 0000000..87e07a0 --- /dev/null +++ b/aws/requirements.txt @@ -0,0 +1,4 @@ +globus-sdk>=1.7.0 +funcx +globus_automate_client + From a9727df9c3d741aa5fc0a469aece031fa9d91dc8 Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Tue, 23 Mar 2021 14:24:51 -0500 Subject: [PATCH 02/69] Add deployment of globus-auth function --- .github/workflows/deploy_lambdas.yml | 8 ++++ aws/funcx-globus-auth.py | 69 ++++++++++++++++++++++++++++ aws/funcx-run.py | 0 3 files changed, 77 insertions(+) create mode 100644 aws/funcx-globus-auth.py create mode 100644 aws/funcx-run.py diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index 17c1aa5..94c6e8b 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -47,3 +47,11 @@ jobs: aws lambda publish-layer-version --layer-name FuncxLayer --zip-file fileb://./funcxLayer.zip --compatible-runtimes python3.6 python3.7 > layer_info.json cat layer_info.json LAYER_VERSION=$(jq ".Version" layer_info.json) + + - name: Upload Globus Auth Function + run: | + cp aws/funcx-globus-auth.py ./lambda_function.py + zip funcx-globus-auth.zip ./lambda_function.py + rm ./lambda_function.py + aws lambda update-function-code --function-name funcx-globus-auth --zip-file fileb://./funcx-globus-auth.zip + diff --git a/aws/funcx-globus-auth.py b/aws/funcx-globus-auth.py new file mode 100644 index 0000000..1e95732 --- /dev/null +++ b/aws/funcx-globus-auth.py @@ -0,0 +1,69 @@ +import globus_sdk +import boto3 + +def get_secret(): + secret_name = "funcX-GlobusAPI" + region_name = "us-east-1" + + # Create a Secrets Manager client + session = boto3.session.Session() + + client = session.client( + service_name='secretsmanager', + region_name=region_name + ) + + get_secret_value_response = client.get_secret_value( + SecretId=secret_name + ) + return eval(get_secret_value_response['SecretString']) + + +def generate_policy(principalId, effect, resource, message="", name=None, identities=[], user_id=None, dependent_token=None): + authResponse = {} + authResponse['principalId'] = principalId + if effect and resource: + policyDocument = {} + policyDocument['Version'] = '2012-10-17' + policyDocument['Statement'] = [ + {'Action': 'execute-api:Invoke', + 'Effect': effect, + 'Resource': resource + } + ] + authResponse['policyDocument'] = policyDocument + authResponse['context'] = { + 'name': name, + 'user_id': user_id, + 'identities': str(identities), + 'globus_dependent_token': str(dependent_token) + } + print("AuthResponse", authResponse) + return authResponse + + +def lambda_handler(event, context): + globus_secrets = get_secret() + + auth_client = globus_sdk.ConfidentialAppAuthClient( + globus_secrets['API_CLIENT_ID'], globus_secrets['API_CLIENT_SECRET']) + + token = event['headers']['Authorization'].replace("Bearer ", "") + + auth_res = auth_client.oauth2_token_introspect(token, include="identities_set") + dependent_token = auth_client.oauth2_get_dependent_tokens(token) + print("Dependent token ", dependent_token) + + if not auth_res: + return generate_policy(None, 'Deny', event['methodArn'], message='User not found') + + if not auth_res['active']: + return generate_policy(None, 'Deny', event['methodArn'], + message='User account not active') + + print("auth_res", auth_res) + return generate_policy(auth_res['username'], 'Allow', event['methodArn'], + name=auth_res["name"], + identities=auth_res["identities_set"], + user_id=auth_res['sub'], + dependent_token=dependent_token) diff --git a/aws/funcx-run.py b/aws/funcx-run.py new file mode 100644 index 0000000..e69de29 From 898a4b2fa9440835f045db819e1288287767d359 Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Tue, 23 Mar 2021 14:33:03 -0500 Subject: [PATCH 03/69] Add funcx-run and introspect lambdas --- .github/workflows/deploy_lambdas.yml | 14 +++++++ aws/action_introspect.py | 57 ++++++++++++++++++++++++++++ aws/funcx-run.py | 25 ++++++++++++ 3 files changed, 96 insertions(+) create mode 100644 aws/action_introspect.py diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index 94c6e8b..fedad7f 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -55,3 +55,17 @@ jobs: rm ./lambda_function.py aws lambda update-function-code --function-name funcx-globus-auth --zip-file fileb://./funcx-globus-auth.zip + - name: Upload Action Introspect Function + run: | + cp aws/action_introspect.py ./lambda_function.py + zip action_introspect.zip ./lambda_function.py + rm ./lambda_function.py + aws lambda update-function-code --function-name action_introspect --zip-file fileb://./action_introspect.zip + + - name: Upload Action Run Function + run: | + cp aws/funcx-run.py ./lambda_function.py + zip funcx-run.zip ./lambda_function.py + rm ./lambda_function.py + aws lambda update-function-code --function-name funcx-run --zip-file fileb://./funcx-run.zip + diff --git a/aws/action_introspect.py b/aws/action_introspect.py new file mode 100644 index 0000000..f05ec28 --- /dev/null +++ b/aws/action_introspect.py @@ -0,0 +1,57 @@ +import json + + +def lambda_handler(event, context): + print(event) + + return { + 'statusCode': 202, + 'body': json.dumps( + { + "globus_auth_scope": "https://auth.globus.org/scopes/b3db7e59-a6f1-4947-95c2-59d6b7a70f8c/action_all", + "title": "FuncX Action Provider", + "subtitle": "Run FuncX", + "admin_contact": "bengal1@illinois.edu", + "synchronous": False, + "input_schema": { + "additionalProperties": False, + "properties": { + "tasks": { + "description": "List of tasks to invoke", + "items": { + "additionalProperties": False, + "properties": { + "endpoint": { + "description": "UUID of Endpoint where the function is to be run", + "type": "string" + }, + "function": { + "description": "UUID of the function to be run", + "type": "string" + }, + "payload": { + "description": "Arguments to function", + "type": "object" + } + } + } + } + + }, + "type": "object" + }, + "keywords": None, + "log_supported": False, + "maximum_deadline": "P30D", + "runnable_by": [ + "all_authenticated_users" + ], + "types": [ + "Action" + ], + "visible_to": [ + "public" + ] + } + ) + } \ No newline at end of file diff --git a/aws/funcx-run.py b/aws/funcx-run.py index e69de29..b4b4159 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -0,0 +1,25 @@ +import json +from funcx.sdk.client import FuncXClient +from globus_sdk import AccessTokenAuthorizer + + +def lambda_handler(event, context): + print(event) + + token = event['headers']['Authorization'].replace("Bearer ", "") + auth = AccessTokenAuthorizer(token) + print("Auth", auth) + FuncXClient.TOKEN_DIR = '/tmp' + fxc = FuncXClient(fx_authorizer=auth) + body = json.loads(event['body']) + print(body) + print(body['body']) + + return { + 'statusCode': 202, + 'body': json.dumps( + { + "result": body['request_id'] + } + ) + } \ No newline at end of file From fbcb5820766639592d40d8de0802600236a4679a Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Tue, 23 Mar 2021 15:21:07 -0500 Subject: [PATCH 04/69] Add some basic instructions --- README.md | 65 +++++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 63 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 244af41..750b2f7 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,63 @@ -# action-provider -Globus Automate Action Provider +# Serverless Globus Automate Action Provider +This repo contains configuration and code for deploying a serverless Globus +Automate Action Provider for submitting funcX tasks. + +## Initial Setup +Create a GlobusAuth app as per the instructions from [Action Provider Tools](https://action-provider-tools.readthedocs.io/en/latest/setting_up_auth.html) + +Save the resulting client ID and secret in [AWS Secrets Manager](https://console.aws.amazon.com/secretsmanager/home?region=us-east-1#!/listSecrets) +You will want to add two values: +- API_CLIENT_ID +- API_CLIENT_SECRET + +The lambda functions here assume that the secret is named `funcX-GlobusAPI` +## API Gateway +The key to the serverless architecture is the AWS API Gateway. Ideally this +would be configured by a CloudFormation template, however for now, these +notes will have to do. + +1. Go to [API Gateway dashboard](https://console.aws.amazon.com/apigateway/main/apis?region=us-east-1) +in the AWS Console. +2. Create new REST API. Use _New API_ since I haven't had luck importing the OpenAPI Spec +3. Click on _Authorizers_ and create a new lambda authorizer based on [aws/funcx-globus-auth.py](aws/funcx-globus-auth.py) +4. For _Lambda Event Payload_ select _Request_ and use `Header: Authorization` - this +takes the bearer token and puts it into the `event` dict. +5. Now go into _Resources_ and add a GET method under `/` +6. Set up as _Lambda Function_ Integration Type +7. Check _Use Lambda Proxy integration_ +8. Select [aws/action_introspect](aws/action_introspect.py) +9. Add a resource named `/run` and add a POST method +10. Select [aws/funcx-run](aws/funcx-run.py) as the lambda function +11. Finally select _Deploy API_ action and create a new stage called `dev` - make +a note of the generated URL + +## Interacting with The Action Provider +You will need to install the Globus Automate cli +```shell script +pip install globus_automate_client +``` + +### View the action +You can retrieve the action document with: +```shell script + globus-automate action introspect --action-url <> --action-scope <> +``` + +## Run the Action +You need to create a body json document that represents an invocation of the action. +```json +{ + "tasks": [{ + "endpoint": "4b116d3c-1703-4f8f-9f6f-39921e5864df", + "function": "4b116d3c-1703-4f8f-9f6f-39921e5864df", + "payload": { + "x": 2 + } + }] +} +``` + +Then invoke the action with +```shell script +globus-automate action run --body sample.json --action-url <> --action-scope <> +``` \ No newline at end of file From f6d5f3f7bbb6a1838fcb81fa653e601e18a586be Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Tue, 23 Mar 2021 16:04:58 -0500 Subject: [PATCH 05/69] Getting the right collection of scopes --- aws/funcx-run.py | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index b4b4159..d52c60e 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -5,12 +5,27 @@ def lambda_handler(event, context): print(event) + name = event['requestContext']['authorizer']['name'] + identities = eval(event['requestContext']['authorizer']['identities']) + user_id = event['requestContext']['authorizer']['user_id'] + user_email = event['requestContext']['authorizer']['principalId'] + depends = eval( + event['requestContext']['authorizer']['globus_dependent_token'].replace("null", + "None")) + print(depends['funcx_service']) - token = event['headers']['Authorization'].replace("Bearer ", "") + token = depends['funcx_service']['access_token'] auth = AccessTokenAuthorizer(token) + + search_token = depends['search.api.globus.org']['access_token'] + search_auth = AccessTokenAuthorizer(search_token) + + openid_token = depends['auth.globus.org']['access_token'] + openid_auth = AccessTokenAuthorizer(openid_token) print("Auth", auth) FuncXClient.TOKEN_DIR = '/tmp' - fxc = FuncXClient(fx_authorizer=auth) + fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, + openid_authorizer=openid_auth) body = json.loads(event['body']) print(body) print(body['body']) From 5fe06b946a9eef602c3976fd1f501e399193a80c Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Tue, 23 Mar 2021 19:49:05 -0500 Subject: [PATCH 06/69] Add dynamo db tracking of requests and status --- .github/workflows/deploy_lambdas.yml | 7 ++++ aws/action-status.py | 48 ++++++++++++++++++++++++++++ aws/funcx-run.py | 21 ++++++++++-- 3 files changed, 73 insertions(+), 3 deletions(-) create mode 100644 aws/action-status.py diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index fedad7f..c49aa39 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -69,3 +69,10 @@ jobs: rm ./lambda_function.py aws lambda update-function-code --function-name funcx-run --zip-file fileb://./funcx-run.zip + - name: Upload Action Status Function + run: | + cp aws/action-status.py ./lambda_function.py + zip action-status.zip ./lambda_function.py + rm ./lambda_function.py + aws lambda update-function-code --function-name action_status --zip-file fileb://./action-status.zip + diff --git a/aws/action-status.py b/aws/action-status.py new file mode 100644 index 0000000..c146757 --- /dev/null +++ b/aws/action-status.py @@ -0,0 +1,48 @@ +import json +import boto3 +from boto3.dynamodb.conditions import Key +from globus_sdk import AccessTokenAuthorizer +from funcx.sdk.client import FuncXClient + + +def lambda_handler(event, context): + dynamodb = boto3.resource('dynamodb') + table = dynamodb.Table('funcx-actions') + + depends = eval( + event['requestContext']['authorizer']['globus_dependent_token'].replace("null", + "None")) + print(depends['funcx_service']) + + token = depends['funcx_service']['access_token'] + auth = AccessTokenAuthorizer(token) + + search_token = depends['search.api.globus.org']['access_token'] + search_auth = AccessTokenAuthorizer(search_token) + + openid_token = depends['auth.globus.org']['access_token'] + openid_auth = AccessTokenAuthorizer(openid_token) + FuncXClient.TOKEN_DIR = '/tmp' + fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, + openid_authorizer=openid_auth) + + parameters = event['pathParameters']['proxy'] + (flow_id, _) = parameters.split('/') + response = table.query( + KeyConditionExpression=Key('flow-id').eq(flow_id) + ) + print(response['Items'][0]) + print(response['Items'][0]['tasks'][0]) + result = None + try: + result = fxc.get_result(response['Items'][0]['tasks'][0]) + except Exception as eek: + result = str(eek) + + print("---->", result) + # TODO implement + print(event) + return { + 'statusCode': 200, + 'body': result + } diff --git a/aws/funcx-run.py b/aws/funcx-run.py index d52c60e..65786f6 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -1,6 +1,7 @@ import json from funcx.sdk.client import FuncXClient from globus_sdk import AccessTokenAuthorizer +import boto3 def lambda_handler(event, context): @@ -22,14 +23,28 @@ def lambda_handler(event, context): openid_token = depends['auth.globus.org']['access_token'] openid_auth = AccessTokenAuthorizer(openid_token) - print("Auth", auth) FuncXClient.TOKEN_DIR = '/tmp' fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, openid_authorizer=openid_auth) + + dynamodb = boto3.resource('dynamodb') + table = dynamodb.Table('funcx-actions') + body = json.loads(event['body']) - print(body) - print(body['body']) + task = body['body']['tasks'][0] + print(task) + res = fxc.run(endpoint_id=task['endpoint'], function_id=task['function']) + print("Funcx", res) + response = table.put_item( + Item={ + 'flow-id': body['request_id'], + 'tasks': [ + res + ] + } + ) + print("Dynamo", response) return { 'statusCode': 202, 'body': json.dumps( From 29c81df84ff1c5d5d4f18874b6cd974d440f6c2c Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Wed, 24 Mar 2021 16:06:14 -0500 Subject: [PATCH 07/69] Update status --- aws/action-status.py | 42 ++++++++++++++------------ aws/funcx-globus-auth.py | 15 +++++++--- aws/funcx-run.py | 64 ++++++++++++++++++++++------------------ aws/requirements.txt | 2 +- 4 files changed, 71 insertions(+), 52 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index c146757..bed4bad 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -9,40 +9,44 @@ def lambda_handler(event, context): dynamodb = boto3.resource('dynamodb') table = dynamodb.Table('funcx-actions') - depends = eval( - event['requestContext']['authorizer']['globus_dependent_token'].replace("null", - "None")) - print(depends['funcx_service']) + auth = AccessTokenAuthorizer(event['requestContext']['authorizer']['funcx_token']) + search_auth = AccessTokenAuthorizer( + event['requestContext']['authorizer']['search_token']) + openid_auth = AccessTokenAuthorizer( + event['requestContext']['authorizer']['openid_token']) - token = depends['funcx_service']['access_token'] - auth = AccessTokenAuthorizer(token) - - search_token = depends['search.api.globus.org']['access_token'] - search_auth = AccessTokenAuthorizer(search_token) - - openid_token = depends['auth.globus.org']['access_token'] - openid_auth = AccessTokenAuthorizer(openid_token) FuncXClient.TOKEN_DIR = '/tmp' fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, openid_authorizer=openid_auth) parameters = event['pathParameters']['proxy'] - (flow_id, _) = parameters.split('/') + (action_id, _) = parameters.split('/') response = table.query( - KeyConditionExpression=Key('flow-id').eq(flow_id) + KeyConditionExpression=Key('action-id').eq(action_id) ) - print(response['Items'][0]) - print(response['Items'][0]['tasks'][0]) + assert "Items" in response + assert len(response['Items']) == 1 + + action_record = response['Items'][0] + print(action_record) + result = None try: - result = fxc.get_result(response['Items'][0]['tasks'][0]) + result = fxc.get_result(action_record['tasks'][0]['task_id']) except Exception as eek: result = str(eek) print("---->", result) - # TODO implement + + result = { + "action_id": action_id, + 'status': 'SUCCEEDED', + 'display_status': 'Function Results Received', + 'details': result + } + print(event) return { 'statusCode': 200, - 'body': result + 'body': json.dumps(result) } diff --git a/aws/funcx-globus-auth.py b/aws/funcx-globus-auth.py index 1e95732..45d2e7c 100644 --- a/aws/funcx-globus-auth.py +++ b/aws/funcx-globus-auth.py @@ -20,6 +20,10 @@ def get_secret(): def generate_policy(principalId, effect, resource, message="", name=None, identities=[], user_id=None, dependent_token=None): + funcx_token = dependent_token.by_resource_server['funcx_service']['access_token'] + search_token = dependent_token.by_resource_server['search.api.globus.org']['access_token'] + openid_token = dependent_token.by_resource_server['auth.globus.org']['access_token'] + authResponse = {} authResponse['principalId'] = principalId if effect and resource: @@ -36,7 +40,9 @@ def generate_policy(principalId, effect, resource, message="", name=None, identi 'name': name, 'user_id': user_id, 'identities': str(identities), - 'globus_dependent_token': str(dependent_token) + 'funcx_token': funcx_token, + 'search_token': search_token, + 'openid_token': openid_token } print("AuthResponse", authResponse) return authResponse @@ -51,8 +57,9 @@ def lambda_handler(event, context): token = event['headers']['Authorization'].replace("Bearer ", "") auth_res = auth_client.oauth2_token_introspect(token, include="identities_set") - dependent_token = auth_client.oauth2_get_dependent_tokens(token) - print("Dependent token ", dependent_token) + depends = auth_client.oauth2_get_dependent_tokens(token) + print(depends) + if not auth_res: return generate_policy(None, 'Deny', event['methodArn'], message='User not found') @@ -66,4 +73,4 @@ def lambda_handler(event, context): name=auth_res["name"], identities=auth_res["identities_set"], user_id=auth_res['sub'], - dependent_token=dependent_token) + dependent_token=depends) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 65786f6..eaa598b 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -2,27 +2,23 @@ from funcx.sdk.client import FuncXClient from globus_sdk import AccessTokenAuthorizer import boto3 +import uuid +import datetime + + +def now_isoformat(): + return datetime.datetime.now().isoformat() def lambda_handler(event, context): print(event) - name = event['requestContext']['authorizer']['name'] - identities = eval(event['requestContext']['authorizer']['identities']) - user_id = event['requestContext']['authorizer']['user_id'] - user_email = event['requestContext']['authorizer']['principalId'] - depends = eval( - event['requestContext']['authorizer']['globus_dependent_token'].replace("null", - "None")) - print(depends['funcx_service']) - - token = depends['funcx_service']['access_token'] - auth = AccessTokenAuthorizer(token) - - search_token = depends['search.api.globus.org']['access_token'] - search_auth = AccessTokenAuthorizer(search_token) - - openid_token = depends['auth.globus.org']['access_token'] - openid_auth = AccessTokenAuthorizer(openid_token) + + auth = AccessTokenAuthorizer(event['requestContext']['authorizer']['funcx_token']) + search_auth = AccessTokenAuthorizer( + event['requestContext']['authorizer']['search_token']) + openid_auth = AccessTokenAuthorizer( + event['requestContext']['authorizer']['openid_token']) + FuncXClient.TOKEN_DIR = '/tmp' fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, openid_authorizer=openid_auth) @@ -31,25 +27,37 @@ def lambda_handler(event, context): table = dynamodb.Table('funcx-actions') body = json.loads(event['body']) + + action_id = str(uuid.uuid4()) + result = { + "action_id": action_id, + 'status': 'ACTIVE', + 'display_status': 'Function Submitted', + 'details': None, + 'monitor_by': body['monitor_by'], + 'manage_by': body['manage_by'], + 'start_time': now_isoformat(), + } + tasks = [] + task = body['body']['tasks'][0] print(task) - res = fxc.run(endpoint_id=task['endpoint'], function_id=task['function']) - print("Funcx", res) + task_id = fxc.run(endpoint_id=task['endpoint'], function_id=task['function']) + print("Funcx", task_id) + + tasks.append({ + "task_id": task_id, + "result": None + }) response = table.put_item( Item={ - 'flow-id': body['request_id'], - 'tasks': [ - res - ] + 'action-id': action_id, + 'tasks': tasks } ) print("Dynamo", response) return { 'statusCode': 202, - 'body': json.dumps( - { - "result": body['request_id'] - } - ) + 'body': json.dumps(result) } \ No newline at end of file diff --git a/aws/requirements.txt b/aws/requirements.txt index 87e07a0..2168185 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -1,4 +1,4 @@ globus-sdk>=1.7.0 funcx globus_automate_client - +globus_action_provider_tools From 06b890b77c91ecc5ed52c567b2b513d15406837e Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Wed, 24 Mar 2021 16:09:39 -0500 Subject: [PATCH 08/69] Drop action_provider_tools --- aws/requirements.txt | 1 - 1 file changed, 1 deletion(-) diff --git a/aws/requirements.txt b/aws/requirements.txt index 2168185..94344cb 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -1,4 +1,3 @@ globus-sdk>=1.7.0 funcx globus_automate_client -globus_action_provider_tools From e776808950a2a3a5e4b8d6629fd93c282af06b8c Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Fri, 26 Mar 2021 13:42:51 -0500 Subject: [PATCH 09/69] Working for multiple tasks. Added example flow --- .gitignore | 2 + aws/action-status.py | 56 +++++++++++++--- aws/funcx-run.py | 26 +++++--- example/deploy_example_flow.py | 73 ++++++++++++++++++++ example/flow_action.py | 22 ++++++ example/globus_automate_flow.py | 115 ++++++++++++++++++++++++++++++++ 6 files changed, 275 insertions(+), 19 deletions(-) create mode 100644 .gitignore create mode 100644 example/deploy_example_flow.py create mode 100644 example/flow_action.py create mode 100644 example/globus_automate_flow.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..1e48398 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +.automatesecrets + diff --git a/aws/action-status.py b/aws/action-status.py index bed4bad..9dc35cc 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -30,19 +30,59 @@ def lambda_handler(event, context): action_record = response['Items'][0] print(action_record) - result = None - try: - result = fxc.get_result(action_record['tasks'][0]['task_id']) - except Exception as eek: - result = str(eek) + # Find the taskIDs where the results are not yet in + running_tasks = list(filter(lambda task_id: bool(task_id), + [key if not action_record['tasks'][key][ + 'result'] else None + for key in action_record['tasks'].keys()])) - print("---->", result) + failure = None + if running_tasks: + for task in running_tasks: + result = None + try: + result = fxc.get_result(task) + print("---->", result) + + except Exception as eek: + print("Faiulure ", type(eek), eek.args) + if str(eek) == 'waiting-for-ep': + result = None + else: + failure = eek + + if result: + action_record['tasks'][task]['result'] = result + + update_response = table.update_item( + Key={ + 'action-id': action_id + }, + UpdateExpression="set tasks=:t", + ExpressionAttributeValues={ + ':t': action_record['tasks'] + }, + ReturnValues="UPDATED_NEW" + ) + + print(update_response) + print(failure) + if failure: + status = "FAILED" + details = failure + else: + status = "ACTIVE" + details = None + else: + status = "SUCCEEDED" + details = [str(action_record['tasks'][tt]['result']) for tt in + action_record['tasks'].keys()] result = { "action_id": action_id, - 'status': 'SUCCEEDED', + 'status': status, 'display_status': 'Function Results Received', - 'details': result + 'details': details } print(event) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index eaa598b..e9e7d01 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -29,26 +29,30 @@ def lambda_handler(event, context): body = json.loads(event['body']) action_id = str(uuid.uuid4()) + monitor_by = body['monitor_by'] if 'monitor_by' in body else None + manage_by = body['manage_by'] if 'manage_by' in body else None + result = { "action_id": action_id, 'status': 'ACTIVE', 'display_status': 'Function Submitted', 'details': None, - 'monitor_by': body['monitor_by'], - 'manage_by': body['manage_by'], + 'monitor_by': monitor_by, + 'manage_by': manage_by, 'start_time': now_isoformat(), } - tasks = [] + tasks = {} + + for task in body['body']['tasks']: + print(task) - task = body['body']['tasks'][0] - print(task) - task_id = fxc.run(endpoint_id=task['endpoint'], function_id=task['function']) - print("Funcx", task_id) + task_id = fxc.run(endpoint_id=task['endpoint'], function_id=task['function'], **task['payload']) - tasks.append({ - "task_id": task_id, - "result": None - }) + print("Funcx", task_id) + + tasks[task_id] = { + "result": None + } response = table.put_item( Item={ diff --git a/example/deploy_example_flow.py b/example/deploy_example_flow.py new file mode 100644 index 0000000..0ca842f --- /dev/null +++ b/example/deploy_example_flow.py @@ -0,0 +1,73 @@ +from globus_automate_flow import GlobusAutomateFlowDef, GlobusAutomateFlow +import json + +import globus_automate_client + + +def flow_def(flow_permissions): + return GlobusAutomateFlowDef( + title="FuncX Example", + description="Show how to invoke FuncX", + visible_to=flow_permissions['flow_permissions'], + runnable_by=flow_permissions['flow_permissions'], + administered_by=flow_permissions['flow_permissions'], + flow_definition={ + "StartAt": "StartSubmission", + "States": { + "StartSubmission": { + "Type": "Action", + "ActionUrl": "https://ippg79abte.execute-api.us-east-1.amazonaws.com/dev", + "ActionScope": "https://auth.globus.org/scopes/b3db7e59-a6f1-4947-95c2-59d6b7a70f8c/action_all", + "Parameters": { + "tasks": [{ + "endpoint": "4b116d3c-1703-4f8f-9f6f-39921e5864df", + "function": "18f8416a-edbd-4f3b-82ef-3c5697a0697a", + "payload": { + "items": [1, 2, 3, 4] + } + }, + { + "endpoint": "4b116d3c-1703-4f8f-9f6f-39921e5864df", + "function": "18f8416a-edbd-4f3b-82ef-3c5697a0697a", + "payload": { + "items": [10, 20, 30, 40] + } + }, + { + "endpoint": "4b116d3c-1703-4f8f-9f6f-39921e5864df", + "function": "74c03996-c4b0-471f-b26b-b596edbf80f9", + "payload": {} + } + ] + }, + "Next": "EndSubmission" + }, + "EndSubmission": { + "Type": "Pass", + "End": True + } + } + } + ) + + +with open(".automatesecrets", 'r') as f: + globus_secrets = json.load(f) + +native_app_id = "417301b1-5101-456a-8a27-423e71a2ae26" # Premade native app ID +flows_client = globus_automate_client.create_flows_client(native_app_id) + +flow = flow_def(flow_permissions={ + "flow_permissions": [ + "urn:globus:groups:id:5fc63928-3752-11e8-9c6f-0e00fd09bf20" + ], + "admin_permissions": [ + "urn:globus:groups:id:5fc63928-3752-11e8-9c6f-0e00fd09bf20" + ], +}) +print(flow.flow_definition) + +mdf_flow = GlobusAutomateFlow.from_flow_def(flows_client, + flow_def=flow) + +mdf_flow.save_flow("mdf_flow_info.json") diff --git a/example/flow_action.py b/example/flow_action.py new file mode 100644 index 0000000..7b0f9f3 --- /dev/null +++ b/example/flow_action.py @@ -0,0 +1,22 @@ +import ast + + +class FlowAction: + def __init__(self, flow, action_id: str): + self.action_id = action_id + self.flow = flow + + def get_status(self): + return self.flow.get_status(self.action_id) + + def get_error_msgs(self): + logs = self.flow.get_flow_logs(self.action_id) + error_msgs = [] + for failure in filter(lambda x: x['code'] == 'ActionFailed', logs['entries']): + # Failures from Search Ingest Action Provider are bundled up as string + # representation of Python dict + cause = ast.literal_eval(failure['details']['cause']) + if 'errors' in cause: + error_msgs.append(cause['errors']) + + return error_msgs diff --git a/example/globus_automate_flow.py b/example/globus_automate_flow.py new file mode 100644 index 0000000..0198986 --- /dev/null +++ b/example/globus_automate_flow.py @@ -0,0 +1,115 @@ +import json +from typing import Mapping, Any, Optional, List + +from globus_automate_client import FlowsClient + +from flow_action import FlowAction + +class GlobusAutomateFlowDef: + def __init__(self, + flow_definition: Mapping[str, Any], + title: str, + subtitle: Optional[str] = None, + description: Optional[str] = None, + keywords: List[str] = [], + visible_to: List[str] = [], + runnable_by: List[str] = [], + administered_by: List[str] = [], + input_schema: Optional[Mapping[str, Any]] = None): + self.flow_definition = flow_definition + self.title = title + self.subtitle = subtitle + self.description = description + self.keywords = keywords + self.visible_to = visible_to + self.runnable_by = runnable_by + self.administered_by = administered_by + self.input_schema = input_schema + + +class GlobusAutomateFlow: + def __init__(self, client: FlowsClient): + self.flows_client = client + self.flow_id = None + self.flow_scope = None + self.saved_flow = None + self.runAsScopes = None + + @classmethod + def from_flow_def(cls, client: FlowsClient, + flow_def: GlobusAutomateFlowDef): + result = GlobusAutomateFlow(client) + result._deploy_mdf_flow(flow_def) + return result + + @classmethod + def from_existing_flow(cls, path: str): + result = GlobusAutomateFlow(None) + result.read_flow(path) + return result + + def set_client(self, client): + self.flows_client = client + + @property + def url(self): + return "https://flows.globus.org/flows/" + self.flow_id + + def __str__(self): + return f'Globus Automate Flow: id={self.flow_id}, scope={self.flow_scope}' + + + def get_status(self, action_id: str): + return self.flows_client.flow_action_status( + self.flow_id, + self.flow_scope, + action_id).data + + def get_flow_logs(self, action_id: str): + return self.flows_client.flow_action_log( + self.flow_id, self.flow_scope, + action_id, + limit=100).data + + def _deploy_mdf_flow(self, mdf_flow_def: GlobusAutomateFlowDef): + flow_deploy_res = self.flows_client.deploy_flow( + flow_definition=mdf_flow_def.flow_definition, + title=mdf_flow_def.title, + subtitle=mdf_flow_def.subtitle, + description=mdf_flow_def.description, + visible_to=mdf_flow_def.visible_to, + runnable_by=mdf_flow_def.runnable_by, + administered_by=mdf_flow_def.administered_by, + # TODO: Make rough schema outline into JSONSchema + input_schema=mdf_flow_def.input_schema, + validate_definition=True, + validate_input_schema=True + ) + self.flow_id = flow_deploy_res["id"] + self.flow_scope = flow_deploy_res["globus_auth_scope"] + self.saved_flow = self.flows_client.get_flow(self.flow_id).data + print(self.runAsScopes) + + def run_flow(self, flow_input: dict): + flow_res = self.flows_client.run_flow(self.flow_id, self.flow_scope, flow_input) + return FlowAction(self, flow_res.data['action_id']) + + def save_flow(self, path): + # Save Flow ID/scope for future use + with open(path, 'w') as f: + flow_info = { + "flow_id": self.flow_id, + "flow_scope": self.flow_scope + } + json.dump(flow_info, f) + + def read_flow(self, path): + # Save Flow ID/scope for future use + with open(path, 'r') as f: + flow_info = json.load(f) + self.flow_id = flow_info['flow_id'] + self.flow_scope = flow_info['flow_scope'] + + def get_scope_for_runAs_role(self, rolename): + print("--->RunAsScopes ", self.runAsScopes[rolename]) + return self.globus_auth.scope_id_from_uri(self.runAsScopes[rolename]) From 3b6b20cab0b873942cf8c17e2bc219fa1e3a5909 Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Fri, 26 Mar 2021 13:53:34 -0500 Subject: [PATCH 10/69] Make a smarter deploy_example_flow --- .gitignore | 140 +++++++++++++++++++++++++++++++++ example/deploy_example_flow.py | 37 ++++++--- 2 files changed, 166 insertions(+), 11 deletions(-) diff --git a/.gitignore b/.gitignore index 1e48398..4d9948b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,142 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + .automatesecrets +example_flow_info.json diff --git a/example/deploy_example_flow.py b/example/deploy_example_flow.py index 0ca842f..173eeea 100644 --- a/example/deploy_example_flow.py +++ b/example/deploy_example_flow.py @@ -4,7 +4,7 @@ import globus_automate_client -def flow_def(flow_permissions): +def flow_def(flow_permissions, endpoint, sum_function, hello_world_function): return GlobusAutomateFlowDef( title="FuncX Example", description="Show how to invoke FuncX", @@ -20,22 +20,22 @@ def flow_def(flow_permissions): "ActionScope": "https://auth.globus.org/scopes/b3db7e59-a6f1-4947-95c2-59d6b7a70f8c/action_all", "Parameters": { "tasks": [{ - "endpoint": "4b116d3c-1703-4f8f-9f6f-39921e5864df", - "function": "18f8416a-edbd-4f3b-82ef-3c5697a0697a", + "endpoint": endpoint, + "function": sum_function, "payload": { "items": [1, 2, 3, 4] } }, { - "endpoint": "4b116d3c-1703-4f8f-9f6f-39921e5864df", - "function": "18f8416a-edbd-4f3b-82ef-3c5697a0697a", + "endpoint": endpoint, + "function": sum_function, "payload": { "items": [10, 20, 30, 40] } }, { - "endpoint": "4b116d3c-1703-4f8f-9f6f-39921e5864df", - "function": "74c03996-c4b0-471f-b26b-b596edbf80f9", + "endpoint": endpoint, + "function": hello_world_function, "payload": {} } ] @@ -50,6 +50,17 @@ def flow_def(flow_permissions): } ) +from funcx.sdk.client import FuncXClient +fxc = FuncXClient() + +def hello_world(): + return "Hello World!" + + +def funcx_sum(items): + import time + time.sleep(15) + return sum(items) with open(".automatesecrets", 'r') as f: globus_secrets = json.load(f) @@ -64,10 +75,14 @@ def flow_def(flow_permissions): "admin_permissions": [ "urn:globus:groups:id:5fc63928-3752-11e8-9c6f-0e00fd09bf20" ], -}) +}, + endpoint="4b116d3c-1703-4f8f-9f6f-39921e5864df", + sum_function=fxc.register_function(funcx_sum), + hello_world_function=fxc.register_function(hello_world)) + print(flow.flow_definition) -mdf_flow = GlobusAutomateFlow.from_flow_def(flows_client, - flow_def=flow) +example_flow = GlobusAutomateFlow.from_flow_def(flows_client, + flow_def=flow) -mdf_flow.save_flow("mdf_flow_info.json") +example_flow.save_flow("example_flow_info.json") From 03052b517591da8f9b490fe3aa8ab2f2005276a2 Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Fri, 26 Mar 2021 14:11:43 -0500 Subject: [PATCH 11/69] Submit tasks as a batch --- aws/funcx-run.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index e9e7d01..78ef7d1 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -42,22 +42,23 @@ def lambda_handler(event, context): 'start_time': now_isoformat(), } tasks = {} + batch = fxc.create_batch() for task in body['body']['tasks']: print(task) + batch.add(endpoint_id=task['endpoint'], function_id=task['function'], + **task['payload']) - task_id = fxc.run(endpoint_id=task['endpoint'], function_id=task['function'], **task['payload']) - - print("Funcx", task_id) - - tasks[task_id] = { - "result": None - } + batch_res = fxc.batch_run(batch) + print(batch_res) + # Create a dynamo record where the primary key is this action's ID + # Tasks is a dict by task_id and contains the eventual results from their + # execution. Where there are no more None results then the action is complete response = table.put_item( Item={ 'action-id': action_id, - 'tasks': tasks + 'tasks': {task_id: {"result": None} for task_id in batch_res} } ) print("Dynamo", response) From 1b97c5333bd8109a6989161de638e578aaa39df1 Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Tue, 30 Mar 2021 14:50:24 -0500 Subject: [PATCH 12/69] Migrate CI job to funcX AWS Account --- .github/workflows/deploy_lambdas.yml | 2 +- aws/action-status.py | 5 +++-- cloud-formation/api_gateway.py | 0 example/deploy_example_flow.py | 2 +- 4 files changed, 5 insertions(+), 4 deletions(-) create mode 100644 cloud-formation/api_gateway.py diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index c49aa39..81ca88c 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -32,7 +32,7 @@ jobs: aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }} aws-region: us-east-1 profile_name: default - project_name: MDFConnect + project_name: FuncXActionProvider runtime: python3.7 - name: Create dependency layer diff --git a/aws/action-status.py b/aws/action-status.py index 9dc35cc..307cc2b 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -8,6 +8,7 @@ def lambda_handler(event, context): dynamodb = boto3.resource('dynamodb') table = dynamodb.Table('funcx-actions') + print(event) auth = AccessTokenAuthorizer(event['requestContext']['authorizer']['funcx_token']) search_auth = AccessTokenAuthorizer( @@ -19,8 +20,8 @@ def lambda_handler(event, context): fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, openid_authorizer=openid_auth) - parameters = event['pathParameters']['proxy'] - (action_id, _) = parameters.split('/') + action_id = event['pathParameters']['action-id'] + response = table.query( KeyConditionExpression=Key('action-id').eq(action_id) ) diff --git a/cloud-formation/api_gateway.py b/cloud-formation/api_gateway.py new file mode 100644 index 0000000..e69de29 diff --git a/example/deploy_example_flow.py b/example/deploy_example_flow.py index 173eeea..ca84963 100644 --- a/example/deploy_example_flow.py +++ b/example/deploy_example_flow.py @@ -16,7 +16,7 @@ def flow_def(flow_permissions, endpoint, sum_function, hello_world_function): "States": { "StartSubmission": { "Type": "Action", - "ActionUrl": "https://ippg79abte.execute-api.us-east-1.amazonaws.com/dev", + "ActionUrl": " https://b6vr4fptui.execute-api.us-east-1.amazonaws.com/test", "ActionScope": "https://auth.globus.org/scopes/b3db7e59-a6f1-4947-95c2-59d6b7a70f8c/action_all", "Parameters": { "tasks": [{ From 11f89e9d512af7f4d8eefde9eb024f15b7649403 Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Thu, 1 Apr 2021 10:32:31 -0500 Subject: [PATCH 13/69] Update README with complete deployment hints --- .gitignore | 3 +- README.md | 120 +++++++++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 119 insertions(+), 4 deletions(-) diff --git a/.gitignore b/.gitignore index 4d9948b..23d005d 100644 --- a/.gitignore +++ b/.gitignore @@ -139,4 +139,5 @@ cython_debug/ .automatesecrets example_flow_info.json - +.python-version +.idea diff --git a/README.md b/README.md index 750b2f7..baa3e26 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,101 @@ You will want to add two values: - API_CLIENT_SECRET The lambda functions here assume that the secret is named `funcX-GlobusAPI` + +## Dynamo DB +We use a single DynamoDB table to relate action_id's to taskIDs. +You will need to create a dynamo DB table called `funcx-actions` - set the +partition key to `action-id` (string). We will need a policy to allow the +lambda functions to interact with this table. This one should do: +```json +{ + "Version": "2012-10-17", + "Statement": [ + { + "Sid": "VisualEditor0", + "Effect": "Allow", + "Action": [ + "dynamodb:BatchGetItem", + "dynamodb:BatchWriteItem", + "dynamodb:ConditionCheckItem", + "dynamodb:PutItem", + "dynamodb:DeleteItem", + "dynamodb:PartiQLUpdate", + "dynamodb:Scan", + "dynamodb:Query", + "dynamodb:UpdateItem", + "dynamodb:DescribeTimeToLive", + "dynamodb:PartiQLSelect", + "dynamodb:DescribeTable", + "dynamodb:PartiQLInsert", + "dynamodb:GetItem", + "dynamodb:UpdateTable", + "dynamodb:PartiQLDelete" + ], + "Resource": "arn:aws:dynamodb:us-east-1:512084481048:table/funcx-actions" + } + ] +} +``` + +The `Resource` ARN should match the table you created. + +An example of the items in this table looks like this: +```json +{ + "action-id": "e9e664d2-d923-4c48-948d-d9b68145077c", + "tasks": { + "5d6ae875-ef80-44eb-8664-4c159ced0c46": { + "result": "Hello World!" + }, + "86017ef8-faaa-4f50-af62-44d1cf02398e": { + "result": 10 + }, + "f2ccbd80-e94b-4af0-afad-f35aff58d4c4": { + "result": 100 + } + } +} +``` + +Action-id is the partition key and matches the GlobusAutomate action that was received +by the service. Tasks is a dictionary of taskID to the results returned from the +function invocation. The `result` is null until something comes back from funcX. + +When all of the results are non-null the action is complete. + + +## Lambda Functions +The lambda functions implement the Action Provider API. They are automatically +deployed to the AWS account by github actions. + +All of the functions get their dependencies from an AWS Lambda Layer called +`FuncxLayer` which is built from [aws/requirements.txt](aws/requirements.txt). + +The action provider functions are fronted by an authorizer function called +`funcx-globus-auth` - This introspect's the bearer token and extracts the +needed dependent tokens. It adds some useful extracted data to the event dict +passed into each lambda function. + +```python + authResponse['context'] = { + 'name': name, + 'user_id': user_id, + 'identities': str(identities), + 'funcx_token': funcx_token, + 'search_token': search_token, + 'openid_token': openid_token + } +``` + +The GitHub Action CI job that deploys these Lambda functions assumes that the +AWS credentials are stored in repository secrets: +* AWS_ACCESS_KEY_ID +* AWS_SECRET_ACCESS_KEY + +I think th GitHubAction assumes that the layer, and the lambda functions already +exist in the account, so you may need to create initial blank values for these. + ## API Gateway The key to the serverless architecture is the AWS API Gateway. Ideally this would be configured by a CloudFormation template, however for now, these @@ -28,7 +123,11 @@ takes the bearer token and puts it into the `event` dict. 8. Select [aws/action_introspect](aws/action_introspect.py) 9. Add a resource named `/run` and add a POST method 10. Select [aws/funcx-run](aws/funcx-run.py) as the lambda function -11. Finally select _Deploy API_ action and create a new stage called `dev` - make +11. Make a new resource called `action_id` and set the path to be `{action_id}` - +this will create a path variable. +12. Under action_id, create a new resource called `status` and add a GET method +which calls `action-status` Lambda +13. Finally select _Deploy API_ action and create a new stage called `dev` - make a note of the generated URL ## Interacting with The Action Provider @@ -43,7 +142,7 @@ You can retrieve the action document with: globus-automate action introspect --action-url <> --action-scope <> ``` -## Run the Action +## Run the Action in Isolation You need to create a body json document that represents an invocation of the action. ```json { @@ -60,4 +159,19 @@ You need to create a body json document that represents an invocation of the act Then invoke the action with ```shell script globus-automate action run --body sample.json --action-url <> --action-scope <> -``` \ No newline at end of file +``` + +## Deploy the Example Flow +Examine the code in [example/deploy_example_flow.py](example/deploy_example_flow.py) +- this will create a flow definition that invokes the action provider. You'll +need to update the `ActionUrl` and `ActionScope` to match your new ActionProvider. + +Note the generated Flow ID. You can launch an instance of this flow with +```shell script + globus-automate flow run -v <> +``` + +Note the flow_id that comes back so you can monitor the progress with +```shell script +globus-automate flow action-status -v --flow-id <> <> +``` From 20544b814416a2e1175c32cf60f1d1db3aef9d83 Mon Sep 17 00:00:00 2001 From: Ryan Date: Thu, 20 May 2021 14:31:31 -0500 Subject: [PATCH 14/69] Bump to 0.2.3 Bump funcx to 0.2.3 --- aws/requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aws/requirements.txt b/aws/requirements.txt index 94344cb..9901c38 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -1,3 +1,3 @@ globus-sdk>=1.7.0 -funcx +funcx==0.2.3 globus_automate_client From b9f37c1b45ecdc1e08835f0784924cb5d6ce6eb1 Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Mon, 24 May 2021 11:56:52 -0500 Subject: [PATCH 15/69] Add python 3.7 to list of valid runtimes --- .github/workflows/deploy_lambdas.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index 81ca88c..b4a6d68 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -44,7 +44,7 @@ jobs: #zip files into current directory zip -r funcxLayer.zip ./python - aws lambda publish-layer-version --layer-name FuncxLayer --zip-file fileb://./funcxLayer.zip --compatible-runtimes python3.6 python3.7 > layer_info.json + aws lambda publish-layer-version --layer-name FuncxLayer --zip-file fileb://./funcxLayer.zip --compatible-runtimes python3.6 python3.7 python3.8 > layer_info.json cat layer_info.json LAYER_VERSION=$(jq ".Version" layer_info.json) From 21a6bc5410064f63982007c80eb72070c0a890ba Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Mon, 24 May 2021 13:25:44 -0500 Subject: [PATCH 16/69] use_offprocess_checker to False --- aws/funcx-run.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 78ef7d1..67f036e 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -21,7 +21,8 @@ def lambda_handler(event, context): FuncXClient.TOKEN_DIR = '/tmp' fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, - openid_authorizer=openid_auth) + openid_authorizer=openid_auth, + use_offprocess_checker=False) dynamodb = boto3.resource('dynamodb') table = dynamodb.Table('funcx-actions') From abc21557b95a0e092a24ba607951edc56c07c250 Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Mon, 24 May 2021 14:12:12 -0500 Subject: [PATCH 17/69] Disable use_offprocess_checker for status call too --- aws/action-status.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/aws/action-status.py b/aws/action-status.py index 307cc2b..8e28ad3 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -18,7 +18,8 @@ def lambda_handler(event, context): FuncXClient.TOKEN_DIR = '/tmp' fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, - openid_authorizer=openid_auth) + openid_authorizer=openid_auth, + use_offprocess_checker=False) action_id = event['pathParameters']['action-id'] From ae84c09cd4ed72267ad2646f67d114aa47327245 Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Wed, 23 Jun 2021 17:18:31 -0500 Subject: [PATCH 18/69] Use a slimmed down version of parsl --- .github/workflows/deploy_lambdas.yml | 5 +++ aws/action-status.py | 53 +++++++++++++++++++--------- slim-parsl/parsl/__init__.py | 0 slim-parsl/parsl/app/__init__.py | 0 slim-parsl/parsl/app/errors.py | 46 ++++++++++++++++++++++++ slim-parsl/requirements.txt | 3 ++ slim-parsl/setup.py | 31 ++++++++++++++++ 7 files changed, 121 insertions(+), 17 deletions(-) create mode 100644 slim-parsl/parsl/__init__.py create mode 100644 slim-parsl/parsl/app/__init__.py create mode 100644 slim-parsl/parsl/app/errors.py create mode 100644 slim-parsl/requirements.txt create mode 100644 slim-parsl/setup.py diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index b4a6d68..59bdda6 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -42,6 +42,11 @@ jobs: #install all dependencies as defined by requirements.txt in the aws directory pip3 install --use-deprecated=legacy-resolver -r aws/requirements.txt -t ./python + # There is a problem with the parsl library in lambda. We have a slimmed + # down version to replace it + rm -Rf python/parsl* + pip3 install --use-deprecated=legacy-resolver -t python ./slim-parsl + #zip files into current directory zip -r funcxLayer.zip ./python aws lambda publish-layer-version --layer-name FuncxLayer --zip-file fileb://./funcxLayer.zip --compatible-runtimes python3.6 python3.7 python3.8 > layer_info.json diff --git a/aws/action-status.py b/aws/action-status.py index 8e28ad3..896951c 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -3,9 +3,21 @@ from boto3.dynamodb.conditions import Key from globus_sdk import AccessTokenAuthorizer from funcx.sdk.client import FuncXClient +from funcx.utils.errors import TaskPending + +from funcx.sdk import VERSION as SDK_VERSION + + +class DecimalEncoder(json.JSONEncoder): + def default(self, obj): + if isinstance(obj, decimal.Decimal): + return int(obj) + return super(DecimalEncoder, self).default(obj) def lambda_handler(event, context): + print("---->", SDK_VERSION) + dynamodb = boto3.resource('dynamodb') table = dynamodb.Table('funcx-actions') print(event) @@ -32,11 +44,12 @@ def lambda_handler(event, context): action_record = response['Items'][0] print(action_record) + task_results = json.loads(action_record['tasks']) # Find the taskIDs where the results are not yet in running_tasks = list(filter(lambda task_id: bool(task_id), - [key if not action_record['tasks'][key][ + [key if not task_results[key][ 'result'] else None - for key in action_record['tasks'].keys()])) + for key in task_results.keys()])) failure = None if running_tasks: @@ -44,17 +57,18 @@ def lambda_handler(event, context): result = None try: result = fxc.get_result(task) - print("---->", result) + print("---->", result, type(result)) - except Exception as eek: - print("Faiulure ", type(eek), eek.args) - if str(eek) == 'waiting-for-ep': - result = None - else: - failure = eek + except TaskPending as eek: + print("Faiulure ", eek) + result = None + except Exception as eek2: + print("Detected an exception: ", eek2) + failure = str(eek2) + result = None if result: - action_record['tasks'][task]['result'] = result + task_results[task]['result'] = result update_response = table.update_item( Key={ @@ -62,23 +76,28 @@ def lambda_handler(event, context): }, UpdateExpression="set tasks=:t", ExpressionAttributeValues={ - ':t': action_record['tasks'] + ':t': json.dumps(task_results, cls=DecimalEncoder) }, ReturnValues="UPDATED_NEW" ) - print(update_response) - print(failure) + print("updated_response", update_response) + print("failure", failure) if failure: status = "FAILED" details = failure + display_status = failure else: status = "ACTIVE" - details = None + details = None else: status = "SUCCEEDED" - details = [str(action_record['tasks'][tt]['result']) for tt in - action_record['tasks'].keys()] + details = json.dumps(task_results) + display_status = "Function Results Received" + print("Success -> ", details) + + # details = [str(action_record['tasks'][tt]['result']) for tt in + # action_record['tasks'].keys()] result = { "action_id": action_id, @@ -87,7 +106,7 @@ def lambda_handler(event, context): 'details': details } - print(event) + print("Status result", result) return { 'statusCode': 200, 'body': json.dumps(result) diff --git a/slim-parsl/parsl/__init__.py b/slim-parsl/parsl/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/slim-parsl/parsl/app/__init__.py b/slim-parsl/parsl/app/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/slim-parsl/parsl/app/errors.py b/slim-parsl/parsl/app/errors.py new file mode 100644 index 0000000..9a56532 --- /dev/null +++ b/slim-parsl/parsl/app/errors.py @@ -0,0 +1,46 @@ +import dill +from six import reraise +from tblib import Traceback +from functools import wraps +from typing import Callable, Union, Any, TypeVar +from types import TracebackType +import logging +logger = logging.getLogger(__name__) + + +class RemoteExceptionWrapper: + def __init__(self, e_type: type, e_value: Exception, traceback: TracebackType) -> None: + + self.e_type = dill.dumps(e_type) + self.e_value = dill.dumps(e_value) + self.e_traceback = Traceback(traceback) + + def reraise(self) -> None: + + t = dill.loads(self.e_type) + + # the type is logged here before deserialising v and tb + # because occasionally there are problems deserialising the + # value (see #785, #548) and the fix is related to the + # specific exception type. + logger.debug("Reraising exception of type {}".format(t)) + + v = dill.loads(self.e_value) + tb = self.e_traceback.as_traceback() + + reraise(t, v, tb) + + +R = TypeVar('R') + + +def wrap_error(func: Callable[..., R]) -> Callable[..., Union[R, RemoteExceptionWrapper]]: + @wraps(func) # type: ignore + def wrapper(*args: object, **kwargs: object) -> Any: + import sys + from funcx.serialize.errors import RemoteExceptionWrapper + try: + return func(*args, **kwargs) # type: ignore + except Exception: + return RemoteExceptionWrapper(*sys.exc_info()) + return wrapper # type: ignore diff --git a/slim-parsl/requirements.txt b/slim-parsl/requirements.txt new file mode 100644 index 0000000..340b840 --- /dev/null +++ b/slim-parsl/requirements.txt @@ -0,0 +1,3 @@ +six +tblib +globus-sdk>=2.0 \ No newline at end of file diff --git a/slim-parsl/setup.py b/slim-parsl/setup.py new file mode 100644 index 0000000..7b625ae --- /dev/null +++ b/slim-parsl/setup.py @@ -0,0 +1,31 @@ +import os + +from setuptools import find_namespace_packages, setup + +version = "1.0" + +with open("requirements.txt") as f: + install_requires = f.readlines() + +setup( + name="slim-parsl", + version=version, + packages=find_namespace_packages(include=["parsl", "parsl.*"]), + description="Slim Parsl that will work with AWS Lambda", + install_requires=install_requires, + python_requires=">=3.6.0", + classifiers=[ + "Development Status :: 3 - Alpha", + "Intended Audience :: Science/Research", + "License :: OSI Approved :: Apache Software License", + "Natural Language :: English", + "Operating System :: OS Independent", + "Programming Language :: Python :: 3", + "Topic :: Scientific/Engineering", + ], + keywords=["funcX", "FaaS", "Function Serving"], + author="funcX team", + author_email="labs@globus.org", + license="Apache License, Version 2.0", + url="https://github.com/funcx-faas/funcx", +) From 43bed770592b2b76c8f80889ea6d9a09ba0dbf70 Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Thu, 24 Jun 2021 09:17:50 -0500 Subject: [PATCH 19/69] action record saved as JSON --- aws/funcx-run.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 67f036e..db11994 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -59,7 +59,7 @@ def lambda_handler(event, context): response = table.put_item( Item={ 'action-id': action_id, - 'tasks': {task_id: {"result": None} for task_id in batch_res} + 'tasks': json.dumps({task_id: {"result": None} for task_id in batch_res}) } ) print("Dynamo", response) From ba1d43ddde0dffcdf19779ad4d6c90194ba4cb87 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Sat, 10 Jul 2021 08:20:14 +1200 Subject: [PATCH 20/69] Remove dumps statement on result --- aws/action-status.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aws/action-status.py b/aws/action-status.py index 896951c..e77d90c 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -92,7 +92,7 @@ def lambda_handler(event, context): details = None else: status = "SUCCEEDED" - details = json.dumps(task_results) + details = task_results display_status = "Function Results Received" print("Success -> ", details) From de9226f700de9feec0992253f9d80ee039fe6a80 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 14 Jul 2021 06:52:35 +1200 Subject: [PATCH 21/69] Reverting to a list of results --- aws/action-status.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index e77d90c..a6606ce 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -96,8 +96,10 @@ def lambda_handler(event, context): display_status = "Function Results Received" print("Success -> ", details) - # details = [str(action_record['tasks'][tt]['result']) for tt in - # action_record['tasks'].keys()] + details = [action_record['tasks'][tt]['result'] for tt in + action_record['tasks'].keys()] + + print("List results ->", details) result = { "action_id": action_id, From dd247a328d3595f01733607c50fcbc19d3d1e649 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 14 Jul 2021 09:39:56 +1200 Subject: [PATCH 22/69] bugfix task_result instead of action_response --- aws/action-status.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index a6606ce..1d24616 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -96,8 +96,8 @@ def lambda_handler(event, context): display_status = "Function Results Received" print("Success -> ", details) - details = [action_record['tasks'][tt]['result'] for tt in - action_record['tasks'].keys()] + details = [task_results[tt]['result'] for tt in + task_results.keys()] print("List results ->", details) From 74a51a7b8d40f196f16b9e235b49bbf0fb9da43b Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 14 Jul 2021 09:58:23 +1200 Subject: [PATCH 23/69] Nest result in a result field. --- aws/action-status.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/aws/action-status.py b/aws/action-status.py index 1d24616..e158ff6 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -96,9 +96,11 @@ def lambda_handler(event, context): display_status = "Function Results Received" print("Success -> ", details) - details = [task_results[tt]['result'] for tt in + all_res = [task_results[tt]['result'] for tt in task_results.keys()] + details = {'result': all_res} + print("List results ->", details) result = { From f90b067bab12f7075826e5ca3aae97ef2b3fca2a Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Thu, 22 Jul 2021 05:50:55 +1200 Subject: [PATCH 24/69] Add decimal import --- aws/action-status.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/aws/action-status.py b/aws/action-status.py index e158ff6..cfa1f06 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -1,5 +1,7 @@ import json import boto3 +import decimal + from boto3.dynamodb.conditions import Key from globus_sdk import AccessTokenAuthorizer from funcx.sdk.client import FuncXClient From 1625cabd399d68365c1145e7d28b4e2ac0fa9b36 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 15 Sep 2021 07:26:43 +1200 Subject: [PATCH 25/69] Add second running_tasks check after polling task status --- aws/action-status.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/aws/action-status.py b/aws/action-status.py index cfa1f06..6d6f5fb 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -92,7 +92,13 @@ def lambda_handler(event, context): else: status = "ACTIVE" details = None - else: + + # Now check again to see if everything is done + running_tasks = list(filter(lambda task_id: bool(task_id), + [key if not task_results[key][ + 'result'] else None + for key in task_results.keys()])) + if running_tasks: status = "SUCCEEDED" details = task_results display_status = "Function Results Received" From 142d0e11d522c3c7d556a734408a8d0bedd11017 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 15 Sep 2021 08:25:50 +1200 Subject: [PATCH 26/69] Fix not running check --- aws/action-status.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aws/action-status.py b/aws/action-status.py index 6d6f5fb..3a2df11 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -98,7 +98,7 @@ def lambda_handler(event, context): [key if not task_results[key][ 'result'] else None for key in task_results.keys()])) - if running_tasks: + if not running_tasks: status = "SUCCEEDED" details = task_results display_status = "Function Results Received" From e6a4d517747bf15961166b463dedc871ca5b56f4 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Tue, 9 Nov 2021 08:21:39 +1300 Subject: [PATCH 27/69] Bump versions to 0.3.4 and include funcx-endpoint --- aws/requirements.txt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/aws/requirements.txt b/aws/requirements.txt index 9901c38..f2736e6 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -1,3 +1,4 @@ globus-sdk>=1.7.0 -funcx==0.2.3 +funcx==0.3.4 +funcx-endpoint==0.3.4 globus_automate_client From 95b6a9238720291831921e341e82a0eb0758d9e4 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Fri, 19 Nov 2021 11:38:19 +1300 Subject: [PATCH 28/69] Return stack trace rather than stringified exception --- aws/action-status.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index 3a2df11..5ab64ca 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -1,6 +1,7 @@ import json import boto3 import decimal +import traceback from boto3.dynamodb.conditions import Key from globus_sdk import AccessTokenAuthorizer @@ -64,9 +65,9 @@ def lambda_handler(event, context): except TaskPending as eek: print("Faiulure ", eek) result = None - except Exception as eek2: - print("Detected an exception: ", eek2) - failure = str(eek2) + except Exception: + failure = traceback.format_exc() + print("Detected an exception: ", failure) result = None if result: From b071418a5db957dffea11d81ab9e8c323583a348 Mon Sep 17 00:00:00 2001 From: Ben Galewsky Date: Fri, 19 Nov 2021 10:41:23 -0600 Subject: [PATCH 29/69] Don't publish to aws on PRs --- .github/workflows/deploy_lambdas.yml | 3 --- 1 file changed, 3 deletions(-) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index 59bdda6..ab1db37 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -6,9 +6,6 @@ on: #when there is a push to the main push: branches: [ serverless ] - #when there is a pull to the main - pull_request: - branches: [ serverless ] jobs: build: From 0db3593d33b52d0d42522c1852401fb190d2cc61 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Thu, 3 Mar 2022 08:18:19 +1300 Subject: [PATCH 30/69] Adds None as the result if none returned. --- aws/action-status.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index 3a2df11..ffe9ba6 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -60,17 +60,13 @@ def lambda_handler(event, context): try: result = fxc.get_result(task) print("---->", result, type(result)) - except TaskPending as eek: print("Faiulure ", eek) - result = None except Exception as eek2: print("Detected an exception: ", eek2) failure = str(eek2) - result = None - if result: - task_results[task]['result'] = result + task_results[task]['result'] = result update_response = table.update_item( Key={ From 822a95e6a70e32790f30df4999767f843644c360 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Thu, 3 Mar 2022 10:04:07 +1300 Subject: [PATCH 31/69] Changing to a completed flag to track running tasks --- aws/action-status.py | 12 ++++++++---- aws/funcx-run.py | 2 +- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index ffe9ba6..4e20442 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -50,23 +50,27 @@ def lambda_handler(event, context): # Find the taskIDs where the results are not yet in running_tasks = list(filter(lambda task_id: bool(task_id), [key if not task_results[key][ - 'result'] else None + 'completed'] else False for key in task_results.keys()])) failure = None if running_tasks: for task in running_tasks: result = None + completed = False try: result = fxc.get_result(task) print("---->", result, type(result)) + completed = True except TaskPending as eek: - print("Faiulure ", eek) + print("Failure ", eek) except Exception as eek2: print("Detected an exception: ", eek2) failure = str(eek2) - + completed = True + task_results[task]['result'] = result + task_results[task]['completed'] = completed update_response = table.update_item( Key={ @@ -92,7 +96,7 @@ def lambda_handler(event, context): # Now check again to see if everything is done running_tasks = list(filter(lambda task_id: bool(task_id), [key if not task_results[key][ - 'result'] else None + 'completed'] else False for key in task_results.keys()])) if not running_tasks: status = "SUCCEEDED" diff --git a/aws/funcx-run.py b/aws/funcx-run.py index db11994..c1cfa52 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -59,7 +59,7 @@ def lambda_handler(event, context): response = table.put_item( Item={ 'action-id': action_id, - 'tasks': json.dumps({task_id: {"result": None} for task_id in batch_res}) + 'tasks': json.dumps({task_id: {"result": None, "completed": False} for task_id in batch_res}) } ) print("Dynamo", response) From 852ea790becdfe3f2ce88dc47c3d5c47b14345b3 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Thu, 3 Mar 2022 11:40:38 +1300 Subject: [PATCH 32/69] Remove unused var --- aws/funcx-run.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index c1cfa52..7cb64ce 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -42,7 +42,7 @@ def lambda_handler(event, context): 'manage_by': manage_by, 'start_time': now_isoformat(), } - tasks = {} + batch = fxc.create_batch() for task in body['body']['tasks']: From 633ed312f6970817a79273fc733e7f253a13a9e1 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Tue, 5 Apr 2022 11:46:47 +1200 Subject: [PATCH 33/69] Adds error handling on submit and supports None payload. Adds log line for task ids and returns task ids as details --- aws/funcx-run.py | 29 +++++++++++++++++++++-------- 1 file changed, 21 insertions(+), 8 deletions(-) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 7cb64ce..86e0305 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -43,15 +43,28 @@ def lambda_handler(event, context): 'start_time': now_isoformat(), } - batch = fxc.create_batch() + status_code = 202 - for task in body['body']['tasks']: - print(task) - batch.add(endpoint_id=task['endpoint'], function_id=task['function'], - **task['payload']) + try: + batch = fxc.create_batch() - batch_res = fxc.batch_run(batch) - print(batch_res) + for task in body['body']['tasks']: + print(task) + payload = task.get('payload', None) + if payload: + batch.add(endpoint_id=task['endpoint'], function_id=task['function'], + **task['payload']) + else: + batch.add(endpoint_id=task['endpoint'], function_id=task['function']) + + batch_res = fxc.batch_run(batch) + print({'action_id': action_id, 'tasks': batch_res}) + result['details'] = batch_res + except Exception as eek: + result['status'] = 'FAILED' + result['display_status'] = 'Failed to submit tasks' + result['details'] = str(eek) + status_code = 400 # Create a dynamo record where the primary key is this action's ID # Tasks is a dict by task_id and contains the eventual results from their @@ -64,6 +77,6 @@ def lambda_handler(event, context): ) print("Dynamo", response) return { - 'statusCode': 202, + 'statusCode': status_code, 'body': json.dumps(result) } \ No newline at end of file From 23653f8126d7dc8fb67f16d9903a19a592bae088 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Tue, 12 Apr 2022 08:21:40 +1200 Subject: [PATCH 34/69] Fix result status message --- aws/action-status.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index 0644c31..5a1cf9d 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -66,7 +66,7 @@ def lambda_handler(event, context): print("Pending ", eek) except Exception as eek2: failure = traceback.format_exc() - print("Detected an exception: ", failure) + print("Detected an exception: ", eek2) completed = True task_results[task]['result'] = result @@ -83,12 +83,13 @@ def lambda_handler(event, context): ReturnValues="UPDATED_NEW" ) + display_status = "Function Active" print("updated_response", update_response) - print("failure", failure) if failure: + print("FAILED ", failure) status = "FAILED" details = failure - display_status = failure + display_status = "Function Failed" else: status = "ACTIVE" details = None @@ -114,7 +115,7 @@ def lambda_handler(event, context): result = { "action_id": action_id, 'status': status, - 'display_status': 'Function Results Received', + 'display_status': display_status, 'details': details } From bf84e7f830037a9a6eac495e8a5b5db00c1b6275 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Tue, 12 Apr 2022 08:39:06 +1200 Subject: [PATCH 35/69] Return the error message as the result --- aws/action-status.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index 5a1cf9d..b980e91 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -54,6 +54,9 @@ def lambda_handler(event, context): 'completed'] else False for key in task_results.keys()])) + status = "SUCCEEDED" + display_status = "Function Results Received" + failure = None if running_tasks: for task in running_tasks: @@ -66,6 +69,7 @@ def lambda_handler(event, context): print("Pending ", eek) except Exception as eek2: failure = traceback.format_exc() + result = failure print("Detected an exception: ", eek2) completed = True @@ -83,15 +87,14 @@ def lambda_handler(event, context): ReturnValues="UPDATED_NEW" ) - display_status = "Function Active" print("updated_response", update_response) if failure: print("FAILED ", failure) status = "FAILED" - details = failure display_status = "Function Failed" else: status = "ACTIVE" + display_status = "Function Active" details = None # Now check again to see if everything is done @@ -100,11 +103,6 @@ def lambda_handler(event, context): 'completed'] else False for key in task_results.keys()])) if not running_tasks: - status = "SUCCEEDED" - details = task_results - display_status = "Function Results Received" - print("Success -> ", details) - all_res = [task_results[tt]['result'] for tt in task_results.keys()] From c5305b66888228c7b6b93148ffece21e2ddb9f6b Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Fri, 22 Apr 2022 06:07:02 +1200 Subject: [PATCH 36/69] Logging submit error --- aws/funcx-run.py | 21 +++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 86e0305..87d55c4 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -45,6 +45,7 @@ def lambda_handler(event, context): status_code = 202 + batch_res = None try: batch = fxc.create_batch() @@ -61,6 +62,7 @@ def lambda_handler(event, context): print({'action_id': action_id, 'tasks': batch_res}) result['details'] = batch_res except Exception as eek: + print('FAILED ', eek) result['status'] = 'FAILED' result['display_status'] = 'Failed to submit tasks' result['details'] = str(eek) @@ -69,14 +71,17 @@ def lambda_handler(event, context): # Create a dynamo record where the primary key is this action's ID # Tasks is a dict by task_id and contains the eventual results from their # execution. Where there are no more None results then the action is complete - response = table.put_item( - Item={ - 'action-id': action_id, - 'tasks': json.dumps({task_id: {"result": None, "completed": False} for task_id in batch_res}) - } - ) - print("Dynamo", response) + if batch_res: + response = table.put_item( + Item={ + 'action-id': action_id, + 'tasks': json.dumps({task_id: {"result": None, "completed": False} for task_id in batch_res}) + } + ) + print("Dynamo", response) + + print("Status result", result) return { 'statusCode': status_code, 'body': json.dumps(result) - } \ No newline at end of file + } From 9861411ff3184ed54773e1e51a68c4473e246cd8 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Sat, 30 Apr 2022 08:28:41 +1200 Subject: [PATCH 37/69] Sets the task group id to the user id --- aws/action-status.py | 4 +++- aws/funcx-run.py | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index b980e91..ac4123c 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -31,9 +31,11 @@ def lambda_handler(event, context): openid_auth = AccessTokenAuthorizer( event['requestContext']['authorizer']['openid_token']) + user_id = event['requestContext']['authorizer']['user_id'] + FuncXClient.TOKEN_DIR = '/tmp' fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, - openid_authorizer=openid_auth, + openid_authorizer=openid_auth, task_group_id=user_id, use_offprocess_checker=False) action_id = event['pathParameters']['action-id'] diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 87d55c4..022c12d 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -19,9 +19,11 @@ def lambda_handler(event, context): openid_auth = AccessTokenAuthorizer( event['requestContext']['authorizer']['openid_token']) + user_id = event['requestContext']['authorizer']['user_id'] + FuncXClient.TOKEN_DIR = '/tmp' fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, - openid_authorizer=openid_auth, + openid_authorizer=openid_auth, task_group_id=user_id, use_offprocess_checker=False) dynamodb = boto3.resource('dynamodb') From 94b693c9a0517847c451af60151da0b154af5a4f Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Sat, 30 Apr 2022 09:51:43 +1200 Subject: [PATCH 38/69] Update reqs --- aws/requirements.txt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/aws/requirements.txt b/aws/requirements.txt index f2736e6..7afeb4a 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -1,4 +1,4 @@ globus-sdk>=1.7.0 -funcx==0.3.4 -funcx-endpoint==0.3.4 +funcx==0.3.9 +funcx-endpoint==0.3.9 globus_automate_client From c3bdea718f6d52ee814728497e42585f6b777079 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Tue, 17 May 2022 10:42:25 +1200 Subject: [PATCH 39/69] Update lambda layer on deploy --- .github/workflows/deploy_lambdas.yml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index ab1db37..a59d9df 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -49,6 +49,7 @@ jobs: aws lambda publish-layer-version --layer-name FuncxLayer --zip-file fileb://./funcxLayer.zip --compatible-runtimes python3.6 python3.7 python3.8 > layer_info.json cat layer_info.json LAYER_VERSION=$(jq ".Version" layer_info.json) + LAYER_VERSION_ARN=$(jq ".LayerVersionArn" layer_info.json) - name: Upload Globus Auth Function run: | @@ -56,6 +57,7 @@ jobs: zip funcx-globus-auth.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name funcx-globus-auth --zip-file fileb://./funcx-globus-auth.zip + aws lambda update-function-configuration --function-name funcx-globus-auth --layers $LAYER_VERSION_ARN - name: Upload Action Introspect Function run: | @@ -63,6 +65,7 @@ jobs: zip action_introspect.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name action_introspect --zip-file fileb://./action_introspect.zip + aws lambda update-function-configuration --function-name action_introspect --layers $LAYER_VERSION_ARN - name: Upload Action Run Function run: | @@ -70,6 +73,7 @@ jobs: zip funcx-run.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name funcx-run --zip-file fileb://./funcx-run.zip + aws lambda update-function-configuration --function-name funcx-run --layers $LAYER_VERSION_ARN - name: Upload Action Status Function run: | @@ -77,4 +81,5 @@ jobs: zip action-status.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name action_status --zip-file fileb://./action-status.zip + aws lambda update-function-configuration --function-name action_status --layers $LAYER_VERSION_ARN From 4443572f8bd1eab47403f57d6ebeb9126fa5c028 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 18 May 2022 09:57:43 +1200 Subject: [PATCH 40/69] Remove layer deployment from globus_auth lambda --- .github/workflows/deploy_lambdas.yml | 1 - 1 file changed, 1 deletion(-) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index a59d9df..8135558 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -57,7 +57,6 @@ jobs: zip funcx-globus-auth.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name funcx-globus-auth --zip-file fileb://./funcx-globus-auth.zip - aws lambda update-function-configuration --function-name funcx-globus-auth --layers $LAYER_VERSION_ARN - name: Upload Action Introspect Function run: | From 544a0fa4355f37e3b9fc1d58c23c3433b25e8a73 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 18 May 2022 10:05:10 +1200 Subject: [PATCH 41/69] Add wait for function to deployment --- .github/workflows/deploy_lambdas.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index 8135558..f2edc6e 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -64,6 +64,7 @@ jobs: zip action_introspect.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name action_introspect --zip-file fileb://./action_introspect.zip + aws lambda function-updated --function-name action_introspect aws lambda update-function-configuration --function-name action_introspect --layers $LAYER_VERSION_ARN - name: Upload Action Run Function @@ -72,6 +73,7 @@ jobs: zip funcx-run.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name funcx-run --zip-file fileb://./funcx-run.zip + aws lambda function-updated --function-name funcx-run aws lambda update-function-configuration --function-name funcx-run --layers $LAYER_VERSION_ARN - name: Upload Action Status Function @@ -80,5 +82,6 @@ jobs: zip action-status.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name action_status --zip-file fileb://./action-status.zip + aws lambda function-updated --function-name action_status aws lambda update-function-configuration --function-name action_status --layers $LAYER_VERSION_ARN From 0e31beddfdaec39fcd6c7241abc2b8101785ded7 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 18 May 2022 10:07:48 +1200 Subject: [PATCH 42/69] Add wait cli command to deployment --- .github/workflows/deploy_lambdas.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index f2edc6e..cc52d91 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -64,7 +64,7 @@ jobs: zip action_introspect.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name action_introspect --zip-file fileb://./action_introspect.zip - aws lambda function-updated --function-name action_introspect + aws lambda wait function-updated --function-name action_introspect aws lambda update-function-configuration --function-name action_introspect --layers $LAYER_VERSION_ARN - name: Upload Action Run Function @@ -73,7 +73,7 @@ jobs: zip funcx-run.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name funcx-run --zip-file fileb://./funcx-run.zip - aws lambda function-updated --function-name funcx-run + aws lambda wait function-updated --function-name funcx-run aws lambda update-function-configuration --function-name funcx-run --layers $LAYER_VERSION_ARN - name: Upload Action Status Function @@ -82,6 +82,6 @@ jobs: zip action-status.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name action_status --zip-file fileb://./action-status.zip - aws lambda function-updated --function-name action_status + aws lambda wait function-updated --function-name action_status aws lambda update-function-configuration --function-name action_status --layers $LAYER_VERSION_ARN From 946a9ff2247d1f0ece858eac067511ef72cda484 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 18 May 2022 10:17:06 +1200 Subject: [PATCH 43/69] Add debug logs to build --- .github/workflows/deploy_lambdas.yml | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index cc52d91..1492aa4 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -64,8 +64,10 @@ jobs: zip action_introspect.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name action_introspect --zip-file fileb://./action_introspect.zip - aws lambda wait function-updated --function-name action_introspect - aws lambda update-function-configuration --function-name action_introspect --layers $LAYER_VERSION_ARN + aws lambda wait function-updated --function-name action_introspect > wait.json + cat wait.json + aws lambda update-function-configuration --function-name action_introspect --layers $LAYER_VERSION_ARN > update_info.json + cat update_info.json - name: Upload Action Run Function run: | From 68bd98a7a6d9b1ebee0a95d25b83a07d387b4509 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 18 May 2022 10:33:52 +1200 Subject: [PATCH 44/69] debugging --- .github/workflows/deploy_lambdas.yml | 45 ++++++++++++++-------------- 1 file changed, 23 insertions(+), 22 deletions(-) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index 1492aa4..f73f5d2 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -51,12 +51,12 @@ jobs: LAYER_VERSION=$(jq ".Version" layer_info.json) LAYER_VERSION_ARN=$(jq ".LayerVersionArn" layer_info.json) - - name: Upload Globus Auth Function - run: | - cp aws/funcx-globus-auth.py ./lambda_function.py - zip funcx-globus-auth.zip ./lambda_function.py - rm ./lambda_function.py - aws lambda update-function-code --function-name funcx-globus-auth --zip-file fileb://./funcx-globus-auth.zip + # - name: Upload Globus Auth Function + # run: | + # cp aws/funcx-globus-auth.py ./lambda_function.py + # zip funcx-globus-auth.zip ./lambda_function.py + # rm ./lambda_function.py + # aws lambda update-function-code --function-name funcx-globus-auth --zip-file fileb://./funcx-globus-auth.zip - name: Upload Action Introspect Function run: | @@ -66,24 +66,25 @@ jobs: aws lambda update-function-code --function-name action_introspect --zip-file fileb://./action_introspect.zip aws lambda wait function-updated --function-name action_introspect > wait.json cat wait.json + echo $LAYER_VERSION_ARN aws lambda update-function-configuration --function-name action_introspect --layers $LAYER_VERSION_ARN > update_info.json cat update_info.json - - name: Upload Action Run Function - run: | - cp aws/funcx-run.py ./lambda_function.py - zip funcx-run.zip ./lambda_function.py - rm ./lambda_function.py - aws lambda update-function-code --function-name funcx-run --zip-file fileb://./funcx-run.zip - aws lambda wait function-updated --function-name funcx-run - aws lambda update-function-configuration --function-name funcx-run --layers $LAYER_VERSION_ARN + # - name: Upload Action Run Function + # run: | + # cp aws/funcx-run.py ./lambda_function.py + # zip funcx-run.zip ./lambda_function.py + # rm ./lambda_function.py + # aws lambda update-function-code --function-name funcx-run --zip-file fileb://./funcx-run.zip + # aws lambda wait function-updated --function-name funcx-run + # aws lambda update-function-configuration --function-name funcx-run --layers $LAYER_VERSION_ARN - - name: Upload Action Status Function - run: | - cp aws/action-status.py ./lambda_function.py - zip action-status.zip ./lambda_function.py - rm ./lambda_function.py - aws lambda update-function-code --function-name action_status --zip-file fileb://./action-status.zip - aws lambda wait function-updated --function-name action_status - aws lambda update-function-configuration --function-name action_status --layers $LAYER_VERSION_ARN + # - name: Upload Action Status Function + # run: | + # cp aws/action-status.py ./lambda_function.py + # zip action-status.zip ./lambda_function.py + # rm ./lambda_function.py + # aws lambda update-function-code --function-name action_status --zip-file fileb://./action-status.zip + # aws lambda wait function-updated --function-name action_status + # aws lambda update-function-configuration --function-name action_status --layers $LAYER_VERSION_ARN From f453832d3b170a8f64ddbdaf810414bd408bdb39 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 18 May 2022 10:38:13 +1200 Subject: [PATCH 45/69] Add state between steps --- .github/workflows/deploy_lambdas.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index f73f5d2..7651876 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -49,7 +49,8 @@ jobs: aws lambda publish-layer-version --layer-name FuncxLayer --zip-file fileb://./funcxLayer.zip --compatible-runtimes python3.6 python3.7 python3.8 > layer_info.json cat layer_info.json LAYER_VERSION=$(jq ".Version" layer_info.json) - LAYER_VERSION_ARN=$(jq ".LayerVersionArn" layer_info.json) + export LAYER_VERSION_ARN=$(jq ".LayerVersionArn" layer_info.json) + echo "::set-env name=LAYER_VERSION_ARN::$LAYER_VERSION_ARN" # - name: Upload Globus Auth Function # run: | From 0284786c7394c364b845df63a7b507ca3422161c Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 18 May 2022 10:43:28 +1200 Subject: [PATCH 46/69] Update github env var usage --- .github/workflows/deploy_lambdas.yml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index 7651876..232f1a9 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -49,8 +49,8 @@ jobs: aws lambda publish-layer-version --layer-name FuncxLayer --zip-file fileb://./funcxLayer.zip --compatible-runtimes python3.6 python3.7 python3.8 > layer_info.json cat layer_info.json LAYER_VERSION=$(jq ".Version" layer_info.json) - export LAYER_VERSION_ARN=$(jq ".LayerVersionArn" layer_info.json) - echo "::set-env name=LAYER_VERSION_ARN::$LAYER_VERSION_ARN" + LAYER_VERSION_ARN=$(jq ".LayerVersionArn" layer_info.json) + echo "LAYER_VERSION_ARN=$LAYER_VERSION_ARN" >> $GITHUB_ENV # - name: Upload Globus Auth Function # run: | @@ -67,8 +67,8 @@ jobs: aws lambda update-function-code --function-name action_introspect --zip-file fileb://./action_introspect.zip aws lambda wait function-updated --function-name action_introspect > wait.json cat wait.json - echo $LAYER_VERSION_ARN - aws lambda update-function-configuration --function-name action_introspect --layers $LAYER_VERSION_ARN > update_info.json + echo ${{ env.action_state }} + aws lambda update-function-configuration --function-name action_introspect --layers ${{ env.action_state }} > update_info.json cat update_info.json # - name: Upload Action Run Function From 464f7cb450a5c7a08c731a599072a846cd734a9d Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 18 May 2022 10:48:11 +1200 Subject: [PATCH 47/69] Typo --- .github/workflows/deploy_lambdas.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index 232f1a9..0efaf1a 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -67,8 +67,8 @@ jobs: aws lambda update-function-code --function-name action_introspect --zip-file fileb://./action_introspect.zip aws lambda wait function-updated --function-name action_introspect > wait.json cat wait.json - echo ${{ env.action_state }} - aws lambda update-function-configuration --function-name action_introspect --layers ${{ env.action_state }} > update_info.json + echo ${{ env.LAYER_VERSION_ARN }} + aws lambda update-function-configuration --function-name action_introspect --layers ${{ env.LAYER_VERSION_ARN }} > update_info.json cat update_info.json # - name: Upload Action Run Function From f038fd48b16699f3b79d252836ac70519c78fa55 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 18 May 2022 11:05:05 +1200 Subject: [PATCH 48/69] Clean up debugging and set deployment for all actions --- .github/workflows/deploy_lambdas.yml | 48 ++++++++++++---------------- 1 file changed, 21 insertions(+), 27 deletions(-) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index 0efaf1a..d976c2f 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -52,12 +52,12 @@ jobs: LAYER_VERSION_ARN=$(jq ".LayerVersionArn" layer_info.json) echo "LAYER_VERSION_ARN=$LAYER_VERSION_ARN" >> $GITHUB_ENV - # - name: Upload Globus Auth Function - # run: | - # cp aws/funcx-globus-auth.py ./lambda_function.py - # zip funcx-globus-auth.zip ./lambda_function.py - # rm ./lambda_function.py - # aws lambda update-function-code --function-name funcx-globus-auth --zip-file fileb://./funcx-globus-auth.zip + - name: Upload Globus Auth Function + run: | + cp aws/funcx-globus-auth.py ./lambda_function.py + zip funcx-globus-auth.zip ./lambda_function.py + rm ./lambda_function.py + aws lambda update-function-code --function-name funcx-globus-auth --zip-file fileb://./funcx-globus-auth.zip - name: Upload Action Introspect Function run: | @@ -65,27 +65,21 @@ jobs: zip action_introspect.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name action_introspect --zip-file fileb://./action_introspect.zip - aws lambda wait function-updated --function-name action_introspect > wait.json - cat wait.json - echo ${{ env.LAYER_VERSION_ARN }} - aws lambda update-function-configuration --function-name action_introspect --layers ${{ env.LAYER_VERSION_ARN }} > update_info.json - cat update_info.json + aws lambda update-function-configuration --function-name action_introspect --layers ${{ env.LAYER_VERSION_ARN }} - # - name: Upload Action Run Function - # run: | - # cp aws/funcx-run.py ./lambda_function.py - # zip funcx-run.zip ./lambda_function.py - # rm ./lambda_function.py - # aws lambda update-function-code --function-name funcx-run --zip-file fileb://./funcx-run.zip - # aws lambda wait function-updated --function-name funcx-run - # aws lambda update-function-configuration --function-name funcx-run --layers $LAYER_VERSION_ARN + - name: Upload Action Run Function + run: | + cp aws/funcx-run.py ./lambda_function.py + zip funcx-run.zip ./lambda_function.py + rm ./lambda_function.py + aws lambda update-function-code --function-name funcx-run --zip-file fileb://./funcx-run.zip + aws lambda update-function-configuration --function-name funcx-run --layers ${{ env.LAYER_VERSION_ARN }} - # - name: Upload Action Status Function - # run: | - # cp aws/action-status.py ./lambda_function.py - # zip action-status.zip ./lambda_function.py - # rm ./lambda_function.py - # aws lambda update-function-code --function-name action_status --zip-file fileb://./action-status.zip - # aws lambda wait function-updated --function-name action_status - # aws lambda update-function-configuration --function-name action_status --layers $LAYER_VERSION_ARN + - name: Upload Action Status Function + run: | + cp aws/action-status.py ./lambda_function.py + zip action-status.zip ./lambda_function.py + rm ./lambda_function.py + aws lambda update-function-code --function-name action_status --zip-file fileb://./action-status.zip + aws lambda update-function-configuration --function-name action_status --layers ${{ env.LAYER_VERSION_ARN }} From 62c5196d5be6c9f0e3744e0ef6a4148651a34915 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 18 May 2022 11:07:07 +1200 Subject: [PATCH 49/69] Add wait steps back into deployment --- .github/workflows/deploy_lambdas.yml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.github/workflows/deploy_lambdas.yml b/.github/workflows/deploy_lambdas.yml index d976c2f..8d102df 100644 --- a/.github/workflows/deploy_lambdas.yml +++ b/.github/workflows/deploy_lambdas.yml @@ -65,6 +65,7 @@ jobs: zip action_introspect.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name action_introspect --zip-file fileb://./action_introspect.zip + aws lambda wait function-updated --function-name action_introspect aws lambda update-function-configuration --function-name action_introspect --layers ${{ env.LAYER_VERSION_ARN }} - name: Upload Action Run Function @@ -73,6 +74,7 @@ jobs: zip funcx-run.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name funcx-run --zip-file fileb://./funcx-run.zip + aws lambda wait function-updated --function-name funcx-run aws lambda update-function-configuration --function-name funcx-run --layers ${{ env.LAYER_VERSION_ARN }} - name: Upload Action Status Function @@ -81,5 +83,6 @@ jobs: zip action-status.zip ./lambda_function.py rm ./lambda_function.py aws lambda update-function-code --function-name action_status --zip-file fileb://./action-status.zip + aws lambda wait function-updated --function-name action_status aws lambda update-function-configuration --function-name action_status --layers ${{ env.LAYER_VERSION_ARN }} From 9a15a24dc1ec49549f250005666ed66ee13b2d4d Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 18 May 2022 11:15:29 +1200 Subject: [PATCH 50/69] Update requirements for environment --- aws/requirements.txt | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/aws/requirements.txt b/aws/requirements.txt index f2736e6..e5640bb 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -1,4 +1,4 @@ -globus-sdk>=1.7.0 -funcx==0.3.4 -funcx-endpoint==0.3.4 +globus-sdk>=3.0.0 +funcx==0.3.9 +funcx-endpoint==0.3.9 globus_automate_client From 30f42f052279aa61c2362746b6fda44a7b2ddc33 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Thu, 26 May 2022 06:34:18 +1200 Subject: [PATCH 51/69] Fix status active overwrite bug --- aws/action-status.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/aws/action-status.py b/aws/action-status.py index ac4123c..ab67428 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -94,7 +94,8 @@ def lambda_handler(event, context): print("FAILED ", failure) status = "FAILED" display_status = "Function Failed" - else: + + if not completed: status = "ACTIVE" display_status = "Function Active" details = None From 0429f3b15bdbd4b11c7ad7f8c4af177ac6748698 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 1 Jun 2022 06:48:09 +1200 Subject: [PATCH 52/69] Add two-week TTLs to the dynamo record --- aws/action-status.py | 6 ++++-- aws/funcx-run.py | 7 +++++-- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index ab67428..53c7b4d 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -1,6 +1,7 @@ import json import boto3 import decimal +import datetime import traceback from boto3.dynamodb.conditions import Key @@ -82,9 +83,10 @@ def lambda_handler(event, context): Key={ 'action-id': action_id }, - UpdateExpression="set tasks=:t", + UpdateExpression="set tasks=:t, ttl=:l", ExpressionAttributeValues={ - ':t': json.dumps(task_results, cls=DecimalEncoder) + ':t': json.dumps(task_results, cls=DecimalEncoder), + ':l': int(datetime.datetime.now().timestamp()) + 1209600 }, ReturnValues="UPDATED_NEW" ) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 022c12d..c62732b 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -72,12 +72,15 @@ def lambda_handler(event, context): # Create a dynamo record where the primary key is this action's ID # Tasks is a dict by task_id and contains the eventual results from their - # execution. Where there are no more None results then the action is complete + # execution. Where there are no more None results then the action is complete. + # Set a TTL for two weeks from now. + if batch_res: response = table.put_item( Item={ 'action-id': action_id, - 'tasks': json.dumps({task_id: {"result": None, "completed": False} for task_id in batch_res}) + 'tasks': json.dumps({task_id: {"result": None, "completed": False} for task_id in batch_res}), + 'ttl': int(datetime.datetime.now().timestamp()) + 1209600 } ) print("Dynamo", response) From 2aa2475b176c855f6156850e1a8bcd1f11ddba9d Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 29 Jun 2022 11:27:25 +1200 Subject: [PATCH 53/69] Rename ttl to fx_ttl --- aws/action-status.py | 2 +- aws/funcx-run.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index 53c7b4d..dccad4d 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -83,7 +83,7 @@ def lambda_handler(event, context): Key={ 'action-id': action_id }, - UpdateExpression="set tasks=:t, ttl=:l", + UpdateExpression="set tasks=:t, fx_ttl=:l", ExpressionAttributeValues={ ':t': json.dumps(task_results, cls=DecimalEncoder), ':l': int(datetime.datetime.now().timestamp()) + 1209600 diff --git a/aws/funcx-run.py b/aws/funcx-run.py index c62732b..71d2a46 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -80,7 +80,7 @@ def lambda_handler(event, context): Item={ 'action-id': action_id, 'tasks': json.dumps({task_id: {"result": None, "completed": False} for task_id in batch_res}), - 'ttl': int(datetime.datetime.now().timestamp()) + 1209600 + 'fx_ttl': int(datetime.datetime.now().timestamp()) + 1209600 } ) print("Dynamo", response) From 7f75775d00d24c1ee47b40633f4d612efc7d0d90 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 17 Aug 2022 09:10:27 +1200 Subject: [PATCH 54/69] Bump to 1.0.0 --- aws/requirements.txt | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/aws/requirements.txt b/aws/requirements.txt index e5640bb..1f9fa9b 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -1,4 +1,4 @@ globus-sdk>=3.0.0 -funcx==0.3.9 -funcx-endpoint==0.3.9 +funcx>=1.0.0 +funcx-endpoint>=1.0.0 globus_automate_client From b1df91cc15ba26046ee0e6c832cd78daae3d66ea Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 17 Aug 2022 09:31:56 +1200 Subject: [PATCH 55/69] Change funcx home to work with 1.0.0 and use /tmp for lambda --- aws/action-status.py | 4 ++-- aws/funcx-run.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index dccad4d..cca0d72 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -34,10 +34,10 @@ def lambda_handler(event, context): user_id = event['requestContext']['authorizer']['user_id'] - FuncXClient.TOKEN_DIR = '/tmp' + home_dir = '/tmp/funcx' fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, openid_authorizer=openid_auth, task_group_id=user_id, - use_offprocess_checker=False) + use_offprocess_checker=False, funcx_home=home_dir) action_id = event['pathParameters']['action-id'] diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 71d2a46..deebb0a 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -21,10 +21,10 @@ def lambda_handler(event, context): user_id = event['requestContext']['authorizer']['user_id'] - FuncXClient.TOKEN_DIR = '/tmp' + home_dir = '/tmp/funcx' fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, openid_authorizer=openid_auth, task_group_id=user_id, - use_offprocess_checker=False) + use_offprocess_checker=False, funcx_home=home_dir) dynamodb = boto3.resource('dynamodb') table = dynamodb.Table('funcx-actions') From 1421045f301771bc1ddedd0c0579207f6cf35e3a Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 17 Aug 2022 09:56:34 +1200 Subject: [PATCH 56/69] Changing token storage _home() function path --- aws/action-status.py | 4 +++- aws/funcx-run.py | 4 ++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/aws/action-status.py b/aws/action-status.py index cca0d72..565ffad 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -10,7 +10,8 @@ from funcx.utils.errors import TaskPending from funcx.sdk import VERSION as SDK_VERSION - +import pathlib +from funcx.sdk.login_manager import tokenstore class DecimalEncoder(json.JSONEncoder): def default(self, obj): @@ -35,6 +36,7 @@ def lambda_handler(event, context): user_id = event['requestContext']['authorizer']['user_id'] home_dir = '/tmp/funcx' + tokenstore._home = lambda: pathlib.PurePath(home_dir) fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, openid_authorizer=openid_auth, task_group_id=user_id, use_offprocess_checker=False, funcx_home=home_dir) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index deebb0a..72b0951 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -4,6 +4,8 @@ import boto3 import uuid import datetime +import pathlib +from funcx.sdk.login_manager import tokenstore def now_isoformat(): @@ -22,6 +24,8 @@ def lambda_handler(event, context): user_id = event['requestContext']['authorizer']['user_id'] home_dir = '/tmp/funcx' + + tokenstore._home = lambda: pathlib.PurePath(home_dir) fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, openid_authorizer=openid_auth, task_group_id=user_id, use_offprocess_checker=False, funcx_home=home_dir) From f63107bf08a699079ce0d5666d292ef76db9c059 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 17 Aug 2022 10:01:40 +1200 Subject: [PATCH 57/69] Change to Path over PurePath for exists() to work. --- aws/action-status.py | 2 +- aws/funcx-run.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index 565ffad..a0dd998 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -36,7 +36,7 @@ def lambda_handler(event, context): user_id = event['requestContext']['authorizer']['user_id'] home_dir = '/tmp/funcx' - tokenstore._home = lambda: pathlib.PurePath(home_dir) + tokenstore._home = lambda: pathlib.Path(home_dir) fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, openid_authorizer=openid_auth, task_group_id=user_id, use_offprocess_checker=False, funcx_home=home_dir) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 72b0951..e8e7c2e 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -25,7 +25,7 @@ def lambda_handler(event, context): home_dir = '/tmp/funcx' - tokenstore._home = lambda: pathlib.PurePath(home_dir) + tokenstore._home = lambda: pathlib.Path(home_dir) fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, openid_authorizer=openid_auth, task_group_id=user_id, use_offprocess_checker=False, funcx_home=home_dir) From 0450340cf14cb3fa26d4e8241e6a19cb13a8d1bb Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 17 Aug 2022 11:07:18 +1200 Subject: [PATCH 58/69] Bump endpoint version to 1.0.1 --- aws/requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aws/requirements.txt b/aws/requirements.txt index 1f9fa9b..bc1f930 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -1,4 +1,4 @@ globus-sdk>=3.0.0 funcx>=1.0.0 -funcx-endpoint>=1.0.0 +funcx-endpoint>=1.0.1 globus_automate_client From cb7bd4c1ddd44bb7833260426c10a33e9df33b66 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 17 Aug 2022 12:18:06 +1200 Subject: [PATCH 59/69] Use custom login manager to manage auth tokens --- aws/funcx-run.py | 25 +++++++++++++++++++++---- 1 file changed, 21 insertions(+), 4 deletions(-) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index e8e7c2e..6147532 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -1,11 +1,24 @@ import json from funcx.sdk.client import FuncXClient -from globus_sdk import AccessTokenAuthorizer +from globus_sdk import AccessTokenAuthorizer, RefreshTokenAuthorizer import boto3 import uuid import datetime import pathlib -from funcx.sdk.login_manager import tokenstore +from funcx.sdk.login_manager import tokenstore, LoginManager + + +class fxLoginManager(LoginManager): + def __init__(self, authorizers, environment=None): + self.authorizers = authorizers + home_dir = '/tmp/funcx' + tokenstore._home = lambda: pathlib.Path(home_dir) + self._token_storage = tokenstore.get_token_storage_adapter(environment=environment) + + def _get_authorizer( + self, resource_server: str + ) -> globus_sdk.RefreshTokenAuthorizer: + return self.authorizers[resource_server] def now_isoformat(): @@ -26,8 +39,12 @@ def lambda_handler(event, context): home_dir = '/tmp/funcx' tokenstore._home = lambda: pathlib.Path(home_dir) - fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, - openid_authorizer=openid_auth, task_group_id=user_id, + + fxmanager = fxLoginManager(authorizers={'funcx_service': auth, + 'search.api.globus.org/all': search_auth, + 'openid': openid_auth}) + + fxc = FuncXClient(login_manager=fxmanager, task_group_id=user_id, use_offprocess_checker=False, funcx_home=home_dir) dynamodb = boto3.resource('dynamodb') From f5d326290f71e1e2a8340c9111b8142721b581de Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 17 Aug 2022 12:20:15 +1200 Subject: [PATCH 60/69] Add login manager to action status --- aws/action-status.py | 25 +++++++++++++++++++++---- 1 file changed, 21 insertions(+), 4 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index a0dd998..dcfb9e2 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -5,13 +5,13 @@ import traceback from boto3.dynamodb.conditions import Key -from globus_sdk import AccessTokenAuthorizer +from globus_sdk import AccessTokenAuthorizer, RefreshTokenAuthorizer from funcx.sdk.client import FuncXClient from funcx.utils.errors import TaskPending from funcx.sdk import VERSION as SDK_VERSION import pathlib -from funcx.sdk.login_manager import tokenstore +from funcx.sdk.login_manager import tokenstore, LoginManager class DecimalEncoder(json.JSONEncoder): def default(self, obj): @@ -20,6 +20,19 @@ def default(self, obj): return super(DecimalEncoder, self).default(obj) +class fxLoginManager(LoginManager): + def __init__(self, authorizers, environment=None): + self.authorizers = authorizers + home_dir = '/tmp/funcx' + tokenstore._home = lambda: pathlib.Path(home_dir) + self._token_storage = tokenstore.get_token_storage_adapter(environment=environment) + + def _get_authorizer( + self, resource_server: str + ) -> globus_sdk.RefreshTokenAuthorizer: + return self.authorizers[resource_server] + + def lambda_handler(event, context): print("---->", SDK_VERSION) @@ -37,8 +50,12 @@ def lambda_handler(event, context): home_dir = '/tmp/funcx' tokenstore._home = lambda: pathlib.Path(home_dir) - fxc = FuncXClient(fx_authorizer=auth, search_authorizer=search_auth, - openid_authorizer=openid_auth, task_group_id=user_id, + + fxmanager = fxLoginManager(authorizers={'funcx_service': auth, + 'search.api.globus.org/all': search_auth, + 'openid': openid_auth}) + + fxc = FuncXClient(login_manager=fxmanager, task_group_id=user_id, use_offprocess_checker=False, funcx_home=home_dir) action_id = event['pathParameters']['action-id'] From 469e2eb596fac1f09b7000fb6e9902735ab99a0a Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 17 Aug 2022 12:26:03 +1200 Subject: [PATCH 61/69] Add globus sdk dependencies --- aws/action-status.py | 1 + aws/funcx-run.py | 1 + 2 files changed, 2 insertions(+) diff --git a/aws/action-status.py b/aws/action-status.py index dcfb9e2..f463783 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -3,6 +3,7 @@ import decimal import datetime import traceback +import globus_sdk from boto3.dynamodb.conditions import Key from globus_sdk import AccessTokenAuthorizer, RefreshTokenAuthorizer diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 6147532..1d79b94 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -5,6 +5,7 @@ import uuid import datetime import pathlib +import globus_sdk from funcx.sdk.login_manager import tokenstore, LoginManager From 2060b556978aa45c2d6122c31419d178d2b637d6 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 17 Aug 2022 12:34:10 +1200 Subject: [PATCH 62/69] Import TaskPending exception --- aws/action-status.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/aws/action-status.py b/aws/action-status.py index f463783..203960e 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -8,7 +8,14 @@ from boto3.dynamodb.conditions import Key from globus_sdk import AccessTokenAuthorizer, RefreshTokenAuthorizer from funcx.sdk.client import FuncXClient -from funcx.utils.errors import TaskPending +# from funcx.utils.errors import TaskPending + +from funcx.errors import ( + FuncxTaskExecutionFailed, + SerializationError, + TaskPending, + handle_response_errors, +) from funcx.sdk import VERSION as SDK_VERSION import pathlib From 014c67bbe74712f26be5719891444f32c1fc7de8 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Wed, 17 Aug 2022 12:41:10 +1200 Subject: [PATCH 63/69] Import new version --- aws/action-status.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index 203960e..ebe4bd8 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -8,7 +8,6 @@ from boto3.dynamodb.conditions import Key from globus_sdk import AccessTokenAuthorizer, RefreshTokenAuthorizer from funcx.sdk.client import FuncXClient -# from funcx.utils.errors import TaskPending from funcx.errors import ( FuncxTaskExecutionFailed, @@ -17,7 +16,7 @@ handle_response_errors, ) -from funcx.sdk import VERSION as SDK_VERSION +from funcx.version import __version__ import pathlib from funcx.sdk.login_manager import tokenstore, LoginManager @@ -42,7 +41,7 @@ def _get_authorizer( def lambda_handler(event, context): - print("---->", SDK_VERSION) + print("---->", __version__) dynamodb = boto3.resource('dynamodb') table = dynamodb.Table('funcx-actions') From 789b788468463768b3b22a4fe74d4d3c6c9c7df1 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Tue, 1 Nov 2022 11:00:56 +1300 Subject: [PATCH 64/69] Move details definition and default to None --- aws/action-status.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/aws/action-status.py b/aws/action-status.py index ebe4bd8..e6b4938 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -85,6 +85,7 @@ def lambda_handler(event, context): status = "SUCCEEDED" display_status = "Function Results Received" + details = None failure = None if running_tasks: @@ -126,7 +127,7 @@ def lambda_handler(event, context): if not completed: status = "ACTIVE" display_status = "Function Active" - details = None + # Now check again to see if everything is done running_tasks = list(filter(lambda task_id: bool(task_id), From 346fb00c9da4432536db852b3f78eca5ae4e44e0 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Thu, 30 Mar 2023 10:18:19 +1300 Subject: [PATCH 65/69] Update version and remove task group --- aws/action-status.py | 3 +-- aws/funcx-run.py | 3 +-- aws/requirements.txt | 4 ++-- 3 files changed, 4 insertions(+), 6 deletions(-) diff --git a/aws/action-status.py b/aws/action-status.py index e6b4938..edcdb1a 100644 --- a/aws/action-status.py +++ b/aws/action-status.py @@ -62,8 +62,7 @@ def lambda_handler(event, context): 'search.api.globus.org/all': search_auth, 'openid': openid_auth}) - fxc = FuncXClient(login_manager=fxmanager, task_group_id=user_id, - use_offprocess_checker=False, funcx_home=home_dir) + fxc = FuncXClient(login_manager=fxmanager, funcx_home=home_dir) action_id = event['pathParameters']['action-id'] diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 1d79b94..786ff3f 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -45,8 +45,7 @@ def lambda_handler(event, context): 'search.api.globus.org/all': search_auth, 'openid': openid_auth}) - fxc = FuncXClient(login_manager=fxmanager, task_group_id=user_id, - use_offprocess_checker=False, funcx_home=home_dir) + fxc = FuncXClient(login_manager=fxmanager, funcx_home=home_dir) dynamodb = boto3.resource('dynamodb') table = dynamodb.Table('funcx-actions') diff --git a/aws/requirements.txt b/aws/requirements.txt index bc1f930..cae66f4 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -1,4 +1,4 @@ globus-sdk>=3.0.0 -funcx>=1.0.0 -funcx-endpoint>=1.0.1 +funcx>=1.0.12 +funcx-endpoint>=1.0.12 globus_automate_client From aa7a25d4397d9b214ed060f5585228ad6b10e734 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Thu, 6 Apr 2023 15:56:20 +1200 Subject: [PATCH 66/69] Pin cryptography wheel requirement for lambda --- aws/requirements.txt | 2 ++ 1 file changed, 2 insertions(+) diff --git a/aws/requirements.txt b/aws/requirements.txt index cae66f4..41e13bd 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -2,3 +2,5 @@ globus-sdk>=3.0.0 funcx>=1.0.12 funcx-endpoint>=1.0.12 globus_automate_client +https://files.pythonhosted.org/packages/1e/85/d5b768b45e564a66fc5ba6344145334208f01d64939adcb8c4032545d164/cryptography-39.0.2-pp38-pypy38_pp73-manylinux_2_28_x86_64.whl + From a4c4243660aef4bc0694ed956d886f6ef8c1fd0c Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Thu, 6 Apr 2023 18:42:44 +1200 Subject: [PATCH 67/69] Try pinning to an old cryptography package. --- aws/requirements.txt | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/aws/requirements.txt b/aws/requirements.txt index 41e13bd..7a89e4f 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -2,5 +2,4 @@ globus-sdk>=3.0.0 funcx>=1.0.12 funcx-endpoint>=1.0.12 globus_automate_client -https://files.pythonhosted.org/packages/1e/85/d5b768b45e564a66fc5ba6344145334208f01d64939adcb8c4032545d164/cryptography-39.0.2-pp38-pypy38_pp73-manylinux_2_28_x86_64.whl - +cryptography==3.4.8 From ae8909eba03d385d253b9d69edd089898221a107 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Thu, 6 Apr 2023 18:50:10 +1200 Subject: [PATCH 68/69] Try manylinux_2_12 --- aws/requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aws/requirements.txt b/aws/requirements.txt index 7a89e4f..0492304 100644 --- a/aws/requirements.txt +++ b/aws/requirements.txt @@ -2,4 +2,4 @@ globus-sdk>=3.0.0 funcx>=1.0.12 funcx-endpoint>=1.0.12 globus_automate_client -cryptography==3.4.8 +https://thirdparty.aboutcode.org/pypi/cryptography-3.4.8-cp36-abi3-manylinux_2_12_x86_64.manylinux2010_x86_64.whl From 37115ddf11160c394611666f6cfbe4a54afc9f80 Mon Sep 17 00:00:00 2001 From: Ryan Chard Date: Thu, 6 Apr 2023 18:57:28 +1200 Subject: [PATCH 69/69] Switch to kwargs for adding values to a batch --- aws/funcx-run.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/aws/funcx-run.py b/aws/funcx-run.py index 786ff3f..0280cfe 100644 --- a/aws/funcx-run.py +++ b/aws/funcx-run.py @@ -77,7 +77,7 @@ def lambda_handler(event, context): payload = task.get('payload', None) if payload: batch.add(endpoint_id=task['endpoint'], function_id=task['function'], - **task['payload']) + kwargs=task['payload']) else: batch.add(endpoint_id=task['endpoint'], function_id=task['function'])