Merge pull request #315 from andersonkrs/master

Add appender for Grafana Loki using the HTTP push API

Author: reidmorrison

CHANGELOG.md

@@ -8,6 +8,7 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 ## [4.17.0]
 
 - Correct `source_code_uri` URL
+- Add appender for Grafana Loki using the HTTP push API.
 - Add :notime as a time_format for Formatters
 - Fix #316: syslog messages contains two timestamps
 - Add appender for CloudWatch Logs

docs/appenders.md

@@ -33,6 +33,7 @@ Log messages can be written to one or more of the following destinations at the
 * Honeybadger Insights
 * CloudWatch Logs
 * Logger, log4r, etc.
+* Grafana Loki
 
 To ensure no log messages are lost it is recommend to use TCP over UDP for logging purposes.
 Due to the architecture of Semantic Logger any performance difference between TCP and UDP will not
@@ -758,6 +759,22 @@ SemanticLogger.add_appender(appender: :honeybadger_insights)
 
 Both appenders use the Honeybadger [gem configuration](https://docs.honeybadger.io/lib/ruby/gem-reference/configuration/).
 
+### Grafana Loki
+
+Sends log messages to [Grafana Loki](https://grafana.com/docs/loki) using its [HTTP push API](https://grafana.com/docs/loki/latest/reference/loki-http-api/#ingest-logs)
+
+```ruby
+SemanticLogger.add_appender(
+  appender: :loki,
+  url: "https://logs-prod-001.grafana.net",
+  username: "grafana_username",
+  password: "grafana_token_here",
+  compress: true
+)
+```
+
+Configure the URL, username and password according to your Grafana Loki instance. The `compress` option can be set to `true` to compress the log messages.
+
 ### CloudWatch Logs
 
 Forward all log messages to CloudWatch Logs.

lib/semantic_logger/appender.rb

@@ -27,6 +27,7 @@ module Appender
     autoload :Udp,                 "semantic_logger/appender/udp"
     autoload :Wrapper,             "semantic_logger/appender/wrapper"
     autoload :SentryRuby,          "semantic_logger/appender/sentry_ruby"
+    autoload :Loki,                "semantic_logger/appender/loki"
     # @formatter:on
 
     # Returns [SemanticLogger::Subscriber] appender for the supplied options

lib/semantic_logger/appender/loki.rb

@@ -0,0 +1,62 @@
+# Forward application metrics to a Loki instance using HTTP push API
+#
+# Example:
+#   SemanticLogger.add_appender(
+#     appender: :loki,
+#     url: "https://logs-prod-001.grafana.net",
+#     username: "grafana_username",
+#     password: "grafana_token_here",
+#     compress: true
+#   )
+module SemanticLogger
+  module Appender
+    class Loki < SemanticLogger::Appender::Http
+      INGESTION_PATH = "loki/api/v1/push".freeze
+
+      # Create Grafana Loki appender.
+      #
+      # Parameters:
+      #   filter: [Regexp|Proc]
+      #     RegExp: Only include log messages where the class name matches the supplied
+      #             regular expression. All other messages will be ignored.
+      #     Proc: Only include log messages where the supplied Proc returns true.
+      #           The Proc must return true or false.
+      #
+      #   host: [String]
+      #     Name of this host to send as a dimension.
+      #     Default: SemanticLogger.host
+      #
+      #   application: [String]
+      #     Name of this application to send as a dimension.
+      #     Default: SemanticLogger.application
+      #
+      #   url: [String]
+      #     Define the loki instance URL.
+      #     Example: https://logs-prod-999.grafana.net
+      #     Default: nil
+      def initialize(url: nil,
+                     formatter: SemanticLogger::Formatters::Loki.new,
+                     header: {"Content-Type" => "application/json"},
+                     path: INGESTION_PATH,
+                     **args,
+                     &block)
+
+        super(url: "#{url}/#{path}", formatter: formatter, header: header, **args, &block)
+      end
+
+      def log(log)
+        message = formatter.call(log, self)
+        puts message
+        logger.trace(message)
+        post(message)
+      end
+
+      # Logs in batches
+      def batch(logs)
+        message = formatter.batch(logs, self)
+        logger.trace(message)
+        post(message)
+      end
+    end
+  end
+end

lib/semantic_logger/formatters.rb

@@ -13,6 +13,7 @@ module Formatters
     autoload :Logfmt,        "semantic_logger/formatters/logfmt"
     autoload :SyslogCee,     "semantic_logger/formatters/syslog_cee"
     autoload :NewRelicLogs,  "semantic_logger/formatters/new_relic_logs"
+    autoload :Loki,          "semantic_logger/formatters/loki"
 
     # Return formatter that responds to call.
     #

lib/semantic_logger/formatters/loki.rb

@@ -0,0 +1,157 @@
+require "json"
+
+module SemanticLogger
+  module Formatters
+    class Loki < Base
+      attr_accessor :stream, :payload_value
+
+      # Returns [String] a single JSON log
+      def call(log, logger)
+        self.logger = logger
+        self.log = log
+
+        {streams: [build_stream]}.to_json
+      end
+
+      # Returns [String] a JSON batch of logs
+      def batch(logs, logger)
+        self.logger = logger
+
+        streams = logs.map do |log|
+          self.log = log
+          build_stream
+        end
+
+        {streams: streams}.to_json
+      end
+
+      private
+
+      def build_stream
+        self.stream = {stream: {pid: pid}, values: [[]]}
+
+        application
+        environment
+        host
+        level
+        thread
+        tags
+        named_tags
+        context
+        time
+        message
+        payload
+        metric
+        duration
+        exception
+
+        stream[:values][0] << payload_value
+        stream
+      end
+
+      def host
+        stream[:stream][:host] = logger.host if log_host && logger.host.to_s
+      end
+
+      def application
+        stream[:stream][:application] = logger.application if log_application && logger&.application
+      end
+
+      def environment
+        stream[:stream][:environment] = logger.environment if log_environment && logger&.environment
+      end
+
+      def level
+        stream[:stream][:level] = log.level
+      end
+
+      def thread
+        stream[:stream][:thread] = log.thread_name if log.thread_name
+      end
+
+      def tags
+        stream[:stream][:tags] = log.tags if log.tags.respond_to?(:empty?) && !log.tags.empty?
+      end
+
+      def named_tags
+        stream[:stream].merge!(log.named_tags) if log.named_tags.respond_to?(:empty?) && !log.named_tags.empty?
+      end
+
+      def context
+        return unless log.context && !log.context.empty?
+
+        log.context.each do |key, value|
+          serialized_value = if value.is_a?(Hash)
+                               value.to_json
+                             else
+                               value.to_s
+                             end
+
+          stream[:stream].merge!(key.to_s => serialized_value)
+        end
+      end
+
+      def time
+        stream[:values][0] << format_time(log)
+      end
+
+      def message
+        stream[:values][0] << (log.message ? log.cleansed_message : "")
+      end
+
+      def format_time(log)
+        log.time.strftime("%s%N")
+      end
+
+      def payload
+        self.payload_value = if log.payload.respond_to?(:empty?) && !log.payload.empty?
+                               # Loki only accepts strings as key and values
+                               stringify_hash(log.payload)
+                             else
+                               {}
+                             end
+      end
+
+      def metric
+        return unless log.metric
+
+        payload_value[:metric] = log.metric
+        payload_value[:metric_value] = log.metric_amount
+      end
+
+      def duration
+        return unless log.duration
+
+        payload_value[:duration] = log.duration.to_s
+        payload_value[:duration_human] = log.duration_human
+      end
+
+      def exception
+        return unless log.exception
+
+        payload_value.merge!(
+          exception_name:    log.exception.class.name,
+          exception_message: log.exception.message,
+          stack_trace:       log.exception.backtrace.to_s
+        )
+      end
+
+      def stringify_hash(hash)
+        result = {}
+
+        hash.each do |key, value|
+          string_key = key.to_s
+
+          result[string_key] = case value
+                               when Hash
+                                 JSON.generate(stringify_hash(value))
+                               else
+                                 value.to_s
+                               end
+        end
+
+        result
+      end
+    end
+  end
+end