init worker reaperrelease/2.17.0

4 files changed, 171 insertions, 113 deletions

diff --git a/Dockerfile b/Dockerfile
new file mode 100644
index 0000000..4b3b904
--- /dev/null
+++ b/Dockerfile
@@ -0,0 +1,10 @@
+FROM node:alpine
+
+RUN mkdir -p /usr/src/app
+WORKDIR /usr/src/app
+
+COPY package.json .
+RUN npm install && npm cache clean --force
+COPY . .
+
+CMD ["node", "worker.js"]
diff --git a/app.js b/app.js
deleted file mode 100644
index d26dd84..0000000
--- a/app.js
+++ /dev/null
@@ -1,113 +0,0 @@
-#!/usr/bin/node
-/* jshint esnext:true */
-"use strict";
-
-const winston = require("winston"),
-    loggly = require("winston-loggly-bulk"),
-    Seq = require("sequelize"),
-    elasticsearch = require("elasticsearch");
-
-const DATABASE_URI = process.env.DATABASE_URI,
-    ELASTIC_URI = process.env.ELASTIC_URI || "localhost:9200",
-    LOGGLY_TOKEN = process.env.LOGGLY_TOKEN,
-    BATCHSIZE = process.env.BATCHSIZE || 50,
-    MAXCONNS = parseInt(process.env.MAXCONNS) || 20;
-
-const logger = new (winston.Logger)({
-    transports: [
-        new (winston.transports.Console)({
-            timestamp: true,
-            colorize: true
-        })
-    ]
-});
-
-// loggly integration
-if (LOGGLY_TOKEN)
-    logger.add(winston.transports.Loggly, {
-        inputToken: LOGGLY_TOKEN,
-        subdomain: "kvahuja",
-        tags: ["backend", "reaper"],
-        json: true
-    });
-
-
-// connect to rabbit & db
-const seq = new Seq(DATABASE_URI, {
-        logging: false,
-        max: MAXCONNS
-    }),
-    model = require("../orm/model")(seq, Seq),
-    elastic = new elasticsearch.Client({ host: ELASTIC_URI, log: "info" });
-
-// assumes `id` exists
-// Sequelize model, index type, index key, key to parent id, hook, custom condition
-async function load(table, type, includes, filter) {
-    // load from biggest id to 0
-    const last_id_r = await model.Keys.findOrCreate({
-            where: { type: "reaper_last_id_fetched", key: type },
-            defaults: { value: 0 }
-        });
-    while (true) {
-        const last_id = last_id_r[0].value,
-            condition = Object.assign({}, filter,
-                { id: { $gt: last_id } });
-
-        logger.info("loading", { type, last_id });
-
-        let data = await table.findAll({
-            where: condition,
-            order: [ [seq.col("id"), "DESC"] ],
-            include: includes,
-            limit: BATCHSIZE,
-            raw: true
-        });
-        if (data.length == 0) break;  // exhausted
-        await last_id_r[0].update({ value: data[data.length-1].id });
-
-        await elastic.bulk({
-            body: [].concat(... data.map((d) => [
-                { index: {
-                    _index: type,
-                    _type: type,
-                    _id: d.api_id || d.id
-                } },
-                d
-            ]) )
-        });
-    }
-
-    logger.info("done.", { type });
-}
-
-(async function() {
-    await Promise.all([
-        load(model.Participant, "participant", [
-            model.ParticipantStats,
-            model.Roster,
-            model.Match,
-
-            model.Region, model.Hero, model.Series, model.GameMode, model.Role
-        ]),
-        load(model.ParticipantPhases, "participant_phases", [ {
-            model: model.Participant,
-            include: [
-                model.Region, model.Hero, model.Series, model.GameMode, model.Role
-            ]
-        } ]),
-        //load(model.Match, "match", [ model.Asset ]),
-
-        load(model.PlayerPoint, "player_point", [ {
-            model: model.Player,
-            include: [ model.Region ]
-        },
-            model.Series, model.Hero, model.GameMode, model.Role
-        ]),
-        load(model.Player, "player", [ model.Region ]),
-    ]);
-})();
-
-process.on("unhandledRejection", (err) => {
-    logger.error(err);
-    //process.exit();
-});
diff --git a/package.json b/package.json
index b0767f4..4b1a4ee 100644
--- a/package.json
+++ b/package.json
@@ -4,6 +4,7 @@
   "description": "",
   "main": "app.js",
   "dependencies": {
+    "amqplib": "^0.5.1",
     "bluebird": "^3.5.0",
     "elasticsearch": "^13.2.0",
     "mysql2": "^1.3.6",
diff --git a/worker.js b/worker.js
new file mode 100644
index 0000000..8cb6595
--- /dev/null
+++ b/worker.js
@@ -0,0 +1,160 @@
+#!/usr/bin/node
+/* jshint esnext:true */
+"use strict";
+
+const amqp = require("amqplib"),
+    Promise = require("bluebird"),
+    winston = require("winston"),
+    loggly = require("winston-loggly-bulk"),
+    Seq = require("sequelize"),
+    elasticsearch = require("elasticsearch");
+
+const RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost",
+    DATABASE_URI = process.env.DATABASE_URI,
+    ELASTICSEARCH_URI = process.env.ELASTICSEARCH_URI || "localhost:9200",
+    QUEUE = process.env.QUEUE || "reap",
+    LOGGLY_TOKEN = process.env.LOGGLY_TOKEN,
+    BATCHSIZE = parseInt(process.env.BATCHSIZE) || 10,
+    IDLE_TIMEOUT = parseInt(process.env.IDLE_TIMEOUT) || 1000,  // ms
+    MAXCONNS = parseInt(process.env.MAXCONNS) || 20;
+
+const logger = new (winston.Logger)({
+    transports: [
+        new (winston.transports.Console)({
+            timestamp: true,
+            colorize: true
+        })
+    ]
+});
+
+// loggly integration
+if (LOGGLY_TOKEN)
+    logger.add(winston.transports.Loggly, {
+        inputToken: LOGGLY_TOKEN,
+        subdomain: "kvahuja",
+        tags: ["backend", "reaper", QUEUE],
+        json: true
+    });
+
+
+amqp.connect(RABBITMQ_URI).then(async (rabbit) => {
+    process.on("SIGINT", () => {
+        rabbit.close();
+        process.exit();
+    });
+
+    // connect to rabbit & db
+    const seq = new Seq(DATABASE_URI, {
+            logging: false,
+            pool: {
+                max: MAXCONNS
+            }
+        }),
+        elastic = new elasticsearch.Client({ host: ELASTICSEARCH_URI, log: "info" });
+
+    const ch = await rabbit.createChannel();
+    await ch.assertQueue(QUEUE, { durable: true });
+    await ch.assertQueue(QUEUE + "_failed", { durable: true });
+    await ch.prefetch(BATCHSIZE);
+
+    logger.info("configuration", {
+        QUEUE, BATCHSIZE, MAXCONNS, IDLE_TIMEOUT
+    });
+
+    const model = require("../orm/model")(seq, Seq);
+
+    let phase_data = new Set();
+    let msg_buffer = new Set();
+    let idle_timer = undefined;
+
+    ch.consume(QUEUE, async (msg) => {
+        const payload = JSON.parse(msg.content);
+        phase_data.add(payload);
+        msg_buffer.add(msg);
+
+        // timeout after last job
+        if (idle_timer != undefined)
+            clearTimeout(idle_timer);
+        idle_timer = setTimeout(tryProcess, IDLE_TIMEOUT);
+        if (phase_data.size == BATCHSIZE)
+            await tryProcess();
+    }, { noAck: false });
+
+    // wrap process() in message handler
+    async function tryProcess() {
+        const msgs = new Set(msg_buffer);
+        msg_buffer.clear();
+
+        logger.info("processing batch");
+
+        // clean up to allow reaper to accept while we wait for db
+        clearTimeout(idle_timer);
+        idle_timer = undefined;
+
+        const phase_objects = new Set(phase_data);
+        phase_data.clear();
+
+        try {
+            await reap(phase_objects);
+
+            logger.info("acking batch", { size: msgs.size });
+            await Promise.map(msgs, async (m) => await ch.ack(m));
+        } catch (err) {
+            // 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 });
+                await ch.nack(m, false, false);
+            });
+        }
+    }
+
+    async function reap(phase_objects) {
+        const db_profiler = logger.startTimer(),
+            data = [].concat(... await Promise.map(phase_objects, async (po) => {
+                // TODO would be better to get all in bulk using the id
+                return await model.ParticipantPhases.findAll({
+                    where: {
+                        "$participant.match_api_id$": po.match_api_id,
+                        start: po.start,
+                        end: po.end
+                    },
+                    include: [ {
+                        model: model.Participant,
+                        include: [
+                            model.Region,
+                            model.Hero,
+                            model.Series,
+                            model.GameMode,
+                            model.Role,
+
+                            model.ParticipantStats,
+                            model.Roster,
+                            model.Match
+                        ]
+                    } ],
+                    raw: true
+                })
+            } ) );
+        db_profiler.done("database transaction");
+
+        const es_profiler = logger.startTimer();
+        await elastic.bulk({
+            body: [].concat(... data.map((d) => [
+                { index: {
+                    _index: "phase",
+                    _type: "phase",
+                    _id: d.id
+                } },
+                d
+            ]) )
+        });
+        es_profiler.done("elastic bulk request");
+    }
+});
+
+process.on("unhandledRejection", (err) => {
+    logger.error(err);
+    process.exit(1);  // fail hard and die
+});