Merge pull request #314 from dtaniwaki/support-prometheus

Support prometheus metrics

Author: minrk

bin/configurable-http-proxy

@@ -87,9 +87,8 @@ cli
   )
   .option("--insecure", "Disable SSL cert verification")
   .option("--host-routing", "Use host routing (host as first level of path)")
-  .option("--statsd-host <host>", "Host to send statsd statistics to")
-  .option("--statsd-port <port>", "Port to send statsd statistics to", parseInt)
-  .option("--statsd-prefix <prefix>", "Prefix to use for statsd statistics")
+  .option("--metrics-ip <ip>", "IP for metrics server", "0.0.0.0")
+  .option("--metrics-port <n>", "Port of metrics server. Defaults to no metrics server")
   .option("--log-level <loglevel>", "Log level (debug, info, warn, error)", "info")
   .option(
     "--timeout <n>",
@@ -266,14 +265,8 @@ options.headers = args.customHeader;
 options.timeout = args.timeout;
 options.proxyTimeout = args.proxyTimeout;
 
-// statsd options
-if (args.statsdHost) {
-  var lynx = require("lynx");
-  options.statsd = new lynx(args.statsdHost, args.statsdPort || 8125, {
-    scope: args.statsdPrefix || "chp",
-  });
-  log.info("Sending metrics to statsd at " + args.statsdHost + ":" + args.statsdPort || 8125);
-}
+// metrics options
+options.enableMetrics = !!args.metricsPort;
 
 // certs need to be provided for https redirection
 if (!options.ssl && options.redirectPort) {
@@ -326,9 +319,14 @@ if (args.ip === "*") {
 listen.ip = args.ip;
 listen.apiIp = args.apiIp || "localhost";
 listen.apiPort = args.apiPort || listen.port + 1;
+listen.metricsIp = args.metricsIp || "0.0.0.0";
+listen.metricsPort = args.metricsPort;
 
 proxy.proxyServer.listen(listen.port, listen.ip);
 proxy.apiServer.listen(listen.apiPort, listen.apiIp);
+if (listen.metricsPort) {
+  proxy.metricsServer.listen(listen.metricsPort, listen.metricsIp);
+}
 
 log.info(
   "Proxying %s://%s:%s to %s",
@@ -343,6 +341,9 @@ log.info(
   listen.apiIp || "*",
   listen.apiPort
 );
+if (listen.metricsPort) {
+  log.info("Serve metrics at %s://%s:%s/metrics", "http", listen.metricsIp, listen.metricsPort);
+}
 
 if (args.pidFile) {
   log.info("Writing pid %s to %s", process.pid, args.pidFile);

lib/configproxy.js

@@ -18,7 +18,8 @@ var http = require("http"),
   util = require("util"),
   URL = require("url"),
   defaultLogger = require("./log").defaultLogger,
-  querystring = require("querystring");
+  querystring = require("querystring"),
+  metrics = require("./metrics");
 
 function bound(that, method) {
   // bind a method, to ensure `this=that` when it is called
@@ -159,23 +160,11 @@ class ConfigurableProxy extends EventEmitter {
       this.errorTarget = this.errorTarget + "/"; // ensure trailing /
     }
     this.errorPath = options.errorPath || path.join(__dirname, "error");
-    if (options.statsd) {
-      this.statsd = options.statsd;
+
+    if (this.options.enableMetrics) {
+      this.metrics = new metrics.Metrics();
     } else {
-      // Mock the statsd object, rather than pepper the codebase with
-      // null checks. FIXME: Maybe use a JS Proxy object (if available?)
-      this.statsd = {
-        increment: function () {},
-        decrement: function () {},
-        timing: function () {},
-        gauge: function () {},
-        set: function () {},
-        createTimer: function () {
-          return {
-            stop: function () {},
-          };
-        },
-      };
+      this.metrics = new metrics.MockMetrics();
     }
 
     if (this.options.defaultTarget) {
@@ -224,6 +213,12 @@ class ConfigurableProxy extends EventEmitter {
       this.apiServer = http.createServer(apiCallback);
     }
 
+    // handle metrics
+    if (this.options.enableMetrics) {
+      var metricsCallback = logErrors(that.handleMetrics);
+      this.metricsServer = http.createServer(metricsCallback);
+    }
+
     // proxy requests separately
     var proxyCallback = logErrors(this.handleProxyWeb);
     if (this.options.ssl) {
@@ -235,7 +230,7 @@ class ConfigurableProxy extends EventEmitter {
     this.proxyServer.on("upgrade", bound(this, this.handleProxyWs));
 
     this.proxy.on("proxyRes", function (proxyRes, req, res) {
-      that.statsd.increment("requests." + proxyRes.statusCode, 1);
+      that.metrics.requestsProxyCount.labels(proxyRes.statusCode).inc();
     });
   }
 
@@ -265,8 +260,8 @@ class ConfigurableProxy extends EventEmitter {
     var that = this;
 
     return this._routes.add(path, data).then(() => {
-      this.updateLastActivity(path);
-      this.log.info("Route added %s -> %s", path, data.target);
+      that.updateLastActivity(path);
+      that.log.info("Route added %s -> %s", path, data.target);
     });
   }
 
@@ -341,7 +336,7 @@ class ConfigurableProxy extends EventEmitter {
 
       res.write(JSON.stringify(results));
       res.end();
-      that.statsd.increment("api.route.get", 1);
+      that.metrics.apiRouteGetCount.inc();
     });
   }
 
@@ -359,7 +354,7 @@ class ConfigurableProxy extends EventEmitter {
     return this.addRoute(path, data).then(function () {
       res.writeHead(201);
       res.end();
-      that.statsd.increment("api.route.add", 1);
+      that.metrics.apiRouteAddCount.inc();
     });
   }
 
@@ -378,19 +373,19 @@ class ConfigurableProxy extends EventEmitter {
       return p.then(() => {
         res.writeHead(code);
         res.end();
-        this.statsd.increment("api.route.delete", 1);
+        this.metrics.apiRouteDeleteCount.inc();
       });
     });
   }
 
   targetForReq(req) {
-    var timer = this.statsd.createTimer("find_target_for_req");
+    var metricsTimerEnd = this.metrics.findTargetForReqSummary.startTimer();
     // return proxy target for a given url path
     var basePath = this.hostRouting ? "/" + parseHost(req) : "";
     var path = basePath + decodeURIComponent(URL.parse(req.url).pathname);
 
     return this._routes.getTarget(path).then(function (route) {
-      timer.stop();
+      metricsTimerEnd();
       if (route) {
         return {
           prefix: route.prefix,
@@ -401,7 +396,7 @@ class ConfigurableProxy extends EventEmitter {
   }
 
   updateLastActivity(prefix) {
-    var timer = this.statsd.createTimer("last_activity_updating");
+    var metricsTimerEnd = this.metrics.lastActivityUpdatingSummary.startTimer();
     var routes = this._routes;
 
     return routes
@@ -411,7 +406,7 @@ class ConfigurableProxy extends EventEmitter {
           return routes.update(prefix, { last_activity: new Date() });
         }
       })
-      .then(timer.stop);
+      .then(metricsTimerEnd);
   }
 
   _handleProxyErrorDefault(code, kind, req, res) {
@@ -430,7 +425,7 @@ class ConfigurableProxy extends EventEmitter {
     // /404?url=%2Fuser%2Ffoo
 
     var errMsg = "";
-    this.statsd.increment("requests." + code, 1);
+    this.metrics.requestsProxyCount.labels(code).inc();
     if (e) {
       // avoid stack traces on known not-our-problem errors:
       // ECONNREFUSED, EHOSTUNREACH (backend isn't there)
@@ -512,7 +507,7 @@ class ConfigurableProxy extends EventEmitter {
           this._handleProxyErrorDefault(code, kind, req, res);
           return;
         }
-        if (res.writableEnded) return; // response already done
+        if (!res.writable) return; // response already done
         if (res.writeHead) res.writeHead(code, { "Content-Type": "text/html" });
         if (res.write) res.write(data);
         if (res.end) res.end();
@@ -603,15 +598,15 @@ class ConfigurableProxy extends EventEmitter {
 
   handleProxyWs(req, socket, head) {
     // Proxy a websocket request
-    this.statsd.increment("requests.ws", 1);
+    this.metrics.requestsWsCount.inc();
     return this.handleProxy("ws", req, socket, head);
   }
 
   handleProxyWeb(req, res) {
     this.handleHealthCheck(req, res);
     if (res.finished) return;
     // Proxy a web request
-    this.statsd.increment("requests.web", 1);
+    this.metrics.requestsWebCount.inc();
     return this.handleProxy("web", req, res);
   }
 
@@ -623,11 +618,18 @@ class ConfigurableProxy extends EventEmitter {
     }
   }
 
+  handleMetrics(req, res) {
+    if (req.url === "/metrics") {
+      return this.metrics.render(res);
+    }
+    fail(req, res, 404);
+  }
+
   handleApiRequest(req, res) {
     // Handle a request to the REST API
-    this.statsd.increment("requests.api", 1);
     if (res) {
       res.on("finish", () => {
+        this.metrics.requestsApiCount.labels(res.statusCode).inc();
         this.logResponse(req, res);
       });
     }

lib/metrics.js

@@ -0,0 +1,113 @@
+"use strict";
+
+var client = require("prom-client");
+
+class Metrics {
+  constructor() {
+    this.register = new client.Registry();
+    client.collectDefaultMetrics({ register: this.register });
+
+    this.apiRouteGetCount = new client.Counter({
+      name: "api_route_get",
+      help: "Count of API route get requests",
+      registers: [this.register],
+    });
+
+    this.apiRouteAddCount = new client.Counter({
+      name: "api_route_add",
+      help: "Count of API route add requests",
+      registers: [this.register],
+    });
+
+    this.apiRouteDeleteCount = new client.Counter({
+      name: "api_route_delete",
+      help: "Count of API route delete requests",
+      registers: [this.register],
+    });
+
+    this.findTargetForReqSummary = new client.Summary({
+      name: "find_target_for_req",
+      help: "Summary of find target requests",
+      registers: [this.register],
+    });
+
+    this.lastActivityUpdatingSummary = new client.Summary({
+      name: "last_activity_updating",
+      help: "Summary of last activity updating requests",
+      registers: [this.register],
+    });
+
+    this.requestsWsCount = new client.Counter({
+      name: "requests_ws",
+      help: "Count of websocket requests",
+      registers: [this.register],
+    });
+
+    this.requestsWebCount = new client.Counter({
+      name: "requests_web",
+      help: "Count of web requests",
+      registers: [this.register],
+    });
+
+    this.requestsProxyCount = new client.Counter({
+      name: "requests_proxy",
+      help: "Count of proxy requests",
+      labelNames: ["status"],
+      registers: [this.register],
+    });
+
+    this.requestsApiCount = new client.Counter({
+      name: "requests_api",
+      help: "Count of API requests",
+      labelNames: ["status"],
+      registers: [this.register],
+    });
+  }
+
+  render(res) {
+    return this.register.metrics().then((s) => {
+      res.writeHead(200, { "Content-Type": this.register.contentType });
+      res.write(s);
+      res.end();
+    });
+  }
+}
+
+class MockMetrics {
+  constructor() {
+    return new Proxy(this, {
+      get(target, name) {
+        const mockCounter = new Proxy(
+          {},
+          {
+            get(target, name) {
+              if (name == "inc") {
+                return () => {};
+              }
+              if (name == "startTimer") {
+                return () => {
+                  return () => {};
+                };
+              }
+              if (name == "labels") {
+                return () => {
+                  return mockCounter;
+                };
+              }
+            },
+          }
+        );
+        return mockCounter;
+      },
+    });
+  }
+
+  render(res) {
+    return Promise.resolve();
+  }
+}
+
+module.exports = {
+  Metrics,
+  MockMetrics,
+};

lib/testutil.js

@@ -128,12 +128,18 @@ exports.setupProxy = function (port, options, paths) {
 
   servers.push(proxy.apiServer);
   servers.push(proxy.proxyServer);
+  if (options.enableMetrics) {
+    servers.push(proxy.metricsServer);
+  }
   proxy.apiServer.on("listening", onlisten);
   proxy.proxyServer.on("listening", onlisten);
 
   addTargets(proxy, paths || ["/"], port + 2).then(function () {
     proxy.proxyServer.listen(port, ip);
     proxy.apiServer.listen(port + 1, ip);
+    if (options.enableMetrics) {
+      proxy.metricsServer.listen(port + 3, ip);
+    }
   });
   return p;
 };