Set threshold for keeping un-processed records

joyvuu-dave · joyvuu-dave · commit 9456a25a2565 · 2025-11-03T13:20:31.000-06:00
Only keep un-processed records if usage event table size is below threshold. This is a safeguard to ensure row count does not grow unbounded in the event of a zombie consumer.
diff --git a/app/jobs/runtime/app_usage_events_cleanup.rb b/app/jobs/runtime/app_usage_events_cleanup.rb
@@ -4,18 +4,19 @@ module VCAP::CloudController
   module Jobs
     module Runtime
       class AppUsageEventsCleanup < VCAP::CloudController::Jobs::CCJob
-        attr_accessor :cutoff_age_in_days
+        attr_accessor :cutoff_age_in_days, :threshold_for_keeping_unprocessed_records
 
-        def initialize(cutoff_age_in_days)
+        def initialize(cutoff_age_in_days, threshold_for_keeping_unprocessed_records)
           @cutoff_age_in_days = cutoff_age_in_days
+          @threshold_for_keeping_unprocessed_records = threshold_for_keeping_unprocessed_records
         end
 
         def perform
           logger = Steno.logger('cc.background')
           logger.info('Cleaning up old AppUsageEvent rows')
 
           repository = Repositories::AppUsageEventRepository.new
-          repository.delete_events_older_than(cutoff_age_in_days)
+          repository.delete_events_older_than(cutoff_age_in_days, threshold_for_keeping_unprocessed_records)
         end
 
         def job_name_in_configuration
diff --git a/app/jobs/v2/services/service_usage_events_cleanup.rb b/app/jobs/v2/services/service_usage_events_cleanup.rb
@@ -4,18 +4,19 @@ module VCAP::CloudController
   module Jobs
     module Services
       class ServiceUsageEventsCleanup < VCAP::CloudController::Jobs::CCJob
-        attr_accessor :cutoff_age_in_days
+        attr_accessor :cutoff_age_in_days, :threshold_for_keeping_unprocessed_records
 
-        def initialize(cutoff_age_in_days)
+        def initialize(cutoff_age_in_days, threshold_for_keeping_unprocessed_records)
           @cutoff_age_in_days = cutoff_age_in_days
+          @threshold_for_keeping_unprocessed_records = threshold_for_keeping_unprocessed_records
         end
 
         def perform
           logger = Steno.logger('cc.background')
           logger.info('Cleaning up old ServiceUsageEvent rows')
 
           repository = Repositories::ServiceUsageEventRepository.new
-          repository.delete_events_older_than(cutoff_age_in_days)
+          repository.delete_events_older_than(cutoff_age_in_days, threshold_for_keeping_unprocessed_records)
         end
 
         def job_name_in_configuration
diff --git a/app/repositories/app_usage_event_repository.rb b/app/repositories/app_usage_event_repository.rb
@@ -151,9 +151,9 @@ def purge_and_reseed_started_apps!
         AppUsageEvent.insert(column_map.keys, usage_query)
       end
 
-      def delete_events_older_than(cutoff_age_in_days)
+      def delete_events_older_than(cutoff_age_in_days, threshold_for_keeping_unprocessed_records)
         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
+                                                      keep_unprocessed_records: true, threshold_for_keeping_unprocessed_records: threshold_for_keeping_unprocessed_records).delete
       end
 
       private
diff --git a/app/repositories/service_usage_event_repository.rb b/app/repositories/service_usage_event_repository.rb
@@ -91,9 +91,10 @@ def purge_and_reseed_service_instances!
         ServiceUsageEvent.insert(column_map.keys, usage_query)
       end
 
-      def delete_events_older_than(cutoff_age_in_days)
+      def delete_events_older_than(cutoff_age_in_days, threshold_for_keeping_unprocessed_records)
         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
+                                                          keep_unprocessed_records: true,
+                                                          threshold_for_keeping_unprocessed_records: threshold_for_keeping_unprocessed_records).delete
       end
     end
   end
diff --git a/config/bosh-lite.yml b/config/bosh-lite.yml
@@ -10,7 +10,7 @@ newrelic_enabled: false
 external_protocol: https
 external_domain: api.bosh-lite.com
 
-system_domain_organization: 
+system_domain_organization:
 system_domain: bosh-lite.com
 app_domains:
   - name: bosh-lite.com
@@ -21,9 +21,11 @@ jobs:
 
 app_usage_events:
   cutoff_age_in_days: 31
