feat: 🎸 Watch ERC-20 token transfers

Add watchers that will watch a specific ERC-20 contract and log the
balances of all transfers.

config.schema.json

@@ -35,6 +35,69 @@
             },
             "type": "object"
         },
+        "BalanceWatcherConfigSchema": {
+            "description": "Balance watcher is a component tracking transfers of a token and reporting balances of its accounts.",
+            "properties": {
+                "blocksMaxChunkSize": {
+                    "description": "Max. number of blocks to fetch at once",
+                    "type": "number"
+                },
+                "contractAddress": {
+                    "description": "The address of the contract to watch.",
+                    "type": "string"
+                },
+                "decimals": {
+                    "description": "The number of decimals to divide balances with.",
+                    "type": "number"
+                },
+                "enabled": {
+                    "description": "Specify `false` to disable the balance watcher",
+                    "type": "boolean"
+                },
+                "maxParallelChunks": {
+                    "description": "Max. number of chunks to process in parallel",
+                    "type": "number"
+                },
+                "pollInterval": {
+                    "description": "Interval in which to look for the latest block number (if not busy processing the backlog)",
+                    "type": ["string", "number"]
+                },
+                "retryWaitTime": {
+                    "anyOf": [
+                        {
+                            "$ref": "#/definitions/ExponentialBackoffConfig"
+                        },
+                        {
+                            "$ref": "#/definitions/LinearBackoffConfig"
+                        },
+                        {
+                            "type": ["string", "number"]
+                        }
+                    ],
+                    "description": "Wait time before retrying to fetch and process blocks after failure"
+                },
+                "startAt": {
+                    "anyOf": [
+                        {
+                            "enum": ["genesis", "latest"],
+                            "type": "string"
+                        },
+                        {
+                            "type": "number"
+                        }
+                    ],
+                    "description": "If no checkpoint exists (yet), this specifies which block should be chosen as the starting point."
+                }
+            },
+            "type": "object"
+        },
+        "BalanceWatchersConfigSchema": {
+            "additionalProperties": {
+                "$ref": "#/definitions/BalanceWatcherConfigSchema"
+            },
+            "description": "Balance watchers is a component tracking transfers of tokens and reporting balances of accounts.",
+            "type": "object"
+        },
         "BlockWatcherConfigSchema": {
             "description": "Block watcher is the component that retrieves blocks, transactions, event logs from the node and sends\nthem to output.",
             "properties": {
@@ -618,6 +681,10 @@
         "SourcetypesSchema": {
             "description": "Configurable set of `sourcetype` field values emitted by ethlogger",
             "properties": {
+                "balance": {
+                    "default": "ethereum:balance",
+                    "type": "string"
+                },
                 "block": {
                     "default": "ethereum:block",
                     "type": "string"
@@ -656,6 +723,10 @@
             "$ref": "#/definitions/AbiRepositoryConfigSchema",
             "description": "ABI repository configuration"
         },
+        "balanceWatchers": {
+            "$ref": "#/definitions/BalanceWatchersConfigSchema",
+            "description": "Balance watchers, tracking balance of ERC-20 token holders"
+        },
         "blockWatcher": {
             "$ref": "#/definitions/BlockWatcherConfigSchema",
             "description": "Block watcher settings, configure how blocks, transactions, event logs are ingested"

docs/configuration.md

@@ -81,6 +81,7 @@ Root configuration schema for ethlogger
 | `abi`             | [`AbiRepository`](#AbiRepository)                                                                                                  | ABI repository configuration                                                                                                                                                                                           |
 | `contractInfo`    | [`ContractInfo`](#ContractInfo)                                                                                                    | Contract info cache settings                                                                                                                                                                                           |
 | `blockWatcher`    | [`BlockWatcher`](#BlockWatcher)                                                                                                    | Block watcher settings, configure how blocks, transactions, event logs are ingested                                                                                                                                    |
+| `balanceWatchers` | [`BalanceWatchers`](#BalanceWatchers)                                                                                              | Balance watchers, tracking balance of ERC-20 token holders                                                                                                                                                             |
 | `nodeMetrics`     | [`NodeMetrics`](#NodeMetrics)                                                                                                      | Settings for the node metrics collector                                                                                                                                                                                |
 | `nodeInfo`        | [`NodeInfo`](#NodeInfo)                                                                                                            | Settings for the node info collector                                                                                                                                                                                   |
 | `pendingTx`       | [`PendingTx`](#PendingTx)                                                                                                          | Settings for collecting pending transactions from node                                                                                                                                                                 |
@@ -141,6 +142,7 @@ Configurable set of `sourcetype` field values emitted by ethlogger
 | `nodeInfo`    | `string` | `"ethereum:node:info"`           |
 | `nodeMetrics` | `string` | `"ethereum:node:metrics"`        |
 | `gethPeer`    | `string` | `"ethereum:geth:peer"`           |
+| `balance`     | `string` | `"ethereum:balance"`             |
 
 ### ConsoleOutput
 
@@ -245,6 +247,29 @@ Block watcher is the component that retrieves blocks, transactions, event logs f
 | `retryWaitTime`              | [`WaitTime`](#WaitTime)     | Wait time before retrying to fetch and process blocks after failure                                                                                           |
 | `decryptPrivateTransactions` | `boolean`                   | For chains/nodes that do support private transactions, this setting instructs block watcher to attempt to load the decrypted payload for private transactions |
 
+### BalanceWatchers
+
+Balance watchers is a component tracking transfers of tokens and reporting balances of accounts.
+
+| Name | Type                                            | Description                                                                |
+| ---- | ----------------------------------------------- | -------------------------------------------------------------------------- |
+| `-`  | map<string,[`BalanceWatcher`](#BalanceWatcher)> | Mapping of name => balancer watcher.<br><br>See BalanceWatcherConfigSchema |
+
+### BalanceWatcher
+
+Balance watcher is a component tracking transfers of a token and reporting balances of its accounts.
+
+| Name                 | Type                        | Description                                                                                       |
+| -------------------- | --------------------------- | ------------------------------------------------------------------------------------------------- |
+| `contractAddress`    | `string`                    | The address of the contract to watch.                                                             |
+| `decimals`           | `number`                    | The number of decimals to divide balances with.                                                   |
+| `startAt`            | [`StartBlock`](#StartBlock) | If no checkpoint exists (yet), this specifies which block should be chosen as the starting point. |
+| `enabled`            | `boolean`                   | Specify `false` to disable the balance watcher                                                    |
+| `pollInterval`       | [`Duration`](#Duration)     | Interval in which to look for the latest block number (if not busy processing the backlog)        |
+| `blocksMaxChunkSize` | `number`                    | Max. number of blocks to fetch at once                                                            |
+| `maxParallelChunks`  | `number`                    | Max. number of chunks to process in parallel                                                      |
+| `retryWaitTime`      | [`WaitTime`](#WaitTime)     | Wait time before retrying to fetch and process blocks after failure                               |
+
 ### NodeMetrics
 
 The node metrics collector retrieves numeric measurements from nodes on a periodic basis.

examples/erc20-tracking/.gitignore

@@ -0,0 +1,2 @@
+/.ethlogger-state.json
+/checkpoints.json

examples/erc20-tracking/README.md

@@ -0,0 +1,32 @@
+# Watching ERC-20 token balances
+
+This is an example showing how to use ethlogger to track the activity of an ERC-20 token.
+
+Ethlogger configuration is provided in the form of [environment variables](../../docs/cli.md#environment-variables) in [docker-compose.yaml](./docker-compose.yaml#L25) and a configuration file, [ethlogger.yaml](./ethlogger.yaml).
+
+## Run
+
+1. Start docker-compose
+
+```sh-session
+$ cd examples/erc20-tracking
+$ docker-compose up -d
+```
+
+2. Wait for all containers to start.
+   You can rely on the output of `docker ps` to see the state of services.
+
+3. Then go to [http://localhost:8000](http://localhost:8000) to explore the data produced by ethlogger.
+   Login using user `admin` and password `changeme`
+
+## Note
+
+> This example is not meant to be used in a production setup.
+> Using the logging driver to log to a container in the same docker-compose stack shouldn't be used in production.
+> Splunk and ethlogger persist data using local volumes and a checkpoints file. If blocks are no longer being ingested, or if you want to change the blockchain you are using, you should clear this state. To start clean, run the following.
+
+```sh-session
+$ docker-compose down
+$ rm checkpoints.json
+$ docker volume prune
+```

examples/erc20-tracking/docker-compose.yaml

@@ -0,0 +1,43 @@
+version: '3.6'
+
+services:
+    splunk:
+        image: splunk/splunk:latest
+        container_name: splunk
+        environment:
+            - SPLUNK_START_ARGS=--accept-license
+            - SPLUNK_HEC_TOKEN=11111111-1111-1111-1111-1111111111113
+            - SPLUNK_PASSWORD=changeme
+            - SPLUNK_APPS_URL=https://github.com/splunk/ethereum-basics/releases/download/latest/ethereum-basics.tgz
+        ports:
+            - 8000:8000
+            - 8088:8088
+        healthcheck:
+            test: ['CMD', 'curl', '-f', 'http://localhost:8000']
+            interval: 5s
+            timeout: 5s
+            retries: 20
+        volumes:
+            - ./splunk.yml:/tmp/defaults/default.yml
+            - /opt/splunk/var
+            - /opt/splunk/etc
+    ethlogger:
+        image: ghcr.io/splunkdlt/ethlogger:latest
+        container_name: ethlogger
+        command: -c /app/ethlogger.yaml
+        environment:
+            - ETH_RPC_URL=https://dai.poa.network
+            # Use these environment variables to connect to infura
+            # - ETH_RPC_URL=https://mainnet.infura.io/v3/<your infura project id>
+            - START_AT_BLOCK=latest
+            - SPLUNK_HEC_URL=https://splunk:8088
+            - SPLUNK_HEC_TOKEN=11111111-1111-1111-1111-1111111111113
+            - SPLUNK_EVENTS_INDEX=main
+            - SPLUNK_METRICS_INDEX=metrics
+            - SPLUNK_INTERNAL_INDEX=metrics
+            - SPLUNK_HEC_REJECT_INVALID_CERTS=false
+        volumes:
+            - ./:/app
+        depends_on:
+            - splunk
+        restart: always

examples/erc20-tracking/ethlogger.yaml

@@ -0,0 +1,11 @@
+balanceWatchers:
+    wxdai:
+        # https://blockscout.com/xdai/mainnet/address/0xe91D153E0b41518A2Ce8Dd3D7944Fa863463a97d/transactions
+        contractAddress: '0xe91d153e0b41518a2ce8dd3d7944fa863463a97d'
+        startAt: 19023469
+blockWatcher:
+    enabled: false
+nodeMetrics:
+    enabled: false
+nodeStats:
+    enabled: false

examples/erc20-tracking/splunk.yml

@@ -0,0 +1,11 @@
+splunk:
+    conf:
+        indexes:
+            directory: /opt/splunk/etc/apps/search/local
+            content:
+                metrics:
+                    coldPath: $SPLUNK_DB/metrics/colddb
+                    datatype: metric
+                    homePath: $SPLUNK_DB/metrics/db
+                    maxTotalDataSizeMB: 512000
+                    thawedPath: $SPLUNK_DB/metrics/thaweddb

scripts/gendocs.ts

@@ -14,8 +14,9 @@ type UnkownType = { type: 'unknown' };
 type LiteralTypeInfo = { type: 'literal'; value: string };
 type PrimitiveTypeInfo = { type: 'primitive'; name: string };
 type ObjectTypeInfo = { type: 'object'; name: string };
+type MapTypeInfo = { type: 'map'; name: string };
 type UnionTypeInfo = Array<TypeInfo>;
-type TypeInfo = LiteralTypeInfo | PrimitiveTypeInfo | ObjectTypeInfo | UnionTypeInfo | UnkownType;
+type TypeInfo = LiteralTypeInfo | PrimitiveTypeInfo | ObjectTypeInfo | UnionTypeInfo | MapTypeInfo | UnkownType;
 
 interface Field {
     name: string;
@@ -29,6 +30,13 @@ interface Field {
 const inlineCode = (s: string) => '`' + s + '`';
 const link = (to: string, label: string) => `[${label}](${to})`;
 
+function formatFieldName(name: string): string {
+    if (name === '__index') {
+        return '-';
+    }
+    return name;
+}
+
 function formatTypeInfo(type: TypeInfo): string {
     if (Array.isArray(type)) {
         return type.map(formatTypeInfo).join(` \\| `);
@@ -40,6 +48,8 @@ function formatTypeInfo(type: TypeInfo): string {
         return inlineCode(type.value);
     } else if (type.type === 'object') {
         return link(`#${type.name}`, inlineCode(type.name));
+    } else if (type.type === 'map') {
+        return 'map<string,' + link(`#${type.name}`, inlineCode(type.name)) + '>';
     }
     throw new Error('INVALID TYPE: ' + JSON.stringify(type));
 }
@@ -175,6 +185,20 @@ function createConfigurationSchemaReference(): string {
                 const memberType = typeChecker.getTypeAtLocation(member.declarations[0]);
                 const resolveType = (type: ts.Type): TypeInfo => {
                     const flags = type.flags;
+                    if (flags & ts.TypeFlags.Any) {
+                        if (member.name == '__index') {
+                            const fieldMapType = member
+                                .getDeclarations()
+                                ?.values()
+                                .next().value.type;
+                            const name = fieldMapType?.typeName.escapedText
+                                ?.replace(/Schema$/, '')
+                                .replace(/Config$/, '');
+                            const mappedType = findType(fieldMapType?.typeName.escapedText, program, typeChecker);
+                            appendSectionForType(mappedType);
+                            return { type: 'map', name };
+                        }
+                    }
                     if (flags & ts.TypeFlags.StringLiteral) {
                         return { type: 'literal', value: JSON.stringify((type as ts.LiteralType).value) };
                     }
@@ -231,7 +255,7 @@ function createConfigurationSchemaReference(): string {
                 const see = member.getJsDocTags().find(t => t.name === 'see')?.text;
                 const defaultValue = member.getJsDocTags().find(t => t.name === 'default')?.text;
                 section.fields.push({
-                    name: member.name,
+                    name: formatFieldName(member.name),
                     type: resolveType(memberType),
                     description: docs.length ? docs.map(d => d.text).join(' ') : undefined,
                     example,