metrics(ender): refine processing and publishing metrics (#3309)

Author: UnbornAztecKing

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

@@ -72,20 +72,34 @@ export class BatchKafkaProducer {
         this.producer.send({ topic: this.topic, messages: this.producerMessages }),
       );
     }
+    const elapsed = Date.now() - startTime;
+
     logger.info({
       at: 'BatchMessenger#sendBatch',
       message: 'Produced kafka batch',
       currentSize: this.currentSize,
+      topic: this.topic,
+      sendTime: elapsed,
+    });
+
+    // avoid logging encoded message contents at INFO level
+    logger.debug({
+      at: 'BatchMessenger#sendBatch',
+      message: 'Produced kafka batch with messages',
+      currentSize: this.currentSize,
       producerMessages: JSON.stringify(this.producerMessages),
       recalculatedCurrentSize: this.producerMessages.reduce(
         (acc: number, msg: ProducerMessage) => acc + msg.value.byteLength,
         0,
       ),
       topic: this.topic,
-      sendTime: Date.now() - startTime,
+      sendTime: elapsed,
     });
+
     stats.gauge(`${config.SERVICE_NAME}.kafka_batch_size`, this.currentSize);
+    // TODO: this metric is useless it only captures creation of promises
     stats.timing(`${config.SERVICE_NAME}.kafka_batch_send_time`, Date.now() - startTime);
+
     this.producerMessages = [];
     this.currentSize = 0;
   }

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

@@ -271,7 +271,7 @@ export class BlockProcessor {
   }
 
   private async processEvents(): Promise<KafkaPublisher> {
-    const kafkaPublisher: KafkaPublisher = new KafkaPublisher();
+    const start: number = Date.now();
 
     await Promise.all(this.sqlEventPromises).then((values) => {
       for (let i: number = 0; i < this.block.events.length; i++) {
@@ -287,7 +287,6 @@ export class BlockProcessor {
       }
     });
 
-    const start: number = Date.now();
     let success = false;
     let resultRow: pg.QueryResultRow;
     try {
@@ -318,17 +317,27 @@ export class BlockProcessor {
     }
 
     // Create a block message from the current block
+    const kafkaPublisher: KafkaPublisher = new KafkaPublisher();
     kafkaPublisher.addEvent(this.createBlockHeightMsg());
 
     // in genesis, handle sync events first, then batched events.
     // in other blocks, handle batched events first, then sync events.
+    const startHandlingEvents: number = Date.now();
+
     if (this.block.height === 0) {
       await this.syncHandlers.process(kafkaPublisher, resultRow);
       await this.batchedHandlers.process(kafkaPublisher, resultRow);
     } else {
       await this.batchedHandlers.process(kafkaPublisher, resultRow);
       await this.syncHandlers.process(kafkaPublisher, resultRow);
     }
+
+    stats.timing(
+      `${config.SERVICE_NAME}.handle_events.timing`,
+      Date.now() - startHandlingEvents,
+      STATS_NO_SAMPLING,
+    );
+
     return kafkaPublisher;
   }
 }

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

@@ -185,6 +185,7 @@ export class KafkaPublisher {
   }
 
   public async publish() {
+    const start = Date.now();
     const allTopicKafkaMessages:
     TopicKafkaMessages[] = this.generateAllTopicKafkaMessages();
 
@@ -204,6 +205,11 @@ export class KafkaPublisher {
         },
       ),
     );
+    stats.timing(
+      `${config.SERVICE_NAME}.kafka_publish.timing`,
+      Date.now() - start,
+      STATS_NO_SAMPLING,
+    );
   }
 
   private generateAllTopicKafkaMessages(): TopicKafkaMessages[] {

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

@@ -95,14 +95,20 @@ export async function onMessage(message: KafkaMessage): Promise<void> {
     );
     const kafkaPublisher: KafkaPublisher = await blockProcessor.process();
 
+    // TODO: KafkaPublisher is not idempotent, so we can't send non-candles messages here
+
     const candlesGenerator: CandlesGenerator = new CandlesGenerator(
       kafkaPublisher,
       DateTime.fromJSDate(indexerTendermintBlock.time!),
       txId,
     );
     const candles: CandleFromDatabase[] = await candlesGenerator.updateCandles();
+
+    // TODO: ideally, use a candles-only kafka publisher to send here
+
     await Transaction.commit(txId);
     stats.gauge(`${config.SERVICE_NAME}.processing_block_height`, indexerTendermintBlock.height);
+
     // Update caches after transaction is committed
     updateBlockCache(blockHeight);
     _.forEach(candles, updateCandleCacheWithCandle);
@@ -114,13 +120,16 @@ export async function onMessage(message: KafkaMessage): Promise<void> {
         'kafkaPublisher.publish',
       );
     }
+
     logger.info({
       at: 'onMessage#onMessage',
       message: 'Successfully processed block',
       height: blockHeight,
     });
     success = true;
+
   } catch (error) {
+
     await Transaction.rollback(txId);
     await refreshDataCaches();
     stats.increment(`${config.SERVICE_NAME}.update_event_tables.failure`, 1);
@@ -143,6 +152,7 @@ export async function onMessage(message: KafkaMessage): Promise<void> {
     }
     // Throw error so the message is not acked and will be reprocessed
     throw error;
+
   } finally {
     const done: number = Date.now();
     stats.timing(