unpkg/server/ingestLogs.js

const parseURL = require("url").parse;
const startOfDay = require("date-fns/start_of_day");
const addDays = require("date-fns/add_days");
const parsePackageURL = require("./utils/parsePackageURL");
const CloudflareAPI = require("./CloudflareAPI");
const StatsAPI = require("./StatsAPI");

const db = require("./RedisClient");

/**
 * Domains we want to analyze.
 */
const DomainNames = [
  "unpkg.com"
  //'npmcdn.com' // We don't have log data on npmcdn.com yet :/
];

/**
 * The window of time to download in a single fetch.
 */
const LogWindowSeconds = 30;

function getSeconds(date) {
  return Math.floor(date.getTime() / 1000);
}

function stringifySeconds(seconds) {
  return new Date(seconds * 1000).toISOString();
}

function toSeconds(millis) {
  return Math.floor(millis / 1000);
}

const oneSecond = 1000;
const oneMinute = oneSecond * 60;
const oneHour = oneMinute * 60;

function computeCounters(stream) {
  return new Promise((resolve, reject) => {
    const counters = {};
    const expireat = {};

    function incr(key, member, by, expiry) {
      counters[key] = counters[key] || {};
      counters[key][member] = (counters[key][member] || 0) + by;
      expireat[key] = expiry;
    }

    stream
      .on("error", reject)
      .on("data", function(entry) {
        const date = new Date(Math.round(entry.timestamp / 1000000));

        const nextDay = startOfDay(addDays(date, 1));
        const sevenDaysLater = getSeconds(addDays(nextDay, 7));
        const thirtyDaysLater = getSeconds(addDays(nextDay, 30));
        const dayKey = StatsAPI.createDayKey(date);

        const clientRequest = entry.clientRequest;
        const edgeResponse = entry.edgeResponse;

        if (edgeResponse.status === 200) {
          // Q: How many requests do we serve for a package per day?
          // Q: How many bytes do we serve for a package per day?
          const url = parsePackageURL(parseURL(clientRequest.uri).pathname);
          const packageName = url && url.packageName;

          if (packageName) {
            incr(
              `stats-packageRequests-${dayKey}`,
              packageName,
              1,
              thirtyDaysLater
            );
            incr(
              `stats-packageBytes-${dayKey}`,
              packageName,
              edgeResponse.bytes,
              thirtyDaysLater
            );
          }
        }

        // Q: How many requests per day do we receive via a protocol?
        const protocol = clientRequest.httpProtocol;

        if (protocol)
          incr(
            `stats-protocolRequests-${dayKey}`,
            protocol,
            1,
            thirtyDaysLater
          );

        // Q: How many requests do we receive from a hostname per day?
        // Q: How many bytes do we serve to a hostname per day?
        const referer = clientRequest.referer;
        const hostname = referer && parseURL(referer).hostname;

        if (hostname) {
          incr(`stats-hostnameRequests-${dayKey}`, hostname, 1, sevenDaysLater);
          incr(
            `stats-hostnameBytes-${dayKey}`,
            hostname,
            edgeResponse.bytes,
            sevenDaysLater
          );
        }
      })
      .on("end", function() {
        resolve({ counters, expireat });
      });
  });
}

function processLogs(stream) {
  return computeCounters(stream).then(({ counters, expireat }) => {
    Object.keys(counters).forEach(key => {
      const values = counters[key];

      Object.keys(values).forEach(member => {
        db.zincrby(key, values[member], member);
      });

      if (expireat[key]) db.expireat(key, expireat[key]);
    });
  });
}

function ingestLogs(zone, startSeconds, endSeconds) {
  return new Promise(resolve => {
    console.log(
      "info: Started ingesting logs for %s from %s to %s",
      zone.name,
      stringifySeconds(startSeconds),
      stringifySeconds(endSeconds)
    );

    const startFetchTime = Date.now();

    resolve(
      CloudflareAPI.getLogs(zone.id, startSeconds, endSeconds).then(stream => {
        const endFetchTime = Date.now();

        console.log(
          "info: Fetched %ds worth of logs for %s in %dms",
          endSeconds - startSeconds,
          zone.name,
          endFetchTime - startFetchTime
        );

        const startProcessTime = Date.now();

        return processLogs(stream).then(() => {
          const endProcessTime = Date.now();

          console.log(
            "info: Processed %ds worth of logs for %s in %dms",
            endSeconds - startSeconds,
            zone.name,
            endProcessTime - startProcessTime
          );
        });
      })
    );
  });
}

function startZone(zone) {
  const startSecondsKey = `ingestLogsWorker-nextStartSeconds-${zone.name.replace(
    ".",
    "-"
  )}`;

  function takeATurn() {
    db.get(startSecondsKey, function(error, value) {
      let startSeconds = value && parseInt(value, 10);

      const now = Date.now();

      // Cloudflare keeps logs around for 72 hours.
      // https://support.cloudflare.com/hc/en-us/articles/216672448-Enterprise-Log-Share-REST-API
      const minSeconds = toSeconds(now - oneHour * 72);

      if (startSeconds == null) {
        startSeconds = minSeconds;
      } else if (startSeconds < minSeconds) {
        console.warn(
          "warning: Dropped logs for %s from %s to %s!",
          zone.name,
          stringifySeconds(startSeconds),
          stringifySeconds(minSeconds)
        );

        startSeconds = minSeconds;
      }

      // The log for a request is typically available within thirty (30) minutes
      // of the request taking place under normal conditions. We deliver logs
      // ordered by the time that the logs were created, i.e. the timestamp of
      // the request when it was received by the edge. Given the order of
      // delivery, we recommend waiting a full thirty minutes to ingest a full
      // set of logs. This will help ensure that any congestion in the log
      // pipeline has passed and a full set of logs can be ingested.
      // https://support.cloudflare.com/hc/en-us/articles/216672448-Enterprise-Log-Share-REST-API
      const maxSeconds = toSeconds(now - oneMinute * 30);

      if (startSeconds < maxSeconds) {
        const endSeconds = startSeconds + LogWindowSeconds;

        ingestLogs(zone, startSeconds, endSeconds).then(
          function() {
            db.set(startSecondsKey, endSeconds);
            setTimeout(takeATurn);
          },
          function(error) {
            console.error(error.stack);
            process.exit(1);
          }
        );
      } else {
        setTimeout(takeATurn, (startSeconds - maxSeconds) * 1000);
      }
    });
  }

  takeATurn();
}

Promise.all(DomainNames.map(CloudflareAPI.getZones)).then(results => {
  const zones = results.reduce((memo, zones) => {
    return memo.concat(zones);
  });

  zones.forEach(startZone);
});