Integrate receipts extraction into ETHWorker #201

YordanPavlov · 2024-07-24T08:25:15Z

This PR integrates the receipts extraction into ETHWorker. To achieve this it does the following steps:

Break the Exporter class into its two main functions, KafkaStorage and ZookeeperState.
Support multiple KafkaStorage instances per running exporter, each mode of operation is associated with one KafkaStorage.
Integrate the ReceiptsWorker into the ETHWorker classes. The latter already extracts receipts, it just needs to format them in the proper format and fill the receipts output topic.

YordanPavlov · 2024-07-24T08:26:51Z

bin/export_env_vars.sh

 export EXPORT_TIMEOUT_MLS=300000
 export CONTRACT_MODE="extract_exact_overwrite"
 export BLOCKCHAIN="eth"
-export KAFKA_TOPIC="erc20_exporter_test_topic"
+#export KAFKA_TOPIC="erc20_exporter_test_topic"
+export KAFKA_TOPIC='native_token_transfers:erc20_exporter_test_topic'


Besides a single topic, we now support multiple ones. The format is comma separate mode:topic_name

this is kind of ugly. I believe we can use several arguments instead of packing multiple values inside a single one.
Can we so something like this?

KAFKA_TOPIC_ERC20 = ... KAFKA_TOPIC_TRANSFER = ...

YordanPavlov · 2024-07-24T08:29:28Z

src/blockchains/construct_worker.ts

@@ -22,8 +21,6 @@ export function constructWorker(blockchain: string, settings: any): BaseWorker {
      return new ETHBlocksWorker(settings);
    case 'matic':
      return new MaticWorker(settings);
-    case 'receipts':
-      return new ReceiptsWorker(settings);


Separate worker is no longer needed. We would use the ETHWorker to extract receipts, we can do it like this:

KAFKA_TOPIC='native_token_transfers:eth_transfers_test_topic,receipts:receipts_test_topic'

If we want to extract transfers and receipts at once. If for some reason we want to have only receipts deploy, we can do:

KAFKA_TOPIC='receipts:receipts_test_topic'

YordanPavlov · 2024-07-24T08:32:51Z

src/blockchains/erc20/erc20_worker.ts

@@ -59,7 +59,7 @@ export class ERC20Worker extends BaseWorker {
    this.allOldContracts = [];
  }

-  async init(exporter?: Exporter) {
+  async init(storage: KafkaStorage | Map<string, KafkaStorage>) {


Workers receive information for which output Kafka topics need to be filled and produce the needed data.

YordanPavlov · 2024-07-24T08:35:06Z

src/blockchains/eth/eth_worker.ts

-      fetchReceipts(this.ethClient, this.web3Wrapper,
-        this.settings.RECEIPTS_API_METHOD, fromBlock, toBlock),
-    ]);
+  async fetchData(fromBlock: number, toBlock: number): Promise<[Trace[], Map<number, ETHBlock>, ETHReceipt[]]> {


We fetch raw data depending on what outputs are needed. Some basic raw data like block times is needed for any output.

WonderBeat · 2024-07-25T10:23:19Z

bin/export_env_vars.sh

@@ -2,11 +2,12 @@ export KAFKA_URL=kafka-hz.stage.san:30911
 export ZOOKEEPER_URL=zookeeper-hz.stage.san:30921
 export NODE_URL=https://ethereum.santiment.net
 export START_BLOCK="15676731"
-export BLOCK_INTERVAL="50"
+export BLOCK_INTERVAL="5"


is it a change required for this PR?

WonderBeat · 2024-07-25T10:25:49Z

bin/export_env_vars.sh

 export EXPORT_TIMEOUT_MLS=300000
 export CONTRACT_MODE="extract_exact_overwrite"
 export BLOCKCHAIN="eth"
-export KAFKA_TOPIC="erc20_exporter_test_topic"
+#export KAFKA_TOPIC="erc20_exporter_test_topic"
+export KAFKA_TOPIC='native_token_transfers:erc20_exporter_test_topic'


this is kind of ugly. I believe we can use several arguments instead of packing multiple values inside a single one.
Can we so something like this?

KAFKA_TOPIC_ERC20 = ... KAFKA_TOPIC_TRANSFER = ...

WonderBeat · 2024-07-25T10:46:14Z

src/blockchains/eth/eth_worker.ts

+        receipts, endOfBlockEvents);
+    }
+    if (this.modes.includes(this.settings.RECEIPTS_MODE)) {
+      result[this.settings.RECEIPTS_MODE] = this.getReceiptsOutput(blocks, receipts);


this is not really modular design because each module has a footprint in the eth_worker file.
imo here we should have something like this:

foreach(module: modules) { result[module] = module.extract(rawData) }

WonderBeat · 2024-07-25T10:48:18Z

src/blockchains/eth/lib/helper_receipts.ts

+
+import { Web3Interface } from './web3_wrapper';
+
+const decodeLog = (log: any, web3Wrapper: Web3Interface) => {


could you please define a return types for this functions?

WonderBeat · 2024-07-25T10:54:38Z

src/main.ts

+      }
+    }
+    else if (typeof workResult === 'object') {
+      if (!(this.kafkaStorage instanceof Map)) {


is it better to always use Map and checking for amount of keys defined than trying to dynamically resolve the type?

WonderBeat · 2024-07-25T10:55:53Z

src/main.ts

+      return this.kafkaStorage.isConnected();
+    }
+    else if (this.kafkaStorage instanceof Map) {
+      return Array.from(this.kafkaStorage.values()).every(storage => storage.isConnected());


hmm. we maintain a separate connection per mode but should we? any benefits?

WonderBeat · 2024-07-25T10:58:47Z

src/main.ts

-          send(response, 500, err.toString());
-        });
+      if (mainInstance.healthcheck()) {
+        return send(response, 200, 'ok');


I guess that this is out of scope for this PR but could we later change this to something more advanced than a simple 200 all the time? Can we measure throughput, ensure we connected to kafka and ZK for example?

YordanPavlov added 7 commits July 10, 2024 12:06

Break 'Exporter' into KafkaStorage and ZookeeperState classes

91c3578

Support multiple topics initial

23cc720

Support multiple topics working

f88493e

ETH native token and receipts unified in ETHWorker

9af0ebb

Construct KafkaStorages correctly. Test fixes.

6263bef

Fetch receipts fixes

d34e0b3

Test fixes. Move KAFKA_TOPIC parsing to util function.

09c33af

YordanPavlov commented Jul 24, 2024

View reviewed changes

Generate output for different modes in separate functions

c63ada0

WonderBeat requested changes Jul 25, 2024

View reviewed changes

YordanPavlov added the WIP label Sep 4, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Integrate receipts extraction into ETHWorker #201

Integrate receipts extraction into ETHWorker #201

YordanPavlov commented Jul 24, 2024

YordanPavlov Jul 24, 2024

WonderBeat Jul 25, 2024

YordanPavlov Jul 24, 2024

YordanPavlov Jul 24, 2024

YordanPavlov Jul 24, 2024

WonderBeat Jul 25, 2024

WonderBeat Jul 25, 2024

WonderBeat Jul 25, 2024

WonderBeat Jul 25, 2024

WonderBeat Jul 25, 2024

WonderBeat Jul 25, 2024

WonderBeat Jul 25, 2024


		import { Web3Interface } from './web3_wrapper';

		const decodeLog = (log: any, web3Wrapper: Web3Interface) => {

Integrate receipts extraction into ETHWorker #201

Are you sure you want to change the base?

Integrate receipts extraction into ETHWorker #201

Conversation

YordanPavlov commented Jul 24, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment