Feat: epoch monitoring

[anomit]

Fixes #40

Checklist

• My branch is up-to-date with upstream/develop branch.
• Everything works and tested for Python 3.8.0 and above.
• I ran pre-commit checks against my changes.
• I've written tests against my changes and all the current present tests are passing.

Current behaviour

There are no APIs that expose the processing status of a snapshot according to the state transitions it goes through before being finalized.

For more details, refer #40

New expected behaviour

The changes included in this PR capture the state transitions as detailed in #40 as per the following data models as an epoch is released until SnapshotFinalized event is received by the processor distributor for the specific epoch.

---

EPOCH_RELEASED

Redis key being set:

pooler/snapshotter/utils/redis/redis_keys.py

Lines 90 to 91 in bcc245d

	def epoch_id_epoch_released_key(epoch_id):
	return f'epochID:{epoch_id}:epochReleased'

pooler/snapshotter/processor_distributor.py

Lines 649 to 652 in bcc245d

	await self._redis_conn.set(
	epoch_id_epoch_released_key(_.epochId),
	int(time.time()),
	)

It contains the UNIX timestamp at which the EpochReleased event was received from the protocol state contract.

---

For each of the state IDs corresponding to a state transition, except EPOCH_RELEASED, a Redis hashtable entry is set against every project ID or type being processed by the snapshotter. The key of this hashtable is in the following format

pooler/snapshotter/utils/redis/redis_keys.py

Lines 94 to 95 in bcc245d

	def epoch_id_project_to_state_mapping(epoch_id, state_id):
	return f'epochID:{epoch_id}:stateID:{state_id}:processingStatus'

and each value against a hashtable entry of a project ID follows the data model:

pooler/snapshotter/utils/models/data_models.py

Lines 48 to 52 in bcc245d

	class SnapshotterStateUpdate(BaseModel):
	status: str
	error: Optional[str] = None
	extra: Optional[Dict[str, Any]] = None
	timestamp: int

---

PRELOAD

For every project type's preloader specifications, the status of all the preloading dependencies being satisfied is captured here:

pooler/snapshotter/processor_distributor.py

Lines 227 to 251 in bcc245d

	for project_type in self._project_type_config_mapping:
	project_config = self._project_type_config_mapping[project_type]
	if not project_config.preload_tasks:
	continue
	self._logger.debug(
	'Expected list of successful preloading for project type {}: {}',
	project_type,
	project_config.preload_tasks,
	)
	if all([t in succesful_preloads for t in project_config.preload_tasks]):
	self._logger.info(
	'Preloading dependency satisfied for project type {} epoch {}. Distributing snapshot build tasks...',
	project_type, epoch.epochId,
	)
	asyncio.ensure_future(
	self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(epoch.epochId, SnapshotterStates.PRELOAD.value),
	mapping={
	project_type: SnapshotterStateUpdate(
	status='success', timestamp=int(time.time())
	).json()
	}
	)
	)
	await self._distribute_callbacks_snapshotting(project_type, epoch)

SNAPSHOT_BUILD

The snapshot builders as configured in projects.json are executed. Also refer to the case study of the current implementation of Pooler for a detailed look at snapshot building for base as well as aggregates.

pooler/snapshotter/utils/snapshot_worker.py

Lines 100 to 120 in bcc245d

	await self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(
	epoch_id=msg_obj.epochId, state_id=SnapshotterStates.SNAPSHOT_BUILD.value,
	),
	mapping={
	project_id: SnapshotterStateUpdate(
	status='failed', error=str(e), timestamp=int(time.time()),
	).json(),
	},
	)
	else:
	await self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(
	epoch_id=msg_obj.epochId, state_id=SnapshotterStates.SNAPSHOT_BUILD.value,
	),
	mapping={
	project_id: SnapshotterStateUpdate(
	status='success', timestamp=int(time.time()),
	).json(),
	},
	)

SNAPSHOT_SUBMIT_PAYLOAD_COMMIT

Captures the status of propagation of the built snapshot to the payload commit service in Audit Protocol for further submission to the protocol state contract.

pooler/snapshotter/utils/generic_worker.py

Lines 166 to 195 in bcc245d

	except Exception as e:
	self._logger.opt(exception=True).error(
	(
	'Exception committing snapshot to commit payload queue:'
	' {} | dump: {}'
	),
	snapshot,
	e,
	)
	await self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(
	epoch.epochId, SnapshotterStates.SNAPSHOT_SUBMIT_PAYLOAD_COMMIT.value,
	),
	mapping={
	project_id: SnapshotterStateUpdate(
	status='failed', error=str(e), timestamp=int(time.time()),
	).json(),
	},
	)
	else:
	await self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(
	epoch.epochId, SnapshotterStates.SNAPSHOT_SUBMIT_PAYLOAD_COMMIT.value,
	),
	mapping={
	project_id: SnapshotterStateUpdate(
	status='success', timestamp=int(time.time()),
	).json(),
	},
	)

SNAPSHOT_SUBMIT_PROTOCOL_CONTRACT

The snapshot submission transaction from the relayer to the protocol state smart contract was successful and a SnapshotSubmitted event was generated

pooler/snapshotter/processor_distributor.py

