diff options
| -rw-r--r-- | .gitignore | 116 | ||||
| -rw-r--r-- | .gitmodules | 3 | ||||
| -rw-r--r-- | Dockerfile | 9 | ||||
| -rw-r--r-- | api.py | 126 | ||||
| m--------- | joblib | 0 | ||||
| -rw-r--r-- | package.json | 19 | ||||
| -rw-r--r-- | requirements.txt | 5 | ||||
| -rw-r--r-- | worker.js | 154 |
8 files changed, 224 insertions, 208 deletions
@@ -1,91 +1,59 @@ -# Byte-compiled / optimized / DLL files -__pycache__/ -*.py[cod] -*$py.class +# Logs +logs +*.log +npm-debug.log* +yarn-debug.log* +yarn-error.log* -# C extensions -*.so +# Runtime data +pids +*.pid +*.seed +*.pid.lock -# Distribution / packaging -.Python -env/ -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ -*.egg-info/ -.installed.cfg -*.egg +# Directory for instrumented libs generated by jscoverage/JSCover +lib-cov -# 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 +# Coverage directory used by tools like istanbul +coverage -# Installer logs -pip-log.txt -pip-delete-this-directory.txt +# nyc test coverage +.nyc_output -# Unit test / coverage reports -htmlcov/ -.tox/ -.coverage -.coverage.* -.cache -nosetests.xml -coverage.xml -*,cover -.hypothesis/ +# Grunt intermediate storage (http://gruntjs.com/creating-plugins#storing-task-files) +.grunt -# Translations -*.mo -*.pot +# Bower dependency directory (https://bower.io/) +bower_components -# Django stuff: -*.log -local_settings.py +# node-waf configuration +.lock-wscript -# Flask stuff: -instance/ -.webassets-cache +# Compiled binary addons (http://nodejs.org/api/addons.html) +build/Release -# Scrapy stuff: -.scrapy +# Dependency directories +node_modules/ +jspm_packages/ -# Sphinx documentation -docs/_build/ +# Typescript v1 declaration files +typings/ -# PyBuilder -target/ +# Optional npm cache directory +.npm -# Jupyter Notebook -.ipynb_checkpoints +# Optional eslint cache +.eslintcache -# pyenv -.python-version +# Optional REPL history +.node_repl_history -# celery beat schedule file -celerybeat-schedule +# Output of 'npm pack' +*.tgz -# dotenv -.env +# Yarn Integrity file +.yarn-integrity -# virtualenv -.venv/ -venv/ -ENV/ - -# Spyder project settings -.spyderproject +# dotenv environment variables file +.env -# Rope project settings -.ropeproject diff --git a/.gitmodules b/.gitmodules index 47695ea..e69de29 100644 --- a/.gitmodules +++ b/.gitmodules @@ -1,3 +0,0 @@ -[submodule "joblib"] - path = joblib - url = https://gitlab.com/vainglorygame/joblib.git 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"] @@ -1,126 +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 "localhost", - "port": os.environ.get("POSTGRESQL_SOURCE_PORT") or 5433, - "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 "localhost", - "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 Cruncher(joblib.worker.Worker): - def __init__(self): - self._con = None - super().__init__(jobtype="crunch") - - 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): - pass - - 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.""" - dimension_id = payload["dimension"] - - logging.debug("%s: crunching dimension %s", - jobid, dimension_id) - - dimension = await self._con.fetchrow(""" - SELECT * FROM stats_dimensions WHERE id=$1 - """, dimension_id) - if dimension is None: - logging.warning("%s: invalid dimension '%s', exiting", - jobid, dimension_id) - raise joblib.worker.JobFailed("invalid dimension '" + dimension_id - + "'", - False) - - table = dimension["dimension_on"] - field = dimension["name"] - value = dimension["value"] - - if table == "hero": - filter_query = "" - if field == "patch": - filter_query = "WHERE match.patch_version='" + value + "'" - if field == "recent_matches": - filter_query = "WHERE match.api_id IN (SELECT api_id FROM match ORDER BY created_at DESC LIMIT " + value + ")" - if field == "skill_tier": - filter_query = "WHERE participant.skill_tier=" + value - - stats = await self._con.fetch(""" - SELECT - heros.id, - SUM(participant.winner::INT)/COUNT(participant.winner)::FLOAT AS win_rate, - COUNT(participant.hero)/(SELECT COUNT(hero) FROM participant)::FLOAT AS pick_rate, - SUM(60*participant.minion_kills/match.duration::FLOAT)/COUNT(participant.farm)::FLOAT AS cs_per_min, - 0 AS gold_per_min - FROM participant - - JOIN participant_stats ON participant.api_id=participant_stats.participant_api_id - JOIN roster ON roster.api_id=participant.roster_api_id - JOIN match ON match.api_id=roster.match_api_id - - JOIN heros ON heros.api_name=hero - """ - + filter_query + - """ - GROUP BY heros.id - """) - - for stat in stats: - stat_id = await self._con.fetchrow(""" - INSERT INTO hero_stats(win_rate, pick_rate, cs_per_min, gold_per_min) - VALUES($1, $2, $3, $4) - RETURNING hero_stats.id - """, stat["win_rate"], stat["pick_rate"], stat["cs_per_min"], - stat["gold_per_min"]) - await self._con.fetch(""" - INSERT INTO hero_dimension(hero_id, dimension_id, stats_id, computed_on) - VALUES($1, $2, $3, NOW()) - """, stat["id"], dimension["id"], - stat_id["id"]) - -async def startup(): - worker = Cruncher() - await worker.connect(db_config, queue_db) - await worker.setup() - await worker.run(batchlimit=1) - - -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 -Subproject 7ced841aa44a7dd7bd2d3a68ceb989ef488ce44 diff --git a/package.json b/package.json new file mode 100644 index 0000000..b3db938 --- /dev/null +++ b/package.json @@ -0,0 +1,19 @@ +{ + "name": "cruncher", + "version": "2.0.0", + "description": "", + "main": "worker.js", + "dependencies": { + "amqplib": "^0.5.1", + "mysql": "^2.13.0", + "object-hash": "^1.1.7", + "sequelize": "^3.30.4", + "sleep-promise": "^2.0.0" + }, + "devDependencies": {}, + "scripts": { + "test": "echo \"Error: no test specified\" && exit 1" + }, + "author": "schneefux", + "license": "UNLICENSED" +} 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..1f1263e --- /dev/null +++ b/worker.js @@ -0,0 +1,154 @@ +#!/usr/bin/node +/* jshint esnext:true */ +"use strict"; + +var amqp = require("amqplib"), + Seq = require("sequelize"), + sleep = require("sleep-promise"), + hash = require("object-hash"); + +var RABBITMQ_URI = process.env.RABBITMQ_URI, + DATABASE_URI = process.env.DATABASE_URI; + +(async () => { + let seq, model, rabbit, ch; + + while (true) { + try { + seq = new Seq(DATABASE_URI), + rabbit = await amqp.connect(RABBITMQ_URI), + ch = await rabbit.createChannel(); + await ch.assertQueue("crunch", {durable: true}); + break; + } catch (err) { + console.error(err); + await sleep(5000); + } + } + model = require("../orm/model")(seq, Seq); + + let queue = [], + timer = undefined; + + await seq.sync(); + + // no batching for cruncher, not worth it + // we'll read a million tables and commit just one record + // once per hour + await ch.prefetch(1); + + ch.consume("crunch", async (msg) => { + let dim = JSON.parse(msg.content), + filter = dim.filter, dimension, filter_hash, + action; + console.log("crunching dimension", dim); + + // we don't want to store duplicate filters + // but we can't reliably compare JSON in SQL either + // so we hash the object and lookup/store a sha1 + filter_hash = hash(dim.filter, { unorderedArrays: true }); + try { + dimension = (await model.StatsDimensions.findOrCreate({ + where: { + dimension_on: dim.dimension_on, + filter_hash: filter_hash + } + }))[0]; + } catch (err) { + console.error(err); + await ch.nack(msg, false, false); + return; + } + + // not implemented -> error + switch (dimension.dimension_on) { + case "hero": + action = calculateHeroStats(dimension.id, filter); + break; + default: + action = async () => { throw "dimension_on not implemented " + dimension.dimension_on; }; + } + + try { + await action; + console.log("acking"); + await ch.ack(msg); + } catch (err) { + console.error(err); + await ch.nack(msg, false, false); // nack and do not requeue + } + }, { noAck: false }); + + async function calculateHeroStats(dimension_id, filter) { + let win_rate, pick_rate, gold_per_min; + + // in literals: q -> column name, e -> function or string + let q = (qry) => seq.dialect.QueryGenerator.quote(qry), + e = (qry) => seq.dialect.QueryGenerator.escape(qry); + + // TODO (workaround) ignore old API data where participant.gold was null + filter["$participant.gold$"] = { $ne: null }; + console.log(filter); + + // alternative for win rate + //[ seq.literal(`${e(seq.fn("sum", seq.cast(seq.col("participant.winner"), "int") ))} / ${e(seq.fn("count", seq.col("participant.id")))}`), "win_rate" ] + + // calculate stats + let total_participants = await model.Participant.count({ + where: filter, + include: [ { + model: model.Roster, + attributes: [], + include: [ { + model: model.Match, + attributes: [] + } ] + }, { + model: model.Heros, + attributes: [] + } ] + }); + + // TODO (workaround) filter unmapped actor<->hero + filter["$hero.id$"] = { $ne: null }; + let data = await model.Participant.findAll({ + where: filter, + attributes: [ + // meta + [ seq.col("hero.name"), "hero_name" ], + [ seq.col("hero.id"), "hero_id" ], + + // stats, see `hero_stats` table + [ seq.literal(`${q(seq.fn("count", seq.col("participant.id") ))} / ${total_participants}`), "pick_rate" ], + [ seq.fn("avg", seq.cast(seq.col("participant.winner"), "int") ), "win_rate" ], + [ seq.fn("sum", seq.literal(`${q("participant.gold")} / ${q("roster.match.duration")}`)), "gold_per_min" ], + [ seq.fn("avg", seq.literal(`60.0 * ${q("participant.minion_kills")} / ${q("roster.match.duration")}`)), "cs_per_min" ] + ], + group: q("hero.name"), + include: [ { + model: model.Roster, + attributes: [], + include: [ { + model: model.Match, + attributes: [] + } ] + }, { + model: model.Heros, + attributes: [] + } ] + }); + console.log("hero stats", filter, data.map((d) => d.dataValues)); + // insert hero x dimension x stats + await seq.transaction({ autocommit: false }, (transaction) => + Promise.all(data.map(async (record) => { + let hero_stat_db = await model.HeroStats.create(record.dataValues); + await model.HeroDimension.create({ + hero_id: record.get("hero_id"), + dimension_id: dimension_id, + stats_id: hero_stat_db.get("id"), + computed_on: new Date() + }, { transaction: transaction }); + })) + ); + } +})(); |