Merge pull request #19455 from overleaf/jpa-metrics

[misc] add metrics for document processing/broadcasting

GitOrigin-RevId: d81de0dfb7a91863547631580f3c85f569718130

services/real-time/app/js/DocumentUpdaterController.js

@@ -105,6 +105,11 @@ module.exports = DocumentUpdaterController = {
     if (clientList.length === 0) {
       return
     }
+
+    update.meta = update.meta || {}
+    const { tsRT: realTimeIngestionTime } = update.meta
+    delete update.meta.tsRT
+
     // send updates to clients
     logger.debug(
       {
@@ -129,6 +134,15 @@ module.exports = DocumentUpdaterController = {
             },
             'distributing update to sender'
           )
+          metrics.histogram(
+            'update-processing-time',
+            performance.now() - realTimeIngestionTime,
+            [
+              0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 20, 50, 100, 200, 500, 1000,
+              2000, 5000, 10000,
+            ],
+            { path: 'sharejs' }
+          )
           client.emit('otUpdateApplied', { v: update.v, doc: update.doc })
         } else if (!update.dup) {
           // Duplicate ops should just be sent back to sending client for acknowledgement

services/real-time/app/js/DocumentUpdaterManager.js

@@ -42,7 +42,16 @@ const DocumentUpdaterManager = {
           return callback(error)
         }
         body = body || {}
-        callback(null, body.lines, body.version, body.ranges, body.ops)
+        callback(
+          null,
+          body.lines,
+          body.version,
+          body.ranges,
+          body.ops,
+          body.ttlInS
+        )
+      } else if (res.statusCode === 422 && body?.firstVersionInRedis) {
+        callback(new ClientRequestedMissingOpsError(422, body))
       } else if ([404, 422].includes(res.statusCode)) {
         callback(new ClientRequestedMissingOpsError(res.statusCode))
       } else {

services/real-time/app/js/Errors.js

@@ -1,9 +1,10 @@
 const OError = require('@overleaf/o-error')
 
 class ClientRequestedMissingOpsError extends OError {
-  constructor(statusCode) {
+  constructor(statusCode, info = {}) {
     super('doc updater could not load requested ops', {
       statusCode,
+      ...info,
     })
   }
 }

services/real-time/app/js/WebsocketController.js

@@ -11,8 +11,28 @@ const {
   JoinLeaveEpochMismatchError,
   NotAuthorizedError,
   NotJoinedError,
+  ClientRequestedMissingOpsError,
 } = require('./Errors')
 
+const JOIN_DOC_CATCH_UP_LENGTH_BUCKETS = [
+  0, 5, 10, 25, 50, 100, 150, 200, 250, 500, 1000,
+]
+const JOIN_DOC_CATCH_UP_AGE = [
+  0,
+  1,
+  2,
+  5,
+  10,
+  20,
+  30,
+  60,
+  120,
+  240,
+  600,
+  60 * 60,
+  24 * 60 * 60,
+].map(x => x * 1000)
+
 let WebsocketController
 module.exports = WebsocketController = {
   // If the protocol version changes when the client reconnects,
@@ -195,6 +215,38 @@ module.exports = WebsocketController = {
       'client joining doc'
     )
 
+    const emitJoinDocCatchUpMetrics = (
+      status,
+      { firstVersionInRedis, version, ttlInS }
+    ) => {
+      if (fromVersion === -1) return // full joinDoc call
+      if (typeof options.age !== 'number') return // old frontend
+      if (!ttlInS) return // old document-updater pod
+
+      const isStale = options.age > ttlInS * 1000
+      const method = isStale ? 'stale' : 'recent'
+      metrics.histogram(
+        'join-doc-catch-up-length',
+        version - fromVersion,
+        JOIN_DOC_CATCH_UP_LENGTH_BUCKETS,
+        { status, method, path: client.transport }
+      )
+      if (firstVersionInRedis) {
+        metrics.histogram(
+          'join-doc-catch-up-length-extra-needed',
+          firstVersionInRedis - fromVersion,
+          JOIN_DOC_CATCH_UP_LENGTH_BUCKETS,
+          { status, method, path: client.transport }
+        )
+      }
+      metrics.histogram(
+        'join-doc-catch-up-age',
+        options.age,
+        JOIN_DOC_CATCH_UP_AGE,
+        { status, path: client.transport }
+      )
+    }
+
     WebsocketController._assertClientAuthorization(
       client,
       docId,
@@ -231,10 +283,14 @@ module.exports = WebsocketController = {
             projectId,
             docId,
             fromVersion,
-            function (error, lines, version, ranges, ops) {
+            function (error, lines, version, ranges, ops, ttlInS) {
               if (error) {
+                if (error instanceof ClientRequestedMissingOpsError) {
+                  emitJoinDocCatchUpMetrics('missing', error.info)
+                }
                 return callback(error)
               }
+              emitJoinDocCatchUpMetrics('success', { version, ttlInS })
               if (client.disconnected) {
                 metrics.inc('editor.join-doc.disconnected', 1, {
                   status: 'after-doc-updater-call',
@@ -503,6 +559,7 @@ module.exports = WebsocketController = {
         }
         update.meta.source = client.publicId
         update.meta.user_id = userId
+        update.meta.tsRT = performance.now()
         metrics.inc('editor.doc-update', 0.3, { status: client.transport })
 
         logger.debug(

services/real-time/app/js/WebsocketLoadBalancer.js

@@ -1,5 +1,6 @@
 const Settings = require('@overleaf/settings')
 const logger = require('@overleaf/logger')
+const Metrics = require('@overleaf/metrics')
 const RedisClientManager = require('./RedisClientManager')
 const SafeJsonParse = require('./SafeJsonParse')
 const EventLogger = require('./EventLogger')
@@ -24,6 +25,11 @@ const RESTRICTED_USER_MESSAGE_TYPE_PASS_LIST = [
   'toggle-track-changes',
   'projectRenamedOrDeletedByExternalSource',
 ]
+const BANDWIDTH_BUCKETS = [0]
+// 64 bytes ... 8MB
+for (let i = 5; i <= 22; i++) {
+  BANDWIDTH_BUCKETS.push(2 << i)
+}
 
 let WebsocketLoadBalancer
 module.exports = WebsocketLoadBalancer = {
@@ -139,6 +145,31 @@ module.exports = WebsocketLoadBalancer = {
         for (const client of clientList) {
           ConnectedUsersManager.refreshClient(message.room_id, client.publicId)
         }
+      } else if (message.message === 'canary-applied-op') {
+        const { ack, broadcast, source, projectId, docId } = message.payload
+
+        const estimateBandwidth = (room, path) => {
+          const seen = new Set()
+          for (const client of io.sockets.clients(room)) {
+            if (seen.has(client.id)) continue
+            seen.add(client.id)
+            let v = client.id === source ? ack : broadcast
+            if (v === 0) {
+              // Acknowledgements with update.dup===true will not get sent to other clients.
+              continue
+            }
+            v += `5:::{"name":"otUpdateApplied","args":[]}`.length
+            Metrics.histogram(
+              'estimated-applied-ops-bandwidth',
+              v,
+              BANDWIDTH_BUCKETS,
+              { path }
+            )
+          }
+        }
+
+        estimateBandwidth(projectId, 'per-project')
+        estimateBandwidth(docId, 'per-doc')
       } else if (message.room_id) {
         if (message._id && Settings.checkEventOrder) {
           const status = EventLogger.checkEventOrder(

services/real-time/test/acceptance/js/ApplyUpdateTests.js

@@ -129,7 +129,7 @@ describe('applyOtUpdate', function () {
           let [update] = Array.from(rest[0])
           update = JSON.parse(update)
           update.op.should.deep.equal(this.update.op)
-          update.meta.should.deep.equal({
+          update.meta.should.include({
             source: this.client.publicId,
             user_id: this.user_id,
           })
@@ -408,7 +408,7 @@ describe('applyOtUpdate', function () {
           let [update] = Array.from(rest[0])
           update = JSON.parse(update)
           update.op.should.deep.equal(this.comment_update.op)
-          update.meta.should.deep.equal({
+          update.meta.should.include({
             source: this.client.publicId,
             user_id: this.user_id,
           })

services/real-time/test/unit/js/DocumentUpdaterControllerTests.js

@@ -55,7 +55,10 @@ describe('DocumentUpdaterController', function () {
         }),
         './EventLogger': (this.EventLogger = { checkEventOrder: sinon.stub() }),
         './HealthCheckManager': { check: sinon.stub() },
-        '@overleaf/metrics': (this.metrics = { inc: sinon.stub() }),
+        '@overleaf/metrics': (this.metrics = {
+          inc: sinon.stub(),
+          histogram: sinon.stub(),
+        }),
         './RoomManager': (this.RoomManager = {
           eventSource: sinon.stub().returns(this.RoomEvents),
         }),