Add health check endpoint (#95)

Co-authored-by: wallrony <[email protected]>

Dockerfile

@@ -29,5 +29,4 @@ COPY create-conf.rb /create-conf.rb
 COPY conf-utils.rb /conf-utils.rb
 COPY start.sh /start.sh
 
-USER root
 CMD ["/start.sh"]

create-conf.rb

@@ -16,6 +16,7 @@ def create_file
     f.write(File.read("#{ETC_DIR}/process.conf"))
     f.write(decode_chunk_events_conf)
     f.write(output_conf)
+    f.write(File.read("#{ETC_DIR}/input-sdm-health-check.conf"))
   end
 end
 

docker-compose.yml

@@ -5,6 +5,7 @@ services:
     ports:
       - "5140:5140"
       - "24321:24321"
+      - "24220:24220"
     environment:
       # IMPORTANT: Do not enclose values in double or single quotes
 

fluentd/etc/input-sdm-health-check.conf

@@ -0,0 +1,5 @@
+<source>
+  @type sdm_health_check
+  bind 0.0.0.0
+  port 24220
+</source>

fluentd/plugins/in_sdm_health_check.rb

@@ -0,0 +1,106 @@
+# frozen_string_literal: true
+
+require 'fluent/plugin/in_monitor_agent'
+
+module Fluent::Plugin
+  class SDMHealthCheckInput < MonitorAgentInput
+    Fluent::Plugin.register_input('sdm_health_check', self)
+
+    def start
+      api_handler = APIHandler.new(self)
+      log.debug "listening monitoring http server on http://#{@bind}:#{@port}/api/health-check for worker#{fluentd_worker_id}"
+      http_server_create_http_server(:in_monitor_http_server_helper, addr: @bind, port: @port, logger: log, default_app: NotFoundJson) do |serv|
+        serv.get('/api/health-check') { |req| health_json(req) }
+      end
+    end
+
+    def health_json(req)
+      begin
+        render_json(
+          {
+            "timestamp" => DateTime.now,
+            "uptime_seconds" => uptime_seconds,
+            "inputs" => input_plugins_health_data(req),
+            "outputs" => output_plugins_health_data(req),
+          },
+          pretty_json: true
+        )
+      rescue Exception => e
+        puts e
+      end
+    end
+
+    def uptime_seconds
+      ps_output = `ps -o etime,args | grep "fluentd -c"`
+      fluentd_info = ps_output.split("\n")[0]
+      elapsed_time_str = fluentd_info.match(/[0-9]+(:[0-9]+)+/)
+      elapsed_time_parts = elapsed_time_str.to_s.split(":")
+      elapsed_seconds = 0
+      time_unit_multiplier = 1
+      for i in 0..(elapsed_time_parts.length - 1) do
+        elapsed_seconds += elapsed_time_parts[elapsed_time_parts.length - 1 - i].to_i * time_unit_multiplier
+        print ">" + elapsed_seconds.to_s
+        time_unit_multiplier *= i >= 2 ? 24 : 60
+      end
+      elapsed_seconds
+    end
+
+    def input_plugins_health_data(req)
+      api_handler = APIHandler.new(self)
+      plugins_json = api_handler.plugins_json(req)
+      plugins_obj = JSON.parse(plugins_json[2])
+
+      health_data = []
+      plugins_obj["plugins"]&.each { |plugin|
+        if plugin["plugin_category"] != 'input'
+          next
+        end
+
+        plugin_health_data = {
+          "type" => plugin["type"],
+          "category" => plugin["plugin_category"],
+        }
+
+        if plugin["config"]["tag"] != nil
+          plugin_health_data["tag"] = plugin["config"]["tag"]
+        end
+
+        health_data << plugin_health_data
+      }
+      health_data
+    end
+
+    def output_plugins_health_data(req)
+      api_handler = APIHandler.new(self)
+      plugins_json = api_handler.plugins_json(req)
+      plugins_obj = JSON.parse(plugins_json[2])
+
+      health_data = []
+      plugins_obj["plugins"]&.each { |plugin|
+        if plugin["plugin_category"] != 'output'
+          next
+        end
+
+        plugin_health_data = {
+          "type" => plugin["type"],
+          "category" => plugin["plugin_category"],
+          "alive" => plugin["retry"] == nil || plugin["retry"].keys.length == 0
+        }
+
+        health_data << plugin_health_data
+      }
+      health_data
+    end
+
+    def render_json(obj, code: 200, pretty_json: nil)
+      body =
+        if pretty_json
+          JSON.pretty_generate(obj)
+        else
+          obj.to_json
+        end
+
+      [code, { 'Content-Type' => 'application/json' }, body]
+    end
+  end
+end

fluentd/scripts/dump_sdm_entities.rb

@@ -26,8 +26,8 @@ def get_audit_activities_rows
     stream_activities
     return
   end
-  datetime_from = DateTime.now - (interval_time + 1.0)/(24*60)
-  datetime_to = DateTime.now - 1.0/(24*60)
+  datetime_from = DateTime.now - (interval_time + 1.0) / (24 * 60)
+  datetime_to = DateTime.now - 1.0 / (24 * 60)
   output = `sdm audit activities -j -e --from "#{datetime_from.to_s}" --to "#{datetime_to.to_s}"`
   output.split("\n")
 end
@@ -57,8 +57,8 @@ def open_activities_stream
   if must_stream_json
     command += " -j"
   end
-  _, out, _, thread = Open3.popen3(command)
-  [out, thread]
+  _, stdout, _, thread = Open3.popen3(command)
+  [stdout, thread]
 end
 
 def send_socket_message(message)

test/e2e/health_check_spec.rb

@@ -0,0 +1,40 @@
+require 'test/unit'
+require 'fluent/test'
+require 'fluent/test/helpers'
+require 'fluent/test/driver/output'
+require 'uri'
+require 'net/http'
+require 'json'
+require_relative '../common'
+require_relative '../helper'
+
+SIGTERM = 15
+
+describe 'health_check' do
+  fluent_thread = nil
+
+  before(:each) do
+    ENV['MONGO_URI'] = 'mongodb://host'
+    generate_fluent_conf('syslog-json', 'stdout mongo')
+    _, stdout, _, fluent_thread = Open3.popen3("fluentd -c #{ETC_DIR}/fluent.conf -p ./fluentd/plugins")
+    wait_for_fluent(stdout)
+    clear_buffer(stdout)
+  end
+
+  after(:each) do
+    Process.kill(SIGTERM, fluent_thread&.pid)
+  end
+
+  it "should return all inputs and outputs status" do
+    uri = URI('http://localhost:24220/api/health-check')
+    res = Net::HTTP.get_response(uri)
+
+    expect(res.is_a?(Net::HTTPSuccess))
+    health_data = JSON.parse(res.body)
+    inputs_types = health_data["inputs"].map { |plugin| plugin["type"] }
+    outputs_types = health_data["outputs"].map { |plugin| plugin["type"] }
+    expect(inputs_types).to include("syslog")
+    expect(outputs_types).to include("stdout")
+    expect(outputs_types).to include("mongo")
+  end
+end