feat: Add usage snapshots API for point-in-time app and service metrics

Introduces new V3 API endpoints for capturing point-in-time usage data:

App Usage Snapshots (/v3/app_usage/snapshots):
- Creates snapshots of all running processes across the platform
- Captures instance counts, memory allocation, and buildpack info
- Data organized by organization and space in paginated chunks

Service Usage Snapshots (/v3/service_usage/snapshots):
- Creates snapshots of all service instances across the platform
- Captures service plan, offering, and broker information
- Supports both managed and user-provided service instances

Both snapshot types:
- Are admin-only operations that run asynchronously via pollable jobs
- Include a checkpoint reference (GUID) to the most recent usage event
- Support automatic cleanup of old and stale snapshots via daily jobs
- Expose Prometheus metrics for generation duration and failure tracking

Author: joyvuu-dave

app/controllers/v3/app_usage_snapshots_controller.rb

@@ -0,0 +1,90 @@
+require 'presenters/v3/app_usage_snapshot_presenter'
+require 'presenters/v3/app_usage_snapshot_chunk_presenter'
+require 'messages/app_usage_snapshots_create_message'
+require 'messages/app_usage_snapshots_list_message'
+require 'fetchers/app_usage_snapshot_list_fetcher'
+require 'jobs/runtime/app_usage_snapshot_generator_job'
+
+class AppUsageSnapshotsController < ApplicationController
+  def index
+    message = AppUsageSnapshotsListMessage.from_params(query_params)
+    unprocessable!(message.errors.full_messages) unless message.valid?
+
+    dataset = AppUsageSnapshot.where(guid: [])
+    dataset = AppUsageSnapshotListFetcher.fetch_all(message, AppUsageSnapshot.dataset) if permission_queryer.can_read_globally?
+
+    render status: :ok, json: Presenters::V3::PaginatedListPresenter.new(
+      presenter: Presenters::V3::AppUsageSnapshotPresenter,
+      paginated_result: SequelPaginator.new.get_page(dataset, message.try(:pagination_options)),
+      path: '/v3/app_usage/snapshots',
+      message: message
+    )
+  end
+
+  def show
+    snapshot_not_found! unless permission_queryer.can_read_globally?
+
+    snapshot = AppUsageSnapshot.first(guid: hashed_params[:guid])
+    snapshot_not_found! unless snapshot
+
+    render status: :ok, json: Presenters::V3::AppUsageSnapshotPresenter.new(snapshot)
+  end
+
+  def create
+    message = AppUsageSnapshotsCreateMessage.new(hashed_params[:body])
+    unprocessable!(message.errors.full_messages) unless message.valid?
+
+    unauthorized! unless permission_queryer.can_write_globally?
+
+    existing_snapshot = AppUsageSnapshot.where(completed_at: nil).first
+    raise CloudController::Errors::ApiError.new_from_details('AppUsageSnapshotGenerationInProgress') if existing_snapshot
+
+    snapshot = AppUsageSnapshot.create(
+      checkpoint_event_guid: nil,
+      created_at: Time.now.utc,
+      completed_at: nil,
+      instance_count: 0,
+      organization_count: 0,
+      space_count: 0,
+      app_count: 0,
+      chunk_count: 0
+    )
+
+    begin
+      job = Jobs::Runtime::AppUsageSnapshotGeneratorJob.new(snapshot.guid)
+      pollable_job = Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(job)
+    rescue StandardError
+      snapshot.destroy
+      raise
+    end
+
+    head :accepted, 'Location' => url_builder.build_url(path: "/v3/jobs/#{pollable_job.guid}")
+  end
+
+  def chunks
+    snapshot_not_found! unless permission_queryer.can_read_globally?
+
+    snapshot = AppUsageSnapshot.first(guid: hashed_params[:guid])
+    snapshot_not_found! unless snapshot
+
+    unprocessable!('Snapshot is still processing') unless snapshot.complete?
+
+    pagination_options = PaginationOptions.from_params(query_params)
+    paginated_result = SequelPaginator.new.get_page(
+      snapshot.app_usage_snapshot_chunks_dataset,
+      pagination_options
+    )
+
+    render status: :ok, json: Presenters::V3::PaginatedListPresenter.new(
+      presenter: Presenters::V3::AppUsageSnapshotChunkPresenter,
+      paginated_result: paginated_result,
+      path: "/v3/app_usage/snapshots/#{snapshot.guid}/chunks"
+    )
+  end
+
+  private
+
+  def snapshot_not_found!
+    resource_not_found!(:app_usage_snapshot)
+  end
+end

app/controllers/v3/service_usage_snapshots_controller.rb

