Merge pull request #12 from lessonnine/TNT-2741/parse-tcp-info

[TNT-2741] Parse TCP INFO binary data

Author: Leszek Zalewski

.github/workflows/build.yml

@@ -7,7 +7,7 @@ jobs:
     strategy:
       matrix:
         os: ['ubuntu-18.04', 'ubuntu-20.04']
-        ruby: ['2.6', '2.7', '3.0']
+        ruby: ['2.6', '2.7', '3.0', '3.1']
 
     runs-on: ${{ matrix.os }}
 

CHANGELOG.md

@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [1.1.0 Beta]
+
+### Added
+
+Different ways to parse `Socket::Option`. Mainly due to the fact that `#inspect` can't
+generate proper data on AWS Fargate, which runs Amazon Linux 2 with 4.14 kernel. So now
+besides `#inspect` there's also `#unpack` that parses binary data and picks proper field.
+
+It depends on the kernel, but new fields are usually added at the end of the `tcp_info`
+struct, so it should more or less stay stable.
+
+You can configure it by passing in `config.socket_parser = :inspect` or
+`config.socket_parser = ->(opt) { your implementation }`.
+
 ## [1.1.0 Alpha]
 
 ### Added

Gemfile.lock

@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    puma-plugin-telemetry (1.1.0.alpha)
+    puma-plugin-telemetry (1.1.0.beta)
       puma (>= 5.0)
 
 GEM

README.md

@@ -62,6 +62,8 @@ Puma::Plugin::Telemetry.configure do |config|
   config.initial_delay = 10
   config.frequency = 30
   config.puma_telemetry = %w[workers.requests_count queue.backlog queue.capacity]
+  config.socket_telemetry!
+  config.socket_parser = :inspect
   config.add_target :io, formatter: :json, io: StringIO.new
   config.add_target :dogstatsd, client: Datadog::Statsd.new(tags: { env: ENV["RAILS_ENV"] })
 end

lib/puma/plugin/telemetry.rb

@@ -52,7 +52,7 @@ def socket_telemetry(telemetry, launcher)
           return telemetry if launcher.nil?
           return telemetry unless config.socket_telemetry?
 
-          telemetry.merge! SocketData.new(launcher.binder.ios)
+          telemetry.merge! SocketData.new(launcher.binder.ios, config.socket_parser)
                                      .metrics
 
           telemetry

lib/puma/plugin/telemetry/config.rb

@@ -62,13 +62,28 @@ class Config
         # - default: false
         attr_accessor :socket_telemetry
 
+        # Symbol representing method to parse the `Socket::Option`, or
+        # the whole implementation as a lambda. Available options:
+        # - `:inspect`, based on the `Socket::Option#inspect` method,
+        #   it's the safest and slowest way to extract the info. `inspect`
+        #   output might not be available, i.e. on AWS Fargate
+        # - `:unpack`, parse binary data given by `Socket::Option`. Fastest
+        #   way (12x compared to `inspect`) but depends on kernel headers
+        #   and fields ordering within the struct. It should almost always
+        #   match though. DEFAULT
+        # - proc/lambda, `Socket::Option` will be given as an argument, it
+        #   should return the value of `unacked` field as an integer.
+        #
+        attr_accessor :socket_parser
+
         def initialize
           @enabled = false
           @initial_delay = 5
           @frequency = 5
           @targets = []
           @puma_telemetry = DEFAULT_PUMA_TELEMETRY
           @socket_telemetry = false
+          @socket_parser = :unpack
         end
 
         def enabled?

lib/puma/plugin/telemetry/data.rb

@@ -100,13 +100,34 @@ def sum_stat(stat)
       class SocketData
         UNACKED_REGEXP = /\ unacked=(?<unacked>\d+)\ /.freeze
 
-        def initialize(ios)
-          @sockets = ios.select { |io| io.respond_to?(:getsockopt) }
+        def initialize(ios, parser)
+          @sockets = ios.select { |io| io.respond_to?(:getsockopt) && io.is_a?(TCPSocket) }
+          @parser =
+            case parser
+            when :inspect then method(:parse_with_inspect)
+            when :unpack then method(:parse_with_unpack)
+            when Proc then parser
+            end
         end
 
         # Number of unacknowledged connections in the sockets, which
         # we know as socket backlog.
         #
+        def unacked
+          @sockets.sum do |socket|
+            @parser.call(socket.getsockopt(Socket::SOL_TCP,
+                                           Socket::TCP_INFO))
+          end
+        end
+
+        def metrics
+          {
+            "sockets.backlog" => unacked
+          }
+        end
+
+        private
+
         # The Socket::Option returned by `getsockopt` doesn't provide
         # any kind of accessors for data inside. It decodes it on demand
         # for `inspect` as strings in C implementation. It looks like
@@ -143,21 +164,104 @@ def initialize(ios)
         #                                     total_retrans=0
         #                                     (128 bytes too long)>
         #
-        # That's why we have to pull the `unacked`  field by parsing
-        # `inspect` output, instead of using something like `opt.unacked`
-        def unacked
-          @sockets.sum do |socket|
-            tcp_info = socket.getsockopt(Socket::SOL_TCP, Socket::TCP_INFO).inspect
-            tcp_match = tcp_info.match(UNACKED_REGEXP)
+        # That's why pulling the `unacked` field by parsing
+        # `inspect` output is one of the ways to retrieve it.
+        #
+        def parse_with_inspect(tcp_info)
+          tcp_match = tcp_info.inspect.match(UNACKED_REGEXP)
 
-            tcp_match[:unacked].to_i
-          end
+          return 0 if tcp_match.nil?
+
+          tcp_match[:unacked].to_i
         end
 
-        def metrics
-          {
-            "sockets.backlog" => unacked
-          }
+        # The above inspect data might not be available everywhere (looking at you
+        # AWS Fargate Host running on kernel 4.14!), but we might still recover it
+        # by manually unpacking the binary data based on linux headers. For example
+        # below is tcp info struct from `linux/tcp.h` header file, from problematic
+        # host rocking kernel 4.14.
+        #
+        #     struct tcp_info {
+        #         __u8    tcpi_state;
+        #         __u8    tcpi_ca_state;
+        #         __u8    tcpi_retransmits;
+        #         __u8    tcpi_probes;
+        #         __u8    tcpi_backoff;
+        #         __u8    tcpi_options;
+        #         __u8    tcpi_snd_wscale : 4, tcpi_rcv_wscale : 4;
+        #         __u8    tcpi_delivery_rate_app_limited:1;
+        #
+        #         __u32   tcpi_rto;
+        #         __u32   tcpi_ato;
+        #         __u32   tcpi_snd_mss;
+        #         __u32   tcpi_rcv_mss;
+        #
+        #         __u32   tcpi_unacked;
+        #         __u32   tcpi_sacked;
+        #         __u32   tcpi_lost;
+        #         __u32   tcpi_retrans;
+        #         __u32   tcpi_fackets;
+        #
+        #         /* Times. */
+        #         __u32   tcpi_last_data_sent;
+        #         __u32   tcpi_last_ack_sent;     /* Not remembered, sorry. */
+        #         __u32   tcpi_last_data_recv;
+        #         __u32   tcpi_last_ack_recv;
+        #
+        #         /* Metrics. */
+        #         __u32   tcpi_pmtu;
+        #         __u32   tcpi_rcv_ssthresh;
+        #         __u32   tcpi_rtt;
+        #         __u32   tcpi_rttvar;
+        #         __u32   tcpi_snd_ssthresh;
+        #         __u32   tcpi_snd_cwnd;
+        #         __u32   tcpi_advmss;
+        #         __u32   tcpi_reordering;
+        #
+        #         __u32   tcpi_rcv_rtt;
+        #         __u32   tcpi_rcv_space;
+        #
+        #         __u32   tcpi_total_retrans;
+        #
+        #         __u64   tcpi_pacing_rate;
+        #         __u64   tcpi_max_pacing_rate;
+        #         __u64   tcpi_bytes_acked;    /* RFC4898 tcpEStatsAppHCThruOctetsAcked */
+        #         __u64   tcpi_bytes_received; /* RFC4898 tcpEStatsAppHCThruOctetsReceived */
+        #         __u32   tcpi_segs_out;       /* RFC4898 tcpEStatsPerfSegsOut */
+        #         __u32   tcpi_segs_in;        /* RFC4898 tcpEStatsPerfSegsIn */
+        #
+        #         __u32   tcpi_notsent_bytes;
+        #         __u32   tcpi_min_rtt;
+        #         __u32   tcpi_data_segs_in;      /* RFC4898 tcpEStatsDataSegsIn */
+        #         __u32   tcpi_data_segs_out;     /* RFC4898 tcpEStatsDataSegsOut */
+        #
+        #         __u64   tcpi_delivery_rate;
+        #
+        #         __u64   tcpi_busy_time;      /* Time (usec) busy sending data */
+        #         __u64   tcpi_rwnd_limited;   /* Time (usec) limited by receive window */
+        #         __u64   tcpi_sndbuf_limited; /* Time (usec) limited by send buffer */
+        #     };
+        #
+        # Now nowing types and order of fields we can easily parse binary data
+        # by using
+        # - `C` flag for `__u8` type - 8-bit unsigned (unsigned char)
+        # - `L` flag for `__u32` type - 32-bit unsigned, native endian (uint32_t)
+        # - `Q` flag for `__u64` type - 64-bit unsigned, native endian (uint64_t)
+        #
+        # Complete `unpack` would look like `C8 L24 Q4 L6 Q4`, but we are only
+        # interested in `unacked` field at the moment, that's why we only parse
+        # till this field by unpacking with `C8 L5`.
+        #
+        # If you find that it's not giving correct results, then please fall back
+        # to inspect, or update this code to accept unpack sequence. But in the
+        # end unpack is preferable, as it's 12x faster than inspect.
+        #
+        # Tested against:
+        # - Amazon Linux 2 with kernel 4.14 & 5.10
+        # - Ubuntu 20.04 with kernel 5.13
+        #
+        def parse_with_unpack(tcp_info)
+          tcp_info.unpack("C8L5").last
         end
       end
     end

lib/puma/plugin/telemetry/version.rb

@@ -3,7 +3,7 @@
 module Puma
   class Plugin
     module Telemetry
-      VERSION = "1.1.0.alpha"
+      VERSION = "1.1.0.beta"
     end
   end
 end

spec/fixtures/config.rb

@@ -4,6 +4,9 @@
 lowlevel_error_handler { |_err| [500, {}, ["error page"]] }
 
 threads 1, 1
+
+bind "unix://#{ENV["BIND_PATH"]}"
+
 plugin "telemetry"
 
 Target = Struct.new(:name) do

spec/fixtures/default.rb

@@ -4,6 +4,9 @@
 lowlevel_error_handler { |_err| [500, {}, ["error page"]] }
 
 threads 1, 1
+
+bind "unix://#{ENV["BIND_PATH"]}"
+
 plugin "telemetry"
 
 Puma::Plugin::Telemetry.configure do |config|