Keep Usage Event records of running apps and services

app/jobs/runtime/events_cleanup.rb

@@ -9,7 +9,7 @@ def initialize(cutoff_age_in_days)
         end
 
         def perform
-          Database::OldRecordCleanup.new(Event, cutoff_age_in_days).delete
+          Database::OldRecordCleanup.new(Event, cutoff_age_in_days:).delete
         end
 
         def job_name_in_configuration

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, 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, 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,22 +3,25 @@
 module Database
   class OldRecordCleanup
     class NoCurrentTimestampError < StandardError; end
-    attr_reader :model, :days_ago, :keep_at_least_one_record
+    attr_reader :model, :cutoff_age_in_days, :keep_at_least_one_record, :keep_running_records
 
-    def initialize(model, days_ago, 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
-      @days_ago = days_ago
+      @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
-      cutoff_date = current_timestamp_from_database - days_ago.to_i.days
+      cutoff_date = current_timestamp_from_database - cutoff_age_in_days.to_i.days
 
       old_records = model.dataset.where(Sequel.lit('created_at < ?', cutoff_date))
       if keep_at_least_one_record
         last_record = model.order(:id).last
         old_records = old_records.where(Sequel.lit('id < ?', last_record.id)) if last_record
       end
+      old_records = exclude_running_records(old_records) if keep_running_records
+
       logger.info("Cleaning up #{old_records.count} #{model.table_name} table rows")
 
       Database::BatchDelete.new(old_records, 1000).delete
@@ -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