feat: add NATS service discovery tools

Gemfile.lock

@@ -1,12 +1,13 @@
 PATH
   remote: .
   specs:
-    leopard (0.2.7)
+    leopard (0.2.9)
       concurrent-ruby (~> 1.1)
       dry-configurable (~> 1.3)
       dry-monads (~> 1.9)
       nats-pure (~> 2.5)
-      semantic_logger (~> 4)
+      semantic_logger (>= 4)
+      trailblazer-operation (~> 0.11)
 
 GEM
   remote: https://rubygems.org/
@@ -31,6 +32,9 @@ GEM
       dry-core (~> 1.1)
       zeitwerk (~> 2.6)
     erb (6.0.3)
+    hashie (5.1.0)
+      logger
+    hirb (0.7.3)
     io-console (0.8.1)
     irb (1.17.0)
       pp (>= 0.6.0)
@@ -114,6 +118,22 @@ GEM
     simplecov-html (0.13.2)
     simplecov_json_formatter (0.1.4)
     stringio (3.2.0)
+    trailblazer-activity (0.17.0)
+      trailblazer-context (~> 0.5.0)
+      trailblazer-option (~> 0.1.0)
+    trailblazer-activity-dsl-linear (1.2.6)
+      trailblazer-activity (>= 0.17.0, < 0.18.0)
+      trailblazer-declarative (>= 0.0.1, < 0.1.0)
+    trailblazer-context (0.5.1)
+      hashie (>= 3.0.0)
+    trailblazer-declarative (0.0.2)
+    trailblazer-developer (0.1.0)
+      hirb
+      trailblazer-activity-dsl-linear (>= 1.2.0, < 1.3.0)
+    trailblazer-operation (0.11.1)
+      trailblazer-activity-dsl-linear (>= 1.2.3, < 1.3.0)
+      trailblazer-developer (>= 0.1.0, < 0.2.0)
+    trailblazer-option (0.1.2)
     tsort (0.2.0)
     unicode-display_width (3.1.5)
       unicode-emoji (~> 4.0, >= 4.0.4)

exe/leopard-service-info

@@ -0,0 +1,20 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+$LOAD_PATH.unshift File.expand_path('../lib', __dir__)
+
+require 'leopard/nats_service_discovery/cli'
+
+cli = Rubyists::Leopard::NatsServiceDiscovery::CLI
+opts = cli.parse(
+  ARGV,
+  banner: 'Usage: leopard-service-info [options]',
+  description: 'Prints raw $SRV.INFO responses as JSON.',
+)
+
+cli.with_client(opts) do |client|
+  result = cli.operation_result!(
+    Rubyists::Leopard::NatsServiceDiscovery::Operation::Info.call(**cli.discovery_options(opts, client)),
+  )
+  cli.print_json(result[:responses])
+end

exe/leopard-service-map

@@ -0,0 +1,29 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+$LOAD_PATH.unshift File.expand_path('../lib', __dir__)
+
+require 'leopard/nats_service_discovery/cli'
+
+cli = Rubyists::Leopard::NatsServiceDiscovery::CLI
+opts = cli.parse(
+  ARGV,
+  banner: 'Usage: leopard-service-map [options]',
+  description: 'Builds a subject-to-endpoint map from $SRV.INFO responses.',
+  json: true,
+)
+
+cli.with_client(opts) do |client|
+  result = cli.operation_result!(
+    Rubyists::Leopard::NatsServiceDiscovery::Operation::SubjectMap.call(**cli.discovery_options(opts, client)),
+  )
+
+  if opts[:json]
+    cli.print_json(result[:subject_map])
+  else
+    rows = result[:subject_map].flat_map do |subject, listeners|
+      listeners.map { |listener| [subject, *listener.values_at('service', 'service_id', 'version', 'endpoint', 'queue_group')] }
+    end
+    cli.print_table(%w[SUBJECT SERVICE SERVICE_ID VERSION ENDPOINT QUEUE], rows)
+  end
+end

exe/leopard-service-stats

