Skip to content

Commit

Permalink
fix: defer bundle/manifest persistence by domain group
Browse files Browse the repository at this point in the history
  • Loading branch information
@maxakuru
maxakuru committed Jun 12, 2024
1 parent ce070ac commit f43e1cd
Show file tree
Hide file tree
Showing 2 changed files with 133 additions and 91 deletions.
142 changes: 91 additions & 51 deletions src/bundler/index.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -131,60 +131,100 @@ export async function importEventsByKey(ctx, rawEventMap, isVirtual = false) {
const { log, attributes: { stats } } = ctx;
const concurrency = getEnvVar(ctx, 'CONCURRENCY_LIMIT', DEFAULT_CONCURRENCY_LIMIT, 'integer');

let totalEvents = 0;
const entries = Object.entries(rawEventMap);
stats[isVirtual ? 'rawKeysVirtual' : 'rawKeys'] = entries.length;
let totalEvents = 0;
await processQueue(
entries,
async ([key, { events, info }]) => {
log.debug(`processing ${events.length} events to file ${key}`);
totalEvents += events.length;
const {
domain, year, month, day, hour,
} = info;

/**
* Sort events further by session ID, same ids get put into a single EventGroup.
* NOTE: session ID is different than Event ID: sessionID == `{event_id}--{event_url_path}`
* This leads to sessions with fewer collisions, since sessions are roughly unique per URL.
*/
/** @type {Record<string, RawRUMEvent[]>} */
const eventsBySessionId = {};
events.forEach((event) => {
// if bundle is virtual, include the domain in the session ID
// since the events being sorted into this key may have different domains
const evUrl = new URL(event.url);
const sessionId = `${event.id}${isVirtual ? `--${evUrl.hostname}` : ''}--${evUrl.pathname}`;
if (!eventsBySessionId[sessionId]) {
eventsBySessionId[sessionId] = [];
}
eventsBySessionId[sessionId].push(event);
});

// get this day's manifest & yesterday's manifest, if needed
const manifest = await Manifest.fromContext(ctx, domain, year, month, day);
const yManifest = hour < 23
? await Manifest.fromContext(ctx, domain, ...yesterday(year, month, day))
: undefined;

const touchedBundles = await addEventsToBundle(
ctx,
info,
eventsBySessionId,
manifest,
yManifest,
);

// save touched manifests and bundles
await Promise.allSettled([
manifest.store(),
yManifest?.store(),
...touchedBundles.map((b) => b.store()),
]);
},
concurrency,
);
stats[isVirtual ? 'totalEventsVirtual' : 'totalEvents'] = totalEvents;
/**
* To avoid repeatedly saving the same manifest/bundle files,
* first we group the key/events pairs by domain, and process
* each domain as a group. Then persist the touched manifests/bundles
* of the domain after processing its keys.
*/

/**
* NOTE: for imports it's possible to exceed memory limits since all events
* will be on the same domain. Possibly need to create a limit and push to new
* groups once that limit is met, like `domain-{n}`.
*
* @type {Record<string, [string, {
* events: RawRUMEvent[];
* info: BundleInfo;
* }][]>}
*/
const groupMap = entries.reduce((acc, [key, val]) => {
const { info, events } = val;
totalEvents += events.length;
if (!acc[info.domain]) {
acc[info.domain] = [];
}
acc[info.domain].push([key, val]);
return acc;
}, {});
const groups = Object.values(groupMap);

stats[`totalEvents${isVirtual ? 'Virtual' : ''}`] = totalEvents;
stats[`importGroups${isVirtual ? 'Virtual' : ''}`] = groups.length;

await processQueue(groups, async (group) => {
/** @type {Set<{store: () => Promise<any>}>} */
const toSave = new Set();

await processQueue(
group,
async ([key, { events, info }]) => {
log.debug(`processing ${events.length} events to file ${key}`);
const {
domain, year, month, day, hour,
} = info;

/**
* Sort events further by session ID, same ids get put into a single EventGroup.
* NOTE: session ID is different than Event ID:
* sessionID == `{event_id}[--{domain}]--{event_url_path}`
* This leads to sessions with fewer collisions, since sessions are roughly unique per URL.
*/
/** @type {Record<string, RawRUMEvent[]>} */
const eventsBySessionId = {};
events.forEach((event) => {
// if bundle is virtual, include the domain in the session ID
// since the events being sorted into this key may have different domains
const evUrl = new URL(event.url);
const sessionId = `${event.id}${isVirtual ? `--${evUrl.hostname}` : ''}--${evUrl.pathname}`;
if (!eventsBySessionId[sessionId]) {
eventsBySessionId[sessionId] = [];
}
eventsBySessionId[sessionId].push(event);
});

// get this day's manifest & yesterday's manifest, if needed
const manifest = await Manifest.fromContext(ctx, domain, year, month, day);
const yManifest = hour < 23
? await Manifest.fromContext(ctx, domain, ...yesterday(year, month, day))
: undefined;

const touchedBundles = await addEventsToBundle(
ctx,
info,
eventsBySessionId,
manifest,
yManifest,
);

toSave.add(manifest);
if (yManifest) {
toSave.add(yManifest);
}
touchedBundles.forEach((b) => toSave.add(b));
},
concurrency,
);

// save touched manifests and bundles
await Promise.allSettled(
[...toSave].map((s) => s.store()),
);
}, concurrency);
}

