[CT-708] Indexer track e2e latency (#1237)

* fwd through message times

* use the var i made

* post processing stat emission

* post-forwarding timestamp

* pass through event type from vulcan

* event type to stat emissions

* test fix function calls

* WIP WIP WIP

* fix tests

* unused import

* test that kafka messages are threaded

Author: jonfung-dydx

indexer/packages/v4-protos/src/index.ts

@@ -16,3 +16,4 @@ export * from './codegen/google/protobuf/timestamp';
 export * from './codegen/dydxprotocol/indexer/protocol/v1/clob';
 export * from './codegen/dydxprotocol/indexer/protocol/v1/subaccount';
 export * from './codegen/dydxprotocol/indexer/shared/removal_reason';
+export * from './utils';

indexer/packages/v4-protos/src/utils.ts

@@ -0,0 +1,12 @@
+import { Timestamp } from './codegen/google/protobuf/timestamp';
+
+export const MILLIS_IN_NANOS: number = 1_000_000;
+export const SECONDS_IN_MILLIS: number = 1_000;
+export function protoTimestampToDate(
+  protoTime: Timestamp,
+): Date {
+  const timeInMillis: number = Number(protoTime.seconds) * SECONDS_IN_MILLIS +
+    Math.floor(protoTime.nanos / MILLIS_IN_NANOS);
+
+  return new Date(timeInMillis);
+}

indexer/services/ender/__tests__/helpers/indexer-proto-helpers.ts

@@ -50,6 +50,7 @@ import {
   IndexerOrderId,
   PerpetualMarketCreateEventV1,
   DeleveragingEventV1,
+  protoTimestampToDate,
 } from '@dydxprotocol-indexer/v4-protos';
 import {
   PerpetualMarketType,
@@ -63,7 +64,6 @@ import {
   generatePerpetualMarketMessage,
   generatePerpetualPositionsContents,
 } from '../../src/helpers/kafka-helper';
-import { protoTimestampToDate } from '../../src/lib/helper';
 import { DydxIndexerSubtypes, VulcanMessage } from '../../src/lib/types';
 
 // TX Hash is SHA256, so is of length 64 hexadecimal without the '0x'.

indexer/services/ender/src/lib/helper.ts

@@ -8,7 +8,6 @@ import {
 import {
   IndexerTendermintEvent,
   IndexerTendermintEvent_BlockEvent,
-  Timestamp,
   OrderFillEventV1,
   MarketEventV1,
   SubaccountUpdateEventV1,
@@ -29,10 +28,6 @@ import Big from 'big.js';
 import _ from 'lodash';
 import { DateTime } from 'luxon';
 
-import {
-  MILLIS_IN_NANOS,
-  SECONDS_IN_MILLIS,
-} from '../constants';
 import {
   AnnotatedSubaccountMessage,
   DydxIndexerSubtypes,
@@ -70,15 +65,6 @@ export function convertToSubaccountMessage(
   return subaccountMessage;
 }
 
-export function protoTimestampToDate(
-  protoTime: Timestamp,
-): Date {
-  const timeInMillis: number = Number(protoTime.seconds) * SECONDS_IN_MILLIS +
-    Math.floor(protoTime.nanos / MILLIS_IN_NANOS);
-
-  return new Date(timeInMillis);
-}
-
 export function dateToDateTime(
   protoTime: Date,
 ): DateTime {

indexer/services/socks/src/lib/message-forwarder.ts

@@ -3,6 +3,7 @@ import {
   logger,
   InfoObject,
   safeJsonStringify,
+  STATS_NO_SAMPLING,
 } from '@dydxprotocol-indexer/base';
 import { updateOnMessageFunction } from '@dydxprotocol-indexer/kafka';
 import { KafkaMessage } from 'kafkajs';
@@ -85,9 +86,10 @@ export class MessageForwarder {
   }
 
   public onMessage(topic: string, message: KafkaMessage): void {
+    const start: number = Date.now();
     stats.timing(
       `${config.SERVICE_NAME}.message_time_in_queue`,
-      Date.now() - Number(message.timestamp),
+      start - Number(message.timestamp),
       config.MESSAGE_FORWARDER_STATSD_SAMPLE_RATE,
       {
         topic,
@@ -125,18 +127,31 @@ export class MessageForwarder {
       return;
     }
 
-    const start: number = Date.now();
+    const startForwardMessage: number = Date.now();
     this.forwardMessage(messageToForward);
     const end: number = Date.now();
     stats.timing(
       `${config.SERVICE_NAME}.forward_message`,
-      end - start,
+      end - startForwardMessage,
       config.MESSAGE_FORWARDER_STATSD_SAMPLE_RATE,
       {
         topic,
         channel: String(channel),
       },
     );
+
+    const originalMessageTimestamp = message.headers?.message_received_timestamp;
+    if (originalMessageTimestamp !== undefined) {
+      stats.timing(
+        `${config.SERVICE_NAME}.message_time_since_received`,
+        startForwardMessage - Number(originalMessageTimestamp),
+        STATS_NO_SAMPLING,
+        {
+          topic,
+          event_type: String(message.headers?.event_type),
+        },
+      );
+    }
   }
 
   public forwardMessage(message: MessageToForward): void {

indexer/services/vulcan/__tests__/handlers/order-place-handler.test.ts

@@ -62,6 +62,7 @@ import { expectCanceledOrderStatus, expectOpenOrderIds, handleInitialOrderPlace
 import { expectOffchainUpdateMessage, expectWebsocketOrderbookMessage, expectWebsocketSubaccountMessage } from '../helpers/websocket-helpers';
 import { OrderbookSide } from '../../src/lib/types';
 import { getOrderIdHash, isLongTermOrder, isStatefulOrder } from '@dydxprotocol-indexer/v4-proto-parser';
+import { defaultKafkaHeaders } from '../helpers/constants';
 import config from '../../src/config';
 
 jest.mock('@dydxprotocol-indexer/base', () => ({
@@ -196,6 +197,12 @@ describe('order-place-handler', () => {
     const replacementMessageIoc: KafkaMessage = createKafkaMessage(
       Buffer.from(Uint8Array.from(OffChainUpdateV1.encode(replacementUpdateIoc).finish())),
     );
+    [replacementMessage, replacementMessageGoodTilBlockTime, replacementMessageConditional,
+      replacementMessageFok, replacementMessageIoc].forEach((message) => {
+      // eslint-disable-next-line no-param-reassign
+      message.headers = defaultKafkaHeaders;
+    });
+
     const dbDefaultOrder: OrderFromDatabase = {
       ...testConstants.defaultOrder,
       id: testConstants.defaultOrderId,
@@ -1225,7 +1232,11 @@ function expectWebsocketMessagesSent(
       version: SUBACCOUNTS_WEBSOCKET_MESSAGE_VERSION,
     });
 
-    expectWebsocketSubaccountMessage(producerSendSpy.mock.calls[callIndex][0], subaccountMessage);
+    expectWebsocketSubaccountMessage(
+      producerSendSpy.mock.calls[callIndex][0],
+      subaccountMessage,
+      defaultKafkaHeaders,
+    );
     callIndex += 1;
   }