Module: Sixty::Instrument::Mongo

Defined in:

lib/sixty/instrument/mongo.rb

Overview

MongoDB.

The third data model this agent measures and the first that is not SQL. Every signal downstream — documents per call, calls per invocation, payload size, error rate — is defined on span attributes rather than on statements, so the feed, the detector and the MCP server need no change. What has to be rebuilt is everything below those attributes: identity, which has no statement to normalize, and the document count, which in a document database arrives in a different shape for every command.

── Why command monitoring here, and not in the JavaScript agent ──────────

@sixty-sh/node deliberately refuses the driver's command events and patches Collection.prototype instead, because in Node a command event fires from the driver's own plumbing after the caller's async context is gone — every operation would be measured correctly and parented to nothing, and attribution is the whole product.

The Ruby driver is synchronous. started and succeeded are published on the calling thread, inside the caller's own stack, so the current span is still the method that issued the query and the parenting is exact. That makes the official monitoring API strictly better here than monkey-patching would be: it is public, versioned, and covers Mongoid, the driver's own cursor round trips and commands issued through paths a collection patch would never see.

── One cursor is one operation, however many round trips it took ────────

A find returning thirty thousand documents at the default batch size is not one exchange with the server: it is the initial command plus roughly three hundred getMore calls, each a network wait the application sits through. The naive reading of command monitoring records three hundred and one operations, and it is wrong in a way that misdirects the reader — the calling method appears to make three hundred database calls, which is the signature of an N+1 the code does not contain, and the fix a fanout finding suggests ("stop looping") is not the fix this needs ("set batchSize").

So a cursor is held open: the find span stays alive until the cursor is exhausted, counting documents and round trips as its getMores arrive, and is emitted once. db_calls then says one call, rows says thirty thousand, and round_trips — the signal the collector added for exactly this — says three hundred. That also matches what the JavaScript agent reports for the same query, which matters: the same regression in the same database should not look like two different findings depending on which language the service happens to be written in.

── No query plans ────────────────────────────────────────────────────────

explain is the highest-value thing this adapter could add — COLLSCAN versus IXSCAN names a cause rather than a symptom. It is absent because Mongo can only explain a filter that still has its values in it, and this file drops values at the moment it sees them. Capturing a plan would mean retaining somebody's query values in order to compose a command out of them. That trade is refused here exactly as it is refused for MySQL.

Defined Under Namespace

Classes: CommandFailed, Subscriber

Constant Summary

IGNORED_COMMANDS =

Commands the driver issues about itself. A handshake is not an operation anybody can act on, and heartbeats would otherwise be the highest-count "query" in every service.

%w[
  ismaster isMaster hello ping buildInfo getnonce authenticate saslStart
  saslContinue logout endSessions getLog hostInfo listDatabases
  connectionStatus getParameter
].freeze

CURSOR_COMMANDS =

Commands that can hand back a cursor rather than an answer.

%w[find aggregate listIndexes listCollections].freeze

ENVELOPE_KEYS =

Keys the driver adds to every command. They describe the session and the topology rather than the query, and including them would make the identity of every operation move with the driver's version.

cursor, batchSize and singleBatch describe how the results are fetched rather than what was asked for, and they are excluded for a sharper reason than tidiness: setting or removing a batch size is precisely the change the round_trips signal exists to report, and an identity that moved with it would re-identify the operation at the moment of the change. The detector would then have nothing to compare — a new operation with no history beside an old one that stopped reporting — and the finding could never fire.

limit is not in this list, and the difference is the point: a limit changes what you asked for.

%w[
  $db lsid txnNumber $clusterTime $readPreference $audit apiVersion
  apiStrict apiDeprecationErrors signature startTransaction autocommit
  readConcern writeConcern comment cursor batchSize singleBatch
].freeze

MAX_IN_FLIGHT =

A command that never completes would otherwise leave its start context behind forever. The cap is per thread and far above any real number of commands in flight on one connection at one time.

64

MAX_OPEN_CURSORS =

And a cursor that is opened and never drained would leave a span open. Past this many, the oldest is emitted with what it has — an incomplete measurement rather than an unbounded one.

128

MAX_CURSOR_AGE_SECONDS =

The same protection in time rather than in count. The Ruby driver kills an abandoned cursor from a finalizer, so breaking out of a loop over a cursor releases nothing until the garbage collector gets to it — which may be never in a process that is not under memory pressure. A cursor nobody has read from in this long is reported with what it has.

