Commit

Author: matthewdunbar

lambda_on_rails.gemspec

@@ -27,6 +27,9 @@ Gem::Specification.new do |spec|
 
   spec.add_dependency "rails", "~> 5.2.2"
   spec.add_dependency "aws-sdk-sqs"
+  spec.add_dependency "aws-sdk-lambda"
+  spec.add_dependency "rubyzip"
   spec.add_development_dependency "sqlite3", "~> 1.3.6"
   spec.add_development_dependency "dotenv-rails"
+  spec.add_development_dependency "pry"
 end

lib/active_job/queue_adapters/lambda_adapter.rb

@@ -1,3 +1,5 @@
+require 'active_job'
+
 module ActiveJob
   module QueueAdapters
     class LambdaAdapter

lib/lambda_on_rails.rb

@@ -1,6 +1,8 @@
 require 'lambda_on_rails/railtie'
 require 'lambda_on_rails/job_runner'
 require 'lambda_on_rails/queue_loader'
+require 'lambda_on_rails/job_uploader'
+require 'lambda_on_rails/job_packager'
 require 'active_job/queue_adapters/lambda_adapter'
 
 module LambdaOnRails

lib/lambda_on_rails/job_packager.rb

@@ -0,0 +1,69 @@
+require_relative './job_wrapper'
+require 'zip'
+require 'tmpdir'
+
+module LambdaOnRails
+  class JobPackager
+    def build_zip_file(job_klass)
+      temp_directory = build_tmp_directory(job_klass)
+      Zip::OutputStream.write_buffer do |io|
+        Dir[File.join(temp_directory, '**', '*')].each do |file|
+          next if File.directory?(file)
+
+          absolute_path = Pathname.new(file)
+          relative_path = absolute_path.relative_path_from(Pathname.new(temp_directory))
+
+          io.put_next_entry(relative_path)
+          io.write(File.read(absolute_path))
+        end
+      end
+    end
+
+    private
+
+    def build_tmp_directory(job_klass)
+      directory = Dir.mktmpdir(job_klass.to_s.gsub(/[^0-9a-z ]/i, '-'))
+      FileUtils.cp(class_source_location(job_klass), directory)
+      FileUtils.cp(job_wrapper_source_location, directory)
+
+      lambda_on_rails_path = '../../'
+
+      File.write(File.join(directory, 'Gemfile'),
+        <<~GEMFILE
+          source 'https://rubygems.org'
+
+          gem 'lambda_on_rails', path: "#{lambda_on_rails_path}"
+        GEMFILE
+      )
+
+      run_bundle_install(directory)
+
+      directory
+    end
+
+    def class_source_location(klass)
+      klass.instance_method(:perform).source_location[0]
+    end
+
+    def class_source_code(klass)
+      File.read(class_source_location(klass))
+    end
+
+    def job_wrapper_source_location
+      method(:handler).source_location[0]
+    end
+
+    def job_wrapper_source_code
+      File.read(job_wrapper_source_location)
+    end
+
+    def run_bundle_install(directory)
+      Bundler.with_clean_env do
+        system("cd #{directory}; docker run -v `pwd`:`pwd` -w `pwd` -i -t lambci/lambda:build-ruby2.5 bundle install")
+        system("cd #{directory}; docker run -v `pwd`:`pwd` -w `pwd` -i -t lambci/lambda:build-ruby2.5 bundle install --deployment")
+        # system("cd #{directory}; bundle install")
+        # system("cd #{directory}; bundle install --deployment")
+      end
+    end
+  end
+end

lib/lambda_on_rails/job_runner.rb

@@ -8,7 +8,7 @@ def initialize(queue_loader = ::LambdaOnRails::QueueLoader.new)
 
     def enqueue(job)
       queue = queue_loader.get_queue(job.enqueue_url)
-      queue.send_message(message_body: job.serialize.to_s)
+      queue.send_message(message_body: Base64.encode64(job.serialize.to_json))
     end
   end
 end

lib/lambda_on_rails/job_uploader.rb