@@ -0,0 +1,34 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+$LOAD_PATH.unshift File.expand_path('../lib', __dir__)
+
+require 'leopard/nats_service_discovery/cli'
+
+cli = Rubyists::Leopard::NatsServiceDiscovery::CLI
+opts = cli.parse(
+  ARGV,
+  banner: 'Usage: leopard-service-stats [options]',
+  description: 'Prints service endpoint stats from $SRV.STATS responses.',
+  json: true,
+)
+
+cli.with_client(opts) do |client|
+  result = cli.operation_result!(
+    Rubyists::Leopard::NatsServiceDiscovery::Operation::Stats.call(**cli.discovery_options(opts, client)),
+  )
+
+  if opts[:json]
+    cli.print_json(result[:responses])
+  else
+    rows = result[:responses].flat_map do |service|
+      Array(service['endpoints']).map do |endpoint|
+        [
+          service['name'], service['id'], endpoint['subject'], endpoint['name'], endpoint['queue_group'],
+          endpoint['num_requests'], endpoint['num_errors'], endpoint['average_processing_time'], endpoint['last_error']
+        ]
+      end
+    end
+    cli.print_table(%w[SERVICE SERVICE_ID SUBJECT ENDPOINT QUEUE REQUESTS ERRORS AVG_NS LAST_ERROR], rows)
+  end
+end

exe/leopard-services

@@ -0,0 +1,27 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+$LOAD_PATH.unshift File.expand_path('../lib', __dir__)
+
+require 'leopard/nats_service_discovery/cli'
+
+cli = Rubyists::Leopard::NatsServiceDiscovery::CLI
+opts = cli.parse(
+  ARGV,
+  banner: 'Usage: leopard-services [options]',
+  description: 'Lists NATS Service API services and their endpoint counts.',
+  json: true,
+)
+
+cli.with_client(opts) do |client|
+  result = cli.operation_result!(
+    Rubyists::Leopard::NatsServiceDiscovery::Operation::Services.call(**cli.discovery_options(opts, client)),
+  )
+
+  if opts[:json]
+    cli.print_json(result[:services])
+  else
+    rows = result[:services].map { |service| service.values_at('name', 'id', 'version', 'endpoints') }
+    cli.print_table(%w[SERVICE ID VERSION ENDPOINTS], rows)
+  end
+end

leopard.gemspec

@@ -39,6 +39,7 @@ Gem::Specification.new do |spec| # rubocop:disable Metrics/BlockLength
   spec.add_dependency 'dry-monads', '~> 1.9'
   spec.add_dependency 'nats-pure', '~> 2.5'
   spec.add_dependency 'semantic_logger', '>= 4'
+  spec.add_dependency 'trailblazer-operation', '~> 0.11'
 
   # For more information and examples about making a new gem, check out our
   # guide at: https://bundler.io/guides/creating_gem.html

lib/leopard/nats_service_discovery.rb

