Class: WideEvent::Store::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/wide_event/store/client.rb

Overview

Minimal standard-library HTTPS client for POST /v1/events (store/internal/httpapi/server.go) and POST /v1/query (store/internal/httpapi/query.go). One instance issues one request at a time. #ingest is called only from WideEvent::Store::Sender's worker thread; #query is called from the wide_events:sql Rails task on whatever thread invokes it (there is no persistent worker for queries).

Defined Under Namespace

Classes: Ack, Error

Constant Summary collapse

OPEN_TIMEOUT = 
1
READ_TIMEOUT = 
2
MAX_RESPONSE_BYTES = 
64 * 1024
LOOPBACK_HOSTS = 
%w[localhost 127.0.0.1 ::1].freeze
INGEST_PATH = 
"/v1/events"
QUERY_PATH = 
"/v1/query"
QUERY_PROTOCOL_VERSION = 
1
QUERY_RESPONSE_CAP =

The store caps its own query responses at 10 MiB (store/internal/query/executor.go's defaultMaxResponseBytes); we read one byte past that so a response that actually reaches the cap+1 byte is distinguishable from one that legitimately ends at (or under) 10 MiB, instead of silently parsing a truncated body.

10 * 1024 * 1024
DEFAULT_QUERY_DEADLINE = 
30
QUERY_FORMATS = 
%w[json csv].freeze
RETRY_AFTER_DEFAULT = 
1
RETRY_AFTER_CAP = 
10
NETWORK_ERRORS = 
[
  Net::OpenTimeout, Net::ReadTimeout, Errno::ECONNREFUSED, Errno::ECONNRESET,
  Errno::EHOSTUNREACH, Errno::ETIMEDOUT, Errno::EPIPE, SocketError,
  OpenSSL::SSL::SSLError, EOFError, IOError
].freeze

Instance Method Summary collapse

Constructor Details

#initialize(url:, ingest_token: nil, query_token: nil, open_timeout: OPEN_TIMEOUT, read_timeout: READ_TIMEOUT) ⇒ Client

Token presence is validated per-operation (in #ingest and #query), not here: a Client built only to query has no ingest token to give it, and vice versa. The constructor still validates the URL, since that's shared by both operations and needed before either can run.

Raises:

  • (ArgumentError) 


61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/wide_event/store/client.rb', line 61

def initialize(url:, ingest_token: nil, query_token: nil, open_timeout: OPEN_TIMEOUT, read_timeout: READ_TIMEOUT)
  @uri = URI.parse(url.to_s)
  raise ArgumentError, "store url must be an absolute http(s) URL" unless @uri.host

  unless https?(@uri) || loopback?(@uri)
    raise ArgumentError, "store url must use HTTPS (plain http is only allowed for loopback testing)"
  end

  @ingest_token = ingest_token.to_s
  @query_token = query_token.to_s
  @open_timeout = open_timeout
  @read_timeout = read_timeout
end

Instance Method Details

#ingest(gzip_body) ⇒ Object

Sends one already-gzipped canonical batch body. Returns an Ack on a 200 response; raises Error (tagged retryable?) otherwise.

Raises:

  • (ArgumentError) 


77
78
79
80
81
 
# File 'lib/wide_event/store/client.rb', line 77

def ingest(gzip_body)
  raise ArgumentError, "store ingest token is required" if @ingest_token.empty?
  code, body = perform(gzip_body)
  handle_response(code, body)
end

#query(sql, format: :json, deadline: DEFAULT_QUERY_DEADLINE) ⇒ Object

Runs one read-only statement. Retries a busy (429) response, bounded by the server's Retry-After header, until deadline monotonic seconds have elapsed overall; policy/limit errors (400/422/503) are never retried here. Returns a QueryResult for format: :json, or the server's CSV text verbatim for format: :csv.

Raises:

  • (ArgumentError) 


88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/wide_event/store/client.rb', line 88

def query(sql, format: :json, deadline: DEFAULT_QUERY_DEADLINE)
  raise ArgumentError, "store query token is required" if @query_token.empty?
  format = format.to_s
  raise ArgumentError, "query format must be :json or :csv" unless QUERY_FORMATS.include?(format)

  deadline_at = monotonic_now + deadline
  loop do
    code, body, retry_after = perform_query(sql, format)
    if code == 429
      sleep(bounded_retry_wait(retry_after, deadline_at))
      next
    end
    return handle_query_response(code, body, format)
  end
end