+  threshold_for_keeping_unprocessed_records: 5000000
 
 service_usage_events:
   cutoff_age_in_days: 31
+  threshold_for_keeping_unprocessed_records: 5000000
 
 audit_events:
   cutoff_age_in_days: 31
@@ -133,8 +135,8 @@ resource_pool:
   maximum_size: 536870912
   resource_directory_key: bosh-lite.com-cc-resources
   cdn:
-    uri: 
-    key_pair_id: 
+    uri:
+    key_pair_id:
     private_key: ""
   fog_connection: {}
 
@@ -150,8 +152,8 @@ packages:
   max_valid_packages_stored: 5
   max_package_size: 1073741824
   cdn:
-    uri: 
-    key_pair_id: 
+    uri:
+    key_pair_id:
     private_key: ""
   fog_connection: {}
 
@@ -165,8 +167,8 @@ droplets:
     password: blobstore-password
   droplet_directory_key: bosh-lite.com-cc-droplets
   cdn:
-    uri: 
-    key_pair_id: 
+    uri:
+    key_pair_id:
     private_key: ""
   fog_connection: {}
   max_staged_droplets_stored: 5
@@ -181,8 +183,8 @@ buildpacks:
     password: blobstore-password
   buildpack_directory_key: bosh-lite.com-cc-buildpacks
   cdn:
-    uri: 
-    key_pair_id: 
+    uri:
+    key_pair_id:
     private_key: ""
   fog_connection: {}
 
@@ -211,7 +213,7 @@ default_app_ssh_access: true
 
 skip_cert_verify: true
 
-install_buildpacks: 
+install_buildpacks:
 - name: staticfile_buildpack
   package: staticfile-buildpack
 - name: java_buildpack
diff --git a/config/cloud_controller.yml b/config/cloud_controller.yml
@@ -38,9 +38,11 @@ jobs:
 
 app_usage_events:
   cutoff_age_in_days: 31
+  threshold_for_keeping_unprocessed_records: 5000000
 
 service_usage_events:
   cutoff_age_in_days: 31
+  threshold_for_keeping_unprocessed_records: 5000000
 
 audit_events:
   cutoff_age_in_days: 31
diff --git a/lib/cloud_controller/clock/scheduler.rb b/lib/cloud_controller/clock/scheduler.rb
@@ -5,9 +5,11 @@
 module VCAP::CloudController
   class Scheduler
     CLEANUPS = [
-      { name: 'app_usage_events', class: Jobs::Runtime::AppUsageEventsCleanup, time: '18:00', arg_from_config: %i[app_usage_events cutoff_age_in_days] },
+      { name: 'app_usage_events', class: Jobs::Runtime::AppUsageEventsCleanup, time: '18:00',
+        arg_from_config: [%i[app_usage_events cutoff_age_in_days], %i[app_usage_events threshold_for_keeping_unprocessed_records]] },
       { name: 'audit_events', class: Jobs::Runtime::EventsCleanup, time: '20:00', arg_from_config: %i[audit_events cutoff_age_in_days] },
-      { name: 'service_usage_events', class: Jobs::Services::ServiceUsageEventsCleanup, time: '22:00', arg_from_config: %i[service_usage_events cutoff_age_in_days] },
+      { name: 'service_usage_events', class: Jobs::Services::ServiceUsageEventsCleanup, time: '22:00',
+        arg_from_config: [%i[service_usage_events cutoff_age_in_days], %i[service_usage_events threshold_for_keeping_unprocessed_records]] },
       { name: 'completed_tasks', class: Jobs::Runtime::PruneCompletedTasks, time: '23:00', arg_from_config: %i[completed_tasks cutoff_age_in_days] },
       { name: 'expired_blob_cleanup', class: Jobs::Runtime::ExpiredBlobCleanup, time: '00:00' },
       { name: 'expired_resource_cleanup', class: Jobs::Runtime::ExpiredResourceCleanup, time: '00:30' },
@@ -87,7 +89,13 @@ def start_daily_jobs
           klass = cleanup_config[:class]
 
           if cleanup_config[:arg_from_config]
-            klass.new(@config.get(*cleanup_config[:arg_from_config]))
+            args = cleanup_config[:arg_from_config]
+            if args.first.is_a?(Array)
+              args = args.map { |arg| @config.get(*arg) }
+              klass.new(*args)
+            else
+              klass.new(@config.get(*args))
+            end
           else
             klass.new
           end
diff --git a/lib/cloud_controller/config_schemas/clock_schema.rb b/lib/cloud_controller/config_schemas/clock_schema.rb
@@ -15,7 +15,8 @@ class ClockSchema < VCAP::Config
             clock: Integer
           },
           app_usage_events: {
-            cutoff_age_in_days: Integer
+            cutoff_age_in_days: Integer,
+            threshold_for_keeping_unprocessed_records: Integer
           },
           audit_events: {
             cutoff_age_in_days: Integer
@@ -206,7 +207,10 @@ class ClockSchema < VCAP::Config
             frequency_in_seconds: Integer
           },
 
