Maintain un-processed usage event records

Allow consumer_guid to be passed into request for AppUsageEvent and ServiceUsageEvent records. This establishes a consumer of these events. Un-processed consumer records will be maintained by cloud controller. Registered consumers are automatically deleted if associated event is removed.

app/access/app_usage_consumer_access.rb

@@ -0,0 +1,79 @@
+module VCAP::CloudController
+  class AppUsageConsumerAccess < BaseAccess
+    def create?(_object, _params=nil)
+      admin_user?
+    end
+
+    def read?(object)
+      return @ok_read if instance_variable_defined?(:@ok_read)
+
+      @ok_read = admin_user? || admin_read_only_user? || global_auditor? || object_is_visible_to_user?(object, context.user)
+    end
+
+    def read_for_update?(_object, _params=nil)
+      admin_user?
+    end
+
+    def can_remove_related_object?(object, params=nil)
+      read_for_update?(object, params)
+    end
+
+    def read_related_object_for_update?(object, params=nil)
+      read_for_update?(object, params)
+    end
+
+    def update?(_object, _params=nil)
+      admin_user?
+    end
+
+    def delete?(_object)
+      admin_user?
+    end
+
+    # These methods should be called first to determine if the user's token has the appropriate scope for the operation
+
+    def read_with_token?(_)
+      admin_user? || admin_read_only_user? || has_read_scope? || global_auditor?
+    end
+
+    def create_with_token?(_)
+      admin_user? || has_write_scope?
+    end
+
+    def read_for_update_with_token?(_)
+      admin_user? || has_write_scope?
+    end
+
+    def can_remove_related_object_with_token?(*)
+      read_for_update_with_token?(*)
+    end
+
+    def read_related_object_for_update_with_token?(*)
+      read_for_update_with_token?(*)
+    end
+
+    def update_with_token?(_)
+      admin_user? || has_write_scope?
+    end
+
+    def delete_with_token?(_)
+      admin_user? || has_write_scope?
+    end
+
+    def index_with_token?(_)
+      admin_user? || admin_read_only_user?
+    end
+
+    def index?(_object_class, _params=nil)
+      admin_user? || admin_read_only_user?
+    end
+
+    def reset?(_object_class)
+      admin_user?
+    end
+
+    def reset_with_token?(_object_class)
+      admin_user?
+    end
+  end
+end

app/access/service_usage_consumer_access.rb

@@ -0,0 +1,79 @@
+module VCAP::CloudController
+  class ServiceUsageConsumerAccess < BaseAccess
+    def create?(_object, _params=nil)
+      admin_user?
+    end
+
+    def read?(object)
+      return @ok_read if instance_variable_defined?(:@ok_read)
+
+      @ok_read = admin_user? || admin_read_only_user? || global_auditor? || object_is_visible_to_user?(object, context.user)
+    end
+
+    def read_for_update?(_object, _params=nil)
+      admin_user?
+    end
+
+    def can_remove_related_object?(object, params=nil)
+      read_for_update?(object, params)
+    end
+
+    def read_related_object_for_update?(object, params=nil)
+      read_for_update?(object, params)
+    end
+
+    def update?(_object, _params=nil)
+      admin_user?
+    end
+
+    def delete?(_object)
+      admin_user?
+    end
+
+    # These methods should be called first to determine if the user's token has the appropriate scope for the operation
+
+    def read_with_token?(_)
+      admin_user? || admin_read_only_user? || has_read_scope? || global_auditor?
+    end
+
+    def create_with_token?(_)
+      admin_user? || has_write_scope?
+    end
+
+    def read_for_update_with_token?(_)
+      admin_user? || has_write_scope?
+    end
+
+    def can_remove_related_object_with_token?(*)
+      read_for_update_with_token?(*)
+    end
+
+    def read_related_object_for_update_with_token?(*)
+      read_for_update_with_token?(*)
+    end
+
+    def update_with_token?(_)
+      admin_user? || has_write_scope?
+    end
+
+    def delete_with_token?(_)
+      admin_user? || has_write_scope?
+    end
+
+    def index_with_token?(_)
+      admin_user? || admin_read_only_user?
+    end
+
+    def index?(_object_class, _params=nil)
+      admin_user? || admin_read_only_user?
+    end
+
+    def reset?(_object_class)
+      admin_user?
+    end
+
+    def reset_with_token?(_object_class)
+      admin_user?
+    end
+  end
+end

app/controllers/v3/app_usage_events_controller.rb

@@ -11,6 +11,20 @@ def index
 
     app_usage_events = AppUsageEventListFetcher.fetch_all(message, AppUsageEvent.dataset) if permission_queryer.can_read_globally?
 
+    if message.consumer_guid && message.after_guid&.first && permission_queryer.can_write_globally?
+      begin
+        consumer = AppUsageConsumer.find_or_create(consumer_guid: message.consumer_guid) do |c|
+          c.last_processed_guid = message.after_guid.first
+        end
+
+        consumer.update(last_processed_guid: message.after_guid.first) if !consumer.new? && consumer.last_processed_guid != message.after_guid.first
+      rescue Sequel::ValidationFailed => e
+        unprocessable!(e.message)
+      rescue Sequel::Error
+        error!('Failed to update consumer tracking', 500)
+      end
+    end
+
     render status: :ok, json: Presenters::V3::PaginatedListPresenter.new(
       presenter: Presenters::V3::AppUsageEventPresenter,
       paginated_result: SequelPaginator.new.get_page(app_usage_events, message.try(:pagination_options)),

app/controllers/v3/service_usage_events_controller.rb

@@ -11,6 +11,20 @@ def index
 
     service_usage_events = ServiceUsageEventListFetcher.fetch_all(message, ServiceUsageEvent.dataset) if permission_queryer.can_read_globally?
 
+    if message.consumer_guid && message.after_guid&.first && permission_queryer.can_write_globally?
+      begin
+        consumer = ServiceUsageConsumer.find_or_create(consumer_guid: message.consumer_guid) do |c|
+          c.last_processed_guid = message.after_guid.first
+        end
+
+        consumer.update(last_processed_guid: message.after_guid.first) if !consumer.new? && consumer.last_processed_guid != message.after_guid.first
+      rescue Sequel::ValidationFailed => e
+        unprocessable!(e.message)
+      rescue Sequel::Error
+        error!('Failed to update consumer tracking', 500)
+      end
+    end
+
     render status: :ok, json: Presenters::V3::PaginatedListPresenter.new(
       presenter: Presenters::V3::ServiceUsageEventPresenter,
       paginated_result: SequelPaginator.new.get_page(service_usage_events, message.try(:pagination_options)),

app/messages/app_usage_events_list_message.rb

@@ -2,8 +2,9 @@
 
 module VCAP::CloudController
   class AppUsageEventsListMessage < ListMessage
-    register_allowed_keys [
-      :after_guid
+    register_allowed_keys %i[
+      after_guid
+      consumer_guid
     ]
 
     validates_with NoAdditionalParamsValidator
@@ -19,7 +20,7 @@ def valid_order_by_values
     end
 
     def self.from_params(params)
-      super(params, %w[after_guid])
+      super(params, %w[after_guid consumer_guid])
     end
   end
 end

app/messages/service_usage_events_list_message.rb

@@ -6,6 +6,7 @@ class ServiceUsageEventsListMessage < ListMessage
       after_guid
       service_instance_types
       service_offering_guids
+      consumer_guid
     ]
 
     validates_with NoAdditionalParamsValidator
@@ -24,7 +25,7 @@ def valid_order_by_values
     end
 
     def self.from_params(params)
-      super(params, %w[after_guid service_instance_types service_offering_guids])
+      super(params, %w[after_guid service_instance_types service_offering_guids consumer_guid])
     end
   end
 end

app/models.rb

@@ -18,6 +18,7 @@
 require 'models/runtime/security_group'
 require 'models/runtime/security_groups_space'
 require 'models/runtime/staging_security_groups_space'
+require 'models/runtime/app_usage_consumer'
 require 'models/runtime/app_usage_event'
 require 'models/runtime/auto_detection_buildpack'
 require 'models/runtime/app_event'
@@ -140,6 +141,7 @@
 require 'models/services/service_plan_visibility'
 require 'models/services/service_plan_annotation_model'
 require 'models/services/service_plan_label_model'
+require 'models/services/service_usage_consumer'
 require 'models/services/service_usage_event'
 require 'models/services/service_key'
 require 'models/services/service_key_label_model'

app/models/runtime/app_usage_consumer.rb

@@ -0,0 +1,26 @@
+module VCAP::CloudController
+  class AppUsageConsumer < Sequel::Model
+    plugin :validation_helpers
+
+    many_to_one :last_processed_event,
+                class: 'VCAP::CloudController::AppUsageEvent',
+                key: :last_processed_guid,
+                primary_key: :guid
+
+    def validate
+      validates_presence %i[consumer_guid last_processed_guid]
+      validates_unique :consumer_guid
+      validates_max_length 255, :consumer_guid, message: 'must be less than 255 characters'
+    end
+
+    def last_processed_guid=(guid)
+      self[:last_processed_guid] = guid
+    end
+
+    def last_processed_guid
+      self[:last_processed_guid]
+    end
+
+    export_attributes :consumer_guid, :last_processed_guid
+  end
+end

app/models/runtime/app_usage_event.rb

@@ -2,6 +2,13 @@ module VCAP::CloudController
   class AppUsageEvent < Sequel::Model
     plugin :serialization
 
+    one_to_many :consumers,
+                class: 'VCAP::CloudController::AppUsageConsumer',
+                key: :last_processed_guid,
+                primary_key: :guid
+
+    add_association_dependencies consumers: :destroy
+
     export_attributes :state, :previous_state,
                       :memory_in_mb_per_instance, :previous_memory_in_mb_per_instance,
                       :instance_count, :previous_instance_count,

app/models/services/service_usage_consumer.rb

@@ -0,0 +1,26 @@
+module VCAP::CloudController
+  class ServiceUsageConsumer < Sequel::Model
+    plugin :validation_helpers
+
+    many_to_one :last_processed_event,
+                class: 'VCAP::CloudController::ServiceUsageEvent',
+                key: :last_processed_guid,
+                primary_key: :guid
+
+    def validate
+      validates_presence %i[consumer_guid last_processed_guid]
+      validates_unique :consumer_guid
+      validates_max_length 255, :consumer_guid, message: 'must be less than 255 characters'
+    end
+
+    def last_processed_guid=(guid)
+      self[:last_processed_guid] = guid
+    end
+
+    def last_processed_guid
+      self[:last_processed_guid]
+    end
+
+    export_attributes :consumer_guid, :last_processed_guid
+  end
+end

app/models/services/service_usage_event.rb

@@ -2,6 +2,13 @@ module VCAP::CloudController
   class ServiceUsageEvent < Sequel::Model
     plugin :serialization
 
+    one_to_many :consumers,
+                class: 'VCAP::CloudController::ServiceUsageConsumer',
+                key: :last_processed_guid,
+                primary_key: :guid
+
+    add_association_dependencies consumers: :destroy
+
     export_attributes :state, :org_guid, :space_guid, :space_name,
                       :service_instance_guid, :service_instance_name, :service_instance_type,
                       :service_plan_guid, :service_plan_name,

app/repositories/app_usage_event_repository.rb

@@ -152,7 +152,8 @@ def purge_and_reseed_started_apps!
       end
 
       def delete_events_older_than(cutoff_age_in_days)
-        Database::OldRecordCleanup.new(AppUsageEvent, cutoff_age_in_days: cutoff_age_in_days, keep_at_least_one_record: true, keep_running_records: true).delete
+        Database::OldRecordCleanup.new(AppUsageEvent, cutoff_age_in_days: cutoff_age_in_days, keep_at_least_one_record: true, keep_running_records: true,
+                                                      keep_unprocessed_records: true).delete
       end
 
       private

app/repositories/service_usage_event_repository.rb

@@ -92,7 +92,8 @@ def purge_and_reseed_service_instances!
       end
 
       def delete_events_older_than(cutoff_age_in_days)
-        Database::OldRecordCleanup.new(ServiceUsageEvent, cutoff_age_in_days: cutoff_age_in_days, keep_at_least_one_record: true, keep_running_records: true).delete
+        Database::OldRecordCleanup.new(ServiceUsageEvent, cutoff_age_in_days: cutoff_age_in_days, keep_at_least_one_record: true, keep_running_records: true,
+                                                          keep_unprocessed_records: true).delete
       end
     end
   end

db/migrations/20241205162359_create_app_usage_consumers.rb

@@ -0,0 +1,9 @@
+Sequel.migration do
+  change do
+    create_table :app_usage_consumers do |_t|
+      VCAP::Migration.common(self)
+      String :consumer_guid, null: false, size: 255
+      String :last_processed_guid, null: false, size: 255
+    end
+  end
+end

db/migrations/20241218113511_create_service_usage_consumers.rb

@@ -0,0 +1,9 @@
+Sequel.migration do
+  change do
+    create_table :service_usage_consumers do |_t|
+      VCAP::Migration.common(self)
+      String :consumer_guid, null: false, size: 255
+      String :last_processed_guid, null: false, size: 255
+    end
+  end
+end