Merge pull request #2900 from loggregator/log-rate-limit-rebased

Feat: log rate limits

app/actions/build_create.rb

@@ -1,3 +1,4 @@
+require 'cloud_controller/backends/quota_validating_staging_log_rate_limit_calculator'
 require 'cloud_controller/backends/quota_validating_staging_memory_calculator'
 require 'cloud_controller/backends/staging_disk_calculator'
 require 'cloud_controller/backends/staging_environment_builder'
@@ -11,12 +12,16 @@ class BuildError < StandardError
     end
     class InvalidPackage < BuildError
     end
-    class SpaceQuotaExceeded < BuildError
+    class MemorySpaceQuotaExceeded < BuildError
     end
-    class OrgQuotaExceeded < BuildError
+    class MemoryOrgQuotaExceeded < BuildError
     end
     class DiskLimitExceeded < BuildError
     end
+    class LogRateLimitSpaceQuotaExceeded < BuildError
+    end
+    class LogRateLimitOrgQuotaExceeded < BuildError
+    end
     class StagingInProgress < BuildError
     end
 
@@ -25,12 +30,14 @@ class StagingInProgress < BuildError
     def initialize(user_audit_info: UserAuditInfo.from_context(SecurityContext),
       memory_limit_calculator: QuotaValidatingStagingMemoryCalculator.new,
       disk_limit_calculator: StagingDiskCalculator.new,
+      log_rate_limit_calculator: QuotaValidatingStagingLogRateLimitCalculator.new,
       environment_presenter: StagingEnvironmentBuilder.new)
 
       @user_audit_info         = user_audit_info
       @memory_limit_calculator = memory_limit_calculator
       @disk_limit_calculator   = disk_limit_calculator
-      @environment_builder     = environment_presenter
+      @log_rate_limit_calculator = log_rate_limit_calculator
+      @environment_builder = environment_presenter
     end
 
     def create_and_stage(package:, lifecycle:, metadata: nil, start_after_staging: false)
