wire worker to emit to fluentd when configured

* also emits app error logs to fluent [finishes #172415438] Co-authored-by: Connor Braa <[email protected]> Co-authored-by: Renee Chu <[email protected]>
geofffranks · Jun 16, 2020 · 6798ef7 · 6798ef7
1 parent b1e1e0e
commit 6798ef7
Show file tree

Hide file tree

Showing 4 changed files with 30 additions and 7 deletions.
diff --git a/lib/app_log_emitter.rb b/lib/app_log_emitter.rb
@@ -13,6 +13,7 @@ def emit(app_id, message)
       end
 
       def emit_error(app_id, message)
+        fluent_emitter.emit(app_id, message) if fluent_emitter
         emitter.emit_error(app_id, message, generate_tags(app_id)) if emitter
       rescue => e
         logger.error('app_event_emitter.emit_error.failed', app_id: app_id, message: message, error: e)

diff --git a/lib/cloud_controller/runner.rb b/lib/cloud_controller/runner.rb
@@ -115,7 +115,7 @@ def start_cloud_controller
       setup_blobstore
       @config.configure_components
 
-      setup_loggregator_emitter
+      setup_app_log_emitter
       @config.set(:external_host, VCAP::HostSystem.new.local_ip(@config.get(:local_route)))
     end
 
@@ -156,13 +156,14 @@ def setup_blobstore
       CloudController::DependencyLocator.instance.buildpack_blobstore.ensure_bucket_exists
     end
 
-    def setup_loggregator_emitter
+    def setup_app_log_emitter
       VCAP::AppLogEmitter.fluent_emitter = fluent_emitter if @config.get(:fluent)
 
       if @config.get(:loggregator) && @config.get(:loggregator, :router)
         VCAP::AppLogEmitter.emitter = LoggregatorEmitter::Emitter.new(@config.get(:loggregator, :router), 'cloud_controller', 'API', @config.get(:index))
-        VCAP::AppLogEmitter.logger = logger
       end
+
+      VCAP::AppLogEmitter.logger = logger
     end
 
     def fluent_emitter

diff --git a/lib/tasks/jobs.rake b/lib/tasks/jobs.rake
@@ -63,12 +63,11 @@ namespace :jobs do
     def start_working
       config = RakeConfig.config
       BackgroundJobEnvironment.new(config).setup_environment(readiness_port)
+
       logger = Steno.logger('cc-worker')
       logger.info("Starting job with options #{@queue_options}")
-      if config.get(:loggregator) && config.get(:loggregator, :router)
-        VCAP::AppLogEmitter.emitter = LoggregatorEmitter::Emitter.new(config.get(:loggregator, :router), 'cloud_controller', 'API', config.get(:index))
-        VCAP::AppLogEmitter.logger = logger
-      end
+
+      setup_app_log_emitter(config, logger)
       Delayed::Worker.destroy_failed_jobs = false
       Delayed::Worker.max_attempts = 3
       Delayed::Worker.logger = logger
@@ -79,6 +78,22 @@ namespace :jobs do
 
     private
 
+    def setup_app_log_emitter(config, logger)
+      VCAP::AppLogEmitter.fluent_emitter = fluent_emitter(config) if config.get(:fluent)
+      if config.get(:loggregator) && config.get(:loggregator, :router)
+        VCAP::AppLogEmitter.emitter = LoggregatorEmitter::Emitter.new(config.get(:loggregator, :router), 'cloud_controller', 'API', config.get(:index))
+      end
+
+      VCAP::AppLogEmitter.logger = logger
+    end
+
+    def fluent_emitter(config)
+      VCAP::FluentEmitter.new(Fluent::Logger::FluentLogger.new(nil,
+        host: config.get(:fluent, :host) || 'localhost',
+        port: config.get(:fluent, :port) || 24224,
+      ))
+    end
+
     def readiness_port
       if is_first_generic_worker_on_machine?
         RakeConfig.config.get(:readiness_port, :cloud_controller_worker)

diff --git a/spec/unit/lib/app_log_emitter_spec.rb b/spec/unit/lib/app_log_emitter_spec.rb
@@ -40,6 +40,12 @@ module VCAP
         AppLogEmitter.emit(app.guid, 'log message')
       end
 
+      it 'emits app event logs to the fluent emitter' do
+        expect(fluent_emitter).to receive(:emit).with(app.guid, 'error')
+
+        AppLogEmitter.emit_error(app.guid, 'error')
+      end
+
       it 'logs errors on failure' do
         expect(fluent_emitter).to receive(:emit).with(app.guid, 'log message').and_raise(StandardError.new('rekt'))
         expect(logger).to receive(:error)