-          service_usage_events: { cutoff_age_in_days: Integer },
+          service_usage_events: {
+            cutoff_age_in_days: Integer,
+            threshold_for_keeping_unprocessed_records: Integer
+          },
           default_app_ssh_access: bool,
           allow_app_ssh_access: bool,
           jobs: {
diff --git a/lib/database/old_record_cleanup.rb b/lib/database/old_record_cleanup.rb
@@ -3,14 +3,16 @@
 module Database
   class OldRecordCleanup
     class NoCurrentTimestampError < StandardError; end
-    attr_reader :model, :days_ago, :keep_at_least_one_record, :keep_running_records, :keep_unprocessed_records
+    attr_reader :model, :days_ago, :keep_at_least_one_record, :keep_running_records, :keep_unprocessed_records, :threshold_for_keeping_unprocessed_records
 
-    def initialize(model, cutoff_age_in_days:, keep_at_least_one_record: false, keep_running_records: false, keep_unprocessed_records: false)
+    def initialize(model, cutoff_age_in_days:, keep_at_least_one_record: false, keep_running_records: false, keep_unprocessed_records: false,
+                   threshold_for_keeping_unprocessed_records: nil)
       @model = model
       @days_ago = cutoff_age_in_days
       @keep_at_least_one_record = keep_at_least_one_record
       @keep_running_records = keep_running_records
       @keep_unprocessed_records = keep_unprocessed_records
+      @threshold_for_keeping_unprocessed_records = threshold_for_keeping_unprocessed_records
     end
 
     def delete
@@ -97,16 +99,22 @@ def exclude_unprocessed_records(old_records)
 
       return old_records unless consumer_model
 
-      # Find the usage event record with the lowest ID
-      # of any that are referenced by a consumer
-      referenced_event = model.
-                         join(consumer_model.table_name.to_sym, last_processed_guid: :guid).
-                         select(Sequel.function(:min, Sequel.qualify(model.table_name, :id)).as(:min_id)).
-                         first
-
-      return old_records if referenced_event[:min_id].nil?
-
-      old_records.where { id < referenced_event[:min_id] }
+      if approximate_row_count(model) < threshold_for_keeping_unprocessed_records.to_i
+        # Find the usage event record with the lowest ID
+        # of any that are referenced by a consumer
+        referenced_event = model.
+                           join(consumer_model.table_name.to_sym, last_processed_guid: :guid).
+                           select(Sequel.function(:min, Sequel.qualify(model.table_name, :id)).as(:min_id)).
+                           first
+
+        return old_records if referenced_event[:min_id].nil?
+
+        old_records.where { id < referenced_event[:min_id] }
+      else
+        # When above threshold, we don't exclude any records
+        # Associated consumers will be automatically deleted by Sequel
+        old_records
+      end
     end
 
     def registered_consumer_model(model)
@@ -116,5 +124,27 @@ def registered_consumer_model(model)
 
       false
     end
+
+    def approximate_row_count(model)
+      case model.db.database_type
+      when :postgres
+        result = model.db[<<-SQL.squish
+          SELECT reltuples::bigint AS estimate
+          FROM pg_class
+          WHERE relname = '#{model.table_name}'
+        SQL
+        ].first
+        result[:estimate].to_i
+      when :mysql, :mysql2
+        result = model.db[<<-SQL.squish
+          SELECT table_rows
+          FROM information_schema.tables
+          WHERE table_schema = DATABASE()
+            AND table_name = '#{model.table_name}'
+        SQL
+        ].first
+        result[:table_rows].to_i
+      end
+    end
   end
 end
diff --git a/spec/fixtures/config/port_8181_config.yml b/spec/fixtures/config/port_8181_config.yml
@@ -30,6 +30,7 @@ jobs:
 
 app_usage_events:
   cutoff_age_in_days: 31
+  threshold_for_keeping_unprocessed_records: 5000000
 
 audit_events:
   cutoff_age_in_days: 31
diff --git a/spec/unit/jobs/runtime/app_usage_events_cleanup_spec.rb b/spec/unit/jobs/runtime/app_usage_events_cleanup_spec.rb
@@ -4,12 +4,13 @@ module VCAP::CloudController
   module Jobs::Runtime
     RSpec.describe AppUsageEventsCleanup, job_context: :worker do
       let(:cutoff_age_in_days) { 30 }
+      let(:threshold_for_keeping_unprocessed_records) { 5_000_000 }
       let(:logger) { double(Steno::Logger, info: nil) }
       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
-        AppUsageEventsCleanup.new(cutoff_age_in_days)
+        AppUsageEventsCleanup.new(cutoff_age_in_days, threshold_for_keeping_unprocessed_records)
       end
 
       before do
diff --git a/spec/unit/jobs/services/service_usage_events_cleanup_spec.rb b/spec/unit/jobs/services/service_usage_events_cleanup_spec.rb
@@ -4,12 +4,13 @@ module VCAP::CloudController
   module Jobs::Services
     RSpec.describe ServiceUsageEventsCleanup, job_context: :worker do
       let(:cutoff_age_in_days) { 30 }
+      let(:threshold_for_keeping_unprocessed_records) { 5_000_000 }
       let(:logger) { double(Steno::Logger, info: nil) }
       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
-        ServiceUsageEventsCleanup.new(cutoff_age_in_days)
+        ServiceUsageEventsCleanup.new(cutoff_age_in_days, threshold_for_keeping_unprocessed_records)
       end
 
       before do
diff --git a/spec/unit/lib/cloud_controller/clock/scheduler_spec.rb b/spec/unit/lib/cloud_controller/clock/scheduler_spec.rb
@@ -16,12 +16,18 @@ module VCAP::CloudController
           jobs: {
             global: { timeout_in_seconds: global_timeout }
           },
-          app_usage_events: { cutoff_age_in_days: 1 },
+          app_usage_events: {
+            cutoff_age_in_days: 1,
+            threshold_for_keeping_unprocessed_records: 5_000_000
+          },
           audit_events: { cutoff_age_in_days: 3 },
           failed_jobs: { frequency_in_seconds: 400, cutoff_age_in_days: 4, max_number_of_failed_delayed_jobs: 10 },
           pollable_jobs: { cutoff_age_in_days: 2 },
           service_operations_initial_cleanup: { frequency_in_seconds: 600 },
-          service_usage_events: { cutoff_age_in_days: 5 },
+          service_usage_events: {
+            cutoff_age_in_days: 5,
+            threshold_for_keeping_unprocessed_records: 5_000_000
+          },
           completed_tasks: { cutoff_age_in_days: 6 },
           pending_droplets: { frequency_in_seconds: 300, expiration_in_seconds: 600 },
           pending_builds: { frequency_in_seconds: 400, expiration_in_seconds: 700 },
@@ -61,7 +67,7 @@ module VCAP::CloudController
 
         expect(clock).to receive(:schedule_daily_job) do |args, &block|
           expect(args).to eql(name: 'app_usage_events', at: '18:00', priority: 0)
-          expect(Jobs::Runtime::AppUsageEventsCleanup).to receive(:new).with(1).and_call_original
+          expect(Jobs::Runtime::AppUsageEventsCleanup).to receive(:new).with(1, 5_000_000).and_call_original
           expect(block.call).to be_instance_of(Jobs::Runtime::AppUsageEventsCleanup)
         end
 
@@ -73,7 +79,7 @@ module VCAP::CloudController
 
         expect(clock).to receive(:schedule_daily_job) do |args, &block|
           expect(args).to eql(name: 'service_usage_events', at: '22:00', priority: 0)
-          expect(Jobs::Services::ServiceUsageEventsCleanup).to receive(:new).with(5).and_call_original
+          expect(Jobs::Services::ServiceUsageEventsCleanup).to receive(:new).with(5, 5_000_000).and_call_original
           expect(block.call).to be_instance_of(Jobs::Services::ServiceUsageEventsCleanup)
         end
 
diff --git a/spec/unit/lib/database/old_record_cleanup_spec.rb b/spec/unit/lib/database/old_record_cleanup_spec.rb
diff --git a/spec/unit/repositories/app_usage_event_repository_spec.rb b/spec/unit/repositories/app_usage_event_repository_spec.rb
diff --git a/spec/unit/repositories/service_usage_event_repository_spec.rb b/spec/unit/repositories/service_usage_event_repository_spec.rb