@@ -0,0 +1,89 @@
+require 'presenters/v3/service_usage_snapshot_presenter'
+require 'presenters/v3/service_usage_snapshot_chunk_presenter'
+require 'messages/service_usage_snapshots_create_message'
+require 'messages/service_usage_snapshots_list_message'
+require 'fetchers/service_usage_snapshot_list_fetcher'
+require 'jobs/runtime/service_usage_snapshot_generator_job'
+
+class ServiceUsageSnapshotsController < ApplicationController
+  def index
+    message = ServiceUsageSnapshotsListMessage.from_params(query_params)
+    unprocessable!(message.errors.full_messages) unless message.valid?
+
+    dataset = ServiceUsageSnapshot.where(guid: [])
+    dataset = ServiceUsageSnapshotListFetcher.fetch_all(message, ServiceUsageSnapshot.dataset) if permission_queryer.can_read_globally?
+
+    render status: :ok, json: Presenters::V3::PaginatedListPresenter.new(
+      presenter: Presenters::V3::ServiceUsageSnapshotPresenter,
+      paginated_result: SequelPaginator.new.get_page(dataset, message.try(:pagination_options)),
+      path: '/v3/service_usage/snapshots',
+      message: message
+    )
+  end
+
+  def show
+    snapshot_not_found! unless permission_queryer.can_read_globally?
+
+    snapshot = ServiceUsageSnapshot.first(guid: hashed_params[:guid])
+    snapshot_not_found! unless snapshot
+
+    render status: :ok, json: Presenters::V3::ServiceUsageSnapshotPresenter.new(snapshot)
+  end
+
+  def create
+    message = ServiceUsageSnapshotsCreateMessage.new(hashed_params[:body])
+    unprocessable!(message.errors.full_messages) unless message.valid?
+
+    unauthorized! unless permission_queryer.can_write_globally?
+
+    existing_snapshot = ServiceUsageSnapshot.where(completed_at: nil).first
+    raise CloudController::Errors::ApiError.new_from_details('ServiceUsageSnapshotGenerationInProgress') if existing_snapshot
+
+    snapshot = ServiceUsageSnapshot.create(
+      checkpoint_event_guid: nil,
+      created_at: Time.now.utc,
+      completed_at: nil,
+      service_instance_count: 0,
+      organization_count: 0,
+      space_count: 0,
+      chunk_count: 0
+    )
+
+    begin
+      job = Jobs::Runtime::ServiceUsageSnapshotGeneratorJob.new(snapshot.guid)
+      pollable_job = Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(job)
+    rescue StandardError
+      snapshot.destroy
+      raise
+    end
+
+    head :accepted, 'Location' => url_builder.build_url(path: "/v3/jobs/#{pollable_job.guid}")
+  end
+
+  def chunks
+    snapshot_not_found! unless permission_queryer.can_read_globally?
+
+    snapshot = ServiceUsageSnapshot.first(guid: hashed_params[:guid])
+    snapshot_not_found! unless snapshot
+
+    unprocessable!('Snapshot is still processing') unless snapshot.complete?
+
+    pagination_options = PaginationOptions.from_params(query_params)
+    paginated_result = SequelPaginator.new.get_page(
+      snapshot.service_usage_snapshot_chunks_dataset,
+      pagination_options
+    )
+
+    render status: :ok, json: Presenters::V3::PaginatedListPresenter.new(
+      presenter: Presenters::V3::ServiceUsageSnapshotChunkPresenter,
+      paginated_result: paginated_result,
+      path: "/v3/service_usage/snapshots/#{snapshot.guid}/chunks"
+    )
+  end
+
+  private
+
+  def snapshot_not_found!
+    resource_not_found!(:service_usage_snapshot)
+  end
+end

app/fetchers/app_usage_snapshot_list_fetcher.rb

@@ -0,0 +1,17 @@
+require 'fetchers/base_list_fetcher'
+
+module VCAP::CloudController
+  class AppUsageSnapshotListFetcher < BaseListFetcher
+    class << self
+      def fetch_all(message, dataset)
+        filter(message, dataset)
+      end
+
+      private
+
+      def filter(message, dataset)
+        super(message, dataset, AppUsageSnapshot)
+      end
+    end
+  end
+end

app/fetchers/service_usage_snapshot_list_fetcher.rb

@@ -0,0 +1,17 @@
+require 'fetchers/base_list_fetcher'
+
+module VCAP::CloudController
+  class ServiceUsageSnapshotListFetcher < BaseListFetcher
+    class << self
+      def fetch_all(message, dataset)
+        filter(message, dataset)
+      end
+
+      private
+
+      def filter(message, dataset)
+        super(message, dataset, ServiceUsageSnapshot)
+      end
+    end
+  end
+end

