Keep usage event records of running apps and services

- Add keep_running_records parameter to OldRecordCleanup
- Implement exclude_running_records method to retain START/CREATED records
  without corresponding STOP/DELETED records
- Enable keep_running_records for AppUsageEvent and ServiceUsageEvent cleanup
- Add indexes for increasing performance on state queries
- Add comprehensive unit tests

The exclude_running_records method identifies
START records that have matching STOP records
while keeping START records without matching STOPs.

This prevents premature deletion of usage event records for long-running
apps and services, ensuring consumers can always determine the true start
time of active resources.

Author: joyvuu-dave

app/repositories/app_usage_event_repository.rb

@@ -152,7 +152,7 @@ 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).delete
+        Database::OldRecordCleanup.new(AppUsageEvent, cutoff_age_in_days: cutoff_age_in_days, keep_at_least_one_record: true, keep_running_records: true).delete
       end
 
       private

app/repositories/service_usage_event_repository.rb

@@ -92,7 +92,7 @@ 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).delete
+        Database::OldRecordCleanup.new(ServiceUsageEvent, cutoff_age_in_days: cutoff_age_in_days, keep_at_least_one_record: true, keep_running_records: true).delete
       end
     end
   end

db/migrations/20251103141149_add_lifecycle_index_to_usage_events.rb

@@ -0,0 +1,68 @@
+Sequel.migration do
+  no_transaction # to use the 'concurrently' option
+
+  up do
+    if database_type == :postgres
+      VCAP::Migration.with_concurrent_timeout(self) do
+        add_index :app_usage_events, %i[state app_guid id],
+                  name: :app_usage_events_lifecycle_index,
+                  if_not_exists: true,
+                  concurrently: true
+
+        add_index :service_usage_events, %i[state service_instance_guid id],
+                  name: :service_usage_events_lifecycle_index,
+                  if_not_exists: true,
+                  concurrently: true
+      end
+
+    elsif database_type == :mysql
+      alter_table :app_usage_events do
+        # rubocop:disable Sequel/ConcurrentIndex
+        add_index %i[state app_guid id], name: :app_usage_events_lifecycle_index unless @db.indexes(:app_usage_events).include?(:app_usage_events_lifecycle_index)
+        # rubocop:enable Sequel/ConcurrentIndex
+      end
+
+      alter_table :service_usage_events do
+        # rubocop:disable Sequel/ConcurrentIndex
+        unless @db.indexes(:service_usage_events).include?(:service_usage_events_lifecycle_index)
+          add_index %i[state service_instance_guid id],
+                    name: :service_usage_events_lifecycle_index
+        end
+        # rubocop:enable Sequel/ConcurrentIndex
+      end
+    end
+  end
+
+  down do
+    if database_type == :postgres
+      VCAP::Migration.with_concurrent_timeout(self) do
+        drop_index :app_usage_events, %i[state app_guid id],
+                   name: :app_usage_events_lifecycle_index,
+                   if_exists: true,
+                   concurrently: true
+
+        drop_index :service_usage_events, %i[state service_instance_guid id],
+                   name: :service_usage_events_lifecycle_index,
+                   if_exists: true,
+                   concurrently: true
+      end
+    end
+
+    if database_type == :mysql
+      alter_table :app_usage_events do
+        # rubocop:disable Sequel/ConcurrentIndex
+        drop_index %i[state app_guid id], name: :app_usage_events_lifecycle_index if @db.indexes(:app_usage_events).include?(:app_usage_events_lifecycle_index)
+        # rubocop:enable Sequel/ConcurrentIndex
+      end
+
+      alter_table :service_usage_events do
+        # rubocop:disable Sequel/ConcurrentIndex
+        if @db.indexes(:service_usage_events).include?(:service_usage_events_lifecycle_index)
+          drop_index %i[state service_instance_guid id],
+                     name: :service_usage_events_lifecycle_index
+        end
+        # rubocop:enable Sequel/ConcurrentIndex
+      end
+    end
+  end
+end

lib/database/old_record_cleanup.rb

@@ -3,12 +3,13 @@
 module Database
   class OldRecordCleanup
     class NoCurrentTimestampError < StandardError; end
-    attr_reader :model, :cutoff_age_in_days, :keep_at_least_one_record
+    attr_reader :model, :cutoff_age_in_days, :keep_at_least_one_record, :keep_running_records
 
-    def initialize(model, cutoff_age_in_days:, keep_at_least_one_record: false)
+    def initialize(model, cutoff_age_in_days:, keep_at_least_one_record: false, keep_running_records: false)
       @model = model
       @cutoff_age_in_days = cutoff_age_in_days
       @keep_at_least_one_record = keep_at_least_one_record
+      @keep_running_records = keep_running_records
     end
 
     def delete
