V3: introduce reoccurring_job to simplify async service create

Author: blgm

app/actions/service_instance_create_managed.rb

@@ -1,5 +1,5 @@
 require 'repositories/service_instance_share_event_repository'
-require 'jobs/v3/services/create_service_instance_job'
+require 'jobs/v3/create_service_instance_job'
 require 'actions/mixins/service_instance_create'
 
 module VCAP::CloudController

app/jobs/enqueuer.rb

@@ -15,8 +15,8 @@ def enqueue
         enqueue_job(@job)
       end
 
-      def enqueue_pollable
-        wrapped_job = PollableJobWrapper.new(@job)
+      def enqueue_pollable(existing_guid: nil)
+        wrapped_job = PollableJobWrapper.new(@job, existing_guid: existing_guid)
 
         if block_given?
           wrapped_job = yield wrapped_job

app/jobs/pollable_job_wrapper.rb

@@ -3,15 +3,32 @@
 module VCAP::CloudController
   module Jobs
     class PollableJobWrapper < WrappingJob
+      attr_reader :existing_guid
+
+      def initialize(handler, existing_guid: nil)
+        @existing_guid = existing_guid
+        super(handler)
+      end
+
       # use custom hook as Job does not have the guid field populated during the normal `enqueue` hook
       def after_enqueue(job)
-        PollableJobModel.create(
-          delayed_job_guid: job.guid,
-          state: PollableJobModel::PROCESSING_STATE,
-          operation: @handler.display_name,
-          resource_guid: @handler.resource_guid,
-          resource_type: @handler.resource_type
-        )
+        if existing_guid && (existing = PollableJobModel.find(guid: existing_guid))
+          existing.update(
+            delayed_job_guid: job.guid,
+            state: PollableJobModel::POLLING_STATE,
+            operation: @handler.display_name,
+            resource_guid: @handler.resource_guid,
+            resource_type: @handler.resource_type
+          )
+        else
+          PollableJobModel.create(
+            delayed_job_guid: job.guid,
+            state: PollableJobModel::PROCESSING_STATE,
+            operation: @handler.display_name,
+            resource_guid: @handler.resource_guid,
+            resource_type: @handler.resource_type
+          )
+        end
       end
 
       def success(job)

app/jobs/reoccurring_job.rb

@@ -0,0 +1,80 @@
+require 'jobs/cc_job'
+
+module VCAP::CloudController
+  module Jobs
+    class ReoccurringJob < VCAP::CloudController::Jobs::CCJob
+      attr_reader :finished, :start_time
+
+      def success(current_delayed_job)
+        pollable_job = PollableJobModel.find_by_delayed_job(current_delayed_job)
+
+        if finished
+          pollable_job.update(state: PollableJobModel::COMPLETE_STATE)
+        elsif next_enqueue_would_exceed_maximum_duration?
+          expire!
+        else
+          enqueue_next_job(pollable_job)
+        end
+      end
+
+      def maximum_duration_seconds
+        @maximum_duration || default_maximum_duration_seconds
+      end
+
+      def maximum_duration_seconds=(duration)
+        @maximum_duration = duration if duration < default_maximum_duration_seconds
+      end
+
+      def polling_interval_seconds
+        @polling_interval || default_polling_interval_seconds
+      end
+
+      def polling_interval_seconds=(interval)
+        @polling_interval = if interval < default_polling_interval_seconds
+                              default_polling_interval_seconds
+                            elsif interval > 24.hours
+                              24.hours
+                            else
+                              interval
+                            end
+      end
+
+      private
+
+      def initialize
+        @start_time = Time.now
+        @finished = false
+      end
+
+      def default_maximum_duration_seconds
+        VCAP::CloudController::Config.config.get(:broker_client_max_async_poll_duration_minutes).minutes
+      end
+
+      def default_polling_interval_seconds
+        Config.config.get(:broker_client_default_async_poll_interval_seconds)
+      end
+
+      def next_enqueue_would_exceed_maximum_duration?
+        Time.now + polling_interval_seconds > start_time + maximum_duration_seconds
+      end
+
+      def finish
+        @finished = true
+      end
+
+      def expire!
+        handle_timeout if self.respond_to?(:handle_timeout)
+        raise CloudController::Errors::ApiError.new_from_details('JobTimeout')
+      end
+
+      def enqueue_next_job(pollable_job)
+        opts = {
+          queue: Jobs::Queues.generic,
+          run_at: Delayed::Job.db_time_now + polling_interval_seconds
+        }
+
+        Jobs::Enqueuer.new(self, opts).enqueue_pollable(existing_guid: pollable_job.guid)
+      end
+    end
+  end
+end

app/jobs/v3/create_service_instance_job.rb