app/jobs/runtime/app_usage_snapshot_cleanup.rb

@@ -0,0 +1,45 @@
+module VCAP::CloudController
+  module Jobs
+    module Runtime
+      class AppUsageSnapshotCleanup < VCAP::CloudController::Jobs::CCJob
+        attr_accessor :cutoff_age_in_days
+
+        def initialize(cutoff_age_in_days)
+          @cutoff_age_in_days = cutoff_age_in_days
+        end
+
+        def perform
+          logger = Steno.logger('cc.background')
+          logger.info("Cleaning up usage snapshots older than #{cutoff_age_in_days} days")
+
+          cutoff_time = Time.now.utc - cutoff_age_in_days.days
+
+          old_completed = AppUsageSnapshot.where(
+            Sequel.lit('created_at < ? AND completed_at IS NOT NULL', cutoff_time)
+          )
+
+          stale_timeout = Time.now.utc - 1.hour
+          stale_in_progress = AppUsageSnapshot.where(
+            Sequel.lit('created_at < ? AND completed_at IS NULL', stale_timeout)
+          )
+
+          completed_count = old_completed.count
+          stale_count = stale_in_progress.count
+
+          old_completed.delete
+          stale_in_progress.delete
+
+          logger.info("Deleted #{completed_count} old completed snapshots and #{stale_count} stale in-progress snapshots")
+        end
+
+        def job_name_in_configuration
+          :app_usage_snapshot_cleanup
+        end
+
+        def max_attempts
+          1
+        end
+      end
+    end
+  end
+end

app/jobs/runtime/app_usage_snapshot_generator_job.rb

@@ -0,0 +1,47 @@
+require 'repositories/app_usage_snapshot_repository'
+
+module VCAP::CloudController
+  module Jobs
+    module Runtime
+      class AppUsageSnapshotGeneratorJob < VCAP::CloudController::Jobs::CCJob
+        attr_reader :resource_guid
+
+        def initialize(snapshot_guid)
+          @resource_guid = snapshot_guid
+        end
+
+        def perform
+          logger = Steno.logger('cc.background')
+          logger.info("Starting usage snapshot generation for snapshot #{@resource_guid}")
+
+          snapshot = AppUsageSnapshot.first(guid: @resource_guid)
+          raise "Snapshot not found: #{@resource_guid}" unless snapshot
+
+          repository = Repositories::AppUsageSnapshotRepository.new
+          repository.populate_snapshot!(snapshot)
+
+          logger.info("Usage snapshot #{snapshot.guid} completed: #{snapshot.instance_count} instances")
+        rescue StandardError => e
+          logger.error("Usage snapshot generation failed: #{e.message}\n#{e.backtrace.join("\n")}")
+          raise
+        end
+
+        def job_name_in_configuration
+          :app_usage_snapshot_generator
+        end
+
+        def max_attempts
+          1
+        end
+
+        def resource_type
+          'app_usage_snapshot'
+        end
+
+        def display_name
+          'app_usage_snapshot.generate'
+        end
+      end
+    end
+  end
+end

app/jobs/runtime/service_usage_snapshot_cleanup.rb

@@ -0,0 +1,45 @@
+module VCAP::CloudController
+  module Jobs
+    module Runtime
+      class ServiceUsageSnapshotCleanup < VCAP::CloudController::Jobs::CCJob
+        attr_accessor :cutoff_age_in_days
+
+        def initialize(cutoff_age_in_days)
+          @cutoff_age_in_days = cutoff_age_in_days
+        end
+
+        def perform
+          logger = Steno.logger('cc.background')
+          logger.info("Cleaning up service usage snapshots older than #{cutoff_age_in_days} days")
+
+          cutoff_time = Time.now.utc - cutoff_age_in_days.days
+
+          old_completed = ServiceUsageSnapshot.where(
+            Sequel.lit('created_at < ? AND completed_at IS NOT NULL', cutoff_time)
+          )
+
+          stale_timeout = Time.now.utc - 1.hour
+          stale_in_progress = ServiceUsageSnapshot.where(
+            Sequel.lit('created_at < ? AND completed_at IS NULL', stale_timeout)
+          )
+
+          completed_count = old_completed.count
+          stale_count = stale_in_progress.count
+
+          old_completed.delete
+          stale_in_progress.delete
+
+          logger.info("Deleted #{completed_count} old completed snapshots and #{stale_count} stale in-progress snapshots")
+        end
+
+        def job_name_in_configuration
+          :service_usage_snapshot_cleanup
+        end
+
+        def max_attempts
+          1
+        end
+      end
+    end
+  end
+end