diff options
| author | schneefux <schneefux+commit@schneefux.xyz> | 2017-07-14 19:03:23 +0200 |
|---|---|---|
| committer | schneefux <schneefux+commit@schneefux.xyz> | 2017-07-14 19:03:23 +0200 |
| commit | a62c2004d7767d7f977050345eae2d850518a19b (patch) | |
| tree | 4fe532d3d3699e4d7072f4a8aa019d5d1c1e3306 /worker.js | |
| parent | a5b43e106ae8a3441690ed7d32c711f0200d4be7 (diff) | |
| download | shrinker-a62c2004d7767d7f977050345eae2d850518a19b.tar.gz shrinker-a62c2004d7767d7f977050345eae2d850518a19b.zip | |
2.14.0 rewrite: split from processor
Diffstat (limited to 'worker.js')
| -rw-r--r-- | worker.js | 537 |
1 files changed, 69 insertions, 468 deletions
@@ -2,11 +2,7 @@ /* jshint esnext:true */ "use strict"; -/* processor inserts API data into the database. - * It listens to the queue `process` and expects a JSON - * which is a match structure or a player structure. - * It will forward notifications to web. - */ +/* shrinker aggregates Telemetry data and inserts it into the database. */ const amqp = require("amqplib"), Promise = require("bluebird"), @@ -14,10 +10,7 @@ const amqp = require("amqplib"), loggly = require("winston-loggly-bulk"), Seq = require("sequelize"), api_name_mappings = require("../orm/mappings").map, - isAbility = require("../orm/mappings").isAbility, - isItem = require("../orm/mappings").isItem, - isHero = require("../orm/mappings").isHero, - sleep = require("sleep-promise"); + isAbility = require("../orm/mappings").isAbility; const RABBITMQ_URI = process.env.RABBITMQ_URI, DATABASE_URI = process.env.DATABASE_URI, @@ -47,72 +40,36 @@ if (LOGGLY_TOKEN) logger.add(winston.transports.Loggly, { inputToken: LOGGLY_TOKEN, subdomain: "kvahuja", - tags: ["backend", "processor", QUEUE], + tags: ["backend", "shrinker", QUEUE], json: true }); -// helpers -const camelCaseRegExp = new RegExp(/([a-z])([A-Z]+)/g); -function camelToSnake(text) { - return text.replace(camelCaseRegExp, (m, $1, $2) => - $1 + "_" + $2.toLowerCase()); -} - -// MadGlory API uses snakeCase, our db uses camel_case -function snakeCaseKeys(obj) { - Object.keys(obj).forEach((key) => { - const new_key = camelToSnake(key); - if (new_key == key) return; - obj[new_key] = obj[key]; - delete obj[key]; - }); - return obj; -} - // split an array into arrays of max chunksize function* chunks(arr) { for (let c=0, len=arr.length; c<len; c+=CHUNKSIZE) yield arr.slice(c, c+CHUNKSIZE); } -// helper to convert API response into flat JSON -// db structure is (almost) 1:1 the API structure -// so we can insert the flat API response as-is -function flatten(obj) { - const attrs = obj.attributes || {}, - stats = attrs.stats || {}; - let o = Object.assign({}, obj, attrs, stats); - o.api_id = o.id; // rename - delete o.id; - delete o.type; - delete o.attributes; - delete o.stats; - delete o.relationships; - return snakeCaseKeys(o); -} - -// main code -(async () => { - let seq, model, rabbit, ch; +amqp.connect(RABBITMQ_URI).then(async (rabbit) => { + global.process.on("SIGINT", () => { + rabbit.close(); + global.process.exit(); + }); // connect to rabbit & db - while (true) { - try { - seq = new Seq(DATABASE_URI, { - logging: false, - max: MAXCONNS - }); - rabbit = await amqp.connect(RABBITMQ_URI, { heartbeat: 120 }); - ch = await rabbit.createChannel(); - await ch.assertQueue(QUEUE, {durable: true}); - break; - } catch (err) { - logger.error("Error connecting", err); - await sleep(5000); - } - } + const seq = new Seq(DATABASE_URI, { + logging: false, + max: MAXCONNS + }); - model = require("../orm/model")(seq, Seq); + const ch = await rabbit.createChannel(); + await ch.assertQueue(QUEUE, { durable: true }); + await ch.assertQueue(QUEUE + "_failed", { 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); + + const model = require("../orm/model")(seq, Seq); // performance logging let load_timer = undefined, @@ -121,67 +78,29 @@ function flatten(obj) { // Maps to quickly convert API names to db ids let item_db_map = new Map(), // "Halcyon Potion" to id - hero_db_map = new Map(), // "*SAW*" to id - series_db_map = new Map(), // date to series id - game_mode_db_map = new Map(), // "ranked" to id - role_db_map = new Map(), // "captain" to id - hero_role_map = new Map(); // SAW.id to "carry" + hero_db_map = new Map(); // "*SAW*" to id // populate maps await Promise.all([ model.Item.findAll() .map((item) => item_db_map.set(item.name, item.id)), model.Hero.findAll() - .map((hero) => hero_db_map.set(hero.name, hero.id)), - model.Series.findAll() - .map((series) => { - if (series.dimension_on == "player") - series_db_map.set(series.name, series.id); - }), - model.GameMode.findAll() - .map((mode) => game_mode_db_map.set(mode.name, mode.id)), - model.Role.findAll() - .map((role) => role_db_map.set(role.name, role.id)) + .map((hero) => hero_db_map.set(hero.name, hero.id)) ]); - - // 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); + if (item_db_map.size == 0 || + hero_db_map.size == 0) { + logger.error("mapping tables are not seeded!!! quitting"); + process.exit(); + } // buffers that will be filled until BATCHSIZE is reached // to make db transactions more efficient - let player_data = new Set(), - match_data = new Set(), - telemetry_data = new Set(), + let telemetry_data = new Set(), msg_buffer = new Set(); ch.consume(QUEUE, async (msg) => { - if (msg.properties.type == "player") { - // bridge sends a single object - player_data.add(JSON.parse(msg.content)); - msg_buffer.add(msg); - } - if (msg.properties.type == "match") { - // apigrabber sends a single object - const match = JSON.parse(msg.content); - // deduplicate and reject immediately - if (await model.Match.count({ where: { api_id: match.id } }) > 0) { - if (msg.properties.headers.notify != undefined) { - await ch.publish("amq.topic", - msg.properties.headers.notify, - new Buffer("matches_dupe")) - // send match_dupe to web player.ign.api_id - await ch.publish("amq.topic", - msg.properties.headers.notify + "." + match.id, - new Buffer("match_dupe")) - } - } else match_data.add(match); - msg_buffer.add(msg); - } - if (msg.properties.type == "telemetry") { - telemetry_data.add(JSON.parse(msg.content)); - msg_buffer.add(msg); - } + telemetry_data.add(JSON.parse(msg.content)); + msg_buffer.add(msg); // fill queue until batchsize or idle // for logging of the time between batch fill and batch process @@ -194,188 +113,39 @@ function flatten(obj) { clearTimeout(idle_timer); idle_timer = setTimeout(process, IDLE_TIMEOUT); // maximum data pressure - if (match_data.size + player_data.size + telemetry_data.size == BATCHSIZE) - await process(); + if (telemetry_data.size == BATCHSIZE) + await tryProcess(); }, { noAck: false }); - // finish a whole batch - async function process() { + // wrap process() in message handler + async function tryProcess() { + const msgs = new Set(msg_buffer); + msg_buffer.clear(); + profiler.done("buffer filled"); profiler = undefined; logger.info("processing batch", { - players: player_data.size, - matches: match_data.size, telemetries: telemetry_data.size }); // clean up to allow processor to accept while we wait for db clearTimeout(idle_timer); clearTimeout(load_timer); - const player_objects = new Set(player_data), - match_objects = new Set(match_data), - telemetry_objects = new Set(telemetry_data), - msgs = new Set(msg_buffer); + } + + // finish a whole batch + async function process() { + const telemetry_objects = new Set(telemetry_data); idle_timer = undefined; load_timer = undefined; - player_data.clear(); - match_data.clear(); telemetry_data.clear(); - msg_buffer.clear(); - - const processed_players = new Set(); // to sort out duplicates // aggregate record objects to do a bulk insert - let match_records = [], - roster_records = [], - participant_records = [], - participant_stats_records = [], - participant_phase_records = [], // Telemetry - player_records = [], - player_records_direct = [], // via `/players` - asset_records = []; - - // populate `_records` - // data from `/players` - // `each` executes serially so there are - // no race conditions within one batch - await Promise.each(player_objects, async (p) => { - const player = flatten(p); - if (processed_players.has(player.api_id)) { - // duplicate within one batch - logger.warn("got player in additional region", - { name: player.name, region: player.shard_id }); - const duplicate = player_records_direct.find((pr) => - pr.api_id == player.api_id); - if (duplicate == undefined) return; // TODO wtf??? - if (duplicate.created_at < player.created_at) { - // replace by newer one as below - player_records_direct.splice( - player_records_direct.indexOf(duplicate), 1); - } else { - logger.warn("ignoring player from additional region", - { name: player.name, region: player.shard_id }); - return; - } - } else { - processed_players.add(player.api_id); - } - // player objects that arrive here came from a search - // with search, updater can't update last_update - player.last_update = seq.fn("NOW"); - logger.info("processing player", - { name: player.name, region: player.shard_id }); - // duplicate in batch and db - // check whether there is a player in db - // that has a more recent `created_at` - // this is only the case with region changes - const count = await model.Player.count({ where: { - api_id: player.api_id, - created_at: { - $gt: player.created_at // equal: just update last_update - } - }}); - if (count > 0) { - logger.warn("ignoring player who seems to have switched from region", - { name: player.name, region: player.shard_id }); - return; - } else player_records_direct.push(player); - }); - - // reject invalid matches (handling API bugs) TODO should happen in apigrabber - match_objects.forEach((match, idx) => { - // it is really `"null"`. - if (match.rosters[0].id == "null") delete match_objects[idx]; - }); - - // data from `/matches` - match_objects.forEach((match) => { - // flatten jsonapi nested response into our db structure-like shape - // also, push missing fields - match.rosters = match.rosters.map((roster) => { - roster.matchApiId = match.id; - // TODO backwards compatibility, all objects have shardId since May 10th - roster.attributes.shardId = roster.attributes.shardId || match.attributes.shardId; - roster.createdAt = match.createdAt; - // TODO API workaround: roster does not have `winner` - if (roster.participants.length > 0) - roster.attributes.stats.winner = roster.participants[0].stats.winner; - else // Blitz 2v0, see 095e86e4-1bd3-11e7-b0b1-0297c91b7699 on eu - roster.attributes.stats.winner = false; - - roster.participants = roster.participants.map((participant) => { - // ! attributes added here need to be added via `calculate_participant_stats` too - participant.attributes.shardId = participant.attributes.shardId || roster.attributes.shardId; - participant.rosterApiId = roster.id; - participant.matchApiId = match.id; - participant.createdAt = roster.createdAt; - participant.playerApiId = participant.player.id; - - // API bug fixes (TODO) - // items on AFK is `null` not `{}` - participant.attributes.stats.itemGrants = participant.attributes.stats.itemGrants || {}; - participant.attributes.stats.itemSells = participant.attributes.stats.itemSells || {}; - participant.attributes.stats.itemUses = participant.attributes.stats.itemUses || {}; - // jungle_kills is `null` in BR - participant.attributes.stats.jungleKills = participant.attributes.stats.jungleKills || 0; - - // map items: names/id -> name -> db - const item_id = ((i) => item_db_map.get(api_name_mappings.get(i))); - let itms = []; - - const pas = participant.attributes.stats; // I'm lazy - // csv - participant.attributes.stats.items = - pas.items.map((i) => item_id(i).toString()).join(","); - // csv with count seperated by ; - participant.attributes.stats.itemGrants = - Object.keys(pas.itemGrants) - .map((key) => item_id(key) + ";" + pas.itemGrants[key]).join(","); - participant.attributes.stats.itemUses = - Object.keys(pas.itemUses) - .map((key) => item_id(key) + ";" + pas.itemUses[key]).join(","); - participant.attributes.stats.itemSells = - Object.keys(pas.itemSells) - .map((key) => item_id(key) + ";" + pas.itemSells[key]).join(","); - - participant.player.attributes.shardId = participant.player.attributes.shardId || participant.attributes.shardId; - participant.player = flatten(participant.player); - return flatten(participant); - }); - return flatten(roster); - }); - match.assets = match.assets.map((asset) => { - asset.matchApiId = match.id; - asset.attributes.shardId = asset.attributes.shardId || match.attributes.shardId; - return flatten(asset); - }); - match = flatten(match); - - // after conversion, create the array of records - match_records.push(match); - match.rosters.forEach((r) => { - roster_records.push(r); - r.participants.forEach((p) => { - const p_pstats = calculate_participant_stats(match, r, p); - // participant gets split into participant and p_stats - participant_records.push(p_pstats[0]); - participant_stats_records.push(p_pstats[1]); - // deduplicate player - // in a batch, it is very likely that players are duplicated - // so this improves performance a bit - if (!processed_players.has(p.player.api_id)) { - processed_players.add(p.player.api_id); - player_records.push(p.player); - } - }); - }); - match.assets.forEach((a) => asset_records.push(a)); - }); - + let participant_phase_records = []; // data from Telemetry, one phase (early/mid/late/…) per obj await Promise.map(telemetry_objects, async (telemetry) => { - if (telemetry.data.length == 0) return; // TODO rm me // api -> telemetry format const sideToTeam = (s) => s == "left/blue"? "Left" : "Right", // yes there is yet another format and yes it's strings @@ -442,13 +212,7 @@ function flatten(obj) { p.actor == t.payload.Killed && p.team == t.payload.KilledTeam)[0]; }); - /* - telemetry.data.forEach((ev) => { // TODO debug - if (ev.payload.Ability == undefined || ev.payload.IsHero == 0) return; - if (!api_name_mappings.has(ev.payload.Ability)) - console.error("ab to name map missing", ev.payload.Ability); - }); - */ + const participants_phase = participants.map((p) => { return { // TODO ban data workaround start: telemetry.start < 0? 0 : telemetry.start, // in seconds @@ -713,58 +477,10 @@ function flatten(obj) { // upsert whole batch in parallel logger.info("inserting batch into db"); await seq.transaction({ autocommit: false }, async (transaction) => { - await Promise.map(chunks(match_records), async (m_r) => - model.Match.bulkCreate(m_r, { - ignoreDuplicates: true, // if this happens, something is wrong - transaction: transaction - }), { concurrency: MAXCONNS } - ); - await Promise.map(chunks(roster_records), async (r_r) => - model.Roster.bulkCreate(r_r, { - ignoreDuplicates: true, - transaction: transaction - }), { concurrency: MAXCONNS } - ); - await Promise.map(chunks(participant_records), async (p_r) => - model.Participant.bulkCreate(p_r, { - ignoreDuplicates: true, - transaction: transaction - }), { concurrency: MAXCONNS } - ); - await Promise.map(chunks(participant_stats_records), async (p_s_r) => - model.ParticipantStats.bulkCreate(p_s_r, { - ignoreDuplicates: true, - transaction: transaction - }), { concurrency: MAXCONNS } - ); await Promise.map(chunks(participant_phase_records), async (p_p_r) => model.ParticipantPhases.bulkCreate(p_p_r, { - /* ignoreDuplicates: true, TODO DEBUG */ - ignoreDuplicates: false, - updateOnDuplicate: [], - transaction: transaction - }), { concurrency: MAXCONNS } - ); - await Promise.map(chunks(player_records), async (pl_r) => - model.Player.bulkCreate(pl_r, { - ignoreDuplicates: true, - transaction: transaction - }), { concurrency: MAXCONNS } - ); - await Promise.map(chunks(player_records_direct), async (p_r_d) => - model.Player.bulkCreate(player_records_direct, { - // if set to [] (all), upsert messes with autoincrement - updateOnDuplicate: [ - "shard_id", "api_id", "name", "last_update", - "created_at", "level", "xp", "lifetime_gold", - "skill_tier" - ], - transaction: transaction - }), { concurrency: MAXCONNS } - ); - await Promise.map(chunks(asset_records), async (a_r) => - model.Asset.bulkCreate(a_r, { ignoreDuplicates: true, + updateOnDuplicate: [], transaction: transaction }), { concurrency: MAXCONNS } ); @@ -773,153 +489,38 @@ function flatten(obj) { logger.info("acking batch", { size: msgs.size }); await Promise.map(msgs, async (m) => await ch.ack(m)); } catch (err) { - // this should only happen for Deadlocks in prod - // it *must not* fail due to broken schema or missing dependency - // TODO: eliminate such cases earlier in the chain - // and immediately NACK those broken matches, requeueing only the rest - logger.error("SQL error", err); - await Promise.map(msgs, async (m) => await ch.nack(m, false, true)); - return; // give up + if (err instanceof Seq.TimeoutError) { + // deadlocks / timeout + logger.error("SQL error", err); + await Promise.map(msgs, async (m) => + await ch.nack(m, false, true)); // retry + } else { + // log, move to error queue and NACK + logger.error(err); + await Promise.map(msgs, async (m) => { + await ch.sendToQueue(QUEUE + "_failed", m.content, { + persistent: true, + headers: m.properties.headers + }); + await ch.nack(m, false, false); + }); + } } transaction_profiler.done("database transaction"); // notify web await Promise.map(msgs, async (m) => { if (m.properties.headers.notify == undefined) return; - let notif = "error"; - // new match - if (m.properties.type == "match") { - notif = "matches_update"; - // TODO this sends match_update for every match in the batch to every player - // notify player.name.api_id about match_update - await Promise.map(match_records, async (mat) => - await ch.publish("amq.topic", - m.properties.headers.notify + "." + mat.api_id, - new Buffer("match_update")) - ); - await ch.publish("amq.topic", m.properties.headers.notify, - new Buffer("match_update")); - } - // player obj updated - if (m.properties.type == "player") notif = "stats_update"; // new phases - if (m.properties.type == "telemetry") { - // notify match.api_id about phase_update - await ch.publish("amq.topic", - m.properties.headers.notify, - new Buffer("phase_update")) - } - - if (m.properties.headers.donotify == true) // TODO remove later - await ch.publish("amq.topic", m.properties.headers.notify, - new Buffer(notif)); + // notify match.api_id about phase_update + await ch.publish("amq.topic", + m.properties.headers.notify, + new Buffer("phase_update")) }); - // …global about new matches - if (match_records.length > 0) - await ch.publish("amq.topic", "global", new Buffer("matches_update")); - // notify follow up services - if (DOANALYZEMATCH) - await Promise.each(match_records, async (m) => - await ch.sendToQueue(ANALYZE_QUEUE, new Buffer(m.api_id), - { persistent: true })); } +}); - // Split participant API data into participant and participant_stats - // Should not need to query db here. - function calculate_participant_stats(match, roster, participant) { - let p_s = {}, // participant_stats_record - p = {}; // participant_record - - // copy all values that are required in db `participant` to `p`/`p_s` here - // meta - p_s.participant_api_id = participant.api_id; - p_s.final = true; // these are the stats at the end of the match - p_s.updated_at = new Date(); - p_s.created_at = new Date(Date.parse(match.created_at)); - p_s.created_at.setMinutes(p_s.created_at.getMinutes() + match.duration / 60); - p_s.items = participant.items; - p_s.item_grants = participant.item_grants; - p_s.item_uses = participant.item_uses; - p_s.item_sells = participant.item_sells; - p_s.duration = match.duration; - p.created_at = match.created_at; - // mappings - // hero names additionally need to be mapped old to new names - // (Sayoc = Taka) - p.hero_id = hero_db_map.get(api_name_mappings.get(participant.actor)); - if (match.patch_version != "") - p.series_id = series_db_map.get("Patch " + match.patch_version); - else { - if (p_s.created_at < new Date("2017-03-28T15:00:00")) - p.series_id = series_db_map.get("Patch 2.2"); - else if (p_s.created_at < new Date("2017-04-26T15:00:00")) - p.series_id = series_db_map.get("Patch 2.3"); - else p.series_id = series_db_map.get("Patch 2.4"); - } - p.game_mode_id = game_mode_db_map.get(match.game_mode); - - // attributes to copy from API to participant - // these don't change over the duration of the match - // (or aren't in Telemetry) - ["api_id", "shard_id", "player_api_id", "roster_api_id", "match_api_id", - "winner", "went_afk", "first_afk_time", - "skin_key", "skill_tier", "level", - "karma_level", "actor"].map((attr) => - p[attr] = participant[attr]); - - // attributes to copy from API to participant_stats - // with Telemetry, these will be calculated in intervals - ["kills", "deaths", "assists", "minion_kills", - "jungle_kills", "non_jungle_minion_kills", - "crystal_mine_captures", "gold_mine_captures", - "kraken_captures", "turret_captures", - "gold", "farm"].map((attr) => - p_s[attr] = participant[attr]); - - let role = classify_role(p_s); - - // score calculations - let impact_score = 50; - switch (role) { - case "carry": - impact_score = -0.47249153 + 0.50145197 * p_s.assists - 0.7136091 * p_s.deaths + 0.18712844 * p_s.kills + 0.00531455 * p_s.farm; - p_s.nacl_score = 1 * p_s.kills + 0.5 * p_s.assists + 0.03 * p_s.farm + 5 * p.winner; - break; - case "jungler": - impact_score = -0.54510754 + 0.19982097 * p_s.assists - 0.35694721 * p_s.deaths + 0.09942473 * p_s.kills + 0.01256313 * p_s.farm; - p_s.nacl_score = 1 * p_s.kills + 0.5 * p_s.assists + 0.04 * p_s.farm + 5 * p.winner; - break; - case "captain": - impact_score = -0.46473539 + 0.09968104 * p_s.assists - 0.38401479 * p_s.deaths + 0.14753133 * p_s.kills + 0.03431293 * p_s.farm; - p_s.nacl_score = 1 * p_s.kills + 1 * p_s.assists + 5 * p.winner; - break; - } - p_s.impact_score = (impact_score - (-4.5038622921659375) ) / (4.431094119937388 - (-4.5038622921659375) ); - - - // classifications - p.role_id = role_db_map.get(role); - - // traits calculations - if (roster.hero_kills == 0) p_s.kill_participation = 0; - else p_s.kill_participation = (p_s.kills + p_s.assists) / roster.hero_kills; - - return [p, p_s]; - } - - // return "captain" "carry" "jungler" - function classify_role(participant_stats) { - const is_captain_score = 2.34365487 + (-0.06188674 * participant_stats.non_jungle_minion_kills) + (-0.10575069 * participant_stats.jungle_kills), // about 88% accurate, trained on Hero.is_captain - is_carry_score = -1.88524473 + (0.05593593 * participant_stats.non_jungle_minion_kills) + (-0.0881661 * participant_stats.jungle_kills), // about 90% accurate, trained on Hero.is_carry - is_jungle_score = -0.78327066 + (-0.03324596 * participant_stats.non_jungle_minion_kills) + (0.10514832 * participant_stats.jungle_kills); // about 88% accurate - if (is_captain_score > is_carry_score && is_captain_score > is_jungle_score) - return "captain"; - if (is_carry_score > is_jungle_score) - return "carry"; - return "jungler"; - } -})(); - -process.on("unhandledRejection", function(reason, promise) { - logger.error(reason); +process.on("unhandledRejection", (err) => { + logger.error(err); + process.exit(1); // fail hard and die }); |