@@ -0,0 +1,153 @@
+require 'jobs/reoccurring_job'
+
+module VCAP::CloudController
+  module V3
+    class CreateServiceInstanceJob < VCAP::CloudController::Jobs::ReoccurringJob
+      attr_reader :warnings
+
+      def initialize(service_instance_guid, arbitrary_parameters: {}, user_audit_info:)
+        super()
+        @service_instance_guid = service_instance_guid
+        @arbitrary_parameters = arbitrary_parameters
+        @user_audit_info = user_audit_info
+        @start_time = Time.now
+        @first_time = true
+        @warnings = []
+      end
+
+      def perform
+        client = VCAP::Services::ServiceClientProvider.provide({ instance: service_instance })
+
+        if first_time
+          compute_maximum_duration
+          send_provision_request(client)
+          compatibility_checks
+          @first_time = false
+        end
+
+        gone! if service_instance.nil?
+
+        operation_in_progress = service_instance.last_operation.type
+        aborted! if operation_in_progress != 'create'
+
+        if service_instance.operation_in_progress?
+          fetch_last_operation(client)
+        end
+
+        if service_instance.last_operation.state == 'succeeded'
+          record_event(service_instance, @arbitrary_parameters)
+          finish
+        elsif service_instance.last_operation.state == 'failed'
+          operation_failed!(service_instance.last_operation.description)
+        end
+      end
+
+      def handle_timeout
+        service_instance.save_and_update_operation(
+          last_operation: {
+            state: 'failed',
+            description: 'Service Broker failed to provision within the required time.',
+          }
+        )
+      end
+
+      def job_name_in_configuration
+        :service_instance_create
+      end
+
+      def max_attempts
+        1
+      end
+
+      def resource_type
+        'service_instances'
+      end
+
+      def resource_guid
+        service_instance_guid
+      end
+
+      def display_name
+        'service_instance.create'
+      end
+
+      private
+
+      attr_reader :service_instance_guid, :arbitrary_parameters, :first_time
+
+      def compute_maximum_duration
+        max_poll_duration_on_plan = service_instance.service_plan.try(:maximum_polling_duration)
+        self.maximum_duration_seconds = max_poll_duration_on_plan if max_poll_duration_on_plan
+      end
+
+      def send_provision_request(client)
+        broker_response = client.provision(
+          service_instance,
+          accepts_incomplete: true,
+          arbitrary_parameters: arbitrary_parameters,
+          maintenance_info: service_instance.service_plan.maintenance_info
+        )
+
+        service_instance.save_with_new_operation(broker_response[:instance], broker_response[:last_operation])
+      rescue => e
+        service_instance.save_with_new_operation({}, {
+          type: 'create',
+          state: 'failed',
+          description: e.message,
+        })
+        raise e
+      end
+
+      def fetch_last_operation(client)
+        last_operation_result = client.fetch_service_instance_last_operation(service_instance)
+        self.polling_interval_seconds = last_operation_result[:retry_after] if last_operation_result[:retry_after]
+
+        service_instance.save_and_update_operation(
+          last_operation: last_operation_result[:last_operation].slice(:state, :description)
+        )
+      rescue HttpRequestError, HttpResponseError, Sequel::Error => e
+        logger = Steno.logger('cc-background')
+        logger.error("There was an error while fetching the service instance operation state: #{e}")
+      end
+
+      def record_event(service_instance, request_attrs)
+        Repositories::ServiceEventRepository.new(@user_audit_info).
+          record_service_instance_event(:create, service_instance, request_attrs)
+      end
+
+      def service_instance
+        ManagedServiceInstance.first(guid: service_instance_guid)
+      end
+
+      def compatibility_checks
+        if service_instance.service_plan.service.volume_service? && volume_services_disabled?
+          @warnings.push({ detail: ServiceInstance::VOLUME_SERVICE_WARNING })
+        end
+
+        if service_instance.service_plan.service.route_service? && route_services_disabled?
+          @warnings.push({ detail: ServiceInstance::ROUTE_SERVICE_WARNING })
+        end
+      end
+
+      def volume_services_disabled?
+        !VCAP::CloudController::Config.config.get(:volume_services_enabled)
+      end
+
+      def route_services_disabled?
+        !VCAP::CloudController::Config.config.get(:route_services_enabled)
+      end
+
+      def gone!
+        raise CloudController::Errors::ApiError.new_from_details('ServiceInstanceNotFound', service_instance_guid)
+      end
+
+      def aborted!
+        raise CloudController::Errors::ApiError.new_from_details('UnableToPerform', 'Create', 'delete in progress')
+      end
+
+      def operation_failed!(msg)
+        raise CloudController::Errors::ApiError.new_from_details('ServiceInstanceProvisionFailed', msg)
+      end
+    end
+  end
+end