@@ -49,6 +56,7 @@ def create_and_stage(package:, lifecycle:, metadata: nil, start_after_staging: f
         app:                   package.app,
         staging_memory_in_mb:  staging_details.staging_memory_in_mb,
         staging_disk_in_mb:    staging_details.staging_disk_in_mb,
+        staging_log_rate_limit: staging_details.staging_log_rate_limit_bytes_per_second,
         created_by_user_guid:  @user_audit_info.user_guid,
         created_by_user_name:  @user_audit_info.user_name,
         created_by_user_email: @user_audit_info.user_email
@@ -117,6 +125,7 @@ def get_staging_details(package, lifecycle)
 
       memory_limit          = get_memory_limit(lifecycle.staging_message.staging_memory_in_mb, app, space, org)
       disk_limit            = get_disk_limit(lifecycle.staging_message.staging_disk_in_mb, app)
+      log_rate_limit        = get_log_rate_limit(lifecycle.staging_message.staging_log_rate_limit_bytes_per_second, app, space, org)
       environment_variables = @environment_builder.build(app,
         space,
         lifecycle,
@@ -128,6 +137,7 @@ def get_staging_details(package, lifecycle)
       staging_details.package               = package
       staging_details.staging_memory_in_mb  = memory_limit
       staging_details.staging_disk_in_mb    = disk_limit
+      staging_details.staging_log_rate_limit_bytes_per_second = log_rate_limit
       staging_details.environment_variables = environment_variables
       staging_details.lifecycle             = lifecycle
       staging_details.isolation_segment     = IsolationSegmentSelector.for_space(space)
@@ -146,9 +156,18 @@ def get_memory_limit(requested_limit, app, space, org)
       limit = requested_limit || app.newest_web_process&.memory
       @memory_limit_calculator.get_limit(limit, space, org)
     rescue QuotaValidatingStagingMemoryCalculator::SpaceQuotaExceeded => e
-      raise SpaceQuotaExceeded.new e.message
+      raise MemorySpaceQuotaExceeded.new e.message
     rescue QuotaValidatingStagingMemoryCalculator::OrgQuotaExceeded => e
-      raise OrgQuotaExceeded.new e.message
+      raise MemoryOrgQuotaExceeded.new e.message
+    end
+
+    def get_log_rate_limit(requested_limit, app, space, org)
+      limit = requested_limit || app.newest_web_process&.log_rate_limit
+      @log_rate_limit_calculator.get_limit(limit, space, org)
+    rescue QuotaValidatingStagingLogRateLimitCalculator::SpaceQuotaExceeded => e
+      raise LogRateLimitSpaceQuotaExceeded.new e.message
+    rescue QuotaValidatingStagingLogRateLimitCalculator::OrgQuotaExceeded => e
+      raise LogRateLimitOrgQuotaExceeded.new e.message
     end
 
     def logger

app/actions/deployment_create.rb

@@ -109,6 +109,7 @@ def clone_existing_web_process(app, revision)
           memory: web_process.memory,
           file_descriptors: web_process.file_descriptors,
           disk_quota: web_process.disk_quota,
+          log_rate_limit: web_process.log_rate_limit,
           metadata: web_process.metadata, # execution_metadata, not labels & annotations metadata!
           detected_buildpack: web_process.detected_buildpack,
           health_check_timeout: web_process.health_check_timeout,

app/actions/organization_quotas_create.rb

@@ -16,6 +16,7 @@ def create(message)
           instance_memory_limit: message.per_process_memory_in_mb || QuotaDefinition::UNLIMITED,
           app_instance_limit: message.total_instances || QuotaDefinition::UNLIMITED,
           app_task_limit: message.per_app_tasks || QuotaDefinition::UNLIMITED,
+          log_rate_limit: message.log_rate_limit_in_bytes_per_second || QuotaDefinition::UNLIMITED,
 
           # Services
           total_services: message.total_service_instances || QuotaDefinition::DEFAULT_TOTAL_SERVICES,

app/actions/organization_quotas_update.rb

@@ -13,6 +13,7 @@ def self.update(quota, message)
         quota.instance_memory_limit = instance_memory_limit(message) if message.apps_limits_message.requested? :per_process_memory_in_mb
         quota.app_instance_limit = app_instance_limit(message) if message.apps_limits_message.requested? :total_instances
         quota.app_task_limit = app_task_limit(message) if message.apps_limits_message.requested? :per_app_tasks
+        quota.log_rate_limit = log_rate_limit(message) if message.apps_limits_message.requested? :log_rate_limit_in_bytes_per_second
 
         quota.total_services = total_services(message) if message.services_limits_message.requested? :total_service_instances
         quota.total_service_keys = total_service_keys(message) if message.services_limits_message.requested? :total_service_keys
@@ -56,6 +57,10 @@ def self.app_task_limit(message)
       default_if_nil(message.per_app_tasks, QuotaDefinition::UNLIMITED)
     end
 
+    def self.log_rate_limit(message)
+      default_if_nil(message.log_rate_limit_in_bytes_per_second, QuotaDefinition::UNLIMITED)
+    end
+
     def self.total_services(message)
       default_if_nil(message.total_service_instances, QuotaDefinition::UNLIMITED)
     end

app/actions/process_scale.rb

@@ -22,7 +22,9 @@ def scale
         @process.instances = @message.instances if @message.requested?(:instances)
         @process.memory = @message.memory_in_mb if @message.requested?(:memory_in_mb)
         @process.disk_quota = @message.disk_in_mb if @message.requested?(:disk_in_mb)
-
+        if @message.requested?(:log_rate_limit_in_bytes_per_second) && !@message.log_rate_limit_in_bytes_per_second.nil?
+          @process.log_rate_limit = @message.log_rate_limit_in_bytes_per_second
+        end
         @process.save
 
         record_audit_event

app/actions/space_diff_manifest.rb

@@ -14,8 +14,10 @@ class << self
       # rubocop:todo Metrics/CyclomaticComplexity
       def generate_diff(app_manifests, space)
         json_diff = []
-        recognized_top_level_keys = AppManifestMessage.allowed_keys.map(&:to_s)
-        app_manifests = convert_byte_measurements_to_mb(app_manifests)
+
+        recognized_top_level_keys = AppManifestMessage.allowed_keys.map(&:to_s).map(&:dasherize)
+
+        app_manifests = normalize_units(app_manifests)
         app_manifests.each_with_index do |manifest_app_hash, index|
           manifest_app_hash = filter_manifest_app_hash(manifest_app_hash)
           existing_app = space.app_models.find { |app| app.name == manifest_app_hash['name'] }
@@ -90,7 +92,7 @@ def filter_manifest_app_hash(manifest_app_hash)
               'memory'
             )
           end
-          manifest_app_hash['sidecars'] = convert_byte_measurements_to_mb(manifest_app_hash['sidecars'])
+          manifest_app_hash['sidecars'] = normalize_units(manifest_app_hash['sidecars'])
           manifest_app_hash = manifest_app_hash.except('sidecars') if manifest_app_hash['sidecars'] == [{}]
         end
         if manifest_app_hash.key? 'processes'
@@ -99,6 +101,7 @@ def filter_manifest_app_hash(manifest_app_hash)
               'type',
               'command',
               'disk_quota',
+              'log-rate-limit-per-second',
               'health-check-http-endpoint',
               'health-check-invocation-timeout',
               'health-check-type',
@@ -107,7 +110,7 @@ def filter_manifest_app_hash(manifest_app_hash)
               'timeout'
             )
           end