60

KEY =

:sixty_mongo_in_flight

Class Attribute Summary

• .config ⇒ Object readonly

  Returns the value of attribute config.

• .subscriber ⇒ Object readonly

  Returns the value of attribute subscriber.

Class Method Summary

• .batch_length(event) ⇒ Object
• .bytes_from(event) ⇒ Object

  Sampled, because serializing a whole reply to measure it would cost more than the query.

• .close_entry(entry, error) ⇒ Object
• .collection_for(command_name, command, event) ⇒ Object

  What the command ran against.

• .continue_cursor(cursor_id, event, error) ⇒ Object
• .cursor_id_from_reply(event) ⇒ Object
• .cursor_id_of(value) ⇒ Object

  BSON::Int64 and Integer both appear here depending on driver version and platform, and they are not eql? — so a map keyed by the raw value would miss on lookup and every cursor would look like a new one.

• .cursor_mutex ⇒ Object
• .cursors ⇒ Object

  Cursors whose spans are still open, keyed by the server's cursor id.

• .describe(event) ⇒ Object

  Everything about a command that has to be read before it runs.

• .exhausted?(event) ⇒ Boolean
• .identity_for(command_name, collection, command) ⇒ Object

  The identity of the operation: the command, what it ran against, and the structure of its arguments.

• .install(config = nil) ⇒ Object
• .installed? ⇒ Boolean
• .kill_cursors(ids) ⇒ Object

  An abandoned cursor: the application stopped reading and the driver told the server so.

• .number(value) ⇒ Object
• .open_cursor(cursor_id, state, event) ⇒ Object

  ---------------------------------------------------------------- state.

• .reply_of(event) ⇒ Object

  A failed command has no reply at all — CommandFailed carries a message where CommandSucceeded carries the server's answer.

• .reset! ⇒ Object

  Exported for tests.

• .rows_from(event, command_name) ⇒ Object

  Documents returned for a read, documents affected for a write — the same meaning rows has for every SQL client in this directory.

• .sweep(max_age: MAX_CURSOR_AGE_SECONDS) ⇒ Object

  Cursors nobody is reading any more, reported rather than held.

Class Attribute Details

.config ⇒ Object (readonly)

Returns the value of attribute config.

# File 'lib/sixty/instrument/mongo.rb', line 115

def config
  @config
end

.subscriber ⇒ Object (readonly)

Returns the value of attribute subscriber.

# File 'lib/sixty/instrument/mongo.rb', line 115

def subscriber
  @subscriber
end

Class Method Details

.batch_length(event) ⇒ Object

# File 'lib/sixty/instrument/mongo.rb', line 313

def batch_length(event)
  cursor = reply_of(event)&.[]('cursor')
  return nil unless cursor.is_a?(Hash)

  batch = cursor['firstBatch'] || cursor['nextBatch']
  batch.is_a?(Array) ? batch.length : nil
end

.bytes_from(event) ⇒ Object

Sampled, because serializing a whole reply to measure it would cost more than the query. Precision does not matter: this exists to catch a payload going from 4KB to 2MB.

# File 'lib/sixty/instrument/mongo.rb', line 357

def bytes_from(event)
  cursor = reply_of(event)&.[]('cursor')
  batch = cursor.is_a?(Hash) ? (cursor['firstBatch'] || cursor['nextBatch']) : nil
  return nil unless batch.is_a?(Array) && !batch.empty?

  # Three documents rather than five: serializing a BSON document to
  # measure it is the most expensive thing this file does per query, and
  # the number only has to be right to an order of magnitude.
  sample = batch.first(3)
  sampled = sample.sum { |doc| doc.to_bson.length }
  ((sampled.to_f / sample.length) * batch.length).round
rescue StandardError
  nil
end

.close_entry(entry, error) ⇒ Object

# File 'lib/sixty/instrument/mongo.rb', line 221

def close_entry(entry, error)
  span = entry[:span]
  span.attrs[:rows] = entry[:rows]
  span.attrs[:round_trips] = entry[:round_trips]
  span.attrs[:bytes] = entry[:bytes] if entry[:bytes].positive?
  Tracer.end_span(span, error)
  Tracer.emit(span)
rescue StandardError
  nil
end

.collection_for(command_name, command, event) ⇒ Object

