gx

chenyc

2025-06-12 7b72ac13a83764a662159d4a49b7fffb90476ecb

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
'use strict';
const KillCursor = require('../connection/commands').KillCursor;
const GetMore = require('../connection/commands').GetMore;
const calculateDurationInMs = require('../../utils').calculateDurationInMs;
const extractCommand = require('../../command_utils').extractCommand;
 
// helper methods
const namespace = command => command.ns;
const databaseName = command => command.ns.split('.')[0];
const generateConnectionId = pool =>
  pool.options ? `${pool.options.host}:${pool.options.port}` : pool.address;
const isLegacyPool = pool => pool.s && pool.queue;
 
const extractReply = (command, reply) => {
  if (command instanceof GetMore) {
    return {
      ok: 1,
      cursor: {
        id: reply.message.cursorId,
        ns: namespace(command),
        nextBatch: reply.message.documents
      }
    };
  }
 
  if (command instanceof KillCursor) {
    return {
      ok: 1,
      cursorsUnknown: command.cursorIds
    };
  }
 
  // is this a legacy find command?
  if (command.query && typeof command.query.$query !== 'undefined') {
    return {
      ok: 1,
      cursor: {
        id: reply.message.cursorId,
        ns: namespace(command),
        firstBatch: reply.message.documents
      }
    };
  }
 
  return reply && reply.result ? reply.result : reply;
};
 
const extractConnectionDetails = pool => {
  if (isLegacyPool(pool)) {
    return {
      connectionId: generateConnectionId(pool)
    };
  }
 
  // APM in the modern pool is done at the `Connection` level, so we rename it here for
  // readability.
  const connection = pool;
  return {
    address: connection.address,
    connectionId: connection.id
  };
};
 
/** An event indicating the start of a given command */
class CommandStartedEvent {
  /**
   * Create a started event
   *
   * @param {Pool} pool the pool that originated the command
   * @param {Object} command the command
   */
  constructor(pool, command) {
    const extractedCommand = extractCommand(command);
    const commandName = extractedCommand.name;
    const connectionDetails = extractConnectionDetails(pool);
 
    Object.assign(this, connectionDetails, {
      requestId: command.requestId,
      databaseName: databaseName(command),
      commandName,
      command: extractedCommand.shouldRedact ? {} : extractedCommand.cmd
    });
  }
}
 
/** An event indicating the success of a given command */
class CommandSucceededEvent {
  /**
   * Create a succeeded event
   *
   * @param {Pool} pool the pool that originated the command
   * @param {Object} command the command
   * @param {Object} reply the reply for this command from the server
   * @param {Array} started a high resolution tuple timestamp of when the command was first sent, to calculate duration
   */
  constructor(pool, command, reply, started) {
    const extractedCommand = extractCommand(command);
    const commandName = extractedCommand.name;
    const connectionDetails = extractConnectionDetails(pool);
 
    Object.assign(this, connectionDetails, {
      requestId: command.requestId,
      commandName,
      duration: calculateDurationInMs(started),
      reply: extractedCommand.shouldRedact ? {} : extractReply(command, reply)
    });
  }
}
 
/** An event indicating the failure of a given command */
class CommandFailedEvent {
  /**
   * Create a failure event
   *
   * @param {Pool} pool the pool that originated the command
   * @param {Object} command the command
   * @param {MongoError|Object} error the generated error or a server error response
   * @param {Array} started a high resolution tuple timestamp of when the command was first sent, to calculate duration
   */
  constructor(pool, command, error, started) {
    const extractedCommand = extractCommand(command);
    const commandName = extractedCommand.name;
    const connectionDetails = extractConnectionDetails(pool);
 
    Object.assign(this, connectionDetails, {
      requestId: command.requestId,
      commandName,
      duration: calculateDurationInMs(started),
      failure: extractedCommand.shouldRedact ? {} : error
    });
  }
}
 
module.exports = {
  CommandStartedEvent,
  CommandSucceededEvent,
  CommandFailedEvent
};