Merge pull request #159 from rstudio/feature-sns

Feature sns

Makefile

@@ -1,5 +1,5 @@
 PLATFORMS := ubuntu-1804 ubuntu-2004 ubuntu-2204 debian-10 debian-11 centos-7 centos-8 rhel-9 opensuse-153 opensuse-154
-SLS_BINARY ?= ./node_modules/serverless/bin/serverless
+SLS_BINARY ?= ./node_modules/serverless/bin/serverless.js
 
 deps:
 	npm install

handler.py

@@ -8,6 +8,27 @@ import botocore
 CRAN_SRC_R3_URL = 'https://cran.rstudio.com/src/base/R-3/'
 CRAN_SRC_R4_URL = 'https://cran.rstudio.com/src/base/R-4/'
 batch_client = boto3.client('batch', region_name='us-east-1')
+sns_client = boto3.client('sns', region_name='us-east-1')
+
+
+class JobDetails:
+    def __init__(self, version, platform):
+        self.version = version
+        self.platform = platform
+
+    @classmethod
+    def from_job_name(cls, job_name):
+        _R, version, platform = job_name.split('-', 2)
+        return cls(version.replace('_', '.'), platform)
+
+    def job_name(self):
+        return f"R-{self.version.replace('.', '_')}-{self.platform}"
+
+    def job_definition_arn(self):
+        return os.environ[f"JOB_DEFINITION_ARN_{self.platform.replace('-','_')}"]
+
+    def to_json(self):
+        return {'version': self.version, 'platform': self.platform}
 
 
 def _to_list(input):
@@ -78,21 +99,20 @@ def _container_overrides(version):
 
 def _submit_job(version, platform):
     """Submit an R build job to AWS Batch."""
-    job_name = '-'.join(['R', version, platform])
+    job_details = JobDetails(version, platform)
-    job_name = job_name.replace('.', '_')
+
-    job_definition_arn = 'JOB_DEFINITION_ARN_{}'.format(platform.replace('-','_'))
     args = {
-        'jobName': job_name,
+        'jobName': job_details.job_name(),
         'jobQueue': os.environ['JOB_QUEUE_ARN'],
-        'jobDefinition': os.environ[job_definition_arn],
+        'jobDefinition': job_details.job_definition_arn(),
-        'containerOverrides': _container_overrides(version)
+        'containerOverrides': _container_overrides(job_details.version)
     }
     if os.environ.get('DRYRUN'):
-        print('DRYRUN: would have queued {}'.format(job_name))
+        print('DRYRUN: would have queued {}'.format(job_details.job_name()))
-        return 'dryrun-no-job-{}'.format(job_name)
+        return 'dryrun-no-job-{}'.format(job_details.job_name())
     else:
         response = batch_client.submit_job(**args)
-        print("Started job for R:{},Platform:{},id:{}".format(version, platform, response['jobId']))
+        print("Started job with details:{},id:{}".format(job_details, response['jobId']))
         return response['jobId']
 
 
@@ -145,3 +165,29 @@ def poll_running_jobs(event, context):
     event['finishedJobCount'] = len(event['failedJobIds']) + len(event['succeededJobIds'])
     event['unfinishedJobCount'] = len(event['jobIds']) - event['finishedJobCount']
     return event
+
+
+def finished(event, _context):
+    """Publish details about successfully finished jobs."""
+
+    # first, if there were no succeeded jobs, return instead of publishing details about builds
+    if len(event['succeededJobIds']) < 1:
+        return event
+
+    # fetch all jobs, removing those which are not in our succeeded id list
+    r = batch_client.list_jobs(jobQueue=os.environ['JOB_QUEUE_ARN'], jobStatus='SUCCEEDED')
+    print(f'r: {r}')
+    succeeded_jobs = [i for i in r['jobSummaryList'] if i['jobId'] in event['succeededJobIds']]
+
+    message = {'versions': []}
+
+    for job in succeeded_jobs:
+        details = JobDetails.from_job_name(job['jobName'])
+        message['versions'].append(vars(details))
+
+    response = sns_client.publish(
+        TargetArn=os.environ['SNS_TOPIC_ARN'],
+        Message=json.dumps(message),
+    )
+    print(f'Published to topic, response:{response}')
+    return event

package.json

@@ -4,9 +4,9 @@
   "version": "0.1.0",
   "dependencies": {},
   "devDependencies": {
-    "serverless": "^1.42.3",
+    "serverless": "^3.21.0",
+    "serverless-python-requirements": "^5.4.0",
     "serverless-pseudo-parameters": "^2.2.0",
-    "serverless-python-requirements": "^4.2.5",
+    "serverless-step-functions": "^3.12.1"
-    "serverless-step-functions": "^1.8.0"
   }
 }

serverless-resources.yml

@@ -323,3 +323,9 @@ rBuildsDevelEventRule:
           Fn::GetAtt: [ rBuildsEventRuleIamRole, Arn ]
         Arn:
           Ref: RBuildsStepFunction
+
+rBuildsTopic:
+  Type: AWS::SNS::Topic
+  Properties:
+    DisplayName: R Builds
+    TopicName: ${self:custom.${self:provider.stage}.snsTopicName}

serverless-stepf.yml

@@ -1,5 +1,4 @@
 rBuilds:
-  Comment: "R Builds"
   id: RBuildsStepFunction
   name: r-builds-${self:provider.stage}
   definition:
@@ -25,7 +24,11 @@ rBuilds:
             Next: Wait
           - Variable: "$.unfinishedJobCount"
             NumericEquals: 0
-            Next: SuccessOrFail
+            Next: Finished
+      Finished:
+        Type: Task
+        Resource: arn:aws:lambda:#{AWS::Region}:#{AWS::AccountId}:function:${self:service}-${self:provider.stage}-finished
+        Next: SuccessOrFail
       SuccessOrFail:
         Type: Choice
         Choices:

serverless.yml

@@ -31,8 +31,15 @@ provider:
        Resource:
         - arn:aws:s3:::${self:custom.${self:provider.stage}.s3Bucket}
         - arn:aws:s3:::${self:custom.${self:provider.stage}.s3Bucket}/*
+    -  Effect: Allow
+       Action:
+         - "sns:Publish"
+       Resource:
+         Ref: rBuildsTopic
   environment:
     S3_BUCKET: ${self:custom.${self:provider.stage}.s3Bucket}
+    SNS_TOPIC_ARN:
+      Ref: rBuildsTopic
     JOB_QUEUE_ARN:
       Ref: rBuildsBatchJobQueue
     JOB_DEFINITION_ARN_ubuntu_1804:
@@ -63,6 +70,8 @@ functions:
     handler: handler.queue_builds
   jobQueueStatus:
     handler: handler.poll_running_jobs
+  finished:
+    handler: handler.finished
 
 
 stepFunctions: