Implement instrumenters for external HTTP requests

Style/StringLiteralsInInterpolation:

  EnforcedStyle: double_quotes

Style/UseLambda:

  Enabled: false

Layout/MultilineMethodCallIndentation:

  Enabled: No

  # Please maintain alphabetical order for dependencies

  spec.add_runtime_dependency "actionpack", ">= 5.0.0", "< 6.1.0"

  spec.add_runtime_dependency "activesupport", ">= 5.0.0", "< 6.1.0"

  spec.add_runtime_dependency "gitlab-pg_query", "~> 1.3"

  spec.add_runtime_dependency "grpc", "~> 1.19" # Be sure to update the "grpc-tools" dev_depenency too

  spec.add_runtime_dependency "jaeger-client", "~> 1.1"

  spec.add_runtime_dependency "opentracing", "~> 0.4"

  spec.add_runtime_dependency "redis", ">3.0.0", "<5.0.0"

  spec.add_runtime_dependency "gitlab-pg_query", "~> 1.3"

  # Please maintain alphabetical order for dev dependencies

  spec.add_development_dependency "excon", "~> 0.78.1"

  spec.add_development_dependency "faraday", "~> 1.2.0"

  spec.add_development_dependency "grpc-tools", "~> 1.19"

  spec.add_development_dependency "httparty", "~> 0.17.3"

  spec.add_development_dependency "httpclient", "~> 2.8.3"

  spec.add_development_dependency "pry", "~> 0.12"

  spec.add_development_dependency "rack", "~> 2.0"

  spec.add_development_dependency "rake", "~> 12.3"

  spec.add_development_dependency "rest-client", "~> 2.1.0"

  spec.add_development_dependency "rspec", "~> 3.8.0"

  spec.add_development_dependency "rspec-parameterized", "~> 0.4"

  spec.add_development_dependency "rubocop", "~> 0.65.0"

# infrastructural concerns, partcularly related to

# observability.

module Labkit

  autoload :System, "labkit/system"

  autoload :Correlation, "labkit/correlation"

  autoload :Context, "labkit/context"

  autoload :Tracing, "labkit/tracing"

  autoload :Logging, "labkit/logging"

  autoload :Middleware, "labkit/middleware"

  # Publishers to publish notifications whenever a HTTP reqeust is made.

  # A broadcasted notification's payload in topic "request.external_http" includes:

  #   + method (String): "GET"

  #   + code (String): "200" # This is the status code read directly from HTTP response

  #   + duration (Float - seconds): 0.234

  #   + host (String): "gitlab.com"

  #   + port (Integer): 80,

  #   + path (String): "/gitlab-org/gitlab"

  #   + scheme (String): "https"

  #   + query (String): "field_a=1&field_b=2"

  #   + fragment (String): "issue-number-1"

  #   + proxy_host (String - Optional): "proxy.gitlab.com"

  #   + proxy_port (Integer - Optional): 80

  #   + exception (Array<String> - Optional): ["Net::ReadTimeout", "Net::ReadTimeout with #<TCPSocket:(closed)>"]

  #   + exception_object (Error Object - Optional): #<Net::ReadTimeout: Net::ReadTimeout>

  #

  # Usage:

  #

  # ActiveSupport::Notifications.subscribe "request.external_http" do |name, started, finished, unique_id, data|

  #   puts "#{name} | #{started} | #{finished} | #{unique_id} | #{data.inspect}"

  # end

  #

  EXTERNAL_HTTP_NOTIFICATION_TOPIC = "request.external_http"

  autoload :NetHttpPublisher, "labkit/net_http_publisher"

  autoload :ExconPublisher, "labkit/excon_publisher"

  autoload :HTTPClientPublisher, "labkit/httpclient_publisher"

end

# rubocop:enable Naming/FileName

# frozen_string_literal: true