@@ -0,0 +1,179 @@
+# frozen_string_literal: true
+
+require 'json'
+require 'nats/client'
+
+module Rubyists
+  module Leopard
+    # Collects NATS Service API monitoring responses from a cluster.
+    #
+    # The NATS Service API uses request/reply subjects such as `$SRV.INFO`.
+    # Cluster-wide requests can receive multiple replies, so this helper creates
+    # an inbox subscription, publishes the request, and collects replies until a
+    # short timeout elapses.
+    class NatsServiceDiscovery
+      # Default NATS Service API monitoring prefix.
+      DEFAULT_PREFIX = '$SRV'
+      # Default idle timeout used to decide that all service replies arrived.
+      DEFAULT_TIMEOUT = 0.25
+
+      attr_reader :client, :prefix
+
+      # @param client [NATS::Client] Connected NATS client.
+      # @param prefix [String] Service API monitoring prefix.
+      def initialize(client:, prefix: DEFAULT_PREFIX)
+        @client = client
+        @prefix = prefix
+      end
+
+      # Collects `$SRV.PING` responses.
+      #
+      # @param name [String, nil] Optional service name filter.
+      # @param id [String, nil] Optional service id filter; requires `name`.
+      # @param timeout [Numeric] Idle timeout while collecting replies.
+      #
+      # @return [Array<Hash>] Parsed ping responses.
+      def ping(name: nil, id: nil, timeout: DEFAULT_TIMEOUT)
+        collect_json(service_subject('PING', name:, id:), timeout:)
+      end
+
+      # Collects `$SRV.INFO` responses.
+      #
+      # @param name [String, nil] Optional service name filter.
+      # @param id [String, nil] Optional service id filter; requires `name`.
+      # @param timeout [Numeric] Idle timeout while collecting replies.
+      #
+      # @return [Array<Hash>] Parsed info responses.
+      def info(name: nil, id: nil, timeout: DEFAULT_TIMEOUT)
+        collect_json(service_subject('INFO', name:, id:), timeout:)
+      end
+
+      # Collects `$SRV.STATS` responses.
+      #
+      # @param name [String, nil] Optional service name filter.
+      # @param id [String, nil] Optional service id filter; requires `name`.
+      # @param timeout [Numeric] Idle timeout while collecting replies.
+      #
+      # @return [Array<Hash>] Parsed stats responses.
+      def stats(name: nil, id: nil, timeout: DEFAULT_TIMEOUT)
+        collect_json(service_subject('STATS', name:, id:), timeout:)
+      end
+
+      # Builds a subject-to-listener map from service info responses.
+      #
+      # @param name [String, nil] Optional service name filter.
+      # @param id [String, nil] Optional service id filter; requires `name`.
+      # @param timeout [Numeric] Idle timeout while collecting replies.
+      #
+      # @return [Hash{String => Array<Hash>}] Endpoint listeners keyed by subject.
+      def endpoint_subject_map(name: nil, id: nil, timeout: DEFAULT_TIMEOUT)
+        info(name:, id:, timeout:).each_with_object({}) do |service, subjects|
+          Array(service['endpoints']).each do |endpoint|
+            subject = endpoint['subject']
+            next if subject.to_s.empty?
+
+            subjects[subject] ||= []
+            subjects[subject] << listener_entry(service, endpoint)
+          end
+        end
+      end
+
+      # Builds a Service API monitoring subject.
+      #
+      # @param verb [String, Symbol] Monitoring verb such as `INFO`.
+      # @param name [String, nil] Optional service name filter.
+      # @param id [String, nil] Optional service id filter; requires `name`.
+      #
+      # @return [String] Monitoring subject.
+      def service_subject(verb, name: nil, id: nil)
+        raise ArgumentError, 'service id requires a service name' if present?(id) && !present?(name)
+
+        parts = [prefix, verb.to_s.upcase]
+        parts << name.to_s if present?(name)
+        parts << id.to_s if present?(id)
+        parts.join('.')
+      end
+
+      private
+
+      # Collects JSON replies for a Service API subject until `timeout` elapses.
+      #
+      # @param subject [String] Service API subject to publish.
+      # @param timeout [Numeric] Idle timeout while waiting for replies.
+      #
+      # @return [Array<Hash>] Parsed response payloads.
+      def collect_json(subject, timeout:)
+        replies = []
+        sub = subscribe_to_inbox
+        client.publish(subject, '', sub.subject)
+        begin
+          collect_replies(sub, replies, timeout)
+        rescue NATS::Timeout
+          replies
+        end
+      ensure
+        sub&.unsubscribe
+      end
+
+      # Subscribes to an ephemeral reply inbox and flushes the subscription.
+      #
+      # @return [NATS::Subscription] Subscription bound to the reply inbox.
+      def subscribe_to_inbox
+        sub = client.subscribe(client.new_inbox)
+        client.flush
+        sub
+      end
+
+      # Appends parsed replies to the provided accumulator until timeout.
+      #
+      # @param sub [NATS::Subscription] Reply subscription.
+      # @param replies [Array<Hash>] Response accumulator.
+      # @param timeout [Numeric] Idle timeout while waiting for replies.
+      #
+      # @return [void]
+      def collect_replies(sub, replies, timeout)
+        loop do
+          msg = sub.next_msg(timeout:)
+          next if no_responders?(msg)
+
+          replies << JSON.parse(msg.data)
+        end
+      end
+
+      # Reports whether a reply is the server no-responders status message.
+      #
+      # @param msg [NATS::Msg] Reply message.
+      #
+      # @return [Boolean]
+      def no_responders?(msg)
+        msg.header && msg.header['Status'] == '503'
+      end
+
+      # Builds a subject-map listener entry from a service and endpoint payload.
+      #
+      # @param service [Hash] Parsed `$SRV.INFO` service response.
+      # @param endpoint [Hash] Endpoint payload from the service response.
+      #
+      # @return [Hash] Listener entry suitable for subject maps.
+      def listener_entry(service, endpoint)
+        {
+          'service' => service['name'],
+          'service_id' => service['id'],
+          'version' => service['version'],
+          'endpoint' => endpoint['name'],
+          'queue_group' => endpoint['queue_group'],
+          'metadata' => endpoint['metadata'],
+        }
+      end
+
+      # Reports whether a value is present for subject construction.
+      #
+      # @param value [Object] Value to check.
+      #
+      # @return [Boolean]
+      def present?(value)
+        !value.nil? && !value.to_s.empty?
+      end
+    end
+  end
+end