Compare commits
9 Commits
main
...
export-mor
Author | SHA1 | Date | |
---|---|---|---|
|
850c32300a | ||
|
c9536d1bde | ||
|
ac376311f4 | ||
|
818a0c7ff4 | ||
|
fa671ae990 | ||
|
686b828332 | ||
|
254e2745ff | ||
|
f13fd9fa39 | ||
|
3f8e3f04af |
4
Tiltfile
4
Tiltfile
@ -107,13 +107,13 @@ local_resource(
|
|||||||
# repo_name="localstack-charts",
|
# repo_name="localstack-charts",
|
||||||
# repo_url="https://localstack.github.io/helm-charts",
|
# repo_url="https://localstack.github.io/helm-charts",
|
||||||
#)
|
#)
|
||||||
|
#
|
||||||
#local_resource(
|
#local_resource(
|
||||||
# 'localstack-port-forward',
|
# 'localstack-port-forward',
|
||||||
# serve_cmd='kubectl port-forward --namespace default svc/localstack 4566:4566',
|
# serve_cmd='kubectl port-forward --namespace default svc/localstack 4566:4566',
|
||||||
# resource_deps=["localstack"]
|
# resource_deps=["localstack"]
|
||||||
#)
|
#)
|
||||||
|
#
|
||||||
#k8s_yaml(configmap_from_dict("mev-inspect-export", inputs = {
|
#k8s_yaml(configmap_from_dict("mev-inspect-export", inputs = {
|
||||||
# "services": "s3",
|
# "services": "s3",
|
||||||
#}))
|
#}))
|
||||||
|
16
cli.py
16
cli.py
@ -159,6 +159,22 @@ def enqueue_s3_export(block_number: int):
|
|||||||
export_actor.send(block_number)
|
export_actor.send(block_number)
|
||||||
|
|
||||||
|
|
||||||
|
@cli.command()
|
||||||
|
@click.argument("after_block", type=int)
|
||||||
|
@click.argument("before_block", type=int)
|
||||||
|
def enqueue_many_s3_exports(after_block: int, before_block: int):
|
||||||
|
broker = connect_broker()
|
||||||
|
export_actor = dramatiq.actor(
|
||||||
|
export_block_task,
|
||||||
|
broker=broker,
|
||||||
|
queue_name=HIGH_PRIORITY_QUEUE,
|
||||||
|
priority=HIGH_PRIORITY,
|
||||||
|
)
|
||||||
|
logger.info(f"Sending blocks {after_block} to {before_block} to queue")
|
||||||
|
for block_number in range(after_block, before_block):
|
||||||
|
export_actor.send(block_number)
|
||||||
|
|
||||||
|
|
||||||
@cli.command()
|
@cli.command()
|
||||||
@click.argument("block_number", type=int)
|
@click.argument("block_number", type=int)
|
||||||
def s3_export(block_number: int):
|
def s3_export(block_number: int):
|
||||||
|
7
mev
7
mev
@ -94,6 +94,13 @@ case "$1" in
|
|||||||
exit 1
|
exit 1
|
||||||
esac
|
esac
|
||||||
;;
|
;;
|
||||||
|
backfill-export)
|
||||||
|
after_block=$2
|
||||||
|
before_block=$3
|
||||||
|
|
||||||
|
echo "Sending $after_block to $before_block export to queue"
|
||||||
|
kubectl exec -ti deploy/mev-inspect -- poetry run enqueue-many-s3-exports $after_block $before_block
|
||||||
|
;;
|
||||||
enqueue-s3-export)
|
enqueue-s3-export)
|
||||||
block_number=$2
|
block_number=$2
|
||||||
|
|
||||||
|
@ -14,23 +14,33 @@ EXPORT_BUCKET_REGION_ENV = "EXPORT_BUCKET_REGION"
|
|||||||
EXPORT_AWS_ACCESS_KEY_ID_ENV = "EXPORT_AWS_ACCESS_KEY_ID"
|
EXPORT_AWS_ACCESS_KEY_ID_ENV = "EXPORT_AWS_ACCESS_KEY_ID"
|
||||||
EXPORT_AWS_SECRET_ACCESS_KEY_ENV = "EXPORT_AWS_SECRET_ACCESS_KEY"
|
EXPORT_AWS_SECRET_ACCESS_KEY_ENV = "EXPORT_AWS_SECRET_ACCESS_KEY"
|
||||||
|
|
||||||
MEV_SUMMARY_EXPORT_QUERY = """
|
supported_tables = [
|
||||||
SELECT to_json(mev_summary)
|
"mev_summary",
|
||||||
FROM mev_summary
|
"arbitrages",
|
||||||
WHERE
|
"liquidations",
|
||||||
block_number = :block_number
|
"sandwiches",
|
||||||
"""
|
"sandwiched_swaps",
|
||||||
|
]
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def export_block(inspect_db_session, block_number: int) -> None:
|
def export_block(inspect_db_session, block_number: int) -> None:
|
||||||
export_bucket_name = get_export_bucket_name()
|
|
||||||
|
for table in supported_tables:
|
||||||
|
|
||||||
|
_export_block_by_table(inspect_db_session, block_number, table)
|
||||||
|
|
||||||
|
|
||||||
|
def _export_block_by_table(inspect_db_session, block_number: int, table: str) -> None:
|
||||||
client = get_s3_client()
|
client = get_s3_client()
|
||||||
object_key = f"mev_summary/flashbots_{block_number}.json"
|
export_bucket_name = get_export_bucket_name()
|
||||||
|
export_statement = _get_export_statement(table)
|
||||||
|
|
||||||
|
object_key = f"{table}/flashbots_{block_number}.json"
|
||||||
|
|
||||||
mev_summary_json_results = inspect_db_session.execute(
|
mev_summary_json_results = inspect_db_session.execute(
|
||||||
statement=MEV_SUMMARY_EXPORT_QUERY,
|
statement=export_statement,
|
||||||
params={
|
params={
|
||||||
"block_number": block_number,
|
"block_number": block_number,
|
||||||
},
|
},
|
||||||
@ -40,7 +50,7 @@ def export_block(inspect_db_session, block_number: int) -> None:
|
|||||||
if first_value is None:
|
if first_value is None:
|
||||||
existing_object_size = _get_object_size(client, export_bucket_name, object_key)
|
existing_object_size = _get_object_size(client, export_bucket_name, object_key)
|
||||||
if existing_object_size is None or existing_object_size == 0:
|
if existing_object_size is None or existing_object_size == 0:
|
||||||
logger.info(f"Skipping block {block_number} - no data")
|
logger.info(f"Skipping {table} for block {block_number} - no data")
|
||||||
return
|
return
|
||||||
|
|
||||||
mev_summary_json_fileobj = BytesIteratorIO(
|
mev_summary_json_fileobj = BytesIteratorIO(
|
||||||
@ -56,6 +66,15 @@ def export_block(inspect_db_session, block_number: int) -> None:
|
|||||||
logger.info(f"Exported to {object_key}")
|
logger.info(f"Exported to {object_key}")
|
||||||
|
|
||||||
|
|
||||||
|
def _get_export_statement(table: str) -> str:
|
||||||
|
return f"""
|
||||||
|
SELECT to_json({table})
|
||||||
|
FROM {table}
|
||||||
|
WHERE
|
||||||
|
block_number = :block_number
|
||||||
|
"""
|
||||||
|
|
||||||
|
|
||||||
def _get_object_size(client, bucket: str, key: str) -> Optional[int]:
|
def _get_object_size(client, bucket: str, key: str) -> Optional[int]:
|
||||||
response = client.list_objects_v2(
|
response = client.list_objects_v2(
|
||||||
Bucket=bucket,
|
Bucket=bucket,
|
||||||
|
@ -43,6 +43,7 @@ fetch-all-prices = 'cli:fetch_all_prices'
|
|||||||
fetch-range = 'cli:fetch_range'
|
fetch-range = 'cli:fetch_range'
|
||||||
s3-export = 'cli:s3_export'
|
s3-export = 'cli:s3_export'
|
||||||
enqueue-s3-export = 'cli:enqueue_s3_export'
|
enqueue-s3-export = 'cli:enqueue_s3_export'
|
||||||
|
enqueue-many-s3-exports = 'cli:enqueue_many_s3_exports'
|
||||||
|
|
||||||
[tool.black]
|
[tool.black]
|
||||||
exclude = '''
|
exclude = '''
|
||||||
|
10
worker.py
10
worker.py
@ -11,7 +11,9 @@ from mev_inspect.queue.middleware import (
|
|||||||
InspectorMiddleware,
|
InspectorMiddleware,
|
||||||
)
|
)
|
||||||
from mev_inspect.queue.tasks import (
|
from mev_inspect.queue.tasks import (
|
||||||
|
HIGH_PRIORITY,
|
||||||
HIGH_PRIORITY_QUEUE,
|
HIGH_PRIORITY_QUEUE,
|
||||||
|
LOW_PRIORITY,
|
||||||
LOW_PRIORITY_QUEUE,
|
LOW_PRIORITY_QUEUE,
|
||||||
export_block_task,
|
export_block_task,
|
||||||
inspect_many_blocks_task,
|
inspect_many_blocks_task,
|
||||||
@ -25,5 +27,9 @@ broker.add_middleware(AsyncMiddleware())
|
|||||||
broker.add_middleware(InspectorMiddleware(os.environ["RPC_URL"]))
|
broker.add_middleware(InspectorMiddleware(os.environ["RPC_URL"]))
|
||||||
dramatiq.set_broker(broker)
|
dramatiq.set_broker(broker)
|
||||||
|
|
||||||
dramatiq.actor(inspect_many_blocks_task, queue_name=HIGH_PRIORITY_QUEUE)
|
dramatiq.actor(
|
||||||
dramatiq.actor(export_block_task, queue_name=LOW_PRIORITY_QUEUE)
|
inspect_many_blocks_task, queue_name=LOW_PRIORITY_QUEUE, priority=LOW_PRIORITY
|
||||||
|
)
|
||||||
|
dramatiq.actor(
|
||||||
|
export_block_task, queue_name=HIGH_PRIORITY_QUEUE, priority=HIGH_PRIORITY
|
||||||
|
)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user