Better logging

zhulik · Feb 9, 2023 · 9531eb0 · 9531eb0
1 parent b8dd946
commit 9531eb0
Show file tree

Hide file tree

Showing 9 changed files with 45 additions and 16 deletions.
diff --git a/Dockerfile b/Dockerfile
@@ -21,10 +21,10 @@ RUN bundle install
 
 FROM base
 
-COPY --from=builder /mnt/vendor .
+COPY --from=builder /mnt .
 
 ADD . .
 
 USER app
 
-CMD ["bundle", "exec", "./exe/nats_streamer", "/config.yaml", "$PATH"]
+CMD ["bundle", "exec", "./exe/nats_streamer", "/config.yaml"]
diff --git a/exe/nats_streamer b/exe/nats_streamer
@@ -3,8 +3,7 @@
 
 require_relative "../lib/nats_streamer"
 
-CONFIG = YAML.load(File.read(ARGV[0]), symbolize_names: true)
-
 Sync do
-  NatsStreamer::Application.new(url: "nats://127.0.0.1:4222", config: NatsStreamer::Config.new(CONFIG)).run
+  config = NatsStreamer::Config.new(YAML.load(File.read(ARGV[0]), symbolize_names: true))
+  NatsStreamer::Application.new(config:).run
 end
diff --git a/lib/nats_streamer/application.rb b/lib/nats_streamer/application.rb
@@ -6,7 +6,6 @@ class NatsStreamer::Application
   include NatsStreamer::Logger
   include Memery
 
-  option :url
   option :config
 
   def run
@@ -19,7 +18,9 @@ def run
 
   private
 
-  memoize def client = NATS.connect(url)
-
-  def jsm = client.jsm
+  memoize def jsm
+    NATS.connect(config.server_url).tap do
+      info { "Connected to #{config.server_url}" }
+    end.jsm
+  end
 end
diff --git a/lib/nats_streamer/config.rb b/lib/nats_streamer/config.rb
@@ -9,6 +9,8 @@ class Subscriber < Dry::Struct
     attribute :params, Types::Hash.map(Types::Coercible::String, Types::String).default({}.freeze)
   end
 
+  attribute :server_url, Types::Coercible::String
+
   attribute :streams, Types::Hash.map(
     Types::Coercible::String,
     Types::Hash.map(

diff --git a/lib/nats_streamer/connection.rb b/lib/nats_streamer/connection.rb
@@ -0,0 +1,10 @@
+# frozen_string_literal: true
+
+class NatsStreamer::Connection
+  def self.build(url)
+    Faraday.new(url) do |f|
+      f.request :json
+      f.response :raise_error
+    end
+  end
+end
diff --git a/lib/nats_streamer/deliverer.rb b/lib/nats_streamer/deliverer.rb
@@ -9,12 +9,12 @@ class NatsStreamer::Deliverer
   # TODO: how to propely unsubscribe?
   option :subscriber
 
-  def deliver(**) = connection.post(".", **)
-
-  memoize def connection
-    Faraday.new(subscriber.url) do |f|
-      f.request :json
-      f.response :raise_error
-    end
+  def deliver(**)
+    info_measure(-> { connection.post(".", **) }) { "Event delivered to #{subscriber.name}: #{_1.round(2)}s" }
+  rescue StandardError => e
+    warn { "Event failed to be delivered to #{subscriber.name}" }
+    warn { e }
   end
+
+  memoize def connection = NatsStreamer::Connection.build(subscriber.url)
 end
diff --git a/lib/nats_streamer/listener.rb b/lib/nats_streamer/listener.rb
@@ -12,6 +12,8 @@ class NatsStreamer::Listener
   option :subscriber
 
   def run
+    info { "Subscribing to #{subject}" }
+
     pull do |msg|
       Async { handle_message(msg) }
     end

diff --git a/lib/nats_streamer/logger.rb b/lib/nats_streamer/logger.rb
@@ -5,5 +5,16 @@ module NatsStreamer::Logger
     define_method(name) do |msg = nil, &block|
       Console.logger.public_send(name, self, msg, &block)
     end
+
+    define_method("#{name}_measure") do |lambda, msg = nil, &block|
+      elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC).then do |start|
+        lambda.call
+        Process.clock_gettime(Process::CLOCK_MONOTONIC) - start
+      end
+
+      return Console.logger.public_send(name, self, msg) { block.call(elapsed) } unless block.nil?
+
+      Console.logger.public_send(name, self, msg)
+    end
   end
 end
diff --git a/lib/nats_streamer/stream.rb b/lib/nats_streamer/stream.rb
@@ -33,8 +33,12 @@ def run
   end
 
   def create_stream!
+    info { "Creating stream '#{name}' with subjects #{subject_names}" }
+
     jsm.add_stream(name:, subjects: subject_names)
   rescue NATS::JetStream::Error::BadRequest
+    warn { "Configuration changed, attempting deleting and recreating stream #{name}" }
+
     jsm.delete_stream(name)
     retry
   end