From 5bc05cb4329eec0c977b9f5df4d6b0014e9051a5 Mon Sep 17 00:00:00 2001 From: "Tj (bougyman) Vanderpoel" Date: Tue, 15 Sep 2026 15:40:51 -0400 Subject: [PATCH] feat: add NATS service discovery tools (#58) --- Gemfile.lock | 24 +- exe/leopard-service-info | 20 ++ exe/leopard-service-map | 29 +++ exe/leopard-service-stats | 34 +++ exe/leopard-services | 27 +++ leopard.gemspec | 1 + lib/leopard/nats_service_discovery.rb | 179 +++++++++++++++ lib/leopard/nats_service_discovery/cli.rb | 208 ++++++++++++++++++ .../nats_service_discovery/operation.rb | 23 ++ .../operation/collect.rb | 35 +++ .../operation/connect.rb | 53 +++++ .../operation/discovery_options.rb | 31 +++ .../nats_service_discovery/operation/info.rb | 25 +++ .../nats_service_discovery/operation/ping.rb | 25 +++ .../operation/services.rb | 52 +++++ .../nats_service_discovery/operation/stats.rb | 25 +++ .../operation/subject_map.rb | 83 +++++++ .../nats_service_discovery/operation/verb.rb | 46 ++++ .../nats_service_discovery_test.rb | 206 +++++++++++++++++ 19 files changed, 1124 insertions(+), 2 deletions(-) create mode 100755 exe/leopard-service-info create mode 100755 exe/leopard-service-map create mode 100755 exe/leopard-service-stats create mode 100755 exe/leopard-services create mode 100644 lib/leopard/nats_service_discovery.rb create mode 100644 lib/leopard/nats_service_discovery/cli.rb create mode 100644 lib/leopard/nats_service_discovery/operation.rb create mode 100644 lib/leopard/nats_service_discovery/operation/collect.rb create mode 100644 lib/leopard/nats_service_discovery/operation/connect.rb create mode 100644 lib/leopard/nats_service_discovery/operation/discovery_options.rb create mode 100644 lib/leopard/nats_service_discovery/operation/info.rb create mode 100644 lib/leopard/nats_service_discovery/operation/ping.rb create mode 100644 lib/leopard/nats_service_discovery/operation/services.rb create mode 100644 lib/leopard/nats_service_discovery/operation/stats.rb create mode 100644 lib/leopard/nats_service_discovery/operation/subject_map.rb create mode 100644 lib/leopard/nats_service_discovery/operation/verb.rb create mode 100644 test/integration/nats_service_discovery_test.rb diff --git a/Gemfile.lock b/Gemfile.lock index f3dad78..717b384 100644 --- a/Gemfile.lock +++ b/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) diff --git a/exe/leopard-service-info b/exe/leopard-service-info new file mode 100755 index 0000000..7acd158 --- /dev/null +++ b/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 diff --git a/exe/leopard-service-map b/exe/leopard-service-map new file mode 100755 index 0000000..3c126e0 --- /dev/null +++ b/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 diff --git a/exe/leopard-service-stats b/exe/leopard-service-stats new file mode 100755 index 0000000..13fe9bb --- /dev/null +++ b/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 diff --git a/exe/leopard-services b/exe/leopard-services new file mode 100755 index 0000000..31ee982 --- /dev/null +++ b/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 diff --git a/leopard.gemspec b/leopard.gemspec index cbd4b6d..c9d5516 100644 --- a/leopard.gemspec +++ b/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 diff --git a/lib/leopard/nats_service_discovery.rb b/lib/leopard/nats_service_discovery.rb new file mode 100644 index 0000000..3bcbfa3 --- /dev/null +++ b/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] 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] 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] 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}] 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] 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] 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 diff --git a/lib/leopard/nats_service_discovery/cli.rb b/lib/leopard/nats_service_discovery/cli.rb new file mode 100644 index 0000000..392bf09 --- /dev/null +++ b/lib/leopard/nats_service_discovery/cli.rb @@ -0,0 +1,208 @@ +# frozen_string_literal: true + +require 'json' +require 'optparse' +require_relative 'operation' + +module Rubyists + module Leopard + class NatsServiceDiscovery + # Shared command-line helpers for NATS service discovery executables. + # + # @api private + module CLI + module_function + + # Parses common NATS discovery CLI options. + # + # @param argv [Array] Command-line arguments. + # @param banner [String] OptionParser banner. + # @param description [String] Command description. + # @param json [Boolean] Whether the command supports `--json`. + # + # @return [Hash] Parsed options. + def parse(argv, banner:, description:, json: false) + opts = default_options(json:) + OptionParser.new do |parser| + parser.banner = banner + parser.separator '' + parser.separator description + parser.separator '' + add_common_options(parser, opts) + end.parse!(argv) + validate!(opts) + opts + end + + # Opens a NATS client for the duration of the provided block. + # + # @param opts [Hash] Parsed CLI options. + # @yieldparam client [NATS::Client] Connected NATS client. + # + # @return [Object] The block return value. + def with_client(opts) + result = Operation::Connect.call( + server: opts[:server], + creds: opts[:creds], + connect_options: { reconnect: false, connect_timeout: opts[:timeout] }, + ) + abort_with(result[:error]) unless result.success? + + yield result[:client] + ensure + result&.[](:client)&.close + end + + # Prints a value as pretty JSON. + # + # @param value [Object] JSON-serializable value. + # + # @return [void] + def print_json(value) + puts JSON.pretty_generate(value) + end + + # Prints rows as a simple aligned table. + # + # @param headers [Array] Column headers. + # @param rows [Array] Table rows. + # + # @return [void] + def print_table(headers, rows) + widths = column_widths(headers, rows) + format = widths.map { |width| "%-#{width}s" }.join(' ') + puts format % headers + puts widths.map { |width| '-' * width }.join(' ') + rows.each { |row| puts format % row } + end + + # Prints an error and exits with a non-zero status. + # + # @param error [Exception] Error to report. + # + # @return [void] + def abort_with(error) + warn "error: #{error.message}" + exit 1 + end + + # Returns a successful operation result or aborts the process. + # + # @param result [Trailblazer::Operation::Railway::Result] Operation result. + # + # @return [Trailblazer::Operation::Railway::Result] + def operation_result!(result) + abort_with(result[:error]) unless result.success? + + result + end + + # Builds discovery operation options from CLI options and a client. + # + # @param opts [Hash] Parsed CLI options. + # @param client [NATS::Client] Connected NATS client. + # + # @return [Hash] Arguments for discovery operations. + def discovery_options(opts, client) + { + client:, + name: opts[:name], + id: opts[:id], + prefix: opts[:prefix], + timeout: opts[:timeout], + } + end + + # Builds default CLI options. + # + # @param json [Boolean] Whether the command supports `--json`. + # + # @return [Hash] Default options. + def default_options(json:) + { + server: ENV.fetch('NATS_URL', 'nats://127.0.0.1:4222'), + timeout: DEFAULT_TIMEOUT, + prefix: DEFAULT_PREFIX, + json: json ? false : nil, + } + end + + # Adds all common options to an OptionParser. + # + # @param parser [OptionParser] Parser to configure. + # @param opts [Hash] Mutable parsed option accumulator. + # + # @return [void] + def add_common_options(parser, opts) + add_connection_options(parser, opts) + add_filter_options(parser, opts) + add_output_options(parser, opts) + end + + # Adds NATS connection options to an OptionParser. + # + # @param parser [OptionParser] Parser to configure. + # @param opts [Hash] Mutable parsed option accumulator. + # + # @return [void] + def add_connection_options(parser, opts) + parser.on('-s', '--server URL', 'NATS server URL. Defaults to ENV[NATS_URL] or local NATS') do |url| + opts[:server] = url + end + parser.on('--creds FILE', 'NATS user credentials file') { |file| opts[:creds] = file } + end + + # Adds service filter options to an OptionParser. + # + # @param parser [OptionParser] Parser to configure. + # @param opts [Hash] Mutable parsed option accumulator. + # + # @return [void] + def add_filter_options(parser, opts) + parser.on('--name NAME', 'Only query services with this name') { |name| opts[:name] = name } + parser.on('--id ID', 'Only query this service id. Requires --name') { |id| opts[:id] = id } + parser.on('--prefix PREFIX', 'Service API prefix. Defaults to $SRV') { |prefix| opts[:prefix] = prefix } + parser.on('-t', '--timeout SECONDS', Float, 'Reply collection idle timeout') { |timeout| opts[:timeout] = timeout } + end + + # Adds output and help options to an OptionParser. + # + # @param parser [OptionParser] Parser to configure. + # @param opts [Hash] Mutable parsed option accumulator. + # + # @return [void] + def add_output_options(parser, opts) + parser.on('--json', 'Print JSON output') { opts[:json] = true } unless opts[:json].nil? + parser.on('-h', '--help', 'Show this help') do + puts parser + exit + end + end + + # Validates parsed CLI options. + # + # @param opts [Hash] Parsed CLI options. + # + # @return [void] + def validate!(opts) + return unless opts[:id] && opts[:name].to_s.empty? + + warn 'error: --id requires --name' + exit 1 + end + + # Computes table column widths. + # + # @param headers [Array] Column headers. + # @param rows [Array] Table rows. + # + # @return [Array] Width for each column. + def column_widths(headers, rows) + headers.each_index.map do |index| + ([headers[index]] + rows.map { |row| row[index].to_s }).map(&:length).max + end + end + end + end + end +end diff --git a/lib/leopard/nats_service_discovery/operation.rb b/lib/leopard/nats_service_discovery/operation.rb new file mode 100644 index 0000000..1cc57ea --- /dev/null +++ b/lib/leopard/nats_service_discovery/operation.rb @@ -0,0 +1,23 @@ +# frozen_string_literal: true + +require_relative '../nats_service_discovery' + +module Rubyists + module Leopard + class NatsServiceDiscovery + # Trailblazer operations for NATS Service API discovery workflows. + module Operation + end + end + end +end + +require_relative 'operation/discovery_options' +require_relative 'operation/connect' +require_relative 'operation/collect' +require_relative 'operation/verb' +require_relative 'operation/ping' +require_relative 'operation/info' +require_relative 'operation/stats' +require_relative 'operation/services' +require_relative 'operation/subject_map' diff --git a/lib/leopard/nats_service_discovery/operation/collect.rb b/lib/leopard/nats_service_discovery/operation/collect.rb new file mode 100644 index 0000000..2e8060f --- /dev/null +++ b/lib/leopard/nats_service_discovery/operation/collect.rb @@ -0,0 +1,35 @@ +# frozen_string_literal: true + +require 'trailblazer/operation' +require_relative '../../nats_service_discovery' + +module Rubyists + module Leopard + class NatsServiceDiscovery + module Operation + # Collects and parses JSON replies for an arbitrary Service API subject. + # + # On success, the result exposes `:responses`. + # On failure, the result exposes `:error`. + class Collect < Trailblazer::Operation + step :collect + + # Collects JSON replies from `subject`. + # + # @param ctx [Hash] Operation context. + # @param client [NATS::Client] Connected NATS client. + # @param subject [String] Service API subject to query. + # @param timeout [Numeric] Idle timeout while waiting for replies. + # + # @return [Boolean] Whether replies were collected successfully. + def collect(ctx, client:, subject:, timeout: DEFAULT_TIMEOUT, **) + ctx[:responses] = NatsServiceDiscovery.new(client:).send(:collect_json, subject, timeout:) + rescue StandardError => e + ctx[:error] = e + false + end + end + end + end + end +end diff --git a/lib/leopard/nats_service_discovery/operation/connect.rb b/lib/leopard/nats_service_discovery/operation/connect.rb new file mode 100644 index 0000000..2acac7d --- /dev/null +++ b/lib/leopard/nats_service_discovery/operation/connect.rb @@ -0,0 +1,53 @@ +# frozen_string_literal: true + +require 'nats/client' +require 'trailblazer/operation' +require_relative '../../nats_service_discovery' + +module Rubyists + module Leopard + class NatsServiceDiscovery + module Operation + # Opens a NATS client connection for discovery operations. + # + # On success, the result exposes `:client`. + # On failure, the result exposes `:error`. + class Connect < Trailblazer::Operation + step :connect + + # Connects to NATS using `:server` or `:nats_url`. + # + # @param ctx [Hash] Operation context. + # + # @return [Boolean] Whether the connection was opened. + def connect(ctx, **) + opts = (ctx[:connect_options] || {}).dup + opts[:user_credentials] = ctx[:creds] if ctx[:creds] + ctx[:client] = NATS.connect(connection_url(ctx), opts) + rescue StandardError => e + ctx[:error] = e + false + end + + private + + # Default NATS URL used when no URL is provided. + # + # @return [String] + def default_url + ENV.fetch('NATS_URL', 'nats://127.0.0.1:4222') + end + + # Resolves the NATS URL from the operation context. + # + # @param ctx [Hash] Operation context. + # + # @return [String] + def connection_url(ctx) + ctx[:nats_url] || ctx[:server] || default_url + end + end + end + end + end +end diff --git a/lib/leopard/nats_service_discovery/operation/discovery_options.rb b/lib/leopard/nats_service_discovery/operation/discovery_options.rb new file mode 100644 index 0000000..6d754ae --- /dev/null +++ b/lib/leopard/nats_service_discovery/operation/discovery_options.rb @@ -0,0 +1,31 @@ +# frozen_string_literal: true + +require_relative '../../nats_service_discovery' + +module Rubyists + module Leopard + class NatsServiceDiscovery + module Operation + # Shared option extraction for operations that compose other discovery operations. + module DiscoveryOptions + private + + # Extracts the common discovery options from a Trailblazer context. + # + # @param ctx [Hash] Operation context. + # + # @return [Hash] Arguments suitable for another discovery operation. + def operation_options(ctx) + { + client: ctx[:client], + prefix: ctx[:prefix], + timeout: ctx[:timeout], + name: ctx[:name], + id: ctx[:id], + } + end + end + end + end + end +end diff --git a/lib/leopard/nats_service_discovery/operation/info.rb b/lib/leopard/nats_service_discovery/operation/info.rb new file mode 100644 index 0000000..71d858a --- /dev/null +++ b/lib/leopard/nats_service_discovery/operation/info.rb @@ -0,0 +1,25 @@ +# frozen_string_literal: true + +require_relative 'verb' + +module Rubyists + module Leopard + class NatsServiceDiscovery + module Operation + # Collects `$SRV.INFO` responses. + # + # On success, the result exposes `:responses`. + class Info < Verb + # Invokes {NatsServiceDiscovery#info}. + # + # @param discovery [NatsServiceDiscovery] Discovery helper. + # + # @return [Array] Parsed info responses. + def public_send_verb(discovery, **) + discovery.info(**) + end + end + end + end + end +end diff --git a/lib/leopard/nats_service_discovery/operation/ping.rb b/lib/leopard/nats_service_discovery/operation/ping.rb new file mode 100644 index 0000000..29c9ffc --- /dev/null +++ b/lib/leopard/nats_service_discovery/operation/ping.rb @@ -0,0 +1,25 @@ +# frozen_string_literal: true + +require_relative 'verb' + +module Rubyists + module Leopard + class NatsServiceDiscovery + module Operation + # Collects `$SRV.PING` responses. + # + # On success, the result exposes `:responses`. + class Ping < Verb + # Invokes {NatsServiceDiscovery#ping}. + # + # @param discovery [NatsServiceDiscovery] Discovery helper. + # + # @return [Array] Parsed ping responses. + def public_send_verb(discovery, **) + discovery.ping(**) + end + end + end + end + end +end diff --git a/lib/leopard/nats_service_discovery/operation/services.rb b/lib/leopard/nats_service_discovery/operation/services.rb new file mode 100644 index 0000000..1e06fb5 --- /dev/null +++ b/lib/leopard/nats_service_discovery/operation/services.rb @@ -0,0 +1,52 @@ +# frozen_string_literal: true + +require 'trailblazer/operation' +require_relative 'discovery_options' +require_relative 'info' + +module Rubyists + module Leopard + class NatsServiceDiscovery + module Operation + # Builds service-list summaries from `$SRV.INFO` responses. + # + # On success, the result exposes `:responses` and `:services`. + class Services < Trailblazer::Operation + include DiscoveryOptions + + step :load_info? + step :summarize + + # Loads raw service info responses for summarization. + # + # @param ctx [Hash] Operation context. + # + # @return [Boolean] Whether service info was loaded. + def load_info?(ctx, **) + result = Info.call(**operation_options(ctx)) + ctx[:error] = result[:error] unless result.success? + ctx[:responses] = result[:responses] + result.success? + end + + # Builds compact service summaries from raw info responses. + # + # @param ctx [Hash] Operation context. + # @param responses [Array] Parsed `$SRV.INFO` responses. + # + # @return [Array] Service summary rows. + def summarize(ctx, responses:, **) + ctx[:services] = responses.map do |service| + { + 'name' => service['name'], + 'id' => service['id'], + 'version' => service['version'], + 'endpoints' => Array(service['endpoints']).size, + } + end + end + end + end + end + end +end diff --git a/lib/leopard/nats_service_discovery/operation/stats.rb b/lib/leopard/nats_service_discovery/operation/stats.rb new file mode 100644 index 0000000..ff59f81 --- /dev/null +++ b/lib/leopard/nats_service_discovery/operation/stats.rb @@ -0,0 +1,25 @@ +# frozen_string_literal: true + +require_relative 'verb' + +module Rubyists + module Leopard + class NatsServiceDiscovery + module Operation + # Collects `$SRV.STATS` responses. + # + # On success, the result exposes `:responses`. + class Stats < Verb + # Invokes {NatsServiceDiscovery#stats}. + # + # @param discovery [NatsServiceDiscovery] Discovery helper. + # + # @return [Array] Parsed stats responses. + def public_send_verb(discovery, **) + discovery.stats(**) + end + end + end + end + end +end diff --git a/lib/leopard/nats_service_discovery/operation/subject_map.rb b/lib/leopard/nats_service_discovery/operation/subject_map.rb new file mode 100644 index 0000000..52e9fac --- /dev/null +++ b/lib/leopard/nats_service_discovery/operation/subject_map.rb @@ -0,0 +1,83 @@ +# frozen_string_literal: true + +require 'trailblazer/operation' +require_relative 'discovery_options' +require_relative 'info' + +module Rubyists + module Leopard + class NatsServiceDiscovery + module Operation + # Builds a subject-to-endpoint listener map from `$SRV.INFO` responses. + # + # On success, the result exposes `:responses` and `:subject_map`. + class SubjectMap < Trailblazer::Operation + include DiscoveryOptions + + step :load_info? + step :map_subjects + + # Loads raw service info responses for mapping. + # + # @param ctx [Hash] Operation context. + # + # @return [Boolean] Whether service info was loaded. + def load_info?(ctx, **) + result = Info.call(**operation_options(ctx)) + ctx[:error] = result[:error] unless result.success? + ctx[:responses] = result[:responses] + result.success? + end + + # Builds a subject-to-listeners map from info responses. + # + # @param ctx [Hash] Operation context. + # @param responses [Array] Parsed `$SRV.INFO` responses. + # + # @return [Hash] Listener entries keyed by NATS subject. + def map_subjects(ctx, responses:, **) + ctx[:subject_map] = responses.each_with_object({}) do |service, subjects| + Array(service['endpoints']).each do |endpoint| + add_listener(subjects, service, endpoint) + end + end + end + + private + + # Adds an endpoint listener to the subject map. + # + # @param subjects [Hash] Subject map accumulator. + # @param service [Hash] Parsed `$SRV.INFO` service response. + # @param endpoint [Hash] Endpoint payload from the service response. + # + # @return [void] + def add_listener(subjects, service, endpoint) + subject = endpoint['subject'] + return if subject.to_s.empty? + + subjects[subject] ||= [] + subjects[subject] << listener_entry(service, endpoint) + end + + # Builds a subject-map listener entry. + # + # @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 + end + end + end + end +end diff --git a/lib/leopard/nats_service_discovery/operation/verb.rb b/lib/leopard/nats_service_discovery/operation/verb.rb new file mode 100644 index 0000000..fb8336d --- /dev/null +++ b/lib/leopard/nats_service_discovery/operation/verb.rb @@ -0,0 +1,46 @@ +# frozen_string_literal: true + +require 'trailblazer/operation' +require_relative '../../nats_service_discovery' + +module Rubyists + module Leopard + class NatsServiceDiscovery + module Operation + # Base class for `$SRV.` operations. + # + # Subclasses choose which Service API verb to invoke. + class Verb < Trailblazer::Operation + step :collect + + # Collects responses for the subclass Service API verb. + # + # @param ctx [Hash] Operation context. + # + # @return [Boolean] Whether responses were collected successfully. + def collect(ctx, **) + discovery = NatsServiceDiscovery.new(client: ctx[:client], prefix: ctx[:prefix] || DEFAULT_PREFIX) + ctx[:responses] = public_send_verb( + discovery, + name: ctx[:name], + id: ctx[:id], + timeout: ctx[:timeout] || DEFAULT_TIMEOUT, + ) + rescue StandardError => e + ctx[:error] = e + false + end + + private + + # Invokes the subclass-specific discovery method. + # + # @raise [NotImplementedError] when a subclass does not implement this. + def public_send_verb(_discovery, **) + raise NotImplementedError, "#{self.class} must define #public_send_verb" + end + end + end + end + end +end diff --git a/test/integration/nats_service_discovery_test.rb b/test/integration/nats_service_discovery_test.rb new file mode 100644 index 0000000..f92ad92 --- /dev/null +++ b/test/integration/nats_service_discovery_test.rb @@ -0,0 +1,206 @@ +# frozen_string_literal: true + +require 'securerandom' +require 'open3' +require 'rbconfig' +require 'timeout' +require_relative '../helper' +require Rubyists::Leopard.libroot / 'leopard/nats_service_discovery/operation' + +class NatsServiceDiscoveryIntegrationTest < Minitest::Test # rubocop:disable Metrics/ClassLength + NATS_URL = ENV.fetch('LEOPARD_NATS_URL', ENV.fetch('NATS_URI', 'nats://127.0.0.1:4222')) + WAIT_TIMEOUT = 5 + + def setup # rubocop:disable Metrics/MethodLength + skip 'NATS broker not available' unless nats_available? + + @token = SecureRandom.hex(4) + @client = NATS.connect(NATS_URL, reconnect: false, connect_timeout: 0.5, max_reconnect_attempts: 0) + @service = @client.services.add( + name: service_name, + version: '1.2.3', + description: 'Discovery integration service', + metadata: { 'schema_registry' => 'proto-schema-registry' }, + ) + add_test_endpoints + @client.flush + end + + def teardown + @service&.stop + @client&.close + end + + def test_info_operation_returns_service_endpoint_contracts # rubocop:disable Metrics/AbcSize, Minitest/MultipleAssertions + result = info_operation + + assert_predicate result, :success? + assert_equal 1, result[:responses].size + assert_equal service_name, info_response['name'] + assert_equal ['proto.sum.Request'], endpoint('sum')['metadata']['request']['message'] + assert_equal [avg_subject, sum_subject], endpoint_subjects + end + + def test_subject_map_operation_groups_listeners_by_subject + result = operation(:SubjectMap) + + assert_predicate result, :success? + sum_endpoints = result[:subject_map][sum_subject].map { |listener| listener['endpoint'] } + + assert_equal %w[sum], sum_endpoints + response_messages = result[:subject_map][avg_subject].map do |listener| + listener['metadata']['response']['message'] + end + + assert_equal ['proto.avg.Response'], response_messages + end + + def test_services_operation_summarizes_info + result = operation(:Services) + + assert_predicate result, :success? + assert_equal [ + { + 'name' => service_name, + 'id' => @service.id, + 'version' => '1.2.3', + 'endpoints' => 2, + }, + ], result[:services] + end + + def test_ping_and_stats_operations_collect_live_service_responses # rubocop:disable Minitest/MultipleAssertions + @client.request(sum_subject, '{}') + + ping = operation(:Ping) + stats = operation(:Stats) + + assert_predicate ping, :success? + assert_predicate stats, :success? + assert_equal 'io.nats.micro.v1.ping_response', ping[:responses].first['type'] + assert_equal 1, stats_endpoint('sum')['num_requests'] + end + + def test_collect_operation_can_query_a_specific_service_subject + subject = "$SRV.INFO.#{service_name}" + result = Rubyists::Leopard::NatsServiceDiscovery::Operation::Collect.call( + client: @client, + subject:, + timeout: 0.5, + ) + + assert_predicate result, :success? + assert_equal service_name, result[:responses].first['name'] + end + + def test_connect_operation_opens_a_client + result = Rubyists::Leopard::NatsServiceDiscovery::Operation::Connect.call( + server: NATS_URL, + connect_options: { reconnect: false, connect_timeout: 0.5, max_reconnect_attempts: 0 }, + ) + + assert_predicate result, :success? + assert_predicate result[:client], :connected? + ensure + result&.[](:client)&.close + end + + def test_missing_service_returns_empty_responses + result = Rubyists::Leopard::NatsServiceDiscovery::Operation::Info.call( + client: @client, + name: "Missing#{@token}", + timeout: 0.05, + ) + + assert_predicate result, :success? + assert_empty result[:responses] + end + + def test_command_line_tools_query_and_format_service_data # rubocop:disable Metrics/AbcSize, Minitest/MultipleAssertions + info = command_json('leopard-service-info') + services = command_json('leopard-services', '--json') + stats = command_json('leopard-service-stats', '--json') + subject_map = command_json('leopard-service-map', '--json') + + assert_equal service_name, info.first['name'] + assert_equal service_name, services.first['name'] + assert_equal service_name, stats.first['name'] + assert_includes subject_map, sum_subject + end + + private + + def nats_available? + nc = NATS.connect(NATS_URL, reconnect: false, connect_timeout: 0.5, max_reconnect_attempts: 0) + nc.close + true + rescue StandardError + false + end + + def command_json(executable, *) + stdout, stderr, status = Open3.capture3( + RbConfig.ruby, "exe/#{executable}", '--server', NATS_URL, '--name', service_name, * + ) + + assert_predicate status, :success?, stderr + JSON.parse(stdout) + end + + def add_test_endpoints + @service.endpoints.add('sum', subject: sum_subject, metadata: sum_metadata) do |message| + message.respond('{"sum":0}') + end + @service.endpoints.add('avg', subject: avg_subject, metadata: avg_metadata) do |message| + message.respond('{"avg":0}') + end + end + + def operation(name) + Rubyists::Leopard::NatsServiceDiscovery::Operation.const_get(name).call( + client: @client, + name: service_name, + timeout: 0.5, + ) + end + + def info_operation + @info_operation ||= operation(:Info) + end + + def info_response + info_operation[:responses].first + end + + def endpoint(name) + info_response['endpoints'].find { |endpoint| endpoint['name'] == name } + end + + def endpoint_subjects + info_response['endpoints'].map { |endpoint| endpoint['subject'] }.sort + end + + def stats_endpoint(name) + operation(:Stats)[:responses].first['endpoints'].find { |endpoint| endpoint['name'] == name } + end + + def service_name + "DiscoveryService#{@token}" + end + + def sum_subject + "leopard.discovery.#{@token}.sum" + end + + def avg_subject + "leopard.discovery.#{@token}.avg" + end + + def sum_metadata + { 'request' => { 'message' => ['proto.sum.Request'] } } + end + + def avg_metadata + { 'response' => { 'message' => 'proto.avg.Response' } } + end +end