changes to connect to k8s efk stack in cloud env

Author: vgenev

config/default.json

@@ -1,9 +1,14 @@
 {
   "efk": {
-    "host": "localhost",
+    "host": "dev1-fluentd.mojaloop.live",
     "port": 24224,
     "timeout": 3.0,
-    "reconnectInterval": 600
+    "reconnectInterval": 600,
+    "requireAckResponse": true
+  },
+  "apm": {
+    "serviceName": "event-stream-processor",
+    "serverUrl": "http://dev1-apm.mojaloop.live:80"
   },
   "PORT": 3082,
   "KAFKA": {

package-lock.json

@@ -14,6 +14,7 @@
     "check-types": "7.4.0",
     "config": "3.0.0",
     "fluent-logger": "^3.3.1",
+    "fluent-logger-stream": "0.0.1",
     "healthcheck-server": "1.0.0",
     "json-rules-engine": "2.3.0",
     "lazy-ass": "1.6.0",

package.json

@@ -1,5 +1,13 @@
 const logger = require('fluent-logger')
 
 module.exports.initLogger = (prefix, options) => {
-  return logger.configure(prefix, options)
+  logger.configure(prefix, options)
+  logger.on('error', (error) => {
+    console.log(error)
+  })
+  logger.on('connect', (v) => {
+    console.log('connected!' + v)
+  })
 }
+
+module.exports.logger = logger

src/lib/efk/index.js

@@ -1,8 +1,12 @@
 const Agent = require('@mojaloop/elastic-apm-node')
 const Tracer = require('@mojaloop/elastic-apm-node-opentracing')
+const Config = require('../config')
+const configuration = Config.util.toObject()
+
+
 
 const agent = (function () {
-  let self = Agent.start({serviceName: 'event-stream-processor'})
+  let self = Agent.start(configuration.apm)
   this.agent = self
   return this
 })()

src/lib/tracer/index.js

@@ -1,9 +1,25 @@
 const Rx = require('rxjs')
-const logger = require('fluent-logger')
+const logger = require('../lib/efk').logger
+const initLogger = require('../lib/efk').initLogger
+
+const Config = require('../lib/config')
+
+const configuration = Config.util.toObject()
+
+// initLogger('fluentd.test', configuration.efk)
 
 const FluentdLoggerObservable = ({ message }) => {
   return Rx.Observable.create(observable => {
-    logger.emit('logstash', message)
+
+    logger.configure('logstash', configuration.efk)
+    logger.on('error', (error) => {
+      console.log(error)
+    })
+    logger.on('connect', () => {
+      console.log('connected!')
+    })
+
+    logger.emit('logstash', message.value, (err) => console.log(`fluentd err ${err}`))
     observable.complete()
   })
 }

src/observables/fluentd-logger.js

@@ -0,0 +1,6 @@
+var message = {message: 'helloworld_new'}
+
+var Logger = require('fluent-logger-stream')
+var logger = new Logger({tag: 'logstash', type: 'http', host: 'dev1-fluentd.mojaloop.live', port: 80}) //in_http
+logger.send('logstash', message)
+

src/observables/log_it.js

@@ -68,7 +68,11 @@ const setup = async () => {
   const sharedMessageObservable = topicObservable.pipe(share())
 
   sharedMessageObservable.subscribe(async props => {
-    Observables.fluentdObservable(props).subscribe(v => console.log(v))
+    Observables.fluentdObservable(props).subscribe({
+      next: v => Logger.info(v),
+      error: (e) => Logger.error(e),
+      completed: () => Logger.info('fluentd log completed')
+    })
   })
 
   const tracingObservable = sharedMessageObservable.pipe(

src/setup.js

@@ -62,11 +62,11 @@ services:
       - setup.kibana.host="kibana:5601"
       - setup.template.enabled=true
       - logging.to_files=false
-    depends_on:
-      elasticsearch:
-        condition: service_healthy
-      kibana:
-        condition: service_healthy
+    # depends_on:
+    #   elasticsearch:
+    #     condition: service_healthy
+    #   kibana:
+    #     condition: service_healthy
 
 
 volumes: