Compare commits
6 Commits
Author | SHA1 | Date | |
---|---|---|---|
7ef36b5c40 | |||
67a8f3fe4d | |||
5ae2c37519 | |||
fcb67ec878 | |||
9e25494f8f | |||
dd8ba4736a |
@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@apiclient.xyz/elasticsearch",
|
||||
"version": "1.0.46",
|
||||
"version": "1.0.49",
|
||||
"private": false,
|
||||
"description": "log to elasticsearch in a kibana compatible format",
|
||||
"main": "dist_ts/index.js",
|
||||
|
@ -59,4 +59,19 @@ tap.test('should delete documents not part of the piping session', async () => {
|
||||
await testElasticDoc.endPipingSession();
|
||||
});
|
||||
|
||||
tap.test('should take and store snapshot', async () => {
|
||||
await testElasticDoc.takeSnapshot(async (iterator, prevSnapshot) => {
|
||||
const aggregationData = [];
|
||||
for await (const doc of iterator) {
|
||||
// Sample aggregation: counting documents
|
||||
aggregationData.push(doc);
|
||||
}
|
||||
const snapshot = {
|
||||
date: new Date().toISOString(),
|
||||
aggregationData,
|
||||
};
|
||||
return snapshot;
|
||||
});
|
||||
});
|
||||
|
||||
tap.start();
|
||||
|
@ -3,6 +3,6 @@
|
||||
*/
|
||||
export const commitinfo = {
|
||||
name: '@apiclient.xyz/elasticsearch',
|
||||
version: '1.0.46',
|
||||
version: '1.0.49',
|
||||
description: 'log to elasticsearch in a kibana compatible format'
|
||||
}
|
||||
|
@ -9,6 +9,13 @@ export interface IElasticDocConstructorOptions {
|
||||
};
|
||||
}
|
||||
|
||||
export interface ISnapshot {
|
||||
date: string;
|
||||
aggregationData: any;
|
||||
}
|
||||
|
||||
export type SnapshotProcessor = (iterator: AsyncIterable<any>, prevSnapshot: ISnapshot | null) => Promise<ISnapshot>;
|
||||
|
||||
export class ElasticDoc {
|
||||
public client: ElasticClient;
|
||||
public index: string;
|
||||
@ -25,7 +32,6 @@ export class ElasticDoc {
|
||||
}
|
||||
|
||||
async startPipingSession() {
|
||||
// Clear the session docs set
|
||||
this.sessionDocs.clear();
|
||||
}
|
||||
|
||||
@ -50,7 +56,6 @@ export class ElasticDoc {
|
||||
response = await this.client.scroll({ scroll_id: response.body._scroll_id, scroll: '1m' });
|
||||
}
|
||||
|
||||
// Batch delete docs
|
||||
for (const docId of allDocIds) {
|
||||
if (!this.sessionDocs.has(docId)) {
|
||||
responseQueue.push({
|
||||
@ -71,7 +76,84 @@ export class ElasticDoc {
|
||||
await this.client.bulk({ refresh: true, body: responseQueue });
|
||||
}
|
||||
|
||||
// Clear the session docs set
|
||||
this.sessionDocs.clear();
|
||||
}
|
||||
|
||||
async takeSnapshot(processIterator: SnapshotProcessor) {
|
||||
const snapshotIndex = `${this.index}_snapshots`;
|
||||
|
||||
const { body: indexExists } = await this.client.indices.exists({ index: snapshotIndex });
|
||||
if (!indexExists) {
|
||||
await this.client.indices.create({
|
||||
index: snapshotIndex,
|
||||
body: {
|
||||
mappings: {
|
||||
properties: {
|
||||
date: {
|
||||
type: 'date'
|
||||
},
|
||||
aggregationData: {
|
||||
type: 'object',
|
||||
enabled: true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
const documentIterator = this.getDocumentIterator();
|
||||
|
||||
const newSnapshot = await processIterator(documentIterator, await this.getLastSnapshot());
|
||||
|
||||
await this.storeSnapshot(newSnapshot);
|
||||
}
|
||||
|
||||
private async getLastSnapshot(): Promise<ISnapshot | null> {
|
||||
const snapshotIndex = `${this.index}_snapshots`;
|
||||
const { body: indexExists } = await this.client.indices.exists({ index: snapshotIndex });
|
||||
|
||||
if (!indexExists) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const response = await this.client.search({
|
||||
index: snapshotIndex,
|
||||
sort: 'date:desc',
|
||||
size: 1
|
||||
});
|
||||
|
||||
if (response.body.hits.hits.length > 0) {
|
||||
const hit = response.body.hits.hits[0];
|
||||
return {
|
||||
date: hit._source.date,
|
||||
aggregationData: hit._source.aggregationData,
|
||||
};
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private async *getDocumentIterator() {
|
||||
let response = await this.client.search({ index: this.index, scroll: '1m', size: this.BATCH_SIZE });
|
||||
while (true) {
|
||||
for (const hit of response.body.hits.hits) {
|
||||
yield hit._source;
|
||||
}
|
||||
|
||||
if (!response.body.hits.hits.length) {
|
||||
break;
|
||||
}
|
||||
|
||||
response = await this.client.scroll({ scroll_id: response.body._scroll_id, scroll: '1m' });
|
||||
}
|
||||
}
|
||||
|
||||
private async storeSnapshot(snapshot: ISnapshot) {
|
||||
await this.client.index({
|
||||
index: `${this.index}_snapshots`,
|
||||
body: snapshot,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
Reference in New Issue
Block a user