Lines 655 to 668 in bcc245d

	elif message_type == 'SnapshotSubmitted':
	try:
	msg_obj: PowerloomSnapshotSubmittedMessage = PowerloomSnapshotSubmittedMessage.parse_raw(message.body)
	except:
	pass
	else:
	await self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(msg_obj.epochId, SnapshotterStates.SNAPSHOT_SUBMIT_PROTOCOL_CONTRACT.value),
	mapping={
	msg_obj.projectId: SnapshotterStateUpdate(
	status='success', timestamp=int(time.time()), extra={'snapshotCid': msg_obj.snapshotCid}
	).json()
	},
	)

SNAPSHOT_FINALIZE

Captures the finalized snapshot accepted against an epoch via a SnapshotFinalized event.

pooler/snapshotter/processor_distributor.py

Lines 475 to 482 in bcc245d

	await self._redis_conn.hset(
	name=epoch_id_project_to_state_mapping(msg_obj.epochId, SnapshotterStates.SNAPSHOT_FINALIZE.value),
	mapping={
	msg_obj.projectId: SnapshotterStateUpdate(
	status='success', timestamp=int(time.time()), extra={'snapshot_cid': msg_obj.snapshotCid}
	).json()
	}
	)

---

Core API ([core_api.py](https://github.com/PowerLoom/pooler/blob/bcc245d228acce504ba803b9b50fd89c8eb05984/snapshotter/core_api.py)) now exposes an internal API endpoint GET /internal/snapshotter/epochProcessingStatus that returns a paginated response of the state transition status of snapshots against project IDs, upto 30 most recent epochs.

Request:

curl -X 'GET' \
  'http://localhost:8002/internal/snapshotter/epochProcessingStatus?page=1&size=10' \
  -H 'accept: application/json'

Sample response:

{
    "items": [
      {
        "epochId": 43523,
        "transitionStatus": {
          "EPOCH_RELEASED": {
            "status": "success",
            "error": null,
            "extra": null,
            "timestamp": 1692530595
          },
          "PRELOAD": {
            "pairContract_pair_total_reserves": {
              "status": "success",
              "error": null,
              "extra": null,
              "timestamp": 1692530595
            },
          },
          "SNAPSHOT_BUILD": {
            "aggregate_24h_stats_lite:35ee1886fa4665255a0d0486c6079c4719c82f0f62ef9e96a98f26fde2e8a106:UNISWAPV2": {
              "status": "success",
              "error": null,
              "extra": null,
              "timestamp": 1692530596
            },
          },
          "SNAPSHOT_SUBMIT_PAYLOAD_COMMIT": {

          },
         "RELAYER_SEND": {

         },
        "SNAPSHOT_FINALIZE": {

        },
      },
    }
   ],
   "total": 30,
   "page": 1,
   "size": 10,
   "pages": 3
}

Change logs

Added

• Core API query endpoint GET /internal/snapshotter/epochProcessingStatus

• Record state transition of a snapshot at every step

• Send health pings to onchain consensus reporting service every 2 seconds

pooler/snapshotter/process_hub_core.py

Lines 266 to 278 in bcc245d

	if settings.reporting.service_url:
	try:
	self._httpx_client.post(
	url=urljoin(settings.reporting.service_url, '/ping'),
	json=SnapshotterPing(instanceID=settings.instance_id).dict(),
	)
	except Exception as e:
	if settings.logs.trace_enabled:
	self._logger.opt(exception=True).error('Error while pinging reporting service: {}', e,)
	else:
	self._logger.error(
	'Error while pinging reporting service: {}', e,
	)

• Send reports on crashed callback workers as well as complete shutdown of running snapshotter instances

pooler/snapshotter/process_hub_core.py

Lines 186 to 199 in bcc245d

	elif signum in [SIGINT, SIGTERM, SIGQUIT]:
	self._shutdown_initiated = True
	if settings.reporting.service_url:
	self._logger.debug('Sending shutdown signal to reporting service')
	send_failure_notifications_sync(
	client=self._httpx_client,
	message=SnapshotterIssue(
	instanceID=settings.instance_id,
	issueType=SnapshotterReportState.SHUTDOWN_INITIATED.value,
	projectID='',
	epochId='',
	timeOfReporting=datetime.now().isoformat(),
	)
	)

pooler/snapshotter/process_hub_core.py

Lines 147 to 166 in bcc245d

	if settings.reporting.service_url:
	send_failure_notifications_sync(
	client=self._httpx_client,
	message=SnapshotterIssue(
	instanceID=settings.instance_id,
	issueType=SnapshotterReportState.CRASHED_CHILD_WORKER.value,
	projectID='',
	epochId='',
	timeOfReporting=datetime.now().isoformat(),
	extra=json.dumps(
	{
	'worker_name': callback_worker_name,
	'pid': pid,
	'worker_class': callback_worker_class,
	'worker_unique_id': callback_worker_unique_id,
	'respawned_pid': worker_obj.pid,
	}
	),
	)
	)

Deployment Instructions

Pull the latest code and restart.

Ensure the issue reporting service URL in config/settings.json is set the base URL of the reporting service entry point of onchain-consensus

pooler/config/settings.example.json

Lines 56 to 59 in bcc245d

	"reporting": {
	"slack_url": "https://slack-reporting-url",
	"service_url": "https://powerloom-reporting-url"
	},

[xadahiya]

Approving since new changes are deployed on one of the Staging instances and have been working without any issues.
Suggested improvements are also incorporated in the latest commits.