metrics: integ with workers analytics

src/commons/envutil.js

@@ -278,3 +278,19 @@ export function gwip6() {
 export function region() {
   return envManager.get("FLY_REGION") || "";
 }
+
+export function metrics() {
+  const nobinding = [null, null];
+
+  if (!envManager) return nobinding;
+
+  // match the binding names as in wrangler.toml
+  if (onCloudflare()) {
+    return [
+      envManager.get("METRICS") || null,
+      envManager.get("BL_METRICS") || null,
+    ];
+  }
+
+  return nobinding;
+}

src/plugins/observability/log-pusher.js

@@ -9,7 +9,9 @@
 import * as util from "../../commons/util.js";
 import * as dnsutil from "../../commons/dnsutil.js";
 import * as envutil from "../../commons/envutil.js";
+import * as rdnsutil from "../rdns-util.js";
 
+const emptyarr = [];
 const emptystring = "";
 // current logpush version
 const ver = "1";
@@ -52,7 +54,8 @@ const logdelim = ",";
  */
 export class LogPusher {
   constructor() {
-    this.log = console.log;
+    this.remotelog = console.log;
+    this.corelog = log.withTags("LogPusher");
     this.sources = envutil.logpushSources();
   }
 
@@ -112,8 +115,11 @@ export class LogPusher {
     // log-id, log-entry
     for (const l of lines) {
       // k:avc,0:cd9i01d9mw,v:1,q:rethinkdns.com,t:AAAA,a:2606:4700::81d4:fa9a
-      this.log(lk + "," + l);
+      this.remotelog(lk + logdelim + l);
     }
+
+    this.corelog.d(`remotelog lines: ${lk} ${lines.length}`);
+    this.rec(lk, all);
   }
 
   getlimit(lklen) {
@@ -214,4 +220,104 @@ export class LogPusher {
     }
     return lines;
   }
+
+  // all => [version, ip, region, host, up, qname, qtype, ans, f]
+  rec(lk, all) {
+    const [m1, m2] = envutil.metrics();
+    if (m1 == null || m2 == null) return;
+
+    const metrics1 = [];
+    const metrics2 = [];
+    const [version, ip, region, up, qname, qtype, ans, f] = all;
+
+    const reqcount = this.key("n", "req");
+    const blockedcount = this.key("n", "blocked");
+    // ans is a multi-value str delimited by pipe
+    const isblocked = this.isansblocked(qtype, ans, f);
+    const blists = this.getblocklists(f);
+    const dom = this.getdomain(qname);
+    // todo: device-id, should it be concatenated with log-key?
+    // todo: faang dominance (sigma?)
+    // todo: geo-ip
+
+    // metric blobs in m1 should never change order; add new blobs at the end
+    metrics1.push(this.met(reqcount, 1.0));
+    metrics1.push(this.met(blockedcount, isblocked ? 1.0 : 0.0));
+    metrics1.push(this.met(ip, 1.0)); // ip hits
+    metrics1.push(this.met(qname, 1.0)); // query count
+    metrics1.push(this.met(region, 1.0)); // total requests
+    metrics1.push(this.met(qtype, 1.0)); // query type count
+    metrics1.push(this.met(dom, 1.0)); // domain count
+
+    if (isblocked) {
+      // metric blobs in m2 can have variable order
+      for (const b of blists) {
+        if (metrics2.length > maxdatapoints) break;
+        const kb = this.key("l", b);
+        metrics2.push(this.met(kb, 1.0));
+      }
+    }
+
+    this.corelog.d(`rec: ${lk} ${metrics1.length} ${metrics2.length}`);
+    // developers.cloudflare.com/analytics/analytics-engine/get-started
+    // indexes are limited to 32 bytes, blobs are limited to 5120 bytes
+    // there can be a maximum of 1 index and 20 blobs, per data point
+    // per cf discord, needn't await / waitUntil as writeDataPoint is
+    // a non-blocking call that returns void (like console.log)
+    m1.writeDataPoint({
+      blobs: metrics1.map((m) => m.blob),
+      doubles: metrics1.map((m) => m.double),
+      indexes: [lk],
+    });
+    if (metrics2.length > 0) {
+      m2.writeDataPoint({
+        blobs: metrics2.map((m) => m.blob),
+        doubles: metrics2.map((m) => m.double),
+        indexes: [lk],
+      });
+    }
+  }
+
+  // d is a domain name like "x.y.z.tld"
+  getdomain(d) {
+    if (util.emptyString(d)) return emptystring;
+    const parts = d.split(".");
+    if (parts.length < 2) return emptystring;
+    return parts.slice(-2).join(".");
+  }
+
+  // flag is of the form f:1:2AOAERQAkAQKAggAAEA
+  getblocklists(flag) {
+    flag = this.valOf(flag);
+    if (util.emptyString(flag)) return emptyarr;
+    return rdnsutil.blocklists(flag);
+  }
+
+  // ansips is a multi-value str delimited by pipe
+  isansblocked(qtype, ansips, flag) {
+    qtype = this.valOf(qtype);
+    ansips = this.valOf(ansips);
+    flag = this.valOf(flag);
+    // empty(answer) => blocked, iff flag is not empty
+    if (!util.emptyString(flag)) {
+      return util.emptyString(ansips);
+    }
+    // for qtypes that don't answer in ips, empty(answer) => blocked
+    if (!dnsutil.queryTypeMayResultInIP(qtype)) {
+      return util.emptyString(ansips);
+    }
+    // when query is blocked, there's only one ansip in ansips
+    for (const ansip of ansips.split(ansdelim)) {
+      if (dnsutil.isIPGrounded(ansip)) return true;
+    }
+    return false;
+  }
+
+  met(k, v = 0) {
+    if (util.emptyString(k)) return {};
+    return {
+      blob: k,
+      double: v,
+    };
+  }
 }

wrangler.toml

@@ -9,6 +9,13 @@ compatibility_date = "2022-11-30"
 send_metrics = false
 minify = false
 
+# uncomment to enable analytics on serverless-dns
+# this binding is not inherited by other worker-envs
+#analytics_engine_datasets = [
+#    { binding = "METRICS", dataset = "SDNS_M0" },
+#    { binding = "BL_METRICS", dataset = "SDNS_BL0" }
+#]
+
 [build]
 command = "npm run build"
 
@@ -59,6 +66,11 @@ routes = [
   "one.rethinkdns.com/*",
   "one.bravedns.com/*",
 ]
+analytics_engine_datasets = [
+    { binding = "METRICS", dataset = "ONE_M0" },
+    { binding = "BL_METRICS", dataset = "ONE_BL0" }
+]
+
 [env.one.vars]
 # just the error and request logs
 LOG_LEVEL = "logpush"