Merge pull request #8 from emanguy/add-health-check

Add health check

.gitignore

@@ -2,4 +2,5 @@ build/
 node_modules/
 .idea/
 .env
-**/*.js
+**/*.js
+yarn-error.log

package.json

@@ -25,6 +25,8 @@
     "cors": "^2.8.5",
     "dotenv": "^5.0.1",
     "express": "^4.16.3",
+    "express-async-handler": "^1.1.4",
+    "http-status-codes": "^1.4.0",
     "morgan": "^1.9.0",
     "redis": "^2.8.0",
     "sse-channel": "^3.0.1",
@@ -39,6 +41,7 @@
     "@types/mocha": "^5.2.0",
     "chai": "^4.1.2",
     "dockerode": "^2.5.5",
-    "mocha": "^5.2.0"
+    "mocha": "^5.2.0",
+    "sleep-promise": "^8.0.1"
   }
 }

src/config.ts

@@ -19,4 +19,4 @@ if (!config.redisPassword) {
     throw new Error("No redis password found in env!");
 }
 
-export default config;
+export default config;

src/controllers/ClientController.ts

@@ -6,14 +6,14 @@ import pushServiceRetriever from "../services/PushService";
 const clientRestController = Router();
 const pushService = pushServiceRetriever();
 
-clientRestController.get("/", (req:WinstonRequest, res:Response) => {
+clientRestController.get("/", (req: WinstonRequest, res: Response) => {
     log.info("Hit base url.", req.winstonMetadata);
     res.status(200).send("Hello world!").end();
 });
 
-clientRestController.get("/register", (req:WinstonRequest, res:Response) => {
+clientRestController.get("/register", (req: WinstonRequest, res: Response) => {
     log.info("Registered new client.", req.winstonMetadata);
     pushService.addClient(req, res);
 });
 
-export default clientRestController;
+export default clientRestController;

src/controllers/HealthController.ts

@@ -0,0 +1,21 @@
+import {Response, Router} from "express";
+import redisServiceRetriever from "../services/RedisUpdaterService";
+import config from "../config";
+import * as asyncHandler from "express-async-handler";
+import {WinstonRequest} from "../middleware/logging-metadata";
+import * as HttpStatus from "http-status-codes";
+
+const healthRestController = Router();
+const redisService = redisServiceRetriever(config);
+
+healthRestController.get("/", asyncHandler(async (req: WinstonRequest, res: Response) => {
+    const connected = await redisService.sendTestMessage();
+
+    if (connected) {
+        res.sendStatus(HttpStatus.OK);
+    } else {
+        res.sendStatus(HttpStatus.INTERNAL_SERVER_ERROR);
+    }
+}));
+
+export default healthRestController;

src/index.ts

@@ -2,11 +2,12 @@ import * as express from "express";
 import {Response} from "express";
 import * as morgan from "morgan";
 import log from "./logger";
-import config from "./config";
 import addLoggingInfo from "./middleware/logging-metadata";
 import jsonParse from "./middleware/json-parse";
 import PushServiceClientController from "./controllers/ClientController";
+import HealthController from "./controllers/HealthController";
 import * as cors from "cors";
+import config from "./config";
 
 const app = express();
 
@@ -18,6 +19,7 @@ app.use(jsonParse);
 app.use(addLoggingInfo);
 
 app.use("/push/", PushServiceClientController);
+app.use("/healthz/", HealthController);
 
 app.get("/", (_, res:Response) => {
     res.status(200).send(`Quest tracker notification service -- version ${process.env.npm_package_version}`);

src/services/PushService.ts

@@ -1,25 +1,25 @@
 import {Request, Response} from "express";
 import {v4 as uuid} from "uuid";
 import redisService, {RedisUpdaterService} from "./RedisUpdaterService";
-import config from "../config";
 import {GenericAdd, GenericDeletion, GenericUpdate} from "common-interfaces/QuestInterfaces";
 import {MessageType} from "common-interfaces/NotificationInterfaces";
+import config from "../config";
 import SseChannel = require("sse-channel");
 
 let serviceInstance: PushService | null = null;
 
 export class PushService {
     private channel:SseChannel;
 
-    constructor(redisService:RedisUpdaterService) {
+    constructor(redisService: RedisUpdaterService) {
         this.channel = new SseChannel({jsonEncode: true, cors: {origins: "*"}});
 
         redisService.toNotifyOnAdd.push(this.addNewItem.bind(this));
         redisService.toNotifyOnUpdate.push(this.updateExistingItem.bind(this));
         redisService.toNotifyOnRemove.push(this.deleteItem.bind(this));
     }
 
-    addClient(req:Request, res:Response) {
+    addClient(req: Request, res: Response) {
         this.channel.addClient(req, res);
     }
 

src/services/RedisUpdaterService.ts

@@ -1,23 +1,39 @@
-import {createClient, RedisClient} from "redis";
+import {createClient, RedisClient, RetryStrategyOptions} from "redis";
 import {Configuration} from "../config";
 import log from "../logger";
 import {GenericAdd, GenericDeletion, GenericUpdate} from "common-interfaces/QuestInterfaces";
+import {RedisTestPayload} from "./customTypes/RedisTestTypes";
 
 type QuestAddedCallback = (update: GenericAdd) => void;
 type QuestUpdatedCallback = (update: GenericUpdate) => void;
 type QuestDeletedCallback = (update: GenericDeletion) => void;
+type TestListenerRegistration  = {
+    id: number;
+    callbackFn: (input: RedisTestPayload) => void;
+};
 
 let serviceSingleton: RedisUpdaterService | null = null;
 
 export class RedisUpdaterService {
-    private subscription:RedisClient;
-    public readonly toNotifyOnAdd:Array<QuestAddedCallback>;
-    public readonly toNotifyOnUpdate:Array<QuestUpdatedCallback>;
-    public readonly toNotifyOnRemove:Array<QuestDeletedCallback>;
+    public readonly toNotifyOnAdd: Array<QuestAddedCallback>;
+    public readonly toNotifyOnUpdate: Array<QuestUpdatedCallback>;
+    public readonly toNotifyOnRemove: Array<QuestDeletedCallback>;
+
+    /**
+     * Base client is used for typical redis stuff. Needed because the client goes into "subscriber mode" on subscribe.
+     *
+     *  @see https://github.com/noderedis/node_redis#publish--subscribe
+     */
+    private baseClient: RedisClient;
+    /** Subscription is used for listening to posted channel messages */
+    private subscription: RedisClient;
+
+    private toNotifyOnTest: Array<TestListenerRegistration>;
 
     private ADD_CHANNEL = "new-quests";
     private UPDATE_CHANNEL = "quest-updates";
     private REMOVE_CHANNEL = "removed-quests";
+    private TEST_CHANNEL = "test-connectivity";
     private RECONNECT_WAIT_TIME = 10000;
 
     constructor(config:Configuration) {
@@ -28,18 +44,24 @@ export class RedisUpdaterService {
         this.toNotifyOnAdd = [];
         this.toNotifyOnUpdate = [];
         this.toNotifyOnRemove = [];
+        this.toNotifyOnTest = [];
 
-        this.subscription = createClient(config.redisUrl, {
-            password: config.redisPassword,
-            retry_strategy: (options) => {
-
-                if (options.attempt > 12) {
-                    log.error("Lost connection with redis after 12 attempts! Shutting down server.");
-                    return Error("Could not connect to redis after 12 attempts.");
-                }
+        let strategy = (options: RetryStrategyOptions) => {
 
-                return this.RECONNECT_WAIT_TIME;
+            if (options.attempt > 12) {
+                log.error("Lost connection with redis after 12 attempts! Shutting down server.");
+                return Error("Could not connect to redis after 12 attempts.");
             }
+
+            return this.RECONNECT_WAIT_TIME;
+        };
+        this.baseClient = createClient(config.redisUrl, {
+            password: config.redisPassword,
+            retry_strategy: strategy
+        });
+        this.subscription = createClient(config.redisUrl, {
+            password: config.redisPassword,
+            retry_strategy: strategy
         });
 
         this.subscribeToRedisTopics();
@@ -51,13 +73,65 @@ export class RedisUpdaterService {
     // This is really only used in testing
     public disconnect() {
         this.subscription.quit();
+        this.baseClient.quit();
+    }
+
+    /**
+     * Sends a test message through redis pubsub with a unique value and waits 3 seconds to receive the value back.
+     * If we don't see the value within 3 seconds, we time out and say we can't reach redis.
+     *
+     * @return A promise which resolves true if we can talk to redis or false if we can't
+     */
+    public sendTestMessage(): Promise<boolean> {
+        return new Promise<boolean>((resolve) => {
+            const randomNum = Math.floor(1000 * Math.random());
+            const payload: RedisTestPayload = {
+                testValue: randomNum
+            };
+            let receivedInput = false;
+            log.debug("Launching promise");
+
+            // Option 1 - Receive response before timeout and our number matches. Resolve with true, our connection to redis is working
+            const listnenerFn = (response: RedisTestPayload) => {
+                // If we got a different value than the one we sent, ignore
+                log.debug("Got callback");
+                if (response.testValue !== randomNum) return;
+                receivedInput = true;
+                this.removeTestListenerByID(randomNum);
+                resolve(true);
+            };
+            // Option 2 - We don't receive a response after 3 seconds, we assume the redis connection isn't working and resolve with false
+            setTimeout(() => {
+                log.debug("Timed out");
+                if (receivedInput) return;
+                log.error(`Health test timeout, did not receive value (${randomNum}).`);
+                this.removeTestListenerByID(randomNum);
+                resolve(false);
+            }, 3000);
+
+            log.debug("Adding listener");
+            // Add the listener and send the message
+            this.toNotifyOnTest.push({
+                id: randomNum,
+                callbackFn: listnenerFn,
+            });
+
+            try {
+                log.debug("Publishing");
+                this.baseClient.publish(this.TEST_CHANNEL, JSON.stringify(payload));
+            } catch (err) {
+                log.error(`Failed to send health test message. Will time out shortly. Problem: ${err.message}`);
+                resolve(false);
+            }
+            log.debug("Done");
+        })
     }
 
     private subscribeToRedisTopics() {
         this.subscription.on("message", (channel:string, message:string) => {
-            let deserializedMessage: GenericAdd|GenericUpdate|GenericDeletion;
-
+            let deserializedMessage: GenericAdd|GenericUpdate|GenericDeletion|RedisTestPayload;
             try {
+                log.debug(`Got message: ${message} channel: ${channel}`);
                 deserializedMessage = JSON.parse(message);
             }
             catch (e) {
@@ -76,14 +150,26 @@ export class RedisUpdaterService {
             if (channel == this.REMOVE_CHANNEL) {
                 this.toNotifyOnRemove.forEach(async (updateFn) => updateFn(<GenericDeletion> deserializedMessage));
             }
+
+            if (channel == this.TEST_CHANNEL) {
+                this.toNotifyOnTest.forEach(async (registration) => registration.callbackFn(<RedisTestPayload> deserializedMessage));
+            }
         });
 
-        this.subscription.subscribe(this.ADD_CHANNEL, this.UPDATE_CHANNEL, this.REMOVE_CHANNEL);
+        this.subscription.subscribe(this.ADD_CHANNEL, this.UPDATE_CHANNEL, this.REMOVE_CHANNEL, this.TEST_CHANNEL);
+    }
+
+    private removeTestListenerByID(id: number) {
+        const listenerIdx = this.toNotifyOnTest.findIndex(registration => registration.id === id);
+        if (listenerIdx === -1) return;
+        this.toNotifyOnTest.splice(listenerIdx, 1);
     }
 
     private logErrors() {
         this.subscription.on("error", (err:Error) => {
             log.warn(`Got an error from the redis connector. Message: ${err.message}`);
+            // Kill the process so K8s can restart it
+            process.exit(1);
         })
     }
 }
@@ -94,4 +180,4 @@ export default (config:Configuration) => {
     }
 
     return serviceSingleton;
-};
+};

src/services/customTypes/RedisTestTypes.ts

@@ -0,0 +1,5 @@
+
+export interface RedisTestPayload {
+    testValue: number;
+}
+

test/Util.ts

@@ -0,0 +1,48 @@
+import * as Docker from "dockerode";
+import {ContainerCreateOptions} from "dockerode";
+import {Configuration} from "../src/config";
+
+interface HostPortBinding {
+    HostPort: string
+}
+
+interface CreateOptionsWithPortBindings extends ContainerCreateOptions {
+    PortBindings: {
+        [key: string]: HostPortBinding[]
+    }
+}
+
+const redisPort = "6379";
+
+export const testConfig: Configuration = {
+    applicationPort: 3000,
+    redisUrl: `redis://localhost:${redisPort}`,
+    environment: "testing",
+    redisPassword: "testRedis"
+};
+
+export function createRedisContainer(docker: Docker): Promise<Docker.Container> {
+    return docker.createContainer(<CreateOptionsWithPortBindings>{
+        Image: "bitnami/redis:4.0.9",
+        Env: [
+            `REDIS_PASSWORD=${testConfig.redisPassword}`
+        ],
+        PortBindings: {
+            "6379/tcp": [{HostPort: redisPort}]
+        }
+    });
+}
+
+export function pullRedisImage(docker: Docker): Promise<void> {
+    return new Promise((resolve, reject) => {
+        docker.pull("bitnami/redis:4.0.9", {}, (err, stream) => {
+            if (err) {
+                console.log("Got an error pulling the image.");
+                reject(err);
+                return;
+            }
+
+            docker.modem.followProgress(stream, (err?: Error) => err ? reject(err) : resolve(), () => {});
+        });
+    });
+}