-          manifest_app_hash['processes'] = convert_byte_measurements_to_mb(manifest_app_hash['processes'])
+          manifest_app_hash['processes'] = normalize_units(manifest_app_hash['processes'])
           manifest_app_hash = manifest_app_hash.except('processes') if manifest_app_hash['processes'] == [{}]
         end
 
@@ -151,14 +154,22 @@ def create_similarity
         end
       end
 
-      def convert_byte_measurements_to_mb(manifest_app_hash)
+      def normalize_units(manifest_app_hash)
         byte_measurement_key_words = ['memory', 'disk-quota', 'disk_quota']
         manifest_app_hash.each_with_index do |process_hash, index|
           byte_measurement_key_words.each do |key|
             value = process_hash[key]
             manifest_app_hash[index][key] = convert_to_mb(value, key) unless value.nil?
           end
         end
+
+        byte_measurement_key_words = ['log-rate-limit-per-second']
+        manifest_app_hash.each_with_index do |process_hash, index|
+          byte_measurement_key_words.each do |key|
+            value = process_hash[key]
+            manifest_app_hash[index][key] = normalize_unit(value, key) unless value.nil?
+          end
+        end
         manifest_app_hash
       end
 
@@ -170,6 +181,18 @@ def convert_to_mb(human_readable_byte_value, attribute_name)
         "#{attribute_name} is not a number"
       end
 
+      def normalize_unit(non_normalized_value, attribute_name)
+        if %w(-1 0).include?(non_normalized_value.to_s)
+          non_normalized_value.to_s
+        else
+          byte_converter.human_readable_byte_value(byte_converter.convert_to_b(non_normalized_value))
+        end
+      rescue ByteConverter::InvalidUnitsError
+        "#{attribute_name} must use a supported unit: B, K, KB, M, MB, G, GB, T, or TB"
+      rescue ByteConverter::NonNumericError
+        "#{attribute_name} is not a number"
+      end
+
       def byte_converter
         ByteConverter.new
       end

app/actions/space_quota_update.rb

@@ -14,6 +14,7 @@ def self.update(quota, message)
         quota.instance_memory_limit = instance_memory_limit(message) if message.apps_limits_message.requested? :per_process_memory_in_mb
         quota.app_instance_limit = app_instance_limit(message) if message.apps_limits_message.requested? :total_instances
         quota.app_task_limit = app_task_limit(message) if message.apps_limits_message.requested? :per_app_tasks
+        quota.log_rate_limit = log_rate_limit(message) if message.apps_limits_message.requested? :log_rate_limit_in_bytes_per_second
 
         quota.total_services = total_services(message) if message.services_limits_message.requested? :total_service_instances
         quota.total_service_keys = total_service_keys(message) if message.services_limits_message.requested? :total_service_keys
@@ -55,6 +56,10 @@ def self.app_task_limit(message)
       default_if_nil(message.per_app_tasks, SpaceQuotaDefinition::UNLIMITED)
     end
 
+    def self.log_rate_limit(message)
+      default_if_nil(message.log_rate_limit_in_bytes_per_second, SpaceQuotaDefinition::UNLIMITED)
+    end
+
     def self.total_services(message)
       default_if_nil(message.total_service_instances, SpaceQuotaDefinition::UNLIMITED)
     end

app/actions/space_quotas_create.rb

@@ -3,6 +3,7 @@ class SpaceQuotasCreate
     class Error < ::StandardError
     end
 
+    # rubocop:todo Metrics/CyclomaticComplexity
     def create(message, organization:)
       space_quota = nil
 
@@ -16,6 +17,7 @@ def create(message, organization:)
           instance_memory_limit: message.per_process_memory_in_mb || SpaceQuotaDefinition::UNLIMITED,
           app_instance_limit: message.total_instances || SpaceQuotaDefinition::UNLIMITED,
           app_task_limit: message.per_app_tasks || SpaceQuotaDefinition::UNLIMITED,
+          log_rate_limit: message.log_rate_limit_in_bytes_per_second || QuotaDefinition::UNLIMITED,
 
           # Services
           total_services: message.total_service_instances || SpaceQuotaDefinition::DEFAULT_TOTAL_SERVICES,
@@ -35,6 +37,7 @@ def create(message, organization:)
     rescue Sequel::ValidationFailed => e
       validation_error!(e, message)
     end
+    # rubocop:enable Metrics/CyclomaticComplexity
 
     private
 

app/actions/task_create.rb

@@ -29,6 +29,7 @@ def create(app, message, user_audit_info, droplet: nil)
           command:               command(message, template_process),
           disk_in_mb:            disk_in_mb(message, template_process),
           memory_in_mb:          memory_in_mb(message, template_process),