module Labkit

  ##

  # A middleware for Excon HTTP library to publish a notification

  # whenever a HTTP request is triggered.

  #

  # Excon supports a middleware system that allows request/response

  # interception freely. Whenever a new Excon connection is created, a list of

  # default middlewares is injected. This list of middlewares can be altered

  # thanks to Excon.defaults accessor. ExconPublisher is inserted into this

  # list. It affects all connections created in future. There is a limitation

  # that this approach doesn't work if a user decides to override the default

  # middleware list. It is unlikely though, at least in the dependency tree of

  # GitLab.

  #

  # ExconPublisher instance is created once and shared between all Excon

  # connections later. Each connection may be triggered by different threads in

  # parallel. In such cases, a connection objects creates multiple sockets for

  # each thread. Therfore in the implementation of this middleware, the

  # instrumation payload for each connection is stored inside a thread-isolated

  # storage.

  #

  # For more information:

  # https://github.com/excon/excon/blob/81a0130537f2f8cd00d6daafb05d02d9a90dc9f7/lib/excon/middlewares/base.rb

  # https://github.com/excon/excon/blob/fa3ec51e9bb062a12846a1cfff09534e76c99f4b/lib/excon/constants.rb#L146

  # https://github.com/excon/excon/blob/fa3ec51e9bb062a12846a1cfff09534e76c99f4b/lib/excon/connection.rb#L474

  class ExconPublisher

    @prepend_mutex = Mutex.new

    def self.labkit_prepend!

      @prepend_mutex.synchronize do

        return if !defined?(Excon) || @prepended

        defaults = Excon.defaults

        defaults[:middlewares] << ExconPublisher

        @prepended = true

      end

    end

    def initialize(stack)

      @stack = stack

      @instrumenter = ActiveSupport::Notifications.instrumenter

    end

    def request_call(datum)

      payload = start_payload(datum)

      store_connection_payload(datum, payload)

      @instrumenter.start(::Labkit::EXTERNAL_HTTP_NOTIFICATION_TOPIC, payload)

      @stack.request_call(datum)

    end

    def response_call(datum)

      payload = fetch_connection_payload(datum)

      return @stack.response_call(datum) if payload.nil?

      calculate_duration(payload)

      payload[:code] = datum[:response][:status].to_s

      @instrumenter.finish(::Labkit::EXTERNAL_HTTP_NOTIFICATION_TOPIC, payload)

      @stack.response_call(datum)

    ensure

      remove_connection_payload(datum)

    end

    def error_call(datum)

      payload = fetch_connection_payload(datum)

      return @stack.error_call(datum) if payload.nil?

      calculate_duration(payload)

      if datum[:error].is_a?(Exception)

        payload[:exception] = [datum[:error].class.name, datum[:error].message]

        payload[:exception_object] = datum[:error]

      elsif datum[:error].is_a?(String)

        exception = StandardError.new(datum[:error])

        payload[:exception] = [exception.class.name, exception.message]

        payload[:exception_object] = exception

      end

      @instrumenter.finish(::Labkit::EXTERNAL_HTTP_NOTIFICATION_TOPIC, payload)

      @stack.error_call(datum)

    ensure

      remove_connection_payload(datum)

    end

    private

    def start_payload(datum)

      payload = {

        method: datum[:method].to_s.upcase,

        host: datum[:host],

        path: datum[:path],

        port: datum[:port],

        scheme: datum[:scheme],

        query: datum[:query],

        start_time: ::Labkit::System.monotonic_time,

      }

      unless datum[:proxy].nil?

        payload[:proxy_host] = datum[:proxy][:host]

        payload[:proxy_port] = datum[:proxy][:port]

      end

      payload

    end

    def calculate_duration(payload)

      start_time = payload.delete(:start_time) || ::Labkit::System.monotonic_time

      payload[:duration] = (::Labkit::System.monotonic_time - start_time).to_f

    end

    def connection_payload

      Thread.current[:__labkit_http_excon_payload] ||= {}

    end

    def store_connection_payload(datum, payload)

      connection_payload[datum[:connection]] = payload

    end

    def fetch_connection_payload(datum)

      connection_payload.fetch(datum[:connection], nil)

    end

    def remove_connection_payload(datum)

      connection_payload.delete(datum[:connection])

    end

  end

end

# frozen_string_literal: true

module Labkit

  ##

  # Prepend to HTTPClient class to publish an ActiveSupport::Notifcation

  # whenever a HTTP request is triggered.

  #

  # Similar to Net::HTTP, this HTTP client redirects all calls to

  # HTTPClient#do_get_block. HTTPClient is prepended with HTTPClientPublisher.

  # Although HTTPClient supports request filter (a kind of middleware), its

  # support is strictly limited. The request and response passed into the

  # filter don't contain connection information. The response doesn't even

  # contain any link to the request object. It's impossible to fit this filter

  # mechanism into our subscribing model.

  #

  # For more information;

  # https://github.com/nahi/httpclient/blob/d3091b095a1b29f65f4531a70a8e581e75be035e/lib/httpclient.rb#L1233

  module HTTPClientPublisher

    @prepend_mutex = Mutex.new

    def self.labkit_prepend!

      @prepend_mutex.synchronize do

        return if !defined?(HTTPClient) || @prepended

        HTTPClient.prepend(self)

        @prepended = true

      end

    end

    def do_get_block(req, proxy, conn, &block)

      start_time = ::Labkit::System.monotonic_time

      ActiveSupport::Notifications.instrument ::Labkit::EXTERNAL_HTTP_NOTIFICATION_TOPIC, create_request_payload(req, proxy) do |payload|

        response =

          begin

            super

          ensure

            payload[:duration] = (::Labkit::System.monotonic_time - start_time).to_f

          end

        payload[:code] = response.status_code.to_s

        response

      end

    end

    private

    def create_request_payload(request, proxy)

      http_header = request.http_header

      payload = {

        method: http_header.request_method,

        host: http_header.request_uri.host,

        path: http_header.request_uri.path,

        port: http_header.request_uri.port,

        scheme: http_header.request_uri.scheme,

        query: http_header.request_uri.query,

        fragment: http_header.request_uri.fragment,

      }

      unless proxy.nil?

        payload[:proxy_host] = proxy.host

        payload[:proxy_port] = proxy.port

      end

      payload

    end

  end

end