Keep usage event records of running apps, service instances, and tasks

App and service usage event cleanup previously pruned every record older than
the cutoff, including the opening STARTED/CREATED event of a resource that is
still running -- which makes it impossible to reconstruct current usage once
that event ages out.

Database::OldRecordCleanup can now optionally keep "running" records. For each
lifecycle a model declares via usage_lifecycles (beginning states, ending
state, guid column), a beginning-state event (STARTED/CREATED/TASK_STARTED,
and the WAS_RUNNING/TASK_WAS_RUNNING baselines) is retained unless:

* a later ending-state event (STOPPED/DELETED/TASK_STOPPED) for the same
  resource also falls outside the retention window -- the run is over; or
* it is a superseded baseline: an earlier beginning of the same run and a
  later beginning both exist outside the window. Consumers only need the first
  beginning of the current run (the true start time) and the latest one (the
  current footprint), so the in-between events written by scaling an app or
  updating a service instance are pruned and cutoff_age_in_days keeps bounding
  the table size for long-running, frequently-changed resources.

The app and service usage event repositories enable the behavior with
keep_running_records: true; requesting it for a model without usage_lifecycles
raises instead of silently deleting the records of running resources. Task
events get their own lifecycle (TASK_STARTED/TASK_WAS_RUNNING -> TASK_STOPPED,
keyed by task_guid), so the start events of long-running tasks survive cleanup
as well. The task baseline state is distinct from WAS_RUNNING because task
events share the app_usage_events table but carry an empty app_guid: reusing
WAS_RUNNING would let the app lifecycle correlate every task baseline through
app_guid = '' and wrongly prune them as superseded baselines of one phantom
app (and the app backfill's stale-row sweep would delete them outright).

Deletion runs in ordered passes -- prunable beginning rows first, while the
rows that make them prunable still exist, then everything else -- so a
beginning row cannot be stranded when its pair is removed in an earlier batch.
The cleanup log line now reports the row counts BatchDelete returns instead of
issuing extra COUNT queries, and BatchDelete fetches each batch's ids in the
same query that checks for emptiness, halving evaluations of the (potentially
expensive) filtered dataset. Also renames the positional days_ago to a
cutoff_age_in_days keyword.

Author: joyvuu-dave

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/models/runtime/app_usage_event.rb

@@ -9,6 +9,23 @@ class AppUsageEvent < Sequel::Model
                       :buildpack_guid, :buildpack_name,
                       :package_state, :previous_package_state, :parent_app_guid,
                       :parent_app_name, :process_type, :task_name, :task_guid
+
+    def self.usage_lifecycles
+      [
+        {
+          beginning_states: [ProcessModel::STARTED, Repositories::AppUsageEventRepository::WAS_RUNNING_EVENT_STATE],
+          ending_state: ProcessModel::STOPPED,
+          guid_column: :app_guid
+        },
+        {
+          beginning_states: [Repositories::AppUsageEventRepository::TASK_STARTED_EVENT_STATE,
+                             Repositories::AppUsageEventRepository::TASK_WAS_RUNNING_EVENT_STATE],
+          ending_state: Repositories::AppUsageEventRepository::TASK_STOPPED_EVENT_STATE,
+          guid_column: :task_guid
+        }
+      ].freeze
+    end
+
     AppUsageEvent.dataset_module do
       def supports_window_functions?
         false

app/models/services/service_usage_event.rb

@@ -7,5 +7,17 @@ class ServiceUsageEvent < Sequel::Model
                       :service_plan_guid, :service_plan_name,
                       :service_guid, :service_label,
                       :service_broker_name, :service_broker_guid
+
+    def self.usage_lifecycles
+      [
+        {
+          beginning_states: [Repositories::ServiceUsageEventRepository::CREATED_EVENT_STATE,
+                             Repositories::ServiceUsageEventRepository::UPDATED_EVENT_STATE,
+                             Repositories::ServiceUsageEventRepository::WAS_RUNNING_EVENT_STATE],
+          ending_state: Repositories::ServiceUsageEventRepository::DELETED_EVENT_STATE,
+          guid_column: :service_instance_guid
+        }
+      ].freeze
+    end
   end
 end

app/repositories/app_usage_event_repository.rb

@@ -4,6 +4,16 @@
 module VCAP::CloudController
   module Repositories
     class AppUsageEventRepository
+      WAS_RUNNING_EVENT_STATE = 'WAS_RUNNING'.freeze
+      TASK_STARTED_EVENT_STATE = 'TASK_STARTED'.freeze
+      TASK_STOPPED_EVENT_STATE = 'TASK_STOPPED'.freeze
+      # Task baselines get their own state (rather than reusing WAS_RUNNING)
+      # because task events share the app_usage_events table with app events but
+      # carry an empty app_guid: a WAS_RUNNING row keyed by app_guid '' would be
+      # correlated with every other task's baseline by the app lifecycle's
+      # cleanup and swept by the app backfill's stale-row sweep.
+      TASK_WAS_RUNNING_EVENT_STATE = 'TASK_WAS_RUNNING'.freeze
+
       def find(guid)
         AppUsageEvent.find(guid:)
       end
@@ -152,7 +162,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

@@ -7,6 +7,7 @@ class ServiceUsageEventRepository
       DELETED_EVENT_STATE = 'DELETED'.freeze
       CREATED_EVENT_STATE = 'CREATED'.freeze
       UPDATED_EVENT_STATE = 'UPDATED'.freeze
+      WAS_RUNNING_EVENT_STATE = 'WAS_RUNNING'.freeze
 
       def find(guid)
         ServiceUsageEvent.find(guid:)
@@ -92,7 +93,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

lib/database/batch_delete.rb

@@ -11,19 +11,21 @@ def delete
       total_count = 0
 
       loop do
-        set = dataset.limit(amount)
-        break if set.empty?
+        # Fetch the batch's ids in the same query that checks for emptiness, so the
+        # (potentially expensive) filtered dataset is evaluated once per batch.
+        ids = dataset.limit(amount).select_map(:id)
+        break if ids.empty?
 
-        total_count += delete_batch(set)
+        total_count += delete_batch(ids)
       end
 
       total_count
     end
 
     private
 
-    def delete_batch(set)
-      dataset.model.where(id: set.select_map(:id)).delete
+    def delete_batch(ids)
+      dataset.model.where(id: ids).delete
     end
   end
 end

lib/database/old_record_cleanup.rb

@@ -3,25 +3,33 @@
 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
 
+    # keep_running_records and keep_at_least_one_record compose: a still-running
+    # resource (a beginning-state event with no later ending-state event for the
+    # same resource) is always retained, and keep_at_least_one_record additionally
+    # protects the single newest row so the table is never fully emptied for
+    # clients that poll the most recent event.
     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
-      logger.info("Cleaning up #{old_records.count} #{model.table_name} table rows")
 
-      Database::BatchDelete.new(old_records, 1000).delete
+      if keep_running_records
+        raise ArgumentError.new("keep_running_records requires #{model} to define .usage_lifecycles") unless model.respond_to?(:usage_lifecycles)
+
+        delete_keeping_running_records(old_records)
+      else
+        old_records = exclude_newest_record(old_records)
+        logger.info("Cleaning up #{old_records.count} #{model.table_name} table rows")
+        Database::BatchDelete.new(old_records, 1000).delete
+      end
     end
 
     private
@@ -35,5 +43,96 @@ def current_timestamp_from_database
     def logger
       @logger ||= Steno.logger('cc.old_record_cleanup')
     end
+
+    # Deletes old records while retaining a usable billing baseline for
+    # still-running resources.
+    #
+    # For each lifecycle of the model, a beginning-state row (e.g.
+    # STARTED/CREATED/WAS_RUNNING) is prunable when:
+    # * a later ending-state row (e.g. STOPPED/DELETED) is also old -- the run is
+    #   over; or
+    # * it is a superseded baseline: an earlier beginning of the same run and a
+    #   later beginning both exist (and are old). Consumers only need the first
+    #   beginning of the current run (the true start time) and the latest one (the
+    #   current footprint); the in-between rows written by scaling/updating a
+    #   running resource carry no baseline information.
+    #
+    # The deletes are ordered deliberately: prunable beginning rows are removed
+    # FIRST, while the rows that make them prunable still exist, so each beginning
+    # stays prunable until it is itself deleted. Only then are the ending rows (and
+    # any other, non-lifecycle states) removed. Reversing the order could strand a
+    # beginning row whose paired ending was deleted in an earlier batch.
+    def delete_keeping_running_records(old_records)
+      lifecycles = model.usage_lifecycles
+      prunable_beginnings = lifecycles.map { |lifecycle| prunable_beginnings_dataset(old_records, lifecycle) }
+
+      # Everything that is not a beginning-state row of some lifecycle (ending rows
+      # plus any other, non-lifecycle states) is unconditionally prunable.
+      all_beginning_states = lifecycles.flat_map { |lifecycle| lifecycle.fetch(:beginning_states) }
+      unconditional_records = exclude_newest_record(old_records.exclude(state: all_beginning_states))
+
+      deleted_count = prunable_beginnings.sum { |dataset| Database::BatchDelete.new(dataset, 1000).delete }
+      deleted_count += Database::BatchDelete.new(unconditional_records, 1000).delete
+
+      logger.info("Cleaned up #{deleted_count} #{model.table_name} table rows")
+    end
+
+    # Builds the dataset of old beginning-state rows that are prunable for one
+    # lifecycle. All correlations use the (state, guid, id) lifecycle index; higher
+    # id implies later creation throughout. The probes only consider OLD rows: a
+    # superseded beginning is kept until the row superseding it is itself old,
+    # which keeps the pruning decision stable for consumers reading within the
+    # retention window.
+    def prunable_beginnings_dataset(old_records, lifecycle)
+      beginning_states = lifecycle.fetch(:beginning_states)
+      ending_state = lifecycle.fetch(:ending_state)
+      guid_column = lifecycle.fetch(:guid_column)
+
+      old_beginnings = old_records.where(state: beginning_states)
+      old_endings = old_records.where(state: ending_state)
+      initial_records = old_beginnings.from_self(alias: :initial_records)
+
+      # The run is over: an ending row for the same resource was created later.
+      matching_ending = old_endings.from_self(alias: :final_records).
+                        where(Sequel[:final_records][guid_column] => Sequel[:initial_records][guid_column]).
+                        where { Sequel[:final_records][:id] > Sequel[:initial_records][:id] }.
+                        select(1).exists
+
+      # Not the run's true start: an earlier beginning of the same run exists,
+      # i.e. one with no ending event between the two.
+      intervening_ending = old_endings.from_self(alias: :intervening_endings).
+                           where(Sequel[:intervening_endings][guid_column] => Sequel[:earlier_beginnings][guid_column]).
+                           where { Sequel[:intervening_endings][:id] > Sequel[:earlier_beginnings][:id] }.
+                           where { Sequel[:intervening_endings][:id] < Sequel[:initial_records][:id] }.
+                           select(1).exists
+      earlier_beginning_in_same_run = old_beginnings.from_self(alias: :earlier_beginnings).
+                                      where(Sequel[:earlier_beginnings][guid_column] => Sequel[:initial_records][guid_column]).
+                                      where { Sequel[:earlier_beginnings][:id] < Sequel[:initial_records][:id] }.
+                                      where(Sequel.~(intervening_ending)).
+                                      select(1).exists
+
+      # Not the latest baseline: a later beginning for the same resource exists.
+      later_beginning = old_beginnings.from_self(alias: :later_beginnings).
+                        where(Sequel[:later_beginnings][guid_column] => Sequel[:initial_records][guid_column]).
+                        where { Sequel[:later_beginnings][:id] > Sequel[:initial_records][:id] }.
+                        select(1).exists
+
+      superseded_baseline = Sequel.&(earlier_beginning_in_same_run, later_beginning)
+      exclude_newest_record(initial_records.where(Sequel.|(matching_ending, superseded_baseline)))
+    end
+
+    # When keep_at_least_one_record is set, never delete the single newest row so
+    # the table always retains at least one record.
+    def exclude_newest_record(records)
+      return records unless keep_at_least_one_record && newest_record_id
+
+      records.where(Sequel.lit('id < ?', newest_record_id))
+    end
+
+    def newest_record_id
+      return @newest_record_id if defined?(@newest_record_id)
+
+      @newest_record_id = model.order(:id).last&.id
+    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