+          log_rate_limit:        log_rate_limit(message, template_process),
           sequence_id:           app.max_task_sequence_id
         )
 
@@ -70,15 +71,15 @@ def command(message, template_process)
     end
 
     def memory_in_mb(message, template_process)
-      return message.memory_in_mb if message.memory_in_mb
+      message.memory_in_mb || template_process.try(:memory) || config.get(:default_app_memory)
+    end
 
-      template_process.present? ? template_process.memory : config.get(:default_app_memory)
+    def log_rate_limit(message, template_process)
+      message.log_rate_limit_in_bytes_per_second || template_process.try(:log_rate_limit) || config.get(:default_app_log_rate_limit_in_bytes_per_second)
     end
 
     def disk_in_mb(message, template_process)
-      return message.disk_in_mb if message.disk_in_mb
-
-      template_process.present? ? template_process.disk_quota : config.get(:default_app_disk_in_mb)
+      message.disk_in_mb || template_process.try(:disk_quota) || config.get(:default_app_disk_in_mb)
     end
 
     def submit_task(task)

app/actions/v2/app_stage.rb

@@ -40,12 +40,16 @@ def stage(process)
         process.last_stager_response = build_creator.staging_response
       rescue Diego::Runner::CannotCommunicateWithDiegoError => e
         logger.error("failed communicating with diego backend: #{e.message}")
-      rescue BuildCreate::SpaceQuotaExceeded => e
+      rescue BuildCreate::MemorySpaceQuotaExceeded => e
         raise CloudController::Errors::ApiError.new_from_details('SpaceQuotaMemoryLimitExceeded', e.message)
-      rescue BuildCreate::OrgQuotaExceeded => e
+      rescue BuildCreate::MemoryOrgQuotaExceeded => e
         raise CloudController::Errors::ApiError.new_from_details('AppMemoryQuotaExceeded', e.message)
       rescue BuildCreate::DiskLimitExceeded
         raise CloudController::Errors::ApiError.new_from_details('AppInvalid', 'too much disk requested')
+      rescue BuildCreate::LogRateLimitSpaceQuotaExceeded => e
+        raise CloudController::Errors::ApiError.new_from_details('SpaceQuotaLogRateLimitExceeded', e.message)
+      rescue BuildCreate::LogRateLimitOrgQuotaExceeded => e
+        raise CloudController::Errors::ApiError.new_from_details('OrgQuotaLogRateLimitExceeded', e.message)
       rescue BuildCreate::BuildError => e
         raise CloudController::Errors::ApiError.new_from_details('AppInvalid', e.message)
       end

app/controllers/runtime/apps_controller.rb

@@ -33,6 +33,7 @@ def self.dependencies
       attribute :docker_credentials, Hash, default: {}
       attribute :debug, String, default: nil
       attribute :disk_quota, Integer, default: nil
+      attribute :log_rate_limit, Integer, default: nil
       attribute :environment_json, Hash, default: {}, redact_in: [:create, :update]
       attribute :health_check_http_endpoint, String, default: nil
       attribute :health_check_type, String, default: 'port'

app/controllers/v3/builds_controller.rb

@@ -60,12 +60,16 @@ def create
     render status: :created, json: Presenters::V3::BuildPresenter.new(build)
   rescue BuildCreate::InvalidPackage => e
     bad_request!(e.message)
-  rescue BuildCreate::SpaceQuotaExceeded => e
+  rescue BuildCreate::MemorySpaceQuotaExceeded => e
     unprocessable!("space's memory limit exceeded: #{e.message}")
-  rescue BuildCreate::OrgQuotaExceeded => e
+  rescue BuildCreate::MemoryOrgQuotaExceeded => e
     unprocessable!("organization's memory limit exceeded: #{e.message}")
   rescue BuildCreate::DiskLimitExceeded
     unprocessable!('disk limit exceeded')
+  rescue BuildCreate::LogRateLimitSpaceQuotaExceeded => e
+    unprocessable!("space's log rate limit exceeded: #{e.message}")
+  rescue BuildCreate::LogRateLimitOrgQuotaExceeded => e
+    unprocessable!("organization's log rate limit exceeded: #{e.message}")
   rescue BuildCreate::StagingInProgress
     raise CloudController::Errors::ApiError.new_from_details('StagingInProgress')
   rescue BuildCreate::BuildError => e

app/controllers/v3/processes_controller.rb

@@ -89,6 +89,7 @@ def scale
         'instance-count' => message.instances,
         'memory-in-mb' => message.memory_in_mb,
         'disk-in-mb' => message.disk_in_mb,
+        'log-rate-in-bytes-per-second' => message.log_rate_limit_in_bytes_per_second,
         'process-type' => @process.type
       }
     )