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

The scheduled usage event cleanup job used to delete every record older than
the configured cutoff age, including the opening STARTED/CREATED event of a
resource that is still running. Once the cleanup deleted that event, nothing
was left to reconstruct what is running right now.

Database::OldRecordCleanup can now optionally keep the records of running
resources. Each model declares its lifecycles via usage_lifecycles: which
states open a run (STARTED/CREATED/TASK_STARTED, plus the
WAS_RUNNING/TASK_WAS_RUNNING baselines), which state closes it
(STOPPED/DELETED/TASK_STOPPED), and which column names the resource. An old
opening event is then only deleted when:

* a closing event for the same resource exists later and is also old -- the
  run is over; or
* it is neither the first opening of the current run nor the resource's
  latest one (again judged only against old rows). Consumers only need the
  first opening (the true start time) and the latest (the current size). The
  ones in between, written each time a running resource is scaled or updated,
  tell a consumer nothing it still needs -- and deleting them is what keeps
  the table size bounded for long-running, frequently-changed resources.

The app and service usage event repositories turn this on with
keep_running_records: true. Asking for it on a model without usage_lifecycles
raises an error instead of silently deleting the records of running
resources. Task events get their own lifecycle (TASK_STARTED/TASK_WAS_RUNNING
-> TASK_STOPPED, matched by task_guid), so the start events of long-running
tasks survive cleanup too. Task baselines use their own TASK_WAS_RUNNING
state because task events carry an empty app_guid: if they said WAS_RUNNING,
the app lifecycle would see them all as events of one app whose guid is ''
and wrongly delete them (and the backfill's repair would write bogus STOPPED
events for that phantom app).

Deletion runs in a deliberate order: first the opening events that are safe
to delete, while the events that make them safe still exist; then everything
else. The reverse order could delete a closing event first and leave its
opening event looking like a still-running resource. The cleanup log line now
reports the row counts BatchDelete returns instead of running extra COUNT
queries, and BatchDelete fetches each batch's ids in the same query that
checks whether anything is left, halving the evaluations of the (potentially
expensive) filtered dataset. Also renames the positional days_ago argument 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,18 @@
 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. If task baselines said WAS_RUNNING, the cleanup
+      # and the backfill's repair would both treat every task baseline as
+      # belonging to a single app whose guid is '' -- the cleanup would wrongly
+      # prune them, and the repair would write bogus STOPPED events for that
+      # phantom app.
+      TASK_WAS_RUNNING_EVENT_STATE = 'TASK_WAS_RUNNING'.freeze
+
       def find(guid)
         AppUsageEvent.find(guid:)
       end
@@ -152,7 +164,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,36 @@
 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
+      return unless keep_running_records && !model.respond_to?(:usage_lifecycles)
+
+      # Fail here, when the caller builds the object, not later when the
+      # cleanup job runs and nobody is watching.
+      raise ArgumentError.new("keep_running_records requires #{model} to define .usage_lifecycles")
     end
 
+    # The two options work together: keep_running_records always keeps the
+    # beginning events that show a resource is still running (those with no
+    # later ending event for the same resource), and keep_at_least_one_record
+    # additionally keeps 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
+        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 +46,101 @@ 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 size); the rows in between, written each time a running
+    #   resource is scaled or updated, tell a consumer nothing it still needs.
+    #
+    # 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 the subqueries use the (state, guid, id) lifecycle index,
+    # and a higher id always means created later. They only look at OLD rows: a
+    # superseded beginning is kept until the row that supersedes it is itself
+    # old, so a consumer reading within the retention window never sees the
+    # decision change underneath it.
+    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 removes nothing here today: a row is only
+      # prunable because a later row for the same resource exists, so a
+      # prunable row can never be the newest row in the table. We keep the
+      # call anyway so keep_at_least_one_record still holds even if the
+      # rules above change.
+      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) { create(:app_usage_event, created_at: (cutoff_age_in_days + 1).days.ago) }
+      let!(:event_before_threshold) { create(:app_usage_event, created_at: (cutoff_age_in_days + 1).days.ago, state: 'STOPPED') }
       let!(:event_after_threshold) { create(:app_usage_event, 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) { create(:service_usage_event, created_at: (cutoff_age_in_days + 1).days.ago) }
+      let!(:event_before_threshold) { create(:service_usage_event, created_at: (cutoff_age_in_days + 1).days.ago, state: 'DELETED') }
       let!(:event_after_threshold) { create(:service_usage_event, created_at: (cutoff_age_in_days - 1).days.ago) }
 
       subject(:job) do