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) })