/**
Expand Down
82 changes: 42 additions & 40 deletions test/bundler/index.test.js
View file
Original file line number Diff line number Diff line change
Expand Up @@ -152,7 +152,7 @@ describe('bundler Tests', () => {
await assertRejectsWithResponse(bundleRUM(ctx), 409);
});

it('should bundle events', async () => {
it.only('should bundle events', async () => {
const logsBody = await fs.readFile(path.resolve(__dirname, 'fixtures', 'list-logs-single.xml'), 'utf-8');
const mockEventResponseBody = mockEventLogFile('example.com');
const bodies = {
Expand Down Expand Up @@ -250,7 +250,7 @@ describe('bundler Tests', () => {
.reply(404)
// store manifest
.put('/example.com/1970/1/1/.manifest.json?x-id=PutObject')
.times(3)
// .times(3)
.reply((_, body) => {
bodies.apex.manifest.push(body);
return [200];
Expand Down Expand Up @@ -299,48 +299,48 @@ describe('bundler Tests', () => {
assert.strictEqual(subdomain.domainkey, (await gzip('TEST-NEW-KEY')).toString('hex'));

// 3 manifest updates & 3 bundles for apex, since events were processed into 3 sessions
// assert.deepStrictEqual(subdomain.manifest[0], { sessions: { '0--/': { hour: 0 } } });
assert.deepEqual(apex.manifest.length, 3);
// but only 1 request should be made since the persist is deferred by domain
assert.deepEqual(apex.manifest.length, 1);
assert.deepEqual(apex.bundle.length, 3);

assert.deepStrictEqual(apex.manifest[0], {
sessions: {
'0--/even': {
hour: 0,
},
'1--/odd': {
hour: 0,
},
'2--/even': {
hour: 0,
},
},
});
// assert.deepStrictEqual(apex.manifest[0], {
// sessions: {
// '0--/even': {
// hour: 0,
// },
// '1--/odd': {
// hour: 0,
// },
// '2--/even': {
// hour: 0,
// },
// },
// });

assert.deepStrictEqual(apex.manifest[1], {
sessions: {
'0--/even': {
hour: 0,
},
'1--/odd': {
hour: 0,
},
'2--/even': {
hour: 0,
},
'3--/odd': {
hour: 1,
},
'4--/even': {
hour: 1,
},
'5--/odd': {
hour: 1,
},
},
});
// assert.deepStrictEqual(apex.manifest[1], {
// sessions: {
// '0--/even': {
// hour: 0,
// },
// '1--/odd': {
// hour: 0,
// },
// '2--/even': {
// hour: 0,
// },
// '3--/odd': {
// hour: 1,
// },
// '4--/even': {
// hour: 1,
// },
// '5--/odd': {
// hour: 1,
// },
// },
// });

assert.deepStrictEqual(apex.manifest[2], {
assert.deepStrictEqual(apex.manifest[0], {
sessions: {
'0--/even': {
hour: 0,
Expand Down Expand Up @@ -501,6 +501,8 @@ describe('bundler Tests', () => {
rawEvents: 10,
logFiles: 1,
domains: 2,
importGroups: 2,
importGroupsVirtual: 0,
newDomains: 1,
rawKeys: 4,
rawKeysVirtual: 0,
Expand Down

0 comments on commit f43e1cd

Please sign in to comment.