@@ -0,0 +1,98 @@
+require_relative './job_wrapper'
+require 'aws-sdk-lambda'
+
+module LambdaOnRails
+  class JobUploader
+    attr_reader :klass
+
+    def initialize(klass)
+      @klass = klass
+    end
+
+    def upload
+      create_queue unless queue_exists?
+
+      if function_exists?
+        update_lambda
+      else
+        create_lambda
+      end
+
+      create_event_source_mapping unless event_source_mapping_exists?
+    end
+
+    private
+
+    def function_exists?
+      lambda_client.get_function(function_name: job_identifier)
+
+      true
+    rescue Aws::Lambda::Errors::ResourceNotFoundException
+      false
+    end
+
+    def queue_exists?
+      sqs_client.get_queue_url(queue_name: "#{job_identifier}-Incoming")
+
+      true
+    rescue Aws::SQS::Errors::NonExistentQueue
+      false
+    end
+
+    def event_source_mapping_exists?
+      lambda_client.list_event_source_mappings(
+        function_name: job_identifier
+      ).event_source_mappings.any?
+    end
+
+    def create_lambda
+      lambda_client.create_function(
+        function_name: job_identifier,
+        runtime: 'ruby2.5',
+        role: klass.role_arn,
+        code: {
+          zip_file: zip_file
+        },
+        handler: 'job_wrapper.handler'
+      )
+    end
+
+    def update_lambda
+      lambda_client.update_function_code(
+        function_name: job_identifier,
+        zip_file: zip_file,
+        publish: true
+      )
+    end
+
+    def create_queue
+      sqs_client.create_queue(queue_name: "#{job_identifier}-Incoming")
+    end
+
+    def create_event_source_mapping
+      queue_url = sqs_client.get_queue_url(queue_name: "#{job_identifier}-Incoming").queue_url
+      queue_attributes = sqs_client.get_queue_attributes(queue_url: queue_url, attribute_names: ["All"])
+
+      lambda_client.create_event_source_mapping(
+        event_source_arn: queue_attributes.attributes['QueueArn'],
+        function_name: job_identifier
+      )
+    end
+
+    def zip_file
+      @zip_file ||= JobPackager.new.build_zip_file(klass).string
+    end
+
+    def job_identifier
+      @job_identifier ||= klass.to_s.gsub(/[^0-9a-z ]/i, '-')
+    end
+
+    def lambda_client
+      @lambda_client ||= Aws::Lambda::Client.new
+    end
+
+    def sqs_client
+      @sqs_client ||= Aws::SQS::Client.new
+    end
+  end
+end

lib/lambda_on_rails/job_wrapper.rb

@@ -0,0 +1,30 @@
+load_paths = Dir["vendor/bundle/**/lib"]
+$LOAD_PATH.unshift(*load_paths)
+
+require 'base64'
+
+def handler(event:, context:)
+  system("bundle install --deployment")
+
+  encoded_job = event['Records'][0]['body']
+  serialized_job = JSON.parse(Base64.decode64(encoded_job))
+
+  require_relative serialized_job['job_class'].underscore
+
+  klass = Object.const_get(serialized_job['job_class'])
+
+  job = klass.new
+  job.perform(*serialized_job['arguments'])
+end
+
+class String
+  def underscore
+    word = self.dup
+    word.gsub!(/::/, '/')
+    word.gsub!(/([A-Z]+)([A-Z][a-z])/,'\1_\2')
+    word.gsub!(/([a-z\d])([A-Z])/,'\1_\2')
+    word.tr!("-", "_")
+    word.downcase!
+    word
+  end
+end

lib/lambda_on_rails/railtie.rb

@@ -1,4 +1,7 @@
 module LambdaOnRails
   class Railtie < ::Rails::Railtie
+    rake_tasks do
+      load 'tasks/lambda_on_rails_tasks.rake'
+    end
   end
 end

lib/tasks/lambda_on_rails_tasks.rake

@@ -1,4 +1,9 @@
-# desc "Explaining what the task does"
-# task :lambda_on_rails do
-#   # Task goes here
-# end
+
+desc 'Uploads ActiveJob code to AWS'
+namespace :lambda_on_rails do
+  task :upload, [:class_name] => :environment do |_task, args|
+    job_class = args[:class_name].constantize
+    job_uploader = LambdaOnRails::JobUploader.new(job_class)
+    job_uploader.upload
+  end
+end

test/dummy/app/jobs/sample_job.rb

@@ -1,8 +1,16 @@
-class SampleJob < ApplicationJob
+class SampleJob < ActiveJob::Base
 
   self.queue_adapter = :lambda
 
+  def perform(name)
+    puts "Hello #{name}, how ya doin?"
+  end
+
   def enqueue_url
