From 6ac4d07004c94cf73134580e7d104e514b17bf82 Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 27 Feb 2017 18:40:33 +0100 Subject: allow writing web-web queries --- .gitignore | 91 +++++++++++++++++++++++++++++++++++++++++++++++ .gitmodules | 3 ++ api.py | 102 +++++++++++++++++++++++++++++++++++++++++++++++++++++ joblib | 1 + queries/player.sql | 9 +++++ 5 files changed, 206 insertions(+) create mode 100644 .gitignore create mode 100644 .gitmodules create mode 100644 api.py create mode 160000 joblib create mode 100644 queries/player.sql diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9a05e2d --- /dev/null +++ b/.gitignore @@ -0,0 +1,91 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +env/ +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*,cover +.hypothesis/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# pyenv +.python-version + +# celery beat schedule file +celerybeat-schedule + +# dotenv +.env + +# virtualenv +.venv/ +venv/ +ENV/ + +# Spyder project settings +.spyderproject + +# Rope project settings +.ropeproject diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000..52948d0 --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "joblib"] + path = joblib + url = https://github.com/vainglorygame/joblib diff --git a/api.py b/api.py new file mode 100644 index 0000000..6af8c42 --- /dev/null +++ b/api.py @@ -0,0 +1,102 @@ +#!/usr/bin/python3 + +import asyncio +import os +import glob +import logging +import json +import asyncpg + +import joblib.joblib + +queue_db = { + "host": os.environ.get("POSTGRESQL_SOURCE_HOST") or "vaindock_postgres_raw", + "port": os.environ.get("POSTGRESQL_SOURCE_PORT") or 5532, + "user": os.environ.get("POSTGRESQL_SOURCE_USER") or "vainraw", + "password": os.environ.get("POSTGRESQL_SOURCE_PASSWORD") or "vainraw", + "database": os.environ.get("POSTGRESQL_SOURCE_DB") or "vainsocial-raw" +} + +db_config = { + "host": os.environ.get("POSTGRESQL_DEST_HOST") or "vaindock_postgres_web", + "port": os.environ.get("POSTGRESQL_DEST_PORT") or 5432, + "user": os.environ.get("POSTGRESQL_DEST_USER") or "vainweb", + "password": os.environ.get("POSTGRESQL_DEST_PASSWORD") or "vainweb", + "database": os.environ.get("POSTGRESQL_DEST_DB") or "vainsocial-web" +} + + +class Worker(object): + def __init__(self): + self._queue = None + self._pool = None + self._queries = {} + + async def connect(self, dbconf, queuedb): + """Connect to database.""" + logging.info("connecting to database") + self._queue = joblib.joblib.JobQueue() + await self._queue.connect(**queuedb) + await self._queue.setup() + self._pool = await asyncpg.create_pool(**dbconf) + + async def setup(self): + """Initialize the database.""" + scriptroot = os.path.realpath( + os.path.join(os.getcwd(), os.path.dirname(__file__))) + for path in glob.glob(scriptroot + "/queries/*.sql"): + # utf-8-sig is used by pgadmin, doesn't hurt to specify + # file names: web target table + table = os.path.splitext(os.path.basename(path))[0] + with open(path, "r", encoding="utf-8-sig") as file: + self._queries[table] = file.read() + logging.info("loaded query '%s'", table) + + async def _execute_job(self, jobid, payload): + """Finish a job.""" + object_id = payload["id"] + table = payload["type"] + if table not in self._queries: + logging.debug("%s: nothing to do", jobid) + return + async with self._pool.acquire() as con: + logging.debug("%s: compiling '%s' from '%s'", + jobid, object_id, table) + async with con.transaction(): + await con.execute(self._queries[table], + object_id) + + async def _work(self): + """Fetch a job and run it.""" + jobid, payload, _ = await self._queue.acquire(jobtype="compile") + if jobid is None: + raise LookupError("no jobs available") + logging.debug("%s: starting job", jobid) + await self._execute_job(jobid, payload) + await self._queue.finish(jobid) + logging.debug("%s: finished job", jobid) + + async def run(self): + """Start jobs forever.""" + while True: + try: + await self._work() + except LookupError: + logging.info("nothing to do, idling") + await asyncio.sleep(10) + + async def start(self, number=1): + """Start jobs in background.""" + for _ in range(number): + asyncio.ensure_future(self.run()) + +async def startup(): + worker = Worker() + await worker.connect(db_config, queue_db) + await worker.setup() + await worker.start(10) + +logging.basicConfig(level=logging.DEBUG) +loop = asyncio.get_event_loop() +loop.run_until_complete(startup()) +loop.run_forever() diff --git a/joblib b/joblib new file mode 160000 index 0000000..31d0588 --- /dev/null +++ b/joblib @@ -0,0 +1 @@ +Subproject commit 31d0588a35c8df17a5d3adcfc25af2b72c896cc7 diff --git a/queries/player.sql b/queries/player.sql new file mode 100644 index 0000000..e847de2 --- /dev/null +++ b/queries/player.sql @@ -0,0 +1,9 @@ +-- count the number of matches we have from that player in the database +-- UPDATE player SET "games_in_db"= +SELECT +COUNT(DISTINCT match."apiId") AS "games_in_db" +FROM player +JOIN participant ON player."apiId" = participant."player_apiId" +JOIN roster ON participant."roster_apiId" = roster."apiId" +JOIN match ON roster."match_apiId" = match."apiId" +WHERE player.id=$1 -- cgit v1.3.1 From 2e0da58ea9e826bea3d6a966400cbd7a7e09236f Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 27 Feb 2017 18:41:24 +0100 Subject: add requirements.txt --- requirements.txt | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 requirements.txt diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..2e1ea39 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +appdirs==1.4.2 +asyncpg==0.8.4 +packaging==16.8 +pyparsing==2.1.10 +six==1.10.0 -- cgit v1.3.1 From 7d34e23ecda9c1bdb5f822d90142d51cf81c0681 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 28 Feb 2017 17:56:22 +0100 Subject: logs: be more quiet --- api.py | 4 ---- 1 file changed, 4 deletions(-) diff --git a/api.py b/api.py index 6af8c42..f668b74 100644 --- a/api.py +++ b/api.py @@ -57,7 +57,6 @@ class Worker(object): object_id = payload["id"] table = payload["type"] if table not in self._queries: - logging.debug("%s: nothing to do", jobid) return async with self._pool.acquire() as con: logging.debug("%s: compiling '%s' from '%s'", @@ -71,10 +70,8 @@ class Worker(object): jobid, payload, _ = await self._queue.acquire(jobtype="compile") if jobid is None: raise LookupError("no jobs available") - logging.debug("%s: starting job", jobid) await self._execute_job(jobid, payload) await self._queue.finish(jobid) - logging.debug("%s: finished job", jobid) async def run(self): """Start jobs forever.""" @@ -82,7 +79,6 @@ class Worker(object): try: await self._work() except LookupError: - logging.info("nothing to do, idling") await asyncio.sleep(10) async def start(self, number=1): -- cgit v1.3.1 From c1f3ee04077420be4ac016d9fc8af9f1ccc8d787 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 28 Feb 2017 18:01:50 +0100 Subject: support split queries --- api.py | 17 ++++++++++------- queries/player.sql | 9 --------- 2 files changed, 10 insertions(+), 16 deletions(-) delete mode 100644 queries/player.sql diff --git a/api.py b/api.py index f668b74..ce186d4 100644 --- a/api.py +++ b/api.py @@ -44,12 +44,15 @@ class Worker(object): """Initialize the database.""" scriptroot = os.path.realpath( os.path.join(os.getcwd(), os.path.dirname(__file__))) - for path in glob.glob(scriptroot + "/queries/*.sql"): + for path in glob.glob(scriptroot + "/queries/*/*.sql"): # utf-8-sig is used by pgadmin, doesn't hurt to specify - # file names: web target table - table = os.path.splitext(os.path.basename(path))[0] + # directory names: web target table + table = os.path.basename(os.path.dirname(path)) with open(path, "r", encoding="utf-8-sig") as file: - self._queries[table] = file.read() + try: + self._queries[table].append(file.read()) + except KeyError: + self._queries[table] = [file.read()] logging.info("loaded query '%s'", table) async def _execute_job(self, jobid, payload): @@ -61,9 +64,9 @@ class Worker(object): async with self._pool.acquire() as con: logging.debug("%s: compiling '%s' from '%s'", jobid, object_id, table) - async with con.transaction(): - await con.execute(self._queries[table], - object_id) + for query in self._queries[table]: + async with con.transaction(): + await con.execute(query, object_id) async def _work(self): """Fetch a job and run it.""" diff --git a/queries/player.sql b/queries/player.sql deleted file mode 100644 index e847de2..0000000 --- a/queries/player.sql +++ /dev/null @@ -1,9 +0,0 @@ --- count the number of matches we have from that player in the database --- UPDATE player SET "games_in_db"= -SELECT -COUNT(DISTINCT match."apiId") AS "games_in_db" -FROM player -JOIN participant ON player."apiId" = participant."player_apiId" -JOIN roster ON participant."roster_apiId" = roster."apiId" -JOIN match ON roster."match_apiId" = match."apiId" -WHERE player.id=$1 -- cgit v1.3.1 From cfa29e50a193137ec7f47fa82429035dfa4a81e7 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 28 Feb 2017 18:25:34 +0100 Subject: reduce default worker idle cooldown --- api.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/api.py b/api.py index ce186d4..24080b2 100644 --- a/api.py +++ b/api.py @@ -82,7 +82,7 @@ class Worker(object): try: await self._work() except LookupError: - await asyncio.sleep(10) + await asyncio.sleep(1) async def start(self, number=1): """Start jobs in background.""" @@ -93,7 +93,7 @@ async def startup(): worker = Worker() await worker.connect(db_config, queue_db) await worker.setup() - await worker.start(10) + await worker.start(4) logging.basicConfig(level=logging.DEBUG) loop = asyncio.get_event_loop() -- cgit v1.3.1 From 4a6f14706e5b837c1ffeaafeaceac2f9971d1420 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 28 Feb 2017 19:06:44 +0100 Subject: add untested example query --- queries/player/count_player_matches.sql | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) create mode 100644 queries/player/count_player_matches.sql diff --git a/queries/player/count_player_matches.sql b/queries/player/count_player_matches.sql new file mode 100644 index 0000000..f7bd29f --- /dev/null +++ b/queries/player/count_player_matches.sql @@ -0,0 +1,16 @@ +-- UNTESTED! +WITH plr AS ( + SELECT + "id", + "apiId", + -- count matches in db + COUNT(DISTINCT participant."apiId") AS "games_in_db" + FROM player + JOIN participant ON player."apiId" = participant."player_apiId" + JOIN roster ON participant."roster_apiId" = roster."apiId" + WHERE player."apiId"=$1 +) +INSERT INTO player_stats("id", "apiId", "games_in_db") + SELECT * FROM plr +ON CONFLICT("apiId") DO +UPDATE SET "games_in_db"=plr."games_in_db" -- cgit v1.3.1 From 90d5f8edb1d5a6cb95a18c48eca98e93cf4d34b7 Mon Sep 17 00:00:00 2001 From: Kapil Viren Ahuja Date: Wed, 1 Mar 2017 20:07:31 +0530 Subject: Some stats calc - we need to add patch version in some core web tables. --- queries/player/count_player_matches.sql | 39 ++++++++++++++++++++------------- 1 file changed, 24 insertions(+), 15 deletions(-) diff --git a/queries/player/count_player_matches.sql b/queries/player/count_player_matches.sql index f7bd29f..2bb549d 100644 --- a/queries/player/count_player_matches.sql +++ b/queries/player/count_player_matches.sql @@ -1,16 +1,25 @@ --- UNTESTED! -WITH plr AS ( - SELECT - "id", - "apiId", - -- count matches in db - COUNT(DISTINCT participant."apiId") AS "games_in_db" - FROM player - JOIN participant ON player."apiId" = participant."player_apiId" - JOIN roster ON participant."roster_apiId" = roster."apiId" - WHERE player."apiId"=$1 +with stats as ( +with plr as ( +select + 'a' as patch_version, + player."apiId" as player_api_id, + match."gameMode" as game_mode, + participant."winner" as winner +from + match, roster, participant, player +where + match."apiId" = roster."match_apiId" and + roster."apiId" = participant."roster_apiId" and + participant."player_apiId" = player."apiId" and + player."apiId" = '6eab29f6-eed5-11e6-ba5f-068789513eb5') +select + (select distinct(player_api_id) from plr), + (select count(*) as played from plr), + (select count(*) as wins from plr where winner = true), + (select count(*) as casual_played from plr where game_mode like '%casual%'), + (select count(*) as casual_wins from plr where game_mode like '%casual%' and winner = true), + (select count(*) as ranked_played from plr where game_mode like '%ranked%'), + (select count(*) as ranked_wins from plr where game_mode like '%ranked%' and winner = true) ) -INSERT INTO player_stats("id", "apiId", "games_in_db") - SELECT * FROM plr -ON CONFLICT("apiId") DO -UPDATE SET "games_in_db"=plr."games_in_db" +INSERT INTO player_stats("player_apiId", played, wins, casual_played, casual_wins, ranked_played, ranked_wins) + SELECT * FROM stats -- cgit v1.3.1 From bd4be8d1e87231b496f91fddbc7230731051d71f Mon Sep 17 00:00:00 2001 From: Kapil Viren Ahuja Date: Wed, 1 Mar 2017 20:10:17 +0530 Subject: fixed player id as parameter --- queries/player/count_player_matches.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/queries/player/count_player_matches.sql b/queries/player/count_player_matches.sql index 2bb549d..ddfbb66 100644 --- a/queries/player/count_player_matches.sql +++ b/queries/player/count_player_matches.sql @@ -11,7 +11,7 @@ where match."apiId" = roster."match_apiId" and roster."apiId" = participant."roster_apiId" and participant."player_apiId" = player."apiId" and - player."apiId" = '6eab29f6-eed5-11e6-ba5f-068789513eb5') + player."apiId" = $1) select (select distinct(player_api_id) from plr), (select count(*) as played from plr), -- cgit v1.3.1 From 5a79b51666eb48ddd5d2b1a8a0b5e42d4383fe16 Mon Sep 17 00:00:00 2001 From: schneefux Date: Wed, 1 Mar 2017 20:13:52 +0100 Subject: log to file --- api.py | 13 +++++++++++-- logs/.gitignore | 1 + 2 files changed, 12 insertions(+), 2 deletions(-) create mode 100644 logs/.gitignore diff --git a/api.py b/api.py index 24080b2..c0e8568 100644 --- a/api.py +++ b/api.py @@ -34,7 +34,7 @@ class Worker(object): async def connect(self, dbconf, queuedb): """Connect to database.""" - logging.info("connecting to database") + logging.warning("connecting to database") self._queue = joblib.joblib.JobQueue() await self._queue.connect(**queuedb) await self._queue.setup() @@ -95,7 +95,16 @@ async def startup(): await worker.setup() await worker.start(4) -logging.basicConfig(level=logging.DEBUG) + +logging.basicConfig( + filename="logs/apigrabber.log", + filemode="a", + level=logging.DEBUG +) +console = logging.StreamHandler() +console.setLevel(logging.WARNING) +logging.getLogger("").addHandler(console) + loop = asyncio.get_event_loop() loop.run_until_complete(startup()) loop.run_forever() diff --git a/logs/.gitignore b/logs/.gitignore new file mode 100644 index 0000000..397b4a7 --- /dev/null +++ b/logs/.gitignore @@ -0,0 +1 @@ +*.log -- cgit v1.3.1 From a5a5ece6c7d4f220853a156e0d5182e123728ecc Mon Sep 17 00:00:00 2001 From: schneefux Date: Wed, 1 Mar 2017 21:25:15 +0100 Subject: use proper log paths for Docker --- api.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/api.py b/api.py index c0e8568..d6289cc 100644 --- a/api.py +++ b/api.py @@ -97,7 +97,10 @@ async def startup(): logging.basicConfig( - filename="logs/apigrabber.log", + filename=os.path.realpath( + os.path.join(os.getcwd(), + os.path.dirname(__file__))) + + "/logs/compiler.log", filemode="a", level=logging.DEBUG ) -- cgit v1.3.1 From 65397e311f6878a0380a0ad2e2f42c20c06890e2 Mon Sep 17 00:00:00 2001 From: Kapil Viren Ahuja Date: Thu, 2 Mar 2017 21:20:12 +0530 Subject: Some stats calc - we need to add patch version in some core web tables. (#5) * Some stats calc - we need to add patch version in some core web tables. * fixed player id as parameter --- queries/player/count_player_matches.sql | 39 ++++++++++++++++++++------------- 1 file changed, 24 insertions(+), 15 deletions(-) diff --git a/queries/player/count_player_matches.sql b/queries/player/count_player_matches.sql index f7bd29f..ddfbb66 100644 --- a/queries/player/count_player_matches.sql +++ b/queries/player/count_player_matches.sql @@ -1,16 +1,25 @@ --- UNTESTED! -WITH plr AS ( - SELECT - "id", - "apiId", - -- count matches in db - COUNT(DISTINCT participant."apiId") AS "games_in_db" - FROM player - JOIN participant ON player."apiId" = participant."player_apiId" - JOIN roster ON participant."roster_apiId" = roster."apiId" - WHERE player."apiId"=$1 +with stats as ( +with plr as ( +select + 'a' as patch_version, + player."apiId" as player_api_id, + match."gameMode" as game_mode, + participant."winner" as winner +from + match, roster, participant, player +where + match."apiId" = roster."match_apiId" and + roster."apiId" = participant."roster_apiId" and + participant."player_apiId" = player."apiId" and + player."apiId" = $1) +select + (select distinct(player_api_id) from plr), + (select count(*) as played from plr), + (select count(*) as wins from plr where winner = true), + (select count(*) as casual_played from plr where game_mode like '%casual%'), + (select count(*) as casual_wins from plr where game_mode like '%casual%' and winner = true), + (select count(*) as ranked_played from plr where game_mode like '%ranked%'), + (select count(*) as ranked_wins from plr where game_mode like '%ranked%' and winner = true) ) -INSERT INTO player_stats("id", "apiId", "games_in_db") - SELECT * FROM plr -ON CONFLICT("apiId") DO -UPDATE SET "games_in_db"=plr."games_in_db" +INSERT INTO player_stats("player_apiId", played, wins, casual_played, casual_wins, ranked_played, ranked_wins) + SELECT * FROM stats -- cgit v1.3.1 From a547f80e220921be7b7c8c5068838cc278b630f5 Mon Sep 17 00:00:00 2001 From: schneefux Date: Thu, 2 Mar 2017 18:36:32 +0100 Subject: process missing stats --- queries/player/count_player_matches.sql | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/queries/player/count_player_matches.sql b/queries/player/count_player_matches.sql index ddfbb66..db5a2de 100644 --- a/queries/player/count_player_matches.sql +++ b/queries/player/count_player_matches.sql @@ -1,7 +1,7 @@ with stats as ( with plr as ( select - 'a' as patch_version, + 'a'::text as "patchVersion", player."apiId" as player_api_id, match."gameMode" as game_mode, participant."winner" as winner @@ -14,12 +14,14 @@ where player."apiId" = $1) select (select distinct(player_api_id) from plr), + (select distinct("patchVersion") as "patchVersion" from plr), (select count(*) as played from plr), (select count(*) as wins from plr where winner = true), (select count(*) as casual_played from plr where game_mode like '%casual%'), (select count(*) as casual_wins from plr where game_mode like '%casual%' and winner = true), (select count(*) as ranked_played from plr where game_mode like '%ranked%'), - (select count(*) as ranked_wins from plr where game_mode like '%ranked%' and winner = true) + (select count(*) as ranked_wins from plr where game_mode like '%ranked%' and winner = true), + (select 0 as streak) ) -INSERT INTO player_stats("player_apiId", played, wins, casual_played, casual_wins, ranked_played, ranked_wins) +INSERT INTO player_stats("player_apiId", "patchVersion", played, wins, casual_played, casual_wins, ranked_played, ranked_wins, streak) SELECT * FROM stats -- cgit v1.3.1 From 814e2ca1d24b91364744f2564c37b2847313acbb Mon Sep 17 00:00:00 2001 From: schneefux Date: Thu, 2 Mar 2017 18:36:42 +0100 Subject: reduce number of workers to 1 to workaround deadlock issues --- api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api.py b/api.py index d6289cc..e7dbfdd 100644 --- a/api.py +++ b/api.py @@ -93,7 +93,7 @@ async def startup(): worker = Worker() await worker.connect(db_config, queue_db) await worker.setup() - await worker.start(4) + await worker.start(1) logging.basicConfig( -- cgit v1.3.1 From 714f26894a1a0b048c9f5a43315ac067deb9f93c Mon Sep 17 00:00:00 2001 From: schneefux Date: Sat, 4 Mar 2017 11:33:28 +0100 Subject: refactor to use joblib worker class --- api.py | 34 ++++++---------------------------- joblib | 2 +- 2 files changed, 7 insertions(+), 29 deletions(-) diff --git a/api.py b/api.py index e7dbfdd..4b3c890 100644 --- a/api.py +++ b/api.py @@ -7,7 +7,7 @@ import logging import json import asyncpg -import joblib.joblib +import joblib.worker queue_db = { "host": os.environ.get("POSTGRESQL_SOURCE_HOST") or "vaindock_postgres_raw", @@ -26,18 +26,16 @@ db_config = { } -class Worker(object): +class Compiler(joblib.worker.Worker): def __init__(self): - self._queue = None self._pool = None self._queries = {} + super().__init__(jobtype="compile") async def connect(self, dbconf, queuedb): """Connect to database.""" logging.warning("connecting to database") - self._queue = joblib.joblib.JobQueue() - await self._queue.connect(**queuedb) - await self._queue.setup() + await super().connect(**queuedb) self._pool = await asyncpg.create_pool(**dbconf) async def setup(self): @@ -55,7 +53,7 @@ class Worker(object): self._queries[table] = [file.read()] logging.info("loaded query '%s'", table) - async def _execute_job(self, jobid, payload): + async def _execute_job(self, jobid, payload, priority): """Finish a job.""" object_id = payload["id"] table = payload["type"] @@ -68,29 +66,9 @@ class Worker(object): async with con.transaction(): await con.execute(query, object_id) - async def _work(self): - """Fetch a job and run it.""" - jobid, payload, _ = await self._queue.acquire(jobtype="compile") - if jobid is None: - raise LookupError("no jobs available") - await self._execute_job(jobid, payload) - await self._queue.finish(jobid) - - async def run(self): - """Start jobs forever.""" - while True: - try: - await self._work() - except LookupError: - await asyncio.sleep(1) - - async def start(self, number=1): - """Start jobs in background.""" - for _ in range(number): - asyncio.ensure_future(self.run()) async def startup(): - worker = Worker() + worker = Compiler() await worker.connect(db_config, queue_db) await worker.setup() await worker.start(1) diff --git a/joblib b/joblib index 31d0588..8d58a4f 160000 --- a/joblib +++ b/joblib @@ -1 +1 @@ -Subproject commit 31d0588a35c8df17a5d3adcfc25af2b72c896cc7 +Subproject commit 8d58a4f5f754b6541af84478a2df3fba70db94be -- cgit v1.3.1 From d6d94e86c7b44c99dd6974453ba27637e8fb7196 Mon Sep 17 00:00:00 2001 From: schneefux Date: Sat, 4 Mar 2017 11:33:46 +0100 Subject: update queries after column name case change --- queries/player/count_player_matches.sql | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/queries/player/count_player_matches.sql b/queries/player/count_player_matches.sql index db5a2de..fbdd283 100644 --- a/queries/player/count_player_matches.sql +++ b/queries/player/count_player_matches.sql @@ -1,20 +1,20 @@ with stats as ( with plr as ( select - 'a'::text as "patchVersion", - player."apiId" as player_api_id, - match."gameMode" as game_mode, - participant."winner" as winner + match.patch_version as patch_version, + player.api_id as player_api_id, + match.game_mode as game_mode, + participant.winner as winner from match, roster, participant, player where - match."apiId" = roster."match_apiId" and - roster."apiId" = participant."roster_apiId" and - participant."player_apiId" = player."apiId" and - player."apiId" = $1) + match.api_id = roster.match_api_id and + roster.api_id = participant.roster_api_id and + participant.player_api_id = player.api_id and + player.api_id = $1) select (select distinct(player_api_id) from plr), - (select distinct("patchVersion") as "patchVersion" from plr), + (select distinct(patch_version) from plr), (select count(*) as played from plr), (select count(*) as wins from plr where winner = true), (select count(*) as casual_played from plr where game_mode like '%casual%'), @@ -23,5 +23,5 @@ select (select count(*) as ranked_wins from plr where game_mode like '%ranked%' and winner = true), (select 0 as streak) ) -INSERT INTO player_stats("player_apiId", "patchVersion", played, wins, casual_played, casual_wins, ranked_played, ranked_wins, streak) +INSERT INTO player_stats(player_api_id, patch_version, played, wins, casual_played, casual_wins, ranked_played, ranked_wins, streak) SELECT * FROM stats -- cgit v1.3.1 From 259dc118f12083fb3040bf8407e903e02579fe57 Mon Sep 17 00:00:00 2001 From: schneefux Date: Sun, 5 Mar 2017 10:36:28 +0100 Subject: run jobs in batches (#7) * run jobs in batches * upsert player_stats --- api.py | 35 +++++++++++++++++++++++---------- joblib | 2 +- queries/player/count_player_matches.sql | 2 ++ 3 files changed, 28 insertions(+), 11 deletions(-) diff --git a/api.py b/api.py index 4b3c890..c88a39a 100644 --- a/api.py +++ b/api.py @@ -53,25 +53,40 @@ class Compiler(joblib.worker.Worker): self._queries[table] = [file.read()] logging.info("loaded query '%s'", table) + + async def _windup(self): + self._con = await self._pool.acquire() + self._tr = self._con.transaction() + await self._tr.start() + + async def _teardown(self, failed): + if failed: + await self._tr.rollback() + else: + await self._tr.commit() + await self._pool.release(self._con) + async def _execute_job(self, jobid, payload, priority): """Finish a job.""" object_id = payload["id"] table = payload["type"] if table not in self._queries: return - async with self._pool.acquire() as con: - logging.debug("%s: compiling '%s' from '%s'", - jobid, object_id, table) - for query in self._queries[table]: - async with con.transaction(): - await con.execute(query, object_id) + logging.debug("%s: compiling '%s' from '%s'", + jobid, object_id, table) + tasks = [] + for query in self._queries[table]: + tasks.append(asyncio.ensure_future( + self._con.execute(query, object_id))) + await asyncio.gather(*tasks) async def startup(): - worker = Compiler() - await worker.connect(db_config, queue_db) - await worker.setup() - await worker.start(1) + for _ in range(1): + worker = Compiler() + await worker.connect(db_config, queue_db) + await worker.setup() + await worker.start(batchlimit=20) logging.basicConfig( diff --git a/joblib b/joblib index 8d58a4f..7799048 160000 --- a/joblib +++ b/joblib @@ -1 +1 @@ -Subproject commit 8d58a4f5f754b6541af84478a2df3fba70db94be +Subproject commit 7799048a5927670b9492f80cdbbf2fd8d319c506 diff --git a/queries/player/count_player_matches.sql b/queries/player/count_player_matches.sql index fbdd283..226cacc 100644 --- a/queries/player/count_player_matches.sql +++ b/queries/player/count_player_matches.sql @@ -25,3 +25,5 @@ select ) INSERT INTO player_stats(player_api_id, patch_version, played, wins, casual_played, casual_wins, ranked_played, ranked_wins, streak) SELECT * FROM stats +ON CONFLICT(player_api_id) DO + UPDATE SET (player_api_id, patch_version, played, wins, casual_played, casual_wins, ranked_played, ranked_wins, streak) = (SELECT * FROM stats) -- cgit v1.3.1 From ec76ec27675f6dc7640003a42f6436b66ebd9898 Mon Sep 17 00:00:00 2001 From: schneefux Date: Sun, 5 Mar 2017 10:41:17 +0100 Subject: update joblib --- joblib | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/joblib b/joblib index 7799048..3229571 160000 --- a/joblib +++ b/joblib @@ -1 +1 @@ -Subproject commit 7799048a5927670b9492f80cdbbf2fd8d319c506 +Subproject commit 3229571141b638d9d5d6985e98b7e61642229b77 -- cgit v1.3.1 From 1d57e2814a6011d6f2fa4d12c12c00b151392a69 Mon Sep 17 00:00:00 2001 From: schneefux Date: Sun, 5 Mar 2017 21:29:23 +0100 Subject: update joblib --- joblib | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/joblib b/joblib index 3229571..4a8262f 160000 --- a/joblib +++ b/joblib @@ -1 +1 @@ -Subproject commit 3229571141b638d9d5d6985e98b7e61642229b77 +Subproject commit 4a8262f5cd9d319817ae7bf7ee69fc587808b0ad -- cgit v1.3.1 From 7832197c4a4259883d728ad2259129bcc5af0ded Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 6 Mar 2017 21:39:01 +0100 Subject: update joblib --- joblib | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/joblib b/joblib index 4a8262f..e6f3ff1 160000 --- a/joblib +++ b/joblib @@ -1 +1 @@ -Subproject commit 4a8262f5cd9d319817ae7bf7ee69fc587808b0ad +Subproject commit e6f3ff106d4ef3570b61748993f712af3a5227b2 -- cgit v1.3.1 From 878ee48a5a60e1f197d38912eb3fa3732fc08a64 Mon Sep 17 00:00:00 2001 From: schneefux Date: Wed, 8 Mar 2017 18:21:41 +0100 Subject: do not use pooling, not needed --- api.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/api.py b/api.py index c88a39a..f75c9a4 100644 --- a/api.py +++ b/api.py @@ -28,7 +28,7 @@ db_config = { class Compiler(joblib.worker.Worker): def __init__(self): - self._pool = None + self._con = None self._queries = {} super().__init__(jobtype="compile") @@ -36,7 +36,7 @@ class Compiler(joblib.worker.Worker): """Connect to database.""" logging.warning("connecting to database") await super().connect(**queuedb) - self._pool = await asyncpg.create_pool(**dbconf) + self._con = await asyncpg.connect(**dbconf) async def setup(self): """Initialize the database.""" @@ -55,7 +55,6 @@ class Compiler(joblib.worker.Worker): async def _windup(self): - self._con = await self._pool.acquire() self._tr = self._con.transaction() await self._tr.start() @@ -64,7 +63,6 @@ class Compiler(joblib.worker.Worker): await self._tr.rollback() else: await self._tr.commit() - await self._pool.release(self._con) async def _execute_job(self, jobid, payload, priority): """Finish a job.""" -- cgit v1.3.1 From dd2cb06dff4e7b60a8a1db1c4f83e259aac3657c Mon Sep 17 00:00:00 2001 From: schneefux Date: Wed, 8 Mar 2017 19:42:28 +0100 Subject: compile up to a whole batch of 50 at once --- api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api.py b/api.py index f75c9a4..873e8be 100644 --- a/api.py +++ b/api.py @@ -84,7 +84,7 @@ async def startup(): worker = Compiler() await worker.connect(db_config, queue_db) await worker.setup() - await worker.start(batchlimit=20) + await worker.start(batchlimit=750) logging.basicConfig( -- cgit v1.3.1 From aaf64697787d13bdd85ce68a676f95c8f6c22a3b Mon Sep 17 00:00:00 2001 From: schneefux Date: Wed, 8 Mar 2017 19:55:13 +0100 Subject: update joblib --- joblib | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/joblib b/joblib index e6f3ff1..7fa00a9 160000 --- a/joblib +++ b/joblib @@ -1 +1 @@ -Subproject commit e6f3ff106d4ef3570b61748993f712af3a5227b2 +Subproject commit 7fa00a9e4f2723f7d93fc1ecc8369ccbbc9e4f76 -- cgit v1.3.1 From 0c62e238a8a572623468ff50d24b5c807319a760 Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 13 Mar 2017 21:14:02 +0100 Subject: use joblib notifications --- joblib | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/joblib b/joblib index 7fa00a9..f0bb72c 160000 --- a/joblib +++ b/joblib @@ -1 +1 @@ -Subproject commit 7fa00a9e4f2723f7d93fc1ecc8369ccbbc9e4f76 +Subproject commit f0bb72ce9b48c2436b3a7052637fd8bdece6a705 -- cgit v1.3.1 From e2d4663729273477817533537320ca856e2400c3 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 14 Mar 2017 18:46:06 +0100 Subject: update joblib --- joblib | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/joblib b/joblib index f0bb72c..7ced841 160000 --- a/joblib +++ b/joblib @@ -1 +1 @@ -Subproject commit f0bb72ce9b48c2436b3a7052637fd8bdece6a705 +Subproject commit 7ced841aa44a7dd7bd2d3a68ceb989ef488ce441 -- cgit v1.3.1 From 0a3421d87d7e55707062f1186cf64f2c54734444 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 14 Mar 2017 18:47:32 +0100 Subject: add participant_stats query (wip) --- queries/participant/stats.sql | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 queries/participant/stats.sql diff --git a/queries/participant/stats.sql b/queries/participant/stats.sql new file mode 100644 index 0000000..05a9697 --- /dev/null +++ b/queries/participant/stats.sql @@ -0,0 +1,13 @@ +WITH stats AS ( + SELECT + api_id, + CASE WHEN roster.hero_kills=0 + THEN 0 + ELSE (kills+assists)/roster.hero_kills::float + END AS kill_participation + FROM participant JOIN roster ON roster.api_id=participant.roster_api_id +) +INSERT INTO participant_stats(participant_api_id, kill_participation) +SELECT * FROM stats +ON CONFLICT(participant_api_id) DO +UPDATE SET(participant_api_id, kill_participation) = (SELECT * FROM stats) -- cgit v1.3.1 From d18b4c4a5f7f8a2d1505bd03c91a3c2c989dadad Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 14 Mar 2017 20:48:43 +0100 Subject: fix participant_stats query --- queries/participant/stats.sql | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/queries/participant/stats.sql b/queries/participant/stats.sql index 05a9697..6a6d540 100644 --- a/queries/participant/stats.sql +++ b/queries/participant/stats.sql @@ -1,13 +1,17 @@ WITH stats AS ( SELECT - api_id, + participant.api_id, + match.patch_version, CASE WHEN roster.hero_kills=0 THEN 0 ELSE (kills+assists)/roster.hero_kills::float - END AS kill_participation - FROM participant JOIN roster ON roster.api_id=participant.roster_api_id + END AS kills_participation + FROM participant + JOIN roster ON roster.api_id=participant.roster_api_id + JOIN match ON match.api_id=roster.match_api_id + WHERE participant.api_id = $1 ) -INSERT INTO participant_stats(participant_api_id, kill_participation) +INSERT INTO participant_stats(participant_api_id, patch_version, kills_participation) SELECT * FROM stats ON CONFLICT(participant_api_id) DO -UPDATE SET(participant_api_id, kill_participation) = (SELECT * FROM stats) +UPDATE SET(participant_api_id, patch_version, kills_participation) = (SELECT * FROM stats) -- cgit v1.3.1 From 1a7f0eb91529148b08eec58ce9cbebf514a9af25 Mon Sep 17 00:00:00 2001 From: schneefux Date: Wed, 15 Mar 2017 14:47:08 +0100 Subject: calculate kda ratio (fix vainglorygame/meta#51) --- queries/participant/stats.sql | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/queries/participant/stats.sql b/queries/participant/stats.sql index 6a6d540..fe4025f 100644 --- a/queries/participant/stats.sql +++ b/queries/participant/stats.sql @@ -5,13 +5,17 @@ WITH stats AS ( CASE WHEN roster.hero_kills=0 THEN 0 ELSE (kills+assists)/roster.hero_kills::float - END AS kills_participation + END AS kills_participation, + CASE WHEN participant.deaths=0 + THEN 0 + ELSE (kills+assists)/deaths::float + END AS kda FROM participant JOIN roster ON roster.api_id=participant.roster_api_id JOIN match ON match.api_id=roster.match_api_id WHERE participant.api_id = $1 ) -INSERT INTO participant_stats(participant_api_id, patch_version, kills_participation) +INSERT INTO participant_stats(participant_api_id, patch_version, kills_participation, kda) SELECT * FROM stats ON CONFLICT(participant_api_id) DO -UPDATE SET(participant_api_id, patch_version, kills_participation) = (SELECT * FROM stats) +UPDATE SET(participant_api_id, patch_version, kills_participation, kda) = (SELECT * FROM stats) -- cgit v1.3.1 From e0cf6bfb6bfd824da141131922e9688caba5eb09 Mon Sep 17 00:00:00 2001 From: schneefux Date: Wed, 15 Mar 2017 15:02:12 +0100 Subject: log debug to console only --- api.py | 12 +----------- logs/.gitignore | 1 - 2 files changed, 1 insertion(+), 12 deletions(-) delete mode 100644 logs/.gitignore diff --git a/api.py b/api.py index 873e8be..b75beb9 100644 --- a/api.py +++ b/api.py @@ -87,17 +87,7 @@ async def startup(): await worker.start(batchlimit=750) -logging.basicConfig( - filename=os.path.realpath( - os.path.join(os.getcwd(), - os.path.dirname(__file__))) + - "/logs/compiler.log", - filemode="a", - level=logging.DEBUG -) -console = logging.StreamHandler() -console.setLevel(logging.WARNING) -logging.getLogger("").addHandler(console) +logging.basicConfig(level=logging.DEBUG) loop = asyncio.get_event_loop() loop.run_until_complete(startup()) diff --git a/logs/.gitignore b/logs/.gitignore deleted file mode 100644 index 397b4a7..0000000 --- a/logs/.gitignore +++ /dev/null @@ -1 +0,0 @@ -*.log -- cgit v1.3.1 From 368077ddd911c487f099d21b1c823f2ed126c7a8 Mon Sep 17 00:00:00 2001 From: schneefux Date: Fri, 17 Mar 2017 18:16:26 +0100 Subject: move repos --- .gitmodules | 2 +- joblib | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.gitmodules b/.gitmodules index 52948d0..47695ea 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,3 +1,3 @@ [submodule "joblib"] path = joblib - url = https://github.com/vainglorygame/joblib + url = https://gitlab.com/vainglorygame/joblib.git diff --git a/joblib b/joblib index 7ced841..542df39 160000 --- a/joblib +++ b/joblib @@ -1 +1 @@ -Subproject commit 7ced841aa44a7dd7bd2d3a68ceb989ef488ce441 +Subproject commit 542df395ce186ccc28d2c4122b7b12024ade8f47 -- cgit v1.3.1 From bb713d3873cba27e3349b31cd8dc35c3c76f9cae Mon Sep 17 00:00:00 2001 From: schneefux Date: Fri, 17 Mar 2017 19:13:19 +0100 Subject: committed wrong joblib ver --- joblib | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/joblib b/joblib index 542df39..7ced841 160000 --- a/joblib +++ b/joblib @@ -1 +1 @@ -Subproject commit 542df395ce186ccc28d2c4122b7b12024ade8f47 +Subproject commit 7ced841aa44a7dd7bd2d3a68ceb989ef488ce441 -- cgit v1.3.1 From 0b0b612c3343eaf0eff86d39b4a4769c52146d1a Mon Sep 17 00:00:00 2001 From: schneefux Date: Sun, 19 Mar 2017 15:39:17 +0100 Subject: kill process when the worker crashes, increase batch limit --- api.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/api.py b/api.py index b75beb9..b1321d5 100644 --- a/api.py +++ b/api.py @@ -80,15 +80,13 @@ class Compiler(joblib.worker.Worker): async def startup(): - for _ in range(1): - worker = Compiler() - await worker.connect(db_config, queue_db) - await worker.setup() - await worker.start(batchlimit=750) + worker = Compiler() + await worker.connect(db_config, queue_db) + await worker.setup() + await worker.run(batchlimit=1000) logging.basicConfig(level=logging.DEBUG) loop = asyncio.get_event_loop() loop.run_until_complete(startup()) -loop.run_forever() -- cgit v1.3.1 From 74c975556c0e067e60f99fb96d64d973210bf786 Mon Sep 17 00:00:00 2001 From: Vainsocial root Date: Sun, 26 Mar 2017 12:25:38 +0000 Subject: increase batch size --- api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api.py b/api.py index b1321d5..fbe2cef 100644 --- a/api.py +++ b/api.py @@ -83,7 +83,7 @@ async def startup(): worker = Compiler() await worker.connect(db_config, queue_db) await worker.setup() - await worker.run(batchlimit=1000) + await worker.run(batchlimit=5000) logging.basicConfig(level=logging.DEBUG) -- cgit v1.3.1 From 392160fdbbb45a93ebc90d57577fb62227de4c5e Mon Sep 17 00:00:00 2001 From: schneefux Date: Sat, 1 Apr 2017 12:02:40 +0200 Subject: rewrite in node --- .gitignore | 126 +++++++++--------------- Dockerfile | 9 ++ api.py | 92 ------------------ joblib | 1 - package.json | 17 ++++ queries/participant/stats.sql | 21 ---- queries/player/count_player_matches.sql | 29 ------ requirements.txt | 5 - worker.js | 163 ++++++++++++++++++++++++++++++++ 9 files changed, 236 insertions(+), 227 deletions(-) create mode 100644 Dockerfile delete mode 100644 api.py delete mode 160000 joblib create mode 100644 package.json delete mode 100644 queries/participant/stats.sql delete mode 100644 queries/player/count_player_matches.sql delete mode 100644 requirements.txt create mode 100644 worker.js diff --git a/.gitignore b/.gitignore index 9a05e2d..00cbbdf 100644 --- a/.gitignore +++ b/.gitignore @@ -1,91 +1,59 @@ -# Byte-compiled / optimized / DLL files -__pycache__/ -*.py[cod] -*$py.class - -# C extensions -*.so - -# Distribution / packaging -.Python -env/ -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ -*.egg-info/ -.installed.cfg -*.egg - -# PyInstaller -# Usually these files are written by a python script from a template -# before PyInstaller builds the exe, so as to inject date/other infos into it. -*.manifest -*.spec - -# Installer logs -pip-log.txt -pip-delete-this-directory.txt - -# Unit test / coverage reports -htmlcov/ -.tox/ -.coverage -.coverage.* -.cache -nosetests.xml -coverage.xml -*,cover -.hypothesis/ - -# Translations -*.mo -*.pot - -# Django stuff: +# Logs +logs *.log -local_settings.py +npm-debug.log* +yarn-debug.log* +yarn-error.log* -# Flask stuff: -instance/ -.webassets-cache +# Runtime data +pids +*.pid +*.seed +*.pid.lock -# Scrapy stuff: -.scrapy +# Directory for instrumented libs generated by jscoverage/JSCover +lib-cov -# Sphinx documentation -docs/_build/ +# Coverage directory used by tools like istanbul +coverage -# PyBuilder -target/ +# nyc test coverage +.nyc_output -# Jupyter Notebook -.ipynb_checkpoints +# Grunt intermediate storage (http://gruntjs.com/creating-plugins#storing-task-files) +.grunt -# pyenv -.python-version +# Bower dependency directory (https://bower.io/) +bower_components -# celery beat schedule file -celerybeat-schedule +# node-waf configuration +.lock-wscript -# dotenv -.env +# Compiled binary addons (http://nodejs.org/api/addons.html) +build/Release + +# Dependency directories +node_modules/ +jspm_packages/ + +# Typescript v1 declaration files +typings/ + +# Optional npm cache directory +.npm -# virtualenv -.venv/ -venv/ -ENV/ +# Optional eslint cache +.eslintcache -# Spyder project settings -.spyderproject +# Optional REPL history +.node_repl_history + +# Output of 'npm pack' +*.tgz + +# Yarn Integrity file +.yarn-integrity + +# dotenv environment variables file +.env -# Rope project settings -.ropeproject diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..9c3debd --- /dev/null +++ b/Dockerfile @@ -0,0 +1,9 @@ +FROM node:7.7-alpine + +RUN mkdir -p /usr/src/app +WORKDIR /usr/src/app + +COPY . /usr/src/app +RUN npm install && npm cache clean + +CMD ["node", "worker.js"] diff --git a/api.py b/api.py deleted file mode 100644 index fbe2cef..0000000 --- a/api.py +++ /dev/null @@ -1,92 +0,0 @@ -#!/usr/bin/python3 - -import asyncio -import os -import glob -import logging -import json -import asyncpg - -import joblib.worker - -queue_db = { - "host": os.environ.get("POSTGRESQL_SOURCE_HOST") or "vaindock_postgres_raw", - "port": os.environ.get("POSTGRESQL_SOURCE_PORT") or 5532, - "user": os.environ.get("POSTGRESQL_SOURCE_USER") or "vainraw", - "password": os.environ.get("POSTGRESQL_SOURCE_PASSWORD") or "vainraw", - "database": os.environ.get("POSTGRESQL_SOURCE_DB") or "vainsocial-raw" -} - -db_config = { - "host": os.environ.get("POSTGRESQL_DEST_HOST") or "vaindock_postgres_web", - "port": os.environ.get("POSTGRESQL_DEST_PORT") or 5432, - "user": os.environ.get("POSTGRESQL_DEST_USER") or "vainweb", - "password": os.environ.get("POSTGRESQL_DEST_PASSWORD") or "vainweb", - "database": os.environ.get("POSTGRESQL_DEST_DB") or "vainsocial-web" -} - - -class Compiler(joblib.worker.Worker): - def __init__(self): - self._con = None - self._queries = {} - super().__init__(jobtype="compile") - - async def connect(self, dbconf, queuedb): - """Connect to database.""" - logging.warning("connecting to database") - await super().connect(**queuedb) - self._con = await asyncpg.connect(**dbconf) - - async def setup(self): - """Initialize the database.""" - scriptroot = os.path.realpath( - os.path.join(os.getcwd(), os.path.dirname(__file__))) - for path in glob.glob(scriptroot + "/queries/*/*.sql"): - # utf-8-sig is used by pgadmin, doesn't hurt to specify - # directory names: web target table - table = os.path.basename(os.path.dirname(path)) - with open(path, "r", encoding="utf-8-sig") as file: - try: - self._queries[table].append(file.read()) - except KeyError: - self._queries[table] = [file.read()] - logging.info("loaded query '%s'", table) - - - async def _windup(self): - self._tr = self._con.transaction() - await self._tr.start() - - async def _teardown(self, failed): - if failed: - await self._tr.rollback() - else: - await self._tr.commit() - - async def _execute_job(self, jobid, payload, priority): - """Finish a job.""" - object_id = payload["id"] - table = payload["type"] - if table not in self._queries: - return - logging.debug("%s: compiling '%s' from '%s'", - jobid, object_id, table) - tasks = [] - for query in self._queries[table]: - tasks.append(asyncio.ensure_future( - self._con.execute(query, object_id))) - await asyncio.gather(*tasks) - - -async def startup(): - worker = Compiler() - await worker.connect(db_config, queue_db) - await worker.setup() - await worker.run(batchlimit=5000) - - -logging.basicConfig(level=logging.DEBUG) - -loop = asyncio.get_event_loop() -loop.run_until_complete(startup()) diff --git a/joblib b/joblib deleted file mode 160000 index 7ced841..0000000 --- a/joblib +++ /dev/null @@ -1 +0,0 @@ -Subproject commit 7ced841aa44a7dd7bd2d3a68ceb989ef488ce441 diff --git a/package.json b/package.json new file mode 100644 index 0000000..0f84dca --- /dev/null +++ b/package.json @@ -0,0 +1,17 @@ +{ + "name": "compiler", + "version": "2.0.0", + "description": "", + "main": "model.js", + "dependencies": { + "amqplib": "^0.5.1", + "mysql": "^2.13.0", + "sequelize": "^3.30.4" + }, + "devDependencies": {}, + "scripts": { + "test": "echo \"Error: no test specified\" && exit 1" + }, + "author": "schneefux", + "license": "UNLICENSED" +} diff --git a/queries/participant/stats.sql b/queries/participant/stats.sql deleted file mode 100644 index fe4025f..0000000 --- a/queries/participant/stats.sql +++ /dev/null @@ -1,21 +0,0 @@ -WITH stats AS ( - SELECT - participant.api_id, - match.patch_version, - CASE WHEN roster.hero_kills=0 - THEN 0 - ELSE (kills+assists)/roster.hero_kills::float - END AS kills_participation, - CASE WHEN participant.deaths=0 - THEN 0 - ELSE (kills+assists)/deaths::float - END AS kda - FROM participant - JOIN roster ON roster.api_id=participant.roster_api_id - JOIN match ON match.api_id=roster.match_api_id - WHERE participant.api_id = $1 -) -INSERT INTO participant_stats(participant_api_id, patch_version, kills_participation, kda) -SELECT * FROM stats -ON CONFLICT(participant_api_id) DO -UPDATE SET(participant_api_id, patch_version, kills_participation, kda) = (SELECT * FROM stats) diff --git a/queries/player/count_player_matches.sql b/queries/player/count_player_matches.sql deleted file mode 100644 index 226cacc..0000000 --- a/queries/player/count_player_matches.sql +++ /dev/null @@ -1,29 +0,0 @@ -with stats as ( -with plr as ( -select - match.patch_version as patch_version, - player.api_id as player_api_id, - match.game_mode as game_mode, - participant.winner as winner -from - match, roster, participant, player -where - match.api_id = roster.match_api_id and - roster.api_id = participant.roster_api_id and - participant.player_api_id = player.api_id and - player.api_id = $1) -select - (select distinct(player_api_id) from plr), - (select distinct(patch_version) from plr), - (select count(*) as played from plr), - (select count(*) as wins from plr where winner = true), - (select count(*) as casual_played from plr where game_mode like '%casual%'), - (select count(*) as casual_wins from plr where game_mode like '%casual%' and winner = true), - (select count(*) as ranked_played from plr where game_mode like '%ranked%'), - (select count(*) as ranked_wins from plr where game_mode like '%ranked%' and winner = true), - (select 0 as streak) -) -INSERT INTO player_stats(player_api_id, patch_version, played, wins, casual_played, casual_wins, ranked_played, ranked_wins, streak) - SELECT * FROM stats -ON CONFLICT(player_api_id) DO - UPDATE SET (player_api_id, patch_version, played, wins, casual_played, casual_wins, ranked_played, ranked_wins, streak) = (SELECT * FROM stats) diff --git a/requirements.txt b/requirements.txt deleted file mode 100644 index 2e1ea39..0000000 --- a/requirements.txt +++ /dev/null @@ -1,5 +0,0 @@ -appdirs==1.4.2 -asyncpg==0.8.4 -packaging==16.8 -pyparsing==2.1.10 -six==1.10.0 diff --git a/worker.js b/worker.js new file mode 100644 index 0000000..6d05e24 --- /dev/null +++ b/worker.js @@ -0,0 +1,163 @@ +#!/usr/bin/node +/* jshint esnext:true */ +'use strict'; + +var amqp = require("amqplib"), + Seq = require("sequelize"); + +var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", + DATABASE_URI = process.env.DATABASE_URI || "sqlite:///db.sqlite", + BATCHSIZE = process.env.PROCESSOR_BATCH || 50 * (1 + 5), // matches + players + teams + IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms + +(async () => { + let seq = new Seq(DATABASE_URI), + model = require("../orm/model")(seq, Seq), + rabbit = await amqp.connect(RABBITMQ_URI), + ch = await rabbit.createChannel(); + + let queue = [], + timer = undefined; + + await seq.sync(); + + await ch.assertQueue("compile", {durable: true}); + // as long as the queue is filled, msg are not ACKed + // server sends as long as there are less than `prefetch` unACKed + await ch.prefetch(BATCHSIZE); + + ch.consume("compile", async (msg) => { + queue.push(msg); + + // fill queue until batchsize or idle + if (timer === undefined) + timer = setTimeout(process, IDLE_TIMEOUT) + if (queue.length == BATCHSIZE) + await process(); + }, { noAck: false }); + + async function process() { + console.log("compiling batch", queue.length); + + // clean up to allow processor to accept while we wait for db + let msgs = queue.slice(); + queue = []; + clearTimeout(timer); + timer = undefined; + + // BEGIN + let transaction = await seq.transaction({ autocommit: false }); + + // UPSERT + try { + // processor sends to queue with a custom "type" so compiler can filter + // m.content: player.api_id + let players = msgs.filter((m) => m.properties.type == "player").map((m) => JSON.parse(m.content)), + participants = msgs.filter((m) => m.properties.type == "participant").map((m) => JSON.parse(m.content)); + + await Promise.all(players.map(async (player) => { + let player_api_id = player.api_id, + player_ext = {}; + + player_ext.player_api_id = player_api_id; + //player_ext.series = "" + + // TODO parallelize + + player_ext.played = await model.Participant.count({ + where: { + player_api_id: player_api_id + } + }); + player_ext.wins = await model.Participant.count({ + where: { + player_api_id: player_api_id, + winner: true + } + }); + + // TODO maybe this can be done in fewer/combined/subqueries + let count_matches_where = async (where) => { + return (await model.Participant.findOne({ + where: where, + attributes: [[seq.fn("COUNT", "$roster.match$"), "count"]], + include: [ { + model: model.Roster, + attributes: [], + include: [ { + model: model.Match, + attributes: [] + } ] + } ] + })).get("count"); + }; + player_ext.played_casual = await count_matches_where({ + player_api_id: player_api_id, + "$roster.match.game_mode$": "casual" + }); + player_ext.played_ranked = await count_matches_where({ + player_api_id: player_api_id, + "$roster.match.game_mode$": "ranked" + }); + player_ext.wins_casual = await count_matches_where({ + player_api_id: player_api_id, + winner: true, + "$roster.match.game_mode$": "casual" + }); + player_ext.wins_ranked = await count_matches_where({ + player_api_id: player_api_id, + winner: true, + "$roster.match.game_mode$": "ranked" + }); + + await model.PlayerExt.upsert(player_ext, { + include: [ model.Participant ], + transaction: transaction + }); + })); + await Promise.all(participants.map(async (api_participant) => { + let participant = await model.Participant.findOne({ + where: { + api_id: api_participant.api_id + }, + attributes: ["api_id", "kills", "assists", "deaths", seq.col("roster.hero_kills")], + include: [ + model.Roster + ] + }), + participant_ext = {}; + + participant_ext.participant_api_id = participant.api_id; + participant_ext.series = "" // TODO rm + + if (participant.roster.hero_kills == 0) + participant_ext.kills_participation = 0; + else + participant_ext.kills_participation = (participant.kills + participant.assists) / participant.roster.hero_kills; + + if (participant.deaths == 0) + participant_ext.kda = 0; + else + participant_ext.kda = (participant.kills + participant.assists) / participant.deaths; + + await model.ParticipantExt.upsert(participant_ext, { + include: [ model.Participant ], + transaction: transaction + }); + })); + + // COMMIT + await transaction.commit(); + console.log("acking batch"); + await ch.ack(msgs.pop(), true); // ack all messages until the last + + // notify web + await Promise.all(players.map(async (p) => await ch.publish("amq.topic", p.name, new Buffer("compile_commit")) )); + } catch (err) { // TODO catch only SQL error, also catch errors in the promises + console.error(err); + await transaction.rollback(); + await ch.nack(msgs.pop(), true, true); // nack all messages until the last and requeue + // TODO don't requeue broken records + } + } +})(); -- cgit v1.3.1 From bbfc647cff598cda5d82d1baee2b3d97b4b551e1 Mon Sep 17 00:00:00 2001 From: schneefux Date: Sat, 1 Apr 2017 12:42:23 +0200 Subject: calculate player.last_match_created_date --- worker.js | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/worker.js b/worker.js index 6d05e24..79b385b 100644 --- a/worker.js +++ b/worker.js @@ -59,6 +59,27 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", let player_api_id = player.api_id, player_ext = {}; + // set last_match_created_date + let lmcd = (await model.Participant.findOne({ + where: { + player_api_id: player_api_id + }, + attributes: [ [seq.col("roster.match.created_at"), "last_match_created_date"] ], + include: [ { + model: model.Roster, + attributes: [], + include: [ { + model: model.Match, + attributes: [] + } ] + } ], + order: [ + [seq.col("last_match_created_date"), "DESC"] + ] + })).get("last_match_created_date"); + await model.Player.update({ last_match_created_date: lmcd }, { where: { api_id: player_api_id } }); + + // calculate "extended" player_ext fields like wins per patch player_ext.player_api_id = player_api_id; //player_ext.series = "" -- cgit v1.3.1 From 81ed09e1d200aaf85e9c7b4ecf3544597a3446ba Mon Sep 17 00:00:00 2001 From: schneefux Date: Sun, 2 Apr 2017 20:26:03 +0200 Subject: insert in batches --- worker.js | 107 ++++++++++++++++++++++++++++++++++++-------------------------- 1 file changed, 62 insertions(+), 45 deletions(-) diff --git a/worker.js b/worker.js index 79b385b..066310d 100644 --- a/worker.js +++ b/worker.js @@ -45,17 +45,20 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", clearTimeout(timer); timer = undefined; - // BEGIN - let transaction = await seq.transaction({ autocommit: false }); - - // UPSERT - try { - // processor sends to queue with a custom "type" so compiler can filter - // m.content: player.api_id - let players = msgs.filter((m) => m.properties.type == "player").map((m) => JSON.parse(m.content)), - participants = msgs.filter((m) => m.properties.type == "participant").map((m) => JSON.parse(m.content)); - - await Promise.all(players.map(async (player) => { + // aggregate & bulk insert + let player_ext_records = [], + participant_ext_records = [], + player_updates = []; // [[what, where]] + + // processor sends to queue with a custom "type" so compiler can filter + // m.content: player.api_id + let players = msgs.filter((m) => m.properties.type == "player").map((m) => JSON.parse(m.content)), + participants = msgs.filter((m) => m.properties.type == "participant").map((m) => JSON.parse(m.content)); + + // collect information and populate _record arrays + await Promise.all([ + // collect player information + Promise.all(players.map(async (player) => { let player_api_id = player.api_id, player_ext = {}; @@ -77,7 +80,8 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", [seq.col("last_match_created_date"), "DESC"] ] })).get("last_match_created_date"); - await model.Player.update({ last_match_created_date: lmcd }, { where: { api_id: player_api_id } }); + // do later in the transaction + player_updates.push([{ last_match_created_date: lmcd }, { where: { api_id: player_api_id } }]); // calculate "extended" player_ext fields like wins per patch player_ext.player_api_id = player_api_id; @@ -85,18 +89,6 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", // TODO parallelize - player_ext.played = await model.Participant.count({ - where: { - player_api_id: player_api_id - } - }); - player_ext.wins = await model.Participant.count({ - where: { - player_api_id: player_api_id, - winner: true - } - }); - // TODO maybe this can be done in fewer/combined/subqueries let count_matches_where = async (where) => { return (await model.Participant.findOne({ @@ -112,6 +104,20 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", } ] })).get("count"); }; + + // TODO run in parallel + player_ext.played = await model.Participant.count({ + where: { + player_api_id: player_api_id + } + }); + player_ext.wins = await model.Participant.count({ + where: { + player_api_id: player_api_id, + winner: true + } + }); + player_ext.played_casual = await count_matches_where({ player_api_id: player_api_id, "$roster.match.game_mode$": "casual" @@ -131,12 +137,10 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", "$roster.match.game_mode$": "ranked" }); - await model.PlayerExt.upsert(player_ext, { - include: [ model.Participant ], - transaction: transaction - }); - })); - await Promise.all(participants.map(async (api_participant) => { + player_ext_records.push(player_ext); + })), + // calculate participant fields + Promise.all(participants.map(async (api_participant) => { let participant = await model.Participant.findOne({ where: { api_id: api_participant.api_id @@ -161,24 +165,37 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", else participant_ext.kda = (participant.kills + participant.assists) / participant.deaths; - await model.ParticipantExt.upsert(participant_ext, { - include: [ model.Participant ], - transaction: transaction - }); - })); - - // COMMIT - await transaction.commit(); - console.log("acking batch"); - await ch.ack(msgs.pop(), true); // ack all messages until the last + participant_ext_records.push(participant_ext); + })) + ]); - // notify web - await Promise.all(players.map(async (p) => await ch.publish("amq.topic", p.name, new Buffer("compile_commit")) )); - } catch (err) { // TODO catch only SQL error, also catch errors in the promises + // load records into db + try { + console.log("inserting batch into db"); + await seq.transaction({ autocommit: false }, (transaction) => { + return Promise.all([ + model.ParticipantExt.bulkCreate(participant_ext_records, { + updateOnDuplicate: [], // all + transaction: transaction + }), + model.PlayerExt.bulkCreate(player_ext_records, { + updateOnDuplicate: [], // all + transaction: transaction + }), + player_updates.map(async (pu) => + await model.Player.update(pu[0], pu[1])) + ]); + }); + } catch (err) { + // this should only happen for deadlocks or non-data related issues console.error(err); - await transaction.rollback(); await ch.nack(msgs.pop(), true, true); // nack all messages until the last and requeue - // TODO don't requeue broken records + return; // give up } + console.log("acking batch"); + await ch.ack(msgs.pop(), true); // ack all messages until the last + + // notify web + await Promise.all(players.map(async (p) => await ch.publish("amq.topic", p.name, new Buffer("compile_commit")) )); } })(); -- cgit v1.3.1 From fda35295a35e807709e1ef8b8d4dfb1e443f4d8a Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 3 Apr 2017 16:05:25 +0200 Subject: redesign web notification --- worker.js | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/worker.js b/worker.js index 066310d..652faf7 100644 --- a/worker.js +++ b/worker.js @@ -196,6 +196,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await ch.ack(msgs.pop(), true); // ack all messages until the last // notify web - await Promise.all(players.map(async (p) => await ch.publish("amq.topic", p.name, new Buffer("compile_commit")) )); + await Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, + new Buffer("stats_update")) )); } })(); -- cgit v1.3.1 From 3ab7ff81923d4472c6aa04e5a9a5f36e64d6fdd8 Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 3 Apr 2017 16:33:28 +0200 Subject: do not log queries --- worker.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/worker.js b/worker.js index 652faf7..0ab63e1 100644 --- a/worker.js +++ b/worker.js @@ -11,7 +11,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms (async () => { - let seq = new Seq(DATABASE_URI), + let seq = new Seq(DATABASE_URI, { logging: () => {} }), model = require("../orm/model")(seq, Seq), rabbit = await amqp.connect(RABBITMQ_URI), ch = await rabbit.createChannel(); -- cgit v1.3.1 From a41d397a57be9e7595baa9310f83d2d6a411175f Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 3 Apr 2017 16:43:29 +0200 Subject: anticipate players with no match history --- worker.js | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/worker.js b/worker.js index 0ab63e1..585333a 100644 --- a/worker.js +++ b/worker.js @@ -63,7 +63,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", player_ext = {}; // set last_match_created_date - let lmcd = (await model.Participant.findOne({ + let record = await model.Participant.findOne({ where: { player_api_id: player_api_id }, @@ -79,9 +79,13 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", order: [ [seq.col("last_match_created_date"), "DESC"] ] - })).get("last_match_created_date"); - // do later in the transaction - player_updates.push([{ last_match_created_date: lmcd }, { where: { api_id: player_api_id } }]); + }); + if (record != null) + // do later in the transaction + player_updates.push([ + { last_match_created_date: record.get("last_match_created_date") }, + { where: { api_id: player_api_id } } + ]); // calculate "extended" player_ext fields like wins per patch player_ext.player_api_id = player_api_id; @@ -91,7 +95,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", // TODO maybe this can be done in fewer/combined/subqueries let count_matches_where = async (where) => { - return (await model.Participant.findOne({ + let record = await model.Participant.findOne({ where: where, attributes: [[seq.fn("COUNT", "$roster.match$"), "count"]], include: [ { @@ -102,7 +106,9 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", attributes: [] } ] } ] - })).get("count"); + }) + if (record == null) return 0; + return record.get("count"); }; // TODO run in parallel -- cgit v1.3.1 From b7b7528a0639e0ad4cd1bdac0d021752ddbf0401 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 11:06:17 +0200 Subject: send notifications for participant stats --- worker.js | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/worker.js b/worker.js index 585333a..11da36f 100644 --- a/worker.js +++ b/worker.js @@ -202,7 +202,11 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await ch.ack(msgs.pop(), true); // ack all messages until the last // notify web - await Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, - new Buffer("stats_update")) )); + await Promise.all([ + Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, + new Buffer("stats_update")) )), + Promise.all(participant_ext_records.map(async (p) => await ch.publish("amq.topic", + "participant." + p.api_id, new Buffer("stats_update")) )) + ]); } })(); -- cgit v1.3.1 From 3406a3e76754859cbee257a61f150160597b73fa Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 12:06:26 +0200 Subject: wait for rabbit and db to start --- package.json | 3 ++- worker.js | 23 +++++++++++++++++------ 2 files changed, 19 insertions(+), 7 deletions(-) diff --git a/package.json b/package.json index 0f84dca..66afc71 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,8 @@ "dependencies": { "amqplib": "^0.5.1", "mysql": "^2.13.0", - "sequelize": "^3.30.4" + "sequelize": "^3.30.4", + "sleep-promise": "^2.0.0" }, "devDependencies": {}, "scripts": { diff --git a/worker.js b/worker.js index 11da36f..a31f97d 100644 --- a/worker.js +++ b/worker.js @@ -3,7 +3,8 @@ 'use strict'; var amqp = require("amqplib"), - Seq = require("sequelize"); + Seq = require("sequelize"), + sleep = require("sleep-promise"); var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", DATABASE_URI = process.env.DATABASE_URI || "sqlite:///db.sqlite", @@ -11,17 +12,27 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms (async () => { - let seq = new Seq(DATABASE_URI, { logging: () => {} }), - model = require("../orm/model")(seq, Seq), - rabbit = await amqp.connect(RABBITMQ_URI), - ch = await rabbit.createChannel(); + let seq, model, rabbit, ch; + + while (true) { + try { + seq = new Seq(DATABASE_URI, { logging: () => {} }), + rabbit = await amqp.connect(RABBITMQ_URI), + ch = await rabbit.createChannel(); + await ch.assertQueue("compile", {durable: true}); + break; + } catch (err) { + console.error(err); + await sleep(5000); + } + } + model = require("../orm/model")(seq, Seq); let queue = [], timer = undefined; await seq.sync(); - await ch.assertQueue("compile", {durable: true}); // as long as the queue is filled, msg are not ACKed // server sends as long as there are less than `prefetch` unACKed await ch.prefetch(BATCHSIZE); -- cgit v1.3.1 From d7fcdb2ee11fa2f5a10f911b6916741ea0672734 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 17:21:31 +0200 Subject: fix player_ext notifications --- worker.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/worker.js b/worker.js index a31f97d..6e4e4cb 100644 --- a/worker.js +++ b/worker.js @@ -217,7 +217,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, new Buffer("stats_update")) )), Promise.all(participant_ext_records.map(async (p) => await ch.publish("amq.topic", - "participant." + p.api_id, new Buffer("stats_update")) )) + "participant." + p.participant_api_id, new Buffer("stats_update")) )) ]); } })(); -- cgit v1.3.1 From de9bd96b5ccda3b9a354cf3e02d49012bdc177d4 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 20:30:33 +0200 Subject: send jobs to compiler --- worker.js | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/worker.js b/worker.js index 6e4e4cb..d0d6ca0 100644 --- a/worker.js +++ b/worker.js @@ -20,6 +20,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", rabbit = await amqp.connect(RABBITMQ_URI), ch = await rabbit.createChannel(); await ch.assertQueue("compile", {durable: true}); + await ch.assertQueue("analyze", {durable: true}); break; } catch (err) { console.error(err); @@ -212,6 +213,14 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", console.log("acking batch"); await ch.ack(msgs.pop(), true); // ack all messages until the last + // notify analyzer + Promise.all(participant_ext_records.map(async (p) => + await ch.sendToQueue("analyze", new Buffer(p.participant_api_id), { + persistent: true, + type: "participant" + }) + )); + // notify web await Promise.all([ Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, -- cgit v1.3.1 From a6b1c4139a485595fd858fa447d893725d43a015 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 21:03:28 +0200 Subject: filter undefined or null objects --- worker.js | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/worker.js b/worker.js index d0d6ca0..b277eca 100644 --- a/worker.js +++ b/worker.js @@ -64,8 +64,10 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", // processor sends to queue with a custom "type" so compiler can filter // m.content: player.api_id - let players = msgs.filter((m) => m.properties.type == "player").map((m) => JSON.parse(m.content)), - participants = msgs.filter((m) => m.properties.type == "participant").map((m) => JSON.parse(m.content)); + let players = msgs.filter((m) => m.properties.type == "player").map( + (m) => JSON.parse(m.content)).filter((p) => p != undefined), + participants = msgs.filter((m) => m.properties.type == "participant").map( + (m) => JSON.parse(m.content)).filter((p) => p != undefined); // collect information and populate _record arrays await Promise.all([ -- cgit v1.3.1 From 7d085bac26ca372a3c9bd097f7c7214af5fbb408 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 21:04:41 +0200 Subject: ack each message individually --- worker.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/worker.js b/worker.js index b277eca..4fcb0bf 100644 --- a/worker.js +++ b/worker.js @@ -209,11 +209,11 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", } catch (err) { // this should only happen for deadlocks or non-data related issues console.error(err); - await ch.nack(msgs.pop(), true, true); // nack all messages until the last and requeue + await Promise.all(msgs.map((m) => ch.nack(m, true)) ); // requeue return; // give up } console.log("acking batch"); - await ch.ack(msgs.pop(), true); // ack all messages until the last + await Promise.all(msgs.map((m) => ch.ack(m)) ); // notify analyzer Promise.all(participant_ext_records.map(async (p) => -- cgit v1.3.1 From 9567fcbd4d41f603fab53864008add884bd315de Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 21:05:09 +0200 Subject: await transaction --- worker.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/worker.js b/worker.js index 4fcb0bf..1efcba7 100644 --- a/worker.js +++ b/worker.js @@ -192,8 +192,8 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", // load records into db try { console.log("inserting batch into db"); - await seq.transaction({ autocommit: false }, (transaction) => { - return Promise.all([ + await seq.transaction({ autocommit: false }, async (transaction) => { + await Promise.all([ model.ParticipantExt.bulkCreate(participant_ext_records, { updateOnDuplicate: [], // all transaction: transaction -- cgit v1.3.1 From 241a277b78d9b3f94ddee550e2d6e7f08fff2a95 Mon Sep 17 00:00:00 2001 From: schneefux Date: Fri, 7 Apr 2017 18:33:14 +0200 Subject: update for new schema --- worker.js | 124 +++++++++++++++----------------------------------------------- 1 file changed, 30 insertions(+), 94 deletions(-) diff --git a/worker.js b/worker.js index 1efcba7..5c1c5bf 100644 --- a/worker.js +++ b/worker.js @@ -8,7 +8,7 @@ var amqp = require("amqplib"), var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", DATABASE_URI = process.env.DATABASE_URI || "sqlite:///db.sqlite", - BATCHSIZE = process.env.PROCESSOR_BATCH || 50 * (1 + 5), // matches + players + teams + BATCHSIZE = process.env.PROCESSOR_BATCH || 50 * (1 + 5), // matches + players IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms (async () => { @@ -58,8 +58,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", timer = undefined; // aggregate & bulk insert - let player_ext_records = [], - participant_ext_records = [], + let participant_stats_records = [], player_updates = []; // [[what, where]] // processor sends to queue with a custom "type" so compiler can filter @@ -73,8 +72,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await Promise.all([ // collect player information Promise.all(players.map(async (player) => { - let player_api_id = player.api_id, - player_ext = {}; + let player_api_id = player.api_id; // set last_match_created_date let record = await model.Participant.findOne({ @@ -100,92 +98,20 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", { last_match_created_date: record.get("last_match_created_date") }, { where: { api_id: player_api_id } } ]); - - // calculate "extended" player_ext fields like wins per patch - player_ext.player_api_id = player_api_id; - //player_ext.series = "" - - // TODO parallelize - - // TODO maybe this can be done in fewer/combined/subqueries - let count_matches_where = async (where) => { - let record = await model.Participant.findOne({ - where: where, - attributes: [[seq.fn("COUNT", "$roster.match$"), "count"]], + })), + // calculate participant fields + Promise.all(participants.map(async (api_participant) => { + participant_stats_records.push(calculate_stats( + await model.Participant.findOne({ + where: { api_id: api_participant.api_id }, include: [ { model: model.Roster, - attributes: [], - include: [ { - model: model.Match, - attributes: [] - } ] + include: [ + model.Match + ] } ] }) - if (record == null) return 0; - return record.get("count"); - }; - - // TODO run in parallel - player_ext.played = await model.Participant.count({ - where: { - player_api_id: player_api_id - } - }); - player_ext.wins = await model.Participant.count({ - where: { - player_api_id: player_api_id, - winner: true - } - }); - - player_ext.played_casual = await count_matches_where({ - player_api_id: player_api_id, - "$roster.match.game_mode$": "casual" - }); - player_ext.played_ranked = await count_matches_where({ - player_api_id: player_api_id, - "$roster.match.game_mode$": "ranked" - }); - player_ext.wins_casual = await count_matches_where({ - player_api_id: player_api_id, - winner: true, - "$roster.match.game_mode$": "casual" - }); - player_ext.wins_ranked = await count_matches_where({ - player_api_id: player_api_id, - winner: true, - "$roster.match.game_mode$": "ranked" - }); - - player_ext_records.push(player_ext); - })), - // calculate participant fields - Promise.all(participants.map(async (api_participant) => { - let participant = await model.Participant.findOne({ - where: { - api_id: api_participant.api_id - }, - attributes: ["api_id", "kills", "assists", "deaths", seq.col("roster.hero_kills")], - include: [ - model.Roster - ] - }), - participant_ext = {}; - - participant_ext.participant_api_id = participant.api_id; - participant_ext.series = "" // TODO rm - - if (participant.roster.hero_kills == 0) - participant_ext.kills_participation = 0; - else - participant_ext.kills_participation = (participant.kills + participant.assists) / participant.roster.hero_kills; - - if (participant.deaths == 0) - participant_ext.kda = 0; - else - participant_ext.kda = (participant.kills + participant.assists) / participant.deaths; - - participant_ext_records.push(participant_ext); + )) })) ]); @@ -194,11 +120,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", console.log("inserting batch into db"); await seq.transaction({ autocommit: false }, async (transaction) => { await Promise.all([ - model.ParticipantExt.bulkCreate(participant_ext_records, { - updateOnDuplicate: [], // all - transaction: transaction - }), - model.PlayerExt.bulkCreate(player_ext_records, { + model.ParticipantStats.bulkCreate(participant_stats_records, { updateOnDuplicate: [], // all transaction: transaction }), @@ -216,7 +138,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await Promise.all(msgs.map((m) => ch.ack(m)) ); // notify analyzer - Promise.all(participant_ext_records.map(async (p) => + Promise.all(participant_stats_records.map(async (p) => await ch.sendToQueue("analyze", new Buffer(p.participant_api_id), { persistent: true, type: "participant" @@ -227,8 +149,22 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await Promise.all([ Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, new Buffer("stats_update")) )), - Promise.all(participant_ext_records.map(async (p) => await ch.publish("amq.topic", + Promise.all(participant_stats_records.map(async (p) => await ch.publish("amq.topic", "participant." + p.participant_api_id, new Buffer("stats_update")) )) ]); } + + // based on the participant db record from the end of the match, + // calculate a participant_stats record and return it + function calculate_stats(participant) { + if (participant == null) { console.error("got nonexisting participant!"); return; } + let participant_stats = {}; + participant_stats.participant_api_id = participant.get("api_id"); + + participant_stats.kills = participant.get("kills"); + + participant_stats.kill_participation = participant.get("kills") / participant.roster.get("hero_kills"); + + return participant_stats; + } })(); -- cgit v1.3.1 From 72165f52411deaf633a7d6c301f0650c94f79671 Mon Sep 17 00:00:00 2001 From: schneefux Date: Fri, 7 Apr 2017 20:23:39 +0200 Subject: fix NaN stat --- worker.js | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/worker.js b/worker.js index 5c1c5bf..84203ec 100644 --- a/worker.js +++ b/worker.js @@ -163,7 +163,8 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", participant_stats.kills = participant.get("kills"); - participant_stats.kill_participation = participant.get("kills") / participant.roster.get("hero_kills"); + if (participant.roster.get("hero_kills") == 0) participant_stats.kill_participation = 0; + else participant_stats.kill_participation = participant.get("kills") / participant.roster.get("hero_kills"); return participant_stats; } -- cgit v1.3.1 From 9932435c28a82164a02f3039b46dd5fb98d5223a Mon Sep 17 00:00:00 2001 From: schneefux Date: Fri, 7 Apr 2017 21:34:31 +0200 Subject: link items to participant --- worker.js | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/worker.js b/worker.js index 84203ec..0c407eb 100644 --- a/worker.js +++ b/worker.js @@ -105,11 +105,16 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await model.Participant.findOne({ where: { api_id: api_participant.api_id }, include: [ { - model: model.Roster, - include: [ - model.Match - ] - } ] + model: model.Roster, + include: [ + model.Match + ] + }, { + model: model.ItemParticipant, + as: "items", + include: [ model.Item ] + } + ] }) )) })) @@ -166,6 +171,14 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", if (participant.roster.get("hero_kills") == 0) participant_stats.kill_participation = 0; else participant_stats.kill_participation = participant.get("kills") / participant.roster.get("hero_kills"); + participant_stats.sustain_score = participant.items.reduce((score, item) => { + if (item.get("action") == "final") { + if (["Eve of Harvest", "Serpent Mask"].indexOf(item.item.get("name")) != -1) + return score + 20; + } + return score; + }, 0); + return participant_stats; } })(); -- cgit v1.3.1