@@ -21,6 +22,8 @@ def delete
       end
       logger.info("Cleaning up #{old_records.count} #{model.table_name} table rows")
 
+      old_records = exclude_running_records(old_records) if keep_running_records
+
       Database::BatchDelete.new(old_records, 1000).delete
     end
 
@@ -35,5 +38,66 @@ def current_timestamp_from_database
     def logger
       @logger ||= Steno.logger('cc.old_record_cleanup')
     end
+
+    def exclude_running_records(old_records)
+      return old_records unless has_duration?(model)
+
+      beginning_string = beginning_string(model)
+      ending_string = ending_string(model)
+      guid_symbol = guid_symbol(model)
+
+      raise "Invalid duration model: #{model}" if beginning_string.nil? || ending_string.nil? || guid_symbol.nil?
+
+      # Create subqueries for START and STOP records within the old records set
+      # Using from_self creates a subquery, allowing us to reference these in complex joins
+      initial_records = old_records.where(state: beginning_string).from_self(alias: :initial_records)
+      final_records = old_records.where(state: ending_string).from_self(alias: :final_records)
+
+      # For each START record, check if there exists a STOP record that:
+      # 1. Has the same resource GUID (app_guid or service_instance_guid)
+      # 2. Was created AFTER the START record (higher ID implies later creation)
+      exists_condition = final_records.where(Sequel[:final_records][guid_symbol] => Sequel[:initial_records][guid_symbol]).where do
+        Sequel[:final_records][:id] > Sequel[:initial_records][:id]
+      end.select(1).exists
+
+      prunable_initial_records = initial_records.where(exists_condition)
+
+      # Include records with states other than START/STOP
+      other_records = old_records.exclude(state: [beginning_string, ending_string])
+
+      # Return the UNION of:
+      # 1. START records that have a matching STOP (safe to delete)
+      # 2. All STOP records (always safe to delete)
+      # 3. Other state records (always safe to delete)
+      prunable_initial_records.union(final_records, all: true).union(other_records, all: true)
+    end
+
+    def has_duration?(model)
+      return true if model == VCAP::CloudController::AppUsageEvent
+      return true if model == VCAP::CloudController::ServiceUsageEvent
+
+      false
+    end
+
+    def beginning_string(model)
+      return VCAP::CloudController::ProcessModel::STARTED if model == VCAP::CloudController::AppUsageEvent
+      return VCAP::CloudController::Repositories::ServiceUsageEventRepository::CREATED_EVENT_STATE if model == VCAP::CloudController::ServiceUsageEvent
+
+      nil
+    end
+
+    def ending_string(model)
+      return VCAP::CloudController::ProcessModel::STOPPED if model == VCAP::CloudController::AppUsageEvent
+      return VCAP::CloudController::Repositories::ServiceUsageEventRepository::DELETED_EVENT_STATE if model == VCAP::CloudController::ServiceUsageEvent
+
+      nil
+    end
+
+    def guid_symbol(model)
+      return :app_guid if model == VCAP::CloudController::AppUsageEvent
+      return :service_instance_guid if model == VCAP::CloudController::ServiceUsageEvent
+
+      nil
+    end
   end
 end

spec/unit/jobs/runtime/app_usage_events_cleanup_spec.rb

@@ -5,7 +5,7 @@ module Jobs::Runtime
     RSpec.describe AppUsageEventsCleanup, job_context: :worker do
       let(:cutoff_age_in_days) { 30 }
       let(:logger) { double(Steno::Logger, info: nil) }
-      let!(:event_before_threshold) { AppUsageEvent.make(created_at: (cutoff_age_in_days + 1).days.ago) }
+      let!(:event_before_threshold) { AppUsageEvent.make(created_at: (cutoff_age_in_days + 1).days.ago, state: 'STOPPED') }
       let!(:event_after_threshold) { AppUsageEvent.make(created_at: (cutoff_age_in_days - 1).days.ago) }
 
       subject(:job) do

spec/unit/jobs/services/service_usage_events_cleanup_spec.rb

@@ -5,7 +5,7 @@ module Jobs::Services
     RSpec.describe ServiceUsageEventsCleanup, job_context: :worker do
       let(:cutoff_age_in_days) { 30 }
       let(:logger) { double(Steno::Logger, info: nil) }
-      let!(:event_before_threshold) { ServiceUsageEvent.make(created_at: (cutoff_age_in_days + 1).days.ago) }
+      let!(:event_before_threshold) { ServiceUsageEvent.make(created_at: (cutoff_age_in_days + 1).days.ago, state: 'DELETED') }
       let!(:event_after_threshold) { ServiceUsageEvent.make(created_at: (cutoff_age_in_days - 1).days.ago) }
 
       subject(:job) do