pass initial kafka timestamp from Ender -> Vulcan for better e2e orde…

…r latency tracking (#1211)

indexer/packages/kafka/__tests__/batch-kafka-producer.test.ts

@@ -1,16 +1,18 @@
 import { KafkaTopics } from '../src';
 import { BatchKafkaProducer, ProducerMessage } from '../src/batch-kafka-producer';
 import { producer } from '../src/producer';
+import { IHeaders } from 'kafkajs';
 import _ from 'lodash';
 
 interface TestMessage {
   key?: string,
   value: string,
+  headers?: IHeaders,
 }
 
 function testMessage2ProducerMessage(data: TestMessage): ProducerMessage {
   const key: Buffer | undefined = data.key === undefined ? undefined : Buffer.from(data.key);
-  return { key, value: Buffer.from(data.value) };
+  return { key, value: Buffer.from(data.value), headers: data.headers };
 }
 
 function testMessage2ProducerMessages(data: TestMessage[]): ProducerMessage[] {
@@ -35,9 +37,9 @@ describe('batch-kafka-producer', () => {
     [
       'will send key if key is not undefined',
       5,
-      [{ key: '1', value: 'a' }, { key: '2', value: 'b' }, { key: '3', value: 'c' }],
+      [{ key: '1', value: 'a' }, { key: '2', value: 'b' }, { key: '3', value: 'c', headers: { timestamp: 'value' } }],
       [[{ key: '1', value: 'a' }, { key: '2', value: 'b' }]],
-      [{ key: '3', value: 'c' }],
+      [{ key: '3', value: 'c', headers: { timestamp: 'value' } }],
     ],
     [
       'will not send message until the batch size is reached',
@@ -104,7 +106,9 @@ describe('batch-kafka-producer', () => {
 
     for (const msg of messages) {
       const key: Buffer | undefined = msg.key === undefined ? undefined : Buffer.from(msg.key);
-      batchProducer.addMessageAndMaybeFlush({ value: Buffer.from(msg.value), key });
+      batchProducer.addMessageAndMaybeFlush(
+        { value: Buffer.from(msg.value), key, headers: msg.headers },
+      );
     }
 
     expect(producerSendMock.mock.calls).toHaveLength(expectedMessagesPerCall.length);

indexer/packages/kafka/src/batch-kafka-producer.ts

@@ -1,5 +1,5 @@
 import { logger } from '@dydxprotocol-indexer/base';
-import { Producer, RecordMetadata } from 'kafkajs';
+import { IHeaders, Producer, RecordMetadata } from 'kafkajs';
 import _ from 'lodash';
 
 import { KafkaTopics } from './types';
@@ -10,6 +10,7 @@ import { KafkaTopics } from './types';
 export type ProducerMessage = {
   key?: Buffer,
   value: Buffer,
+  headers?: IHeaders,
 };
 
 /**
@@ -52,7 +53,7 @@ export class BatchKafkaProducer {
     if (this.currentSize + msgBuffer.byteLength + keyByteLength > this.maxBatchSizeBytes) {
       this.sendBatch();
     }
-    this.producerMessages.push({ key: message.key, value: msgBuffer });
+    this.producerMessages.push({ key: message.key, value: msgBuffer, headers: message.headers });
     this.currentSize += msgBuffer.byteLength;
     this.currentSize += keyByteLength;
   }

indexer/services/bazooka/src/vulcan-helpers.ts

@@ -22,6 +22,7 @@ import {
 } from '@dydxprotocol-indexer/v4-protos';
 import { Long } from '@dydxprotocol-indexer/v4-protos/build/codegen/helpers';
 import Big from 'big.js';
+import { IHeaders } from 'kafkajs';
 import _ from 'lodash';
 
 import config from './config';
@@ -30,6 +31,7 @@ import { ZERO } from './constants';
 interface VulcanMessage {
   key: Buffer,
   value: OffChainUpdateV1,
+  headers?: IHeaders,
 }
 
 type IndexerOrderIdMap = { [orderUuid: string]: IndexerOrderId };
@@ -129,6 +131,7 @@ export async function sendStatefulOrderMessages() {
       return {
         key: message.key,
         value: Buffer.from(Uint8Array.from(OffChainUpdateV1.encode(message.value).finish())),
+        headers: message.headers,
       };
     });
 

indexer/services/ender/__tests__/handlers/stateful-order/conditional-order-triggered-handler.test.ts

@@ -143,6 +143,7 @@ describe('conditionalOrderTriggeredHandler', () => {
       producerSendMock,
       orderId: conditionalOrderId,
       offchainUpdate: expectedOffchainUpdate,
+      headers: { message_received_timestamp: kafkaMessage.timestamp, event_type: 'ConditionalOrderTriggered' },
     });
   });
 

indexer/services/ender/__tests__/handlers/stateful-order/stateful-order-placement-handler.test.ts

@@ -180,6 +180,7 @@ describe('statefulOrderPlacementHandler', () => {
       producerSendMock,
       orderId: defaultOrder.orderId!,
       offchainUpdate: expectedOffchainUpdate,
+      headers: { message_received_timestamp: kafkaMessage.timestamp, event_type: 'StatefulOrderPlacement' },
     });
   });
 

indexer/services/ender/__tests__/handlers/stateful-order/stateful-order-removal-handler.test.ts

@@ -21,7 +21,10 @@ import { DydxIndexerSubtypes } from '../../../src/lib/types';
 import {
   defaultDateTime,
   defaultHeight,
-  defaultOrderId, defaultPreviousHeight, defaultTime, defaultTxHash,
+  defaultOrderId,
+  defaultPreviousHeight,
+  defaultTime,
+  defaultTxHash,
 } from '../../helpers/constants';
 import { createKafkaMessageFromStatefulOrderEvent } from '../../helpers/kafka-helpers';
 import { updateBlockCache } from '../../../src/caches/block-cache';
@@ -130,6 +133,7 @@ describe('statefulOrderRemovalHandler', () => {
       producerSendMock,
       orderId: defaultOrderId,
       offchainUpdate: expectedOffchainUpdate,
+      headers: { message_received_timestamp: kafkaMessage.timestamp, event_type: 'StatefulOrderRemoval' },
     });
   });
 

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

@@ -51,7 +51,7 @@ import {
   PerpetualMarketCreateEventV1,
   DeleveragingEventV1,
 } from '@dydxprotocol-indexer/v4-protos';
-import { Message, ProducerRecord } from 'kafkajs';
+import { IHeaders, Message, ProducerRecord } from 'kafkajs';
 import _ from 'lodash';
 
 import {
@@ -318,10 +318,12 @@ export function expectVulcanKafkaMessage({
   producerSendMock,
   orderId,
   offchainUpdate,
+  headers,
 }: {
   producerSendMock: jest.SpyInstance,
   orderId: IndexerOrderId,
   offchainUpdate: OffChainUpdateV1,
+  headers?: IHeaders,
 }): void {
   expect(producerSendMock.mock.calls.length).toBeGreaterThanOrEqual(1);
   expect(producerSendMock.mock.calls[0].length).toBeGreaterThanOrEqual(1);
@@ -339,7 +341,6 @@ export function expectVulcanKafkaMessage({
     vulcanProducerRecord.messages,
     (message: Message): VulcanMessage => {
       expect(Buffer.isBuffer(message.value));
-
       const messageValueBinary: Uint8Array = new Uint8Array(
         // Can assume Buffer, since we check above that it is a buffer
         message.value as Buffer,
@@ -348,13 +349,15 @@ export function expectVulcanKafkaMessage({
       return {
         key: message.key as Buffer,
         value: OffChainUpdateV1.decode(messageValueBinary),
+        headers: message.headers,
       };
     },
   );
 
   expect(vulcanMessages).toContainEqual({
     key: getOrderIdHash(orderId),
     value: offchainUpdate,
+    headers,
   });
 }
 

indexer/services/ender/__tests__/lib/block-processor.test.ts

@@ -121,6 +121,7 @@ describe('block-processor', () => {
     const blockProcessor: BlockProcessor = new BlockProcessor(
       block,
       txId,
+      defaultDateTime.toString(),
     );
     blockProcessor.batchedHandlers = batchedHandlers;
     blockProcessor.syncHandlers = syncHandlers;
@@ -149,6 +150,7 @@ describe('block-processor', () => {
     const blockProcessor: BlockProcessor = new BlockProcessor(
       block,
       txId,
+      defaultDateTime.toString(),
     );
     blockProcessor.batchedHandlers = batchedHandlers;
     blockProcessor.syncHandlers = syncHandlers;

indexer/services/ender/src/handlers/handler.ts

@@ -18,6 +18,7 @@ import {
   OffChainUpdateV1,
   SubaccountId,
 } from '@dydxprotocol-indexer/v4-protos';
+import { IHeaders } from 'kafkajs';
 import { DateTime } from 'luxon';
 import * as pg from 'pg';
 
@@ -33,6 +34,7 @@ export type HandlerInitializer = new (
   indexerTendermintEvent: IndexerTendermintEvent,
   txId: number,
   event: EventMessage,
+  messageReceivedTimestamp?: string,
 ) => Handler<EventMessage>;
 
 /**
@@ -49,20 +51,23 @@ export abstract class Handler<T> {
   blockEventIndex: number;
   event: T;
   abstract eventType: string;
+  messageReceivedTimestamp?: string;
 
   constructor(
     block: IndexerTendermintBlock,
     blockEventIndex: number,
     indexerTendermintEvent: IndexerTendermintEvent,
     txId: number,
     event: T,
+    messageReceivedTimestamp?: string,
   ) {
     this.block = block;
     this.blockEventIndex = blockEventIndex;
     this.indexerTendermintEvent = indexerTendermintEvent;
     this.timestamp = DateTime.fromJSDate(block.time!);
     this.txId = txId;
     this.event = event;
+    this.messageReceivedTimestamp = messageReceivedTimestamp;
   }
 
   /**
@@ -178,6 +183,7 @@ export abstract class Handler<T> {
   protected generateConsolidatedVulcanKafkaEvent(
     key: Buffer,
     offChainUpdate: OffChainUpdateV1,
+    headers?: IHeaders,
   ): ConsolidatedKafkaEvent {
     stats.increment(`${config.SERVICE_NAME}.create_vulcan_kafka_event`, 1);
 
@@ -186,6 +192,7 @@ export abstract class Handler<T> {
       message: {
         key,
         value: offChainUpdate,
+        headers,
       },
     };
   }

indexer/services/ender/src/handlers/stateful-order/conditional-order-triggered-handler.ts

@@ -54,6 +54,10 @@ export class ConditionalOrderTriggeredHandler extends
       this.generateConsolidatedVulcanKafkaEvent(
         getOrderIdHash(order.orderId!),
         offChainUpdate,
+        {
+          message_received_timestamp: this.messageReceivedTimestamp,
+          event_type: 'ConditionalOrderTriggered',
+        },
       ),
     ];
   }

indexer/services/ender/src/handlers/stateful-order/stateful-order-placement-handler.ts

@@ -54,6 +54,10 @@ export class StatefulOrderPlacementHandler extends
     kafakEvents.push(this.generateConsolidatedVulcanKafkaEvent(
       getOrderIdHash(order.orderId!),
       offChainUpdate,
+      {
+        message_received_timestamp: this.messageReceivedTimestamp,
+        event_type: 'StatefulOrderPlacement',
+      },
     ));
 
     return kafakEvents;

indexer/services/ender/src/handlers/stateful-order/stateful-order-removal-handler.ts

@@ -42,6 +42,10 @@ export class StatefulOrderRemovalHandler extends
       this.generateConsolidatedVulcanKafkaEvent(
         getOrderIdHash(orderIdProto),
         offChainUpdate,
+        {
+          message_received_timestamp: this.messageReceivedTimestamp,
+          event_type: 'StatefulOrderRemoval',
+        },
       ),
     ];
   }

indexer/services/ender/src/lib/block-processor.ts

@@ -77,13 +77,16 @@ export class BlockProcessor {
   txId: number;
   batchedHandlers: BatchedHandlers;
   syncHandlers: SyncHandlers;
+  messageReceivedTimestamp: string;
 
   constructor(
     block: IndexerTendermintBlock,
     txId: number,
+    messageReceivedTimestamp: string,
   ) {
     this.block = block;
     this.txId = txId;
+    this.messageReceivedTimestamp = messageReceivedTimestamp;
     this.sqlBlock = {
       ...this.block,
       events: new Array(this.block.events.length),
@@ -205,6 +208,7 @@ export class BlockProcessor {
     const handlers: Handler<EventMessage>[] = validator.createHandlers(
       eventProto.indexerTendermintEvent,
       this.txId,
+      this.messageReceivedTimestamp,
     );
 
     _.map(handlers, (handler: Handler<EventMessage>) => {

indexer/services/ender/src/lib/kafka-publisher.ts

@@ -247,6 +247,7 @@ export class KafkaPublisher {
           return {
             key: message.key,
             value: Buffer.from(Uint8Array.from(OffChainUpdateV1.encode(message.value).finish())),
+            headers: message.headers,
           };
         }),
       });

indexer/services/ender/src/lib/on-message.ts

@@ -83,6 +83,7 @@ export async function onMessage(message: KafkaMessage): Promise<void> {
     const blockProcessor: BlockProcessor = new BlockProcessor(
       indexerTendermintBlock,
       txId,
+      message.timestamp,
     );
     const kafkaPublisher: KafkaPublisher = await blockProcessor.process();
 

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

@@ -33,6 +33,7 @@ import {
   DeleveragingEventV1,
   TradingRewardsEventV1,
 } from '@dydxprotocol-indexer/v4-protos';
+import { IHeaders } from 'kafkajs';
 import Long from 'long';
 
 // Type sourced from protocol:
@@ -217,6 +218,7 @@ export interface AnnotatedSubaccountMessage extends SubaccountMessage {
 export interface VulcanMessage {
   key: Buffer,
   value: OffChainUpdateV1,
+  headers?: IHeaders,
 }
 
 export type ConsolidatedKafkaEvent = {

indexer/services/ender/src/validators/asset-validator.ts

@@ -10,6 +10,7 @@ export class AssetValidator extends Validator<AssetCreateEventV1> {
   public createHandlers(
     indexerTendermintEvent: IndexerTendermintEvent,
     txId: number,
+    _: string,
   ): Handler<AssetCreateEventV1>[] {
     const handler: Handler<AssetCreateEventV1> = new AssetCreationHandler(
       this.block,

indexer/services/ender/src/validators/deleveraging-validator.ts

@@ -38,6 +38,7 @@ export class DeleveragingValidator extends Validator<DeleveragingEventV1> {
   public createHandlers(
     indexerTendermintEvent: IndexerTendermintEvent,
     txId: number,
+    _: string,
   ): Handler<DeleveragingEventV1>[] {
     return [
       new DeleveragingHandler(

indexer/services/ender/src/validators/funding-validator.ts

@@ -42,6 +42,7 @@ export class FundingValidator extends Validator<FundingEventV1> {
   public createHandlers(
     indexerTendermintEvent: IndexerTendermintEvent,
     txId: number,
+    _: string,
   ): Handler<FundingEventMessage>[] {
     const handler: Handler<FundingEventMessage> = new FundingHandler(
       this.block,

indexer/services/ender/src/validators/liquidity-tier-validator.ts

@@ -34,6 +34,7 @@ export class LiquidityTierValidator extends Validator<LiquidityTierUpsertEventV1
   public createHandlers(
     indexerTendermintEvent: IndexerTendermintEvent,
     txId: number,
+    _: string,
   ): Handler<LiquidityTierUpsertEventV1>[] {
     const handler: Handler<LiquidityTierUpsertEventV1> = new LiquidityTierHandler(
       this.block,

indexer/services/ender/src/validators/market-validator.ts

@@ -104,6 +104,7 @@ export class MarketValidator extends Validator<MarketEventV1> {
   public createHandlers(
     indexerTendermintEvent: IndexerTendermintEvent,
     txId: number,
+    __: string,
   ): Handler<MarketEventV1>[] {
     const Initializer:
     HandlerInitializer | undefined = this.getHandlerInitializer();

indexer/services/ender/src/validators/order-fill-validator.ts

@@ -93,6 +93,7 @@ export class OrderFillValidator extends Validator<OrderFillEventV1> {
   public createHandlers(
     indexerTendermintEvent: IndexerTendermintEvent,
     txId: number,
+    __: string,
   ): Handler<OrderFillWithLiquidity>[] {
     const orderFillEventsWithLiquidity: OrderFillEventWithLiquidity[] = [
       {

indexer/services/ender/src/validators/perpetual-market-validator.ts

@@ -38,6 +38,7 @@ export class PerpetualMarketValidator extends Validator<PerpetualMarketCreateEve
   public createHandlers(
     indexerTendermintEvent: IndexerTendermintEvent,
     txId: number,
+    _: string,
   ): Handler<PerpetualMarketCreateEventV1>[] {
     const handler: Handler<PerpetualMarketCreateEventV1> = new PerpetualMarketCreationHandler(
       this.block,