|
|
|
@ -71,26 +71,40 @@ export class ElasticDoc {
|
|
|
|
|
async startPipingSession(options: { onlyNew?: boolean }) {
|
|
|
|
|
this.sessionDocs.clear();
|
|
|
|
|
this.onlyNew = options.onlyNew;
|
|
|
|
|
|
|
|
|
|
if (this.onlyNew) {
|
|
|
|
|
// Retrieve and store the latest timestamp
|
|
|
|
|
try {
|
|
|
|
|
const response = await this.client.search({
|
|
|
|
|
index: this.index,
|
|
|
|
|
sort: '@timestamp:desc',
|
|
|
|
|
size: 1,
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
// If the search query succeeded, the index exists.
|
|
|
|
|
const hit = response.body.hits.hits[0];
|
|
|
|
|
this.latestTimestamp = hit?._source?.['@timestamp'] || null;
|
|
|
|
|
|
|
|
|
|
if (this.latestTimestamp) {
|
|
|
|
|
console.log(`Working in "onlyNew" mode. Hence we are omitting documents prior to ${this.latestTimestamp}`);
|
|
|
|
|
} else {
|
|
|
|
|
`Working in "onlyNew" mode, but no documents found in index ${this.index}. Hence processing all documents now.`
|
|
|
|
|
console.log(`Working in "onlyNew" mode, but no documents found in index ${this.index}. Hence processing all documents now.`);
|
|
|
|
|
}
|
|
|
|
|
} catch (e) {
|
|
|
|
|
// If the search query failed, the index likely doesn't exist or some other error occurred.
|
|
|
|
|
if (e.meta && e.meta.statusCode === 404) {
|
|
|
|
|
console.log(`Index ${this.index} does not exist. Working in "onlyNew" mode, but will process all documents as the index is empty.`);
|
|
|
|
|
} else {
|
|
|
|
|
console.log(`An error occurred while trying to retrieve the latest timestamp: ${e}`);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
this.latestTimestamp = null;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
async pipeDocument(optionsArg: { docId: string; timestamp?: string | number; doc: any }) {
|
|
|
|
|
// If 'onlyNew' is true, compare the document timestamp with the latest timestamp
|
|
|
|
|
|
|
|
|
|
await this.ensureIndexExists(optionsArg.doc);
|
|
|
|
|
|
|
|
|
|
const documentBody = {
|
|
|
|
@ -98,6 +112,7 @@ export class ElasticDoc {
|
|
|
|
|
...(optionsArg.timestamp && { '@timestamp': optionsArg.timestamp }),
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
// If 'onlyNew' is true, compare the document timestamp with the latest timestamp
|
|
|
|
|
if (this.onlyNew) {
|
|
|
|
|
if (this.latestTimestamp && optionsArg.timestamp <= this.latestTimestamp) {
|
|
|
|
|
// Omit the document
|
|
|
|
|