diff --git a/Procfile b/Procfile index 78b170a..7eee027 100644 --- a/Procfile +++ b/Procfile @@ -1,2 +1,3 @@ web: node server.js ingest_logs: node workers/ingest-logs.js +ingest_stats: node workers/ingest-stats.js diff --git a/workers/ingest-logs.js b/workers/ingest-logs.js index 3f9bde5..0630e85 100644 --- a/workers/ingest-logs.js +++ b/workers/ingest-logs.js @@ -28,8 +28,8 @@ invariant( * Domains we want to analyze. */ const DomainNames = [ - //'npmcdn.com', // We don't have log data on npmcdn.com yet :/ 'unpkg.com' + //'npmcdn.com' // We don't have log data on npmcdn.com yet :/ ] /** @@ -96,12 +96,12 @@ const computeCounters = (stream) => const date = new Date(Math.round(entry.timestamp / 1000000)) const dayKey = `${date.getUTCFullYear()}-${date.getUTCMonth()}-${date.getUTCDate()}` const hourKey = `${dayKey}-${date.getUTCHours()}` - const minuteKey = `${hourKey}-${date.getUTCMinutes()}` + // const minuteKey = `${hourKey}-${date.getUTCMinutes()}` // Q: How many requests do we receive per day/hour/minute? - incrCounter(`stats-requests-${dayKey}`) - incrCounter(`stats-requests-${hourKey}`) - incrCounter(`stats-requests-${minuteKey}`) + // incrCounter(`stats-requests-${dayKey}`) // Done by ingest_stats worker + // incrCounter(`stats-requests-${hourKey}`) // Done by ingest_stats worker + // incrCounter(`stats-requests-${minuteKey}`) // Done by ingest_stats worker // Q: How many requests are served by origin/cache/edge per day/hour? if (entry.origin) { diff --git a/workers/ingest-stats.js b/workers/ingest-stats.js new file mode 100644 index 0000000..6bcf3e0 --- /dev/null +++ b/workers/ingest-stats.js @@ -0,0 +1,290 @@ +require('isomorphic-fetch') +const parseURL = require('url').parse +const invariant = require('invariant') +const redis = require('redis') + +const CloudflareEmail = process.env.CLOUDFLARE_EMAIL +const CloudflareKey = process.env.CLOUDFLARE_KEY +const RedisURL = process.env.REDIS_URL + +invariant( + CloudflareEmail, + 'Missing the $CLOUDFLARE_EMAIL environment variable' +) + +invariant( + CloudflareKey, + 'Missing the $CLOUDFLARE_KEY environment variable' +) + +invariant( + RedisURL, + 'Missing the $REDIS_URL environment variable' +) + +/** + * Domains we want to analyze. + */ +const DomainNames = [ + 'unpkg.com', + 'npmcdn.com' +] + +const db = redis.createClient(RedisURL) + +const getZones = (domain) => + fetch(`https://api.cloudflare.com/client/v4/zones?name=${domain}`, { + method: 'GET', + headers: { + 'X-Auth-Email': CloudflareEmail, + 'X-Auth-Key': CloudflareKey + } + }).then(res => res.json()) + .then(data => data.result) + +const getZoneAnalyticsDashboard = (zoneId, since) => + fetch(`https://api.cloudflare.com/client/v4/zones/${zoneId}/analytics/dashboard?since=${since}&continuous=true`, { + method: 'GET', + headers: { + 'X-Auth-Email': CloudflareEmail, + 'X-Auth-Key': CloudflareKey + } + }).then(res => res.json()) + .then(data => data.result) + +const oneSecond = 1000 +const oneMinute = oneSecond * 60 +const oneHour = oneMinute * 60 +const oneDay = oneHour * 24 + +const oneMinuteSeconds = 60 +const oneHourSeconds = oneMinuteSeconds * 60 +const oneDaySeconds = oneHourSeconds * 24 + +const reduceResults = (memo, results) => { + Object.keys(results).forEach(key => { + const value = results[key] + + if (typeof value === 'object' && value) { + memo[key] = reduceResults(memo[key] || {}, value) + } else if (typeof value === 'number') { + memo[key] = (memo[key] || 0) + results[key] + } + }) + + return memo +} + +const ingestStatsForZones = (zones, since, processDashboard) => + new Promise(resolve => { + const zoneNames = zones.map(zone => zone.name).join(', ') + + console.log( + 'LOG: start ingesting stats for zones %s since %d', + zoneNames, + since + ) + + const startFetchTime = Date.now() + + resolve( + Promise.all( + zones.map(zone => getZoneAnalyticsDashboard(zone.id, since)) + ).then( + results => { + const endFetchTime = Date.now() + + console.log( + 'LOG: fetched zone analytics dashboards for %s since %d in %dms', + zoneNames, + since, + endFetchTime - startFetchTime + ) + + // We don't have per-minute dashboards available for npmcdn.com yet, + // so the dashboard for that domain will be null when querying for + // per-minute data. Just filter it out here for now. + results = results.filter(Boolean) + + return results.reduce(reduceResults) + } + ).then( + dashboard => { + const startProcessTime = Date.now() + + return processDashboard(dashboard).then(() => { + const endProcessTime = Date.now() + + console.log( + 'LOG: processed zone analytics dashboards for %s since %d in %dms', + zoneNames, + since, + endProcessTime - startProcessTime + ) + }) + } + ) + ) + }) + +const ingestPerDayStats = (zones) => + ingestStatsForZones(zones, -10080, processPerDayDashboard) + +const processPerDayDashboard = (dashboard) => + Promise.all(dashboard.timeseries.map(processPerDayTimeseries)) + +const processPerDayTimeseries = (ts) => + new Promise(resolve => { + const since = new Date(ts.since) + const until = new Date(ts.until) + + invariant( + since.getUTCHours() === 0 && since.getUTCMinutes() === 0 && since.getUTCSeconds() === 0, + 'ERROR: per-day timeseries.since must begin exactly on the day' + ) + + invariant( + (until - since) === oneDay, + 'ERROR: per-day timeseries must span exactly one day' + ) + + const dayKey = `${since.getUTCFullYear()}-${since.getUTCMonth()}-${since.getUTCDate()}` + + // Q: How many requests do we serve per day? + db.set(`stats-requests-${dayKey}`, ts.requests.all) + // Q: How many requests do we serve per day from the cache? + db.set(`stats-requestsFromCache-${dayKey}`, ts.requests.cached) + + // Q: How much bandwidth do we serve per day? + db.set(`stats-bandwidth-${dayKey}`, ts.bandwidth.all) + // Q: How much bandwidth do we serve per day from the cache? + db.set(`stats-bandwidthFromCache-${dayKey}`, ts.bandwidth.cached) + + // Q: How many errors do we serve per day? + const httpStatus = ts.requests.http_status + const errors = Object.keys(httpStatus).reduce((memo, status) => { + return parseInt(status, 10) >= 500 ? memo + httpStatus[status] : memo + }, 0) + + db.set(`stats-errors-${dayKey}`, errors) + + // Q: How many requests do we serve to a country per day? + // Q: How much bandwidth do we serve to a country per day? + const requestsByCountry = [] + const bandwidthByCountry = [] + + Object.keys(ts.requests.country).forEach(country => { + const requests = ts.requests.country[country] + const bandwidth = ts.bandwidth.country[country] + + // Include only countries who made at least 1M requests. + if (requests > 1000000) { + requestsByCountry.push(requests, country) + bandwidthByCountry.push(bandwidth, country) + } + }) + + if (requestsByCountry.length) + db.zadd([ `stats-requestsByCountry-${dayKey}`, ...requestsByCountry ]) + + if (bandwidthByCountry.length) + db.zadd([ `stats-bandwidthByCountry-${dayKey}`, ...bandwidthByCountry ]) + + resolve() + }) + +const ingestPerHourStats = (zones) => + ingestStatsForZones(zones, -1440, processPerHourDashboard) + +const processPerHourDashboard = (dashboard) => + Promise.all(dashboard.timeseries.map(processPerHourTimeseries)) + +const processPerHourTimeseries = (ts) => + new Promise(resolve => { + const since = new Date(ts.since) + const until = new Date(ts.until) + + invariant( + since.getUTCMinutes() === 0 && since.getUTCSeconds() === 0, + 'ERROR: per-hour timeseries.since must begin exactly on the hour' + ) + + invariant( + (until - since) === oneHour, + 'ERROR: per-hour timeseries must span exactly one hour' + ) + + const hourKey = `${since.getUTCFullYear()}-${since.getUTCMonth()}-${since.getUTCDate()}-${since.getUTCHours()}` + + // Q: How many requests do we serve per hour? (expire after 7 days) + db.setex(`stats-requests-${hourKey}`, (oneDaySeconds * 7), ts.requests.all) + // Q: How many requests do we serve per hour from the cache? (expire after 7 days) + db.setex(`stats-requestsFromCache-${hourKey}`, (oneDaySeconds * 7), ts.requests.cached) + + // Q: How much bandwidth do we serve per hour? (expire after 7 days) + db.setex(`stats-bandwidth-${hourKey}`, (oneDaySeconds * 7), ts.bandwidth.all) + // Q: How much bandwidth do we serve per hour from the cache? (expire after 7 days) + db.setex(`stats-bandwidthFromCache-${hourKey}`, (oneDaySeconds * 7), ts.bandwidth.cached) + + resolve() + }) + +const ingestPerMinuteStats = (zones) => + ingestStatsForZones(zones, -30, processPerMinuteDashboard) + +const processPerMinuteDashboard = (dashboard) => + Promise.all(dashboard.timeseries.map(processPerMinuteTimeseries)) + +const processPerMinuteTimeseries = (ts) => + new Promise(resolve => { + const since = new Date(ts.since) + const until = new Date(ts.until) + + invariant( + since.getUTCSeconds() === 0, + 'ERROR: per-minute timeseries.since must begin exactly on the minute' + ) + + invariant( + (until - since) === oneMinute, + 'ERROR: per-minute timeseries must span exactly one minute' + ) + + const minuteKey = `${since.getUTCFullYear()}-${since.getUTCMonth()}-${since.getUTCDate()}-${since.getUTCHours()}-${since.getUTCMinutes()}` + + // Q: How many requests do we serve per minute? (expire after 1 day) + db.setex(`stats-requests-${minuteKey}`, oneDaySeconds, ts.requests.all) + // Q: How many requests do we serve per minute from the cache? (expire after 1 day) + db.setex(`stats-requestsFromCache-${minuteKey}`, oneDaySeconds, ts.requests.cached) + + // Q: How much bandwidth do we serve per minute? (expire after 1 day) + db.setex(`stats-bandwidth-${minuteKey}`, oneDaySeconds, ts.bandwidth.all) + // Q: How much bandwidth do we serve per minute from the cache? (expire after 1 day) + db.setex(`stats-bandwidthFromCache-${minuteKey}`, oneDaySeconds, ts.bandwidth.cached) + + resolve() + }) + +const startZones = (zones) => { + const takePerMinuteTurn = () => + ingestPerMinuteStats(zones) + + const takePerHourTurn = () => + ingestPerHourStats(zones) + + const takePerDayTurn = () => + ingestPerDayStats(zones) + + takePerMinuteTurn() + takePerHourTurn() + takePerDayTurn() + + setInterval(takePerMinuteTurn, oneMinute) + setInterval(takePerHourTurn, oneHour) + setInterval(takePerDayTurn, oneDay / 2) +} + +Promise.all(DomainNames.map(getZones)).then(results => { + const zones = results.reduce((memo, zones) => memo.concat(zones)) + startZones(zones) +})