-    'https://sqs.us-east-1.amazonaws.com/567419588983/rails_active_job'
+    'https://sqs.us-east-1.amazonaws.com/789533204773/SampleJob-Incoming'
+  end
+
+  def self.role_arn
+    'arn:aws:iam::789533204773:role/lambda_sqs_basic_execution'
   end
 end

test/lambda_on_rails/job_packager_test.rb

@@ -0,0 +1,22 @@
+require 'test_helper'
+require_relative './test_job'
+require 'zip'
+require 'aws-sdk-lambda'
+
+class LambdaOnRails::JobPackager::Test < ActiveSupport::TestCase
+  test 'creates a zip file with the source code' do
+    job_packager = ::LambdaOnRails::JobPackager.new
+
+    job_packager.stub(:run_bundle_install, true, ['a']) do
+      @zip_file = job_packager.build_zip_file(LambdaOnRails::TestJob)
+    end
+
+    Zip::File.open_buffer(@zip_file) do |data|
+      job = data.select { |entry| entry.name == 'test_job.rb' }.first
+      assert(job.get_input_stream.read.include?('class LambdaOnRails::TestJob < ActiveJob::Base'))
+
+      job_wrapper = data.select { |entry| entry.name == 'job_wrapper.rb' }.first
+      assert(job_wrapper.get_input_stream.read.include?('def handler'))
+    end
+  end
+end

test/lambda_on_rails/job_runner_test.rb

@@ -8,7 +8,7 @@ class LambdaOnRails::JobRunner::Test < ActiveSupport::TestCase
       true,
       [
         {
-          message_body: '{}'
+          message_body: Base64.encode64({}.to_json)
         }
       ]
     )

test/lambda_on_rails/job_uploader_test.rb

@@ -0,0 +1,70 @@
+require 'test_helper'
+require_relative './test_job'
+require 'zip'
+require 'aws-sdk-lambda'
+
+class LambdaOnRails::JobUploader::Test < ActiveSupport::TestCase
+  test 'creates a Lambda function and SQS queue' do
+    job_uploader = ::LambdaOnRails::JobUploader.new(LambdaOnRails::TestJob)
+
+    sqs_client = Minitest::Mock.new
+    def sqs_client.get_queue_url(options)
+      @count ||= 0
+      @count += 1
+      if @count > 1
+        OpenStruct.new(queue_url: 'https://sqs')
+      else
+        raise Aws::SQS::Errors::NonExistentQueue.new(nil, nil)
+      end
+    end
+
+    sqs_client.expect :create_queue, OpenStruct.new(queue_url: 'https://sqs'), [{
+      queue_name: 'LambdaOnRails--TestJob-Incoming'
+    }]
+
+    sqs_client.expect :get_queue_attributes, OpenStruct.new(attributes: { 'QueueArn' => 'arn:test:queue' }), [{
+      queue_url: 'https://sqs',
+      attribute_names: ['All']
+    }]
+
+    lambda_client = Minitest::Mock.new
+    def lambda_client.get_function(options); raise Aws::Lambda::Errors::ResourceNotFoundException.new(nil, nil) end
+    lambda_client.expect :create_function, nil, [{
+      function_name: 'LambdaOnRails--TestJob',
+      code: {
+        zip_file: 'ABCD'
+      },
+      runtime: 'ruby2.5',
+      role: 'role:12345/us-east1',
+      handler: 'job_wrapper.handler'
+    }]
+
+    lambda_client.expect :create_event_source_mapping, nil, [{
+      event_source_arn: 'arn:test:queue',
+      function_name: 'LambdaOnRails--TestJob'
+    }]
+
+    lambda_client.expect :list_event_source_mappings, OpenStruct.new(event_source_mappings: []), [{
+      function_name: 'LambdaOnRails--TestJob'
+    }]
+
+    job_packager = Minitest::Mock.new
+    job_packager.expect :build_zip_file, StringIO.new('ABCD'), [LambdaOnRails::TestJob]
+
+    class LambdaOnRails::TestJob
+      def self.role_arn; 'role:12345/us-east1' end
+    end
+
+    LambdaOnRails::JobPackager.stub(:new, job_packager) do
+      Aws::SQS::Client.stub(:new, sqs_client) do
+        Aws::Lambda::Client.stub(:new, lambda_client) do
+          job_uploader.upload
+        end
+      end
+    end
+
+    assert_mock job_packager
+    assert_mock lambda_client
+    assert_mock sqs_client
+  end
+end