What the command ran against. For most commands the collection is the value of the command key itself ({find: 'orders', ...}); getMore names a cursor there and carries the collection separately.

# File 'lib/sixty/instrument/mongo.rb', line 272

def collection_for(command_name, command, event)
  value = command_name == 'getMore' ? command['collection'] : command[command_name]
  return value if value.is_a?(String) && !value.empty?

  database = event.database_name.to_s
  database.empty? ? 'collection' : database
end

.continue_cursor(cursor_id, event, error) ⇒ Object

# File 'lib/sixty/instrument/mongo.rb', line 184

def continue_cursor(cursor_id, event, error)
  entry = cursor_mutex.synchronize { cursors[cursor_id] }
  return false unless entry

  entry[:round_trips] += 1
  entry[:rows] += batch_length(event) || 0
  entry[:bytes] += bytes_from(event) || 0
  entry[:touched_at] = Tracer.monotonic_ms

  if error || exhausted?(event)
    cursor_mutex.synchronize { cursors.delete(cursor_id) }
    close_entry(entry, error)
  end
  true
end

.cursor_id_from_reply(event) ⇒ Object

# File 'lib/sixty/instrument/mongo.rb', line 321

def cursor_id_from_reply(event)
  cursor = reply_of(event)&.[]('cursor')
  cursor.is_a?(Hash) ? cursor_id_of(cursor['id']) : nil
end

.cursor_id_of(value) ⇒ Object

BSON::Int64 and Integer both appear here depending on driver version and platform, and they are not eql? — so a map keyed by the raw value would miss on lookup and every cursor would look like a new one.

# File 'lib/sixty/instrument/mongo.rb', line 334

def cursor_id_of(value)
  return nil if value.nil?

  value.respond_to?(:value) ? value.value.to_i : value.to_i
rescue StandardError
  nil
end

.cursor_mutex ⇒ Object

# File 'lib/sixty/instrument/mongo.rb', line 151

def cursor_mutex
  @cursor_mutex ||= Mutex.new
end

.cursors ⇒ Object

Cursors whose spans are still open, keyed by the server's cursor id.

Process-wide rather than thread-local: a cursor may be handed to another thread to drain, and a span that could only be closed by the thread that opened it would leak in exactly that case.

# File 'lib/sixty/instrument/mongo.rb', line 147

def cursors
  @cursors ||= {}
end

.describe(event) ⇒ Object

Everything about a command that has to be read before it runs.

# File 'lib/sixty/instrument/mongo.rb', line 235

def describe(event)
  command = event.command || {}
  command_name = event.command_name.to_s
  collection = collection_for(command_name, command, event)
  identity = identity_for(command_name, collection, command)

  {
    command_name: command_name,
    name: "#{command_name}:#{collection}",
    identity: identity,
    frames: Stack.capture(identity),
    cursor_id: command_name == 'getMore' ? cursor_id_of(command['getMore']) : nil
  }
end

.exhausted?(event) ⇒ Boolean

Returns:

• (Boolean)

# File 'lib/sixty/instrument/mongo.rb', line 326

def exhausted?(event)
  id = cursor_id_from_reply(event)
  id.nil? || id.zero?
end

.identity_for(command_name, collection, command) ⇒ Object

The identity of the operation: the command, what it ran against, and the structure of its arguments. See Sixty::Shape — no value in the command has a path into this string.

# File 'lib/sixty/instrument/mongo.rb', line 253

def identity_for(command_name, collection, command)
  parts = []
  # A pipeline is ordered structure, so every stage counts and in order.
  pipeline = command['pipeline'] || command[:pipeline]
  parts << Shape.shape_of_sequence(pipeline) if pipeline.is_a?(Array)

  rest = command.reject do |key, _value|
    name = key.to_s
    ENVELOPE_KEYS.include?(name) || name == command_name || name == 'pipeline'
  end
  shape = Shape.shape_of(rest)
  parts << "{#{shape}}" unless shape.empty?

  "#{command_name} #{collection} #{parts.join(' ')}".strip
end

.install(config = nil) ⇒ Object

# File 'lib/sixty/instrument/mongo.rb', line 117

def install(config = nil)
  return false if @installed
  return false unless defined?(::Mongo::Monitoring::Global)

  @config = config
  @subscriber = Subscriber.new
  # Global, so clients created later are covered. Clients that already
  # exist — Mongoid builds its own during boot, which may be before
  # this runs — are subscribed to individually below.
  ::Mongo::Monitoring::Global.subscribe(::Mongo::Monitoring::COMMAND, @subscriber)
  subscribe_existing_clients
  Sixty.before_flush { sweep }
  @installed = true
end

.installed? ⇒ Boolean

Returns:

• (Boolean)

# File 'lib/sixty/instrument/mongo.rb', line 132

def installed?
  @installed == true
end

.kill_cursors(ids) ⇒ Object

An abandoned cursor: the application stopped reading and the driver told the server so. The measurement ends where the reading did.

# File 'lib/sixty/instrument/mongo.rb', line 214

def kill_cursors(ids)
  ids.each do |id|
    entry = cursor_mutex.synchronize { cursors.delete(id) }
    close_entry(entry, nil) if entry
  end
end

.number(value) ⇒ Object

# File 'lib/sixty/instrument/mongo.rb', line 372

def number(value)
  value.is_a?(Numeric) ? value.to_i : nil
end

.open_cursor(cursor_id, state, event) ⇒ Object

---------------------------------------------------------------- state

# File 'lib/sixty/instrument/mongo.rb', line 157

def open_cursor(cursor_id, state, event)
  span = Tracer.start_span(
    kind: Tracer::KIND_DB,
    name: state[:name],
    attrs: { normalized_sql: state[:identity] }
  )
  span.attrs[:frames] = state[:frames] if state[:frames]
  # Back-dated by the command's own duration, so the span covers the
  # round trip that opened the cursor as well as the ones that follow.
  span.start -= event.duration.to_f * 1000.0

  entry = {
    span: span,
    rows: batch_length(event) || 0,
    round_trips: 1,
    bytes: bytes_from(event) || 0,
    touched_at: Tracer.monotonic_ms
  }
  evicted = nil
  cursor_mutex.synchronize do
    cursors[cursor_id] = entry
    evicted = cursors.shift if cursors.size > MAX_OPEN_CURSORS
  end
  close_entry(evicted[1], nil) if evicted
  entry
end

.reply_of(event) ⇒ Object

A failed command has no reply at all — CommandFailed carries a message where CommandSucceeded carries the server's answer. Reading it unguarded is how the first version of this file dropped every failed query: the error reached the application correctly and the operation that produced it was never recorded.

# File 'lib/sixty/instrument/mongo.rb', line 347

def reply_of(event)
  return nil unless event.respond_to?(:reply)

  reply = event.reply
  reply.is_a?(Hash) ? reply : nil
end

.reset! ⇒ Object

Exported for tests.

# File 'lib/sixty/instrument/mongo.rb', line 137

def reset!
  @installed = false
  @cursors = nil
end

.rows_from(event, command_name) ⇒ Object

Documents returned for a read, documents affected for a write — the same meaning rows has for every SQL client in this directory. A rows drift that meant one thing on Postgres and another on Mongo would be a chart nobody can read.

# File 'lib/sixty/instrument/mongo.rb', line 286

def rows_from(event, command_name)
  batch = batch_length(event)
  return batch if batch

  reply = reply_of(event)
  return nil unless reply

  case command_name
  when 'count', 'countDocuments'
    # A count returns one number, so one document came back. Reporting
    # the count itself would say that counting a million-document
    # collection returned a million documents, and the first collection
    # to grow would look like the regression this product exists to
    # report.
    1
  when 'distinct'
    values = reply['values']
    values.is_a?(Array) ? values.length : 1
  when 'findAndModify'
    reply['value'].nil? ? 0 : 1
  when 'update'
    number(reply['nModified']) || number(reply['n'])
  else
    number(reply['n'])
  end
end

.sweep(max_age: MAX_CURSOR_AGE_SECONDS) ⇒ Object

Cursors nobody is reading any more, reported rather than held. Runs on the agent's flush thread, like every other piece of housekeeping here.

# File 'lib/sixty/instrument/mongo.rb', line 202

def sweep(max_age: MAX_CURSOR_AGE_SECONDS)
  deadline = Tracer.monotonic_ms - (max_age * 1000.0)
  stale = cursor_mutex.synchronize do
    cursors.select { |_id, entry| entry[:touched_at] < deadline }
           .each_key { |id| cursors.delete(id) }
  end
  stale.each_value { |entry| close_entry(entry, nil) }
  stale.size
end