make app event logging sendable to fluentd

* rename VCAP::Loggregator -> VCAP::AppLogEmitter

[#172415438]

Co-authored-by: Connor Braa <[email protected]>
Co-authored-by: Sannidhi Jalukar <[email protected]>
Co-authored-by: Piyali Banerjee <[email protected]>

Gemfile

@@ -7,6 +7,7 @@ gem 'clockwork', require: false
 gem 'cloudfront-signer'
 gem 'em-http-request', '~> 1.1'
 gem 'eventmachine', '~> 1.0.9'
+gem 'fluent-logger'
 gem 'googleapis-common-protos'
 gem 'hashdiff'
 gem 'httpclient'

Gemfile.lock

@@ -499,6 +499,7 @@ DEPENDENCIES
   debase (>= 0.2.2.beta14)
   em-http-request (~> 1.1)
   eventmachine (~> 1.0.9)
+  fluent-logger
   fog-aliyun
   fog-aws
   fog-azure-rm!

app/controllers/runtime/app_bits_download_controller.rb

@@ -18,7 +18,7 @@ def download(guid)
       process = find_guid_and_validate_access(:read, guid)
       blob_dispatcher.send_or_redirect(guid: process.latest_package.guid)
     rescue CloudController::Errors::BlobNotFound
-      VCAP::Loggregator.emit_error(guid, "Could not find package for #{guid}")
+      VCAP::AppLogEmitter.emit_error(guid, "Could not find package for #{guid}")
       logger.error "could not find package for #{guid}"
       raise CloudController::Errors::ApiError.new_from_details('AppPackageNotFound', guid)
     end

app/repositories/app_event_repository.rb

@@ -15,7 +15,7 @@ class AppEventRepository
       SYSTEM_ACTOR_HASH = { guid: 'system', type: 'system', name: 'system', user_name: 'system' }.freeze
 
       def create_app_exit_event(app, droplet_exited_payload)
-        VCAP::Loggregator.emit(app.guid, "App instance exited with guid #{app.guid} payload: #{droplet_exited_payload}")
+        VCAP::AppLogEmitter.emit(app.guid, "App instance exited with guid #{app.guid} payload: #{droplet_exited_payload}")
 
         actor    = { name: app.name, guid: app.guid, type: 'app' }
         metadata = droplet_exited_payload.slice('instance', 'index', 'cell_id', 'exit_status', 'exit_description', 'reason')
@@ -26,7 +26,7 @@ def create_app_exit_event(app, droplet_exited_payload)
 
       def record_app_update(app, space, user_audit_info, request_attrs, manifest_triggered: false)
         audit_hash = app_audit_hash(request_attrs)
-        VCAP::Loggregator.emit(app.guid, "Updated app with guid #{app.guid} (#{audit_hash})")
+        VCAP::AppLogEmitter.emit(app.guid, "Updated app with guid #{app.guid} (#{audit_hash})")
 
         actor    = { name: user_audit_info.user_email, guid: user_audit_info.user_guid, user_name: user_audit_info.user_name, type: 'user' }
         metadata = add_manifest_triggered(manifest_triggered, {
@@ -37,52 +37,52 @@ def record_app_update(app, space, user_audit_info, request_attrs, manifest_trigg
 
       def record_app_map_droplet(app, space, user_audit_info, request_attrs)
         audit_hash = app_audit_hash(request_attrs)
-        VCAP::Loggregator.emit(app.guid, "Updated app with guid #{app.guid} (#{audit_hash})")
+        VCAP::AppLogEmitter.emit(app.guid, "Updated app with guid #{app.guid} (#{audit_hash})")
 
         actor    = { name: user_audit_info.user_email, guid: user_audit_info.user_guid, user_name: user_audit_info.user_name, type: 'user' }
         metadata = { request: audit_hash }
         create_app_audit_event('audit.app.droplet.mapped', app, space, actor, metadata)
       end
 
       def record_app_apply_manifest(app, space, user_audit_info, manifest_request_yaml)
-        VCAP::Loggregator.emit(app.guid, "Applied manifest to app with guid #{app.guid} (#{manifest_request_yaml})")
+        VCAP::AppLogEmitter.emit(app.guid, "Applied manifest to app with guid #{app.guid} (#{manifest_request_yaml})")
 
         actor    = { name: user_audit_info.user_email, guid: user_audit_info.user_guid, user_name: user_audit_info.user_name, type: 'user' }
         metadata = { request: { manifest: manifest_request_yaml } }
         create_app_audit_event('audit.app.apply_manifest', app, space, actor, metadata)
       end
 
       def record_app_create(app, space, user_audit_info, request_attrs)
-        VCAP::Loggregator.emit(app.guid, "Created app with guid #{app.guid}")
+        VCAP::AppLogEmitter.emit(app.guid, "Created app with guid #{app.guid}")
 
         actor    = { name: user_audit_info.user_email, guid: user_audit_info.user_guid, user_name: user_audit_info.user_name, type: 'user' }
         metadata = { request: app_audit_hash(request_attrs) }
         create_app_audit_event('audit.app.create', app, space, actor, metadata)
       end
 
       def record_app_start(app, user_audit_info)
-        VCAP::Loggregator.emit(app.guid, "Starting app with guid #{app.guid}")
+        VCAP::AppLogEmitter.emit(app.guid, "Starting app with guid #{app.guid}")
 
         actor = { name: user_audit_info.user_email, guid: user_audit_info.user_guid, user_name: user_audit_info.user_name, type: 'user' }
         create_app_audit_event('audit.app.start', app, app.space, actor, nil)
       end
 
       def record_app_restart(app, user_audit_info)
-        VCAP::Loggregator.emit(app.guid, "Restarted app with guid #{app.guid}")
+        VCAP::AppLogEmitter.emit(app.guid, "Restarted app with guid #{app.guid}")
 
         actor = { name: user_audit_info.user_email, guid: user_audit_info.user_guid, type: 'user', user_name: user_audit_info.user_name }
         create_app_audit_event('audit.app.restart', app, app.space, actor, nil)
       end
 
       def record_app_stop(app, user_audit_info)
-        VCAP::Loggregator.emit(app.guid, "Stopping app with guid #{app.guid}")
+        VCAP::AppLogEmitter.emit(app.guid, "Stopping app with guid #{app.guid}")
 
         actor = { name: user_audit_info.user_email, guid: user_audit_info.user_guid, user_name: user_audit_info.user_name, type: 'user' }
         create_app_audit_event('audit.app.stop', app, app.space, actor, nil)
       end
 
       def record_app_delete_request(app, space, user_audit_info, recursive=nil)
-        VCAP::Loggregator.emit(app.guid, "Deleted app with guid #{app.guid}")
+        VCAP::AppLogEmitter.emit(app.guid, "Deleted app with guid #{app.guid}")
 
         actor    = { name: user_audit_info.user_email, guid: user_audit_info.user_guid, user_name: user_audit_info.user_name, type: 'user' }
         metadata = nil

app/repositories/build_event_repository.rb

@@ -2,7 +2,7 @@ module VCAP::CloudController
   module Repositories
     class BuildEventRepository
       def self.record_build_create(build, user_audit_info, v3_app_name, space_guid, org_guid)
-        VCAP::Loggregator.emit(build.app_guid, "Creating build for app with guid #{build.app_guid}")
+        VCAP::AppLogEmitter.emit(build.app_guid, "Creating build for app with guid #{build.app_guid}")
 
         metadata = {
           build_guid: build.guid,

app/repositories/deployment_event_repository.rb

@@ -2,7 +2,7 @@ module VCAP::CloudController
   module Repositories
     class DeploymentEventRepository
       def self.record_create(deployment, droplet, user_audit_info, v3_app_name, space_guid, org_guid, params, type)
-        VCAP::Loggregator.emit(deployment.app_guid, "Creating deployment for app with guid #{deployment.app_guid}")
+        VCAP::AppLogEmitter.emit(deployment.app_guid, "Creating deployment for app with guid #{deployment.app_guid}")
 
         metadata = {
           deployment_guid: deployment.guid,
@@ -29,7 +29,7 @@ def self.record_create(deployment, droplet, user_audit_info, v3_app_name, space_
       end
 
       def self.record_cancel(deployment, droplet, user_audit_info, v3_app_name, space_guid, org_guid)
-        VCAP::Loggregator.emit(deployment.app_guid, "Cancelling deployment for app with guid #{deployment.app_guid}")
+        VCAP::AppLogEmitter.emit(deployment.app_guid, "Cancelling deployment for app with guid #{deployment.app_guid}")
 
         metadata = {
           deployment_guid: deployment.guid,

app/repositories/droplet_event_repository.rb

@@ -2,7 +2,7 @@ module VCAP::CloudController
   module Repositories
     class DropletEventRepository
       def self.record_create_by_staging(droplet, user_audit_info, v3_app_name, space_guid, org_guid)
-        VCAP::Loggregator.emit(droplet.app_guid, "Creating droplet for app with guid #{droplet.app_guid}")
+        VCAP::AppLogEmitter.emit(droplet.app_guid, "Creating droplet for app with guid #{droplet.app_guid}")
 
         metadata = {
           droplet_guid: droplet.guid,
@@ -26,7 +26,7 @@ def self.record_create_by_staging(droplet, user_audit_info, v3_app_name, space_g
       end
 
       def self.record_create(droplet, user_audit_info, v3_app_name, space_guid, org_guid)
-        VCAP::Loggregator.emit(droplet.app_guid, "Creating droplet for app with guid #{droplet.app_guid}")
+        VCAP::AppLogEmitter.emit(droplet.app_guid, "Creating droplet for app with guid #{droplet.app_guid}")
 
         metadata = {
           droplet_guid: droplet.guid,
@@ -49,7 +49,7 @@ def self.record_create(droplet, user_audit_info, v3_app_name, space_guid, org_gu
       end
 
       def self.record_create_by_copying(new_droplet_guid, source_droplet_guid, user_audit_info, v3_app_guid, v3_app_name, space_guid, org_guid)
-        VCAP::Loggregator.emit(v3_app_guid, "Creating droplet for app with guid #{v3_app_guid}")
+        VCAP::AppLogEmitter.emit(v3_app_guid, "Creating droplet for app with guid #{v3_app_guid}")
 
         metadata = {
           droplet_guid: new_droplet_guid,
@@ -75,7 +75,7 @@ def self.record_create_by_copying(new_droplet_guid, source_droplet_guid, user_au
       end
 
       def self.record_delete(droplet, user_audit_info, v3_app_name, space_guid, org_guid)
-        VCAP::Loggregator.emit(droplet.app_guid, "Deleting droplet for app with guid #{droplet.app_guid}")
+        VCAP::AppLogEmitter.emit(droplet.app_guid, "Deleting droplet for app with guid #{droplet.app_guid}")
 
         metadata = { droplet_guid: droplet.guid }
 
@@ -97,7 +97,7 @@ def self.record_delete(droplet, user_audit_info, v3_app_name, space_guid, org_gu
 
       # Emit this event once we have droplet download capability
       def self.record_download(droplet, user_audit_info, v3_app_name, space_guid, org_guid)
-        VCAP::Loggregator.emit(droplet.app_guid, "Downloading droplet for app with guid #{droplet.app_guid}")
+        VCAP::AppLogEmitter.emit(droplet.app_guid, "Downloading droplet for app with guid #{droplet.app_guid}")
 
         metadata = { droplet_guid: droplet.guid }
 
@@ -118,7 +118,7 @@ def self.record_download(droplet, user_audit_info, v3_app_name, space_guid, org_
       end
 
       def self.record_upload(droplet, user_audit_info, v3_app_name, space_guid, org_guid)
-        VCAP::Loggregator.emit(droplet.app_guid, "Uploading droplet for app with guid #{droplet.app_guid}")
+        VCAP::AppLogEmitter.emit(droplet.app_guid, "Uploading droplet for app with guid #{droplet.app_guid}")
 
         metadata = { droplet_guid: droplet.guid }
 

app/repositories/package_event_repository.rb

@@ -3,7 +3,7 @@ module Repositories
     class PackageEventRepository
       def self.record_app_package_create(package, user_audit_info, request_attrs)
         app_guid = request_attrs.delete('app_guid')
-        VCAP::Loggregator.emit(app_guid, "Adding app package for app with guid #{app_guid}")
+        VCAP::AppLogEmitter.emit(app_guid, "Adding app package for app with guid #{app_guid}")
 
         metadata = {
           package_guid: package.guid,
@@ -16,7 +16,7 @@ def self.record_app_package_create(package, user_audit_info, request_attrs)
 
       def self.record_app_package_copy(package, user_audit_info, source_package_guid)
         app = package.app
-        VCAP::Loggregator.emit(app.guid, "Adding app package for app with guid #{app.guid} copied from package with guid #{source_package_guid}")
+        VCAP::AppLogEmitter.emit(app.guid, "Adding app package for app with guid #{app.guid} copied from package with guid #{source_package_guid}")
         metadata = {
           package_guid: package.guid,
           request:      {
@@ -29,31 +29,31 @@ def self.record_app_package_copy(package, user_audit_info, source_package_guid)
       end
 
       def self.record_app_package_upload(package, user_audit_info)
-        VCAP::Loggregator.emit(package.app.guid, "Uploading app package for app with guid #{package.app.guid}")
+        VCAP::AppLogEmitter.emit(package.app.guid, "Uploading app package for app with guid #{package.app.guid}")
         metadata = { package_guid: package.guid }
         type     = 'audit.app.package.upload'
 
         create_event(package, type, user_audit_info, metadata)
       end
 
       def self.record_app_upload_bits(package, user_audit_info)
-        VCAP::Loggregator.emit(package.app.guid, "Uploading bits for app with guid #{package.app.guid}")
+        VCAP::AppLogEmitter.emit(package.app.guid, "Uploading bits for app with guid #{package.app.guid}")
         metadata = { package_guid: package.guid }
         type     = 'audit.app.upload-bits'
 
         create_event(package, type, user_audit_info, metadata)
       end
 
       def self.record_app_package_delete(package, user_audit_info)
-        VCAP::Loggregator.emit(package.app.guid, "Deleting app package for app with guid #{package.app.guid}")
+        VCAP::AppLogEmitter.emit(package.app.guid, "Deleting app package for app with guid #{package.app.guid}")
         metadata = { package_guid: package.guid }
         type     = 'audit.app.package.delete'
 
         create_event(package, type, user_audit_info, metadata)
       end
 
       def self.record_app_package_download(package, user_audit_info)
-        VCAP::Loggregator.emit(package.app.guid, "Downloading app package for app with guid #{package.app.guid}")
+        VCAP::AppLogEmitter.emit(package.app.guid, "Downloading app package for app with guid #{package.app.guid}")
         metadata = { package_guid: package.guid }
         type     = 'audit.app.package.download'
 

app/repositories/process_event_repository.rb

@@ -6,7 +6,7 @@ class ProcessEventRepository
       extend AppManifestEventMixins
 
       def self.record_create(process, user_audit_info, manifest_triggered: false)
-        VCAP::Loggregator.emit(process.app.guid, "Added process: \"#{process.type}\"")
+        VCAP::AppLogEmitter.emit(process.app.guid, "Added process: \"#{process.type}\"")
 
         metadata = add_manifest_triggered(manifest_triggered, {
           process_guid: process.guid,
@@ -24,7 +24,7 @@ def self.record_create(process, user_audit_info, manifest_triggered: false)
       end
 
       def self.record_delete(process, user_audit_info)
-        VCAP::Loggregator.emit(process.app.guid, "Deleting process: \"#{process.type}\"")
+        VCAP::AppLogEmitter.emit(process.app.guid, "Deleting process: \"#{process.type}\"")
 
         create_event(
           process:        process,
@@ -40,7 +40,7 @@ def self.record_delete(process, user_audit_info)
       end
 
       def self.record_update(process, user_audit_info, request, manifest_triggered: false)
-        VCAP::Loggregator.emit(process.app.guid, "Updating process: \"#{process.type}\"")
+        VCAP::AppLogEmitter.emit(process.app.guid, "Updating process: \"#{process.type}\"")
 
         request           = request.dup.symbolize_keys
         request[:command] = Presenters::Censorship::PRIVATE_DATA_HIDDEN if request.key?(:command)
@@ -61,7 +61,7 @@ def self.record_update(process, user_audit_info, request, manifest_triggered: fa
       end
 
       def self.record_scale(process, user_audit_info, request, manifest_triggered: false)
-        VCAP::Loggregator.emit(process.app.guid, "Scaling process: \"#{process.type}\"")
+        VCAP::AppLogEmitter.emit(process.app.guid, "Scaling process: \"#{process.type}\"")
 
         metadata = add_manifest_triggered(manifest_triggered, {
           process_guid: process.guid,
@@ -80,7 +80,7 @@ def self.record_scale(process, user_audit_info, request, manifest_triggered: fal
       end
 
       def self.record_terminate(process, user_audit_info, index)
-        VCAP::Loggregator.emit(process.app.guid, "Terminating process: \"#{process.type}\", index: \"#{index}\"")
+        VCAP::AppLogEmitter.emit(process.app.guid, "Terminating process: \"#{process.type}\", index: \"#{index}\"")
 
         create_event(
           process:        process,
@@ -97,7 +97,7 @@ def self.record_terminate(process, user_audit_info, index)
       end
 
       def self.record_crash(process, crash_payload)
-        VCAP::Loggregator.emit(process.app.guid, "Process has crashed with type: \"#{process.type}\"")
+        VCAP::AppLogEmitter.emit(process.app.guid, "Process has crashed with type: \"#{process.type}\"")
 
         create_event(
           process:    process,

app/repositories/revision_event_repository.rb

@@ -2,7 +2,7 @@ module VCAP::CloudController
   module Repositories
     class RevisionEventRepository
       def self.record_create(revision, app, user_audit_info)
-        VCAP::Loggregator.emit(revision.app_guid, "Creating revision for app with guid #{app.guid}")
+        VCAP::AppLogEmitter.emit(revision.app_guid, "Creating revision for app with guid #{app.guid}")
 
         Event.create(
           type:              'audit.app.revision.create',

config/newrelic/initializers/new_relic_loggregator_instrumentation.rb

@@ -1,9 +1,9 @@
 require 'new_relic/agent/method_tracer'
-require 'loggregator'
+require 'app_log_emitter'
 
 module CCInitializers
   def self.new_relic_loggregator_instrumentation(_)
-    VCAP::Loggregator.class_eval do
+    VCAP::AppLogEmitter.class_eval do
       include ::NewRelic::Agent::MethodTracer
 
       class << self

lib/loggregator.rb → lib/app_log_emitter.rb

@@ -1,18 +1,21 @@
+require 'fluent_emitter'
+
 module VCAP
-  class Loggregator
+  class AppLogEmitter
     class << self
-      attr_accessor :emitter, :logger
+      attr_accessor :emitter, :logger, :fluent_emitter
 
       def emit(app_id, message)
+        fluent_emitter.emit(app_id, message) if fluent_emitter
         emitter.emit(app_id, message, generate_tags(app_id)) if emitter
       rescue => e
-        logger.error('loggregator_emitter.emit.failed', app_id: app_id, message: message, error: e)
+        logger.error('app_event_emitter.emit.failed', app_id: app_id, message: message, error: e)
       end
 
       def emit_error(app_id, message)
         emitter.emit_error(app_id, message, generate_tags(app_id)) if emitter
       rescue => e
-        logger.error('loggregator_emitter.emit_error.failed', app_id: app_id, message: message, error: e)
+        logger.error('app_event_emitter.emit_error.failed', app_id: app_id, message: message, error: e)
       end
 
       private

lib/cloud_controller/blob_sender/missing_blob_handler.rb

@@ -2,7 +2,7 @@ module CloudController
   module BlobSender
     class MissingBlobHandler
       def handle_missing_blob!(app_guid, name)
-        VCAP::Loggregator.emit_error(app_guid, "Did not find #{name} for app with guid: #{app_guid}")
+        VCAP::AppLogEmitter.emit_error(app_guid, "Did not find #{name} for app with guid: #{app_guid}")
         logger.error "could not find #{name} for #{app_guid}"
         raise CloudController::Errors::ApiError.new_from_details('StagingError', "#{name} not found for #{app_guid}")
       end

lib/cloud_controller/config_schemas/api_schema.rb

@@ -222,6 +222,11 @@ class ApiSchema < VCAP::Config
             internal_url: String,
           },
 
+          optional(:fluent) => {
+            optional(:host) => String,
+            optional(:port) => Integer,
+          },
+
           doppler: {
             url: String
           },

lib/cloud_controller/diego/staging_completion_handler.rb

@@ -69,7 +69,7 @@ def handle_failure(payload, with_start)
           )
         end
 
-        VCAP::Loggregator.emit_error(build.app_guid, "Failed to stage build: #{payload[:error][:message]}")
+        VCAP::AppLogEmitter.emit_error(build.app_guid, "Failed to stage build: #{payload[:error][:message]}")
       end
 
       # with_start is true when v2 staging causes apps to start