Skip to content

Very big update to query multi-opensearch #138

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
atheurer merged 6 commits into from
Jul 31, 2025
Merged

Very big update to query multi-opensearch #138

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
e73830d
Very big update to query multi-opensearch
atheurer Jul 24, 2025
b06f2a0
Fix memory usage for get metrics
atheurer Jul 31, 2025
5c95f9f
remove object-sizeof
atheurer Jul 31, 2025
655ada1
output cleanup
atheurer Jul 31, 2025
f209178
fix delete-run, minor cleanups
atheurer Jul 31, 2025
175bbe5
usage memUsage()
atheurer Jul 31, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
280 changes: 186 additions & 94 deletions queries/cdmq/add-run.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,15 +1,17 @@
//# vim: autoindent tabstop=2 shiftwidth=2 expandtab softtabstop=2 filetype=javascript

var cdm = require('./cdm');
var fs = require('fs');
var xz = require('xz');
var readline = require('readline');
var path = require('path');
var program = require('commander');
var instances = []; // opensearch instances
const { XzReadableStream } = require('xz-decompress');
const { Readable } = require('stream');
const cdm = require('./cdm');
const fs = require('fs');
const xz = require('xz');
const readline = require('readline');
const path = require('path');
const program = require('commander');
const instances = []; // opensearch instances

function save_host(host) {
var host_info = { host: host, header: { 'Content-Type': 'application/json' } };
const host_info = { host: host, header: { 'Content-Type': 'application/json' } };
instances.push(host_info);
}

Expand Down Expand Up @@ -37,10 +39,25 @@ function save_ver(ver) {
}
}

// Read an xz file and decompress to string
async function decompressXzFile(filename) {
// Create a readable stream from file
const fileStream = Readable.toWeb(fs.createReadStream(filename));

// Decompress using XzReadableStream
const decompressedResponse = new Response(new XzReadableStream(fileStream));

// Get the decompressed content as string
const text = await decompressedResponse.text();
return text;
}

async function readNdjsonXzToString(filePath) {
return new Promise((resolve, reject) => {
try {
const fileStream = fs.createReadStream(filePath);
const fileStream = fs.createReadStream(filePath, {
highWaterMark: 5 * 1024 * 1024
Comment thread
k-rister marked this conversation as resolved.
});
const decompressStream = new xz.Decompressor();

let decompressedString = '';
Expand All @@ -52,37 +69,43 @@ async function readNdjsonXzToString(filePath) {
});

decompressStream.on('end', () => {
fileStream.destroy();
resolve(decompressedString);
});

decompressStream.on('error', (error) => {
fileStream.destroy();
reject(error);
});

fileStream.on('error', (error) => {
fileStream.destroy();
reject(error);
});
} catch (error) {
reject(error);
}
} /*finally {
// Always cleanup, regardless of success or failure
if (fileStream && !fileStream.destroyed) {
fileStream.destroy();
}
}*/
});
}

async function processDir(instance, dir, mode) {
var jsonArr = [];
var files = fs.readdirSync(dir);

for (var i = 0; i < files.length; i++) {
var regExp = /\.ndjson\.xz$/;
var matches = regExp.exec(files[i]);
// skip non ndjson.xz files
if (!matches) {
continue;
}
const jsonArr = [];
const info = { runIds: {} };
const allFiles = fs.readdirSync(dir);
const regExp = /\.ndjson\.xz$/;
const xzFiles = allFiles.filter((item) => regExp.test(item));

for (var i = 0; i < xzFiles.length; i++) {
const regExp = /\.ndjson\.xz$/;
const filePath = path.join(program.dir, xzFiles[i]);
try {
const filePath = path.join(program.dir, files[i]);
const decompressedData = await readNdjsonXzToString(filePath);
var lines = decompressedData.split('\n');
const decompressedData = await decompressXzFile(filePath);
const lines = decompressedData.split('\n');
for (var j = 0; j < lines.length; j++) {
// TODO: validate JSON syntax and possible validate document schema?
if (lines[j] != '') {
Expand All @@ -92,51 +115,100 @@ async function processDir(instance, dir, mode) {
} catch (error) {
console.error('Error processing NDJSON.XZ file:', error);
}
}
cdm.memUsage();

if (mode == 'index') {
var responses = await esJsonArrRequest(instance, '', '/_bulk', jsonArr);
} else if (mode == 'getinfo') {
var info = { runIds: [], indices: {} };
console.log('going to process ' + jsonArr.length + ' lines');
for (var k = 1; k < jsonArr.length; k += 2) {
try {
var action = JSON.parse(jsonArr[k - 1]);
var doc = JSON.parse(jsonArr[k]);
} catch (jsonError) {
console.log('Could not porse: [' + jsonArr[k] + ']');
continue;
}
runId = doc['run']['run-uuid'];
if (!info['runIds'].includes(runId)) {
info['runIds'].push(runId);
}
// { "index": { "_index": "cdmv8dev-metric_data" } }
if (Object.keys(action).includes('index') && Object.keys(action['index']).includes('_index')) {
const indexName = action['index']['_index'];
const regExp = /^cdm-*(v7dev|v8dev|v9dev)-([^@]+)(@\d\d\d\d\.\d\d)*$/;
var matches = regExp.exec(indexName);
if (matches) {
cdmVer = matches[1];
if (!Object.keys(info['indices']).includes(cdmVer)) {
info['indices'][cdmVer] = [];
// After so much data is accumulated or at the last file, index what we have then clear the array.
if (jsonArr.length > 4000000 || i == xzFiles.length) {
Comment thread
k-rister marked this conversation as resolved.
console.log(
mode +
': After reading ' +
i +
' of ' +
(xzFiles.length + 1) +
' files, going to process ' +
jsonArr.length +
' lines'
);

if (mode == 'index') {
// The final argument for esJsonArrRequest, 'yearDotMonth', is not
// required here because the jsonArr includes the exact index name
// in the 'action' part of the array
const responses = await esJsonArrRequest(instance, '', '/_bulk', jsonArr);
} else if (mode == 'getinfo') {
//const info = { runIds: [], indices: {} };
for (var k = 1; k < jsonArr.length; k += 2) {
try {
var action = JSON.parse(jsonArr[k - 1]);
var doc = JSON.parse(jsonArr[k]);
} catch (jsonError) {
console.log('Could not porse: [' + jsonArr[k] + ']');
continue;
}
if (!info['indices'][cdmVer].includes(indexName)) {
debuglog('going to add indexname to info[indices][' + cdmVer + ']: ' + indexName);
info['indices'][cdmVer].push(indexName);
let runId = doc['run']['run-uuid'];
if (!Object.keys(info['runIds']).includes(runId)) {
info['runIds'][runId] = { indices: {}, yearDotMonth: '' };
}
// example action: { "index": { "_index": "cdmv8dev-metric_data" } }
// exmaple doc:
// {"cdm":{"ver":"v9dev"},
// "run":{"run-uuid":"c0e04edb-ddbc-4081-8bbc-9b9e84e6538d"}}
if (Object.keys(action).includes('index') && Object.keys(action['index']).includes('_index')) {
const indexName = action['index']['_index'];
const regExp = /^cdm-*(v7dev|v8dev|v9dev)-([^@]+)(@\d\d\d\d\.\d\d)*$/;
const matches = regExp.exec(indexName);
//console.log("doc:\n" + JSON.stringify(doc, null, 2));
if (matches) {
const cdmVer = matches[1];
let yearDotMonth = '';
let runId = '';
if (cdmVer == 'v7dev') {
runId = doc['run']['id'];
} else {
// v8dev and newer use run-uuid
runId = doc['run']['run-uuid'];
yearDotMonth = matches[3];
}
if (!Object.keys(info['runIds'][runId]).includes('indices')) {
info['runIds'][runId]['indices'] = {};
}
if (!Object.keys(info['runIds'][runId]['indices']).includes(cdmVer)) {
info['runIds'][runId]['indices'][cdmVer] = [];
}
if (!info['runIds'][runId]['indices'][cdmVer].includes(indexName)) {
debuglog(
'going to add indexname to info[runIds][' + runId + '][indices][' + cdmVer + ']: ' + indexName
);
info['runIds'][runId]['indices'][cdmVer].push(indexName);
}
if (yearDotMonth != '') {
if (info['runIds'][runId]['yearDotMonth'] == '') {
info['runIds'][runId]['yearDotMonth'] = yearDotMonth;
} else {
if (info['runIds'][runId]['yearDotMonth'] != yearDotMonth) {
console.log('ERROR: found more than one year and month for same runId [' + runId + ']');
process.exit(1);
}
}
}
} else {
console.log('ERROR: the index name [' + indexName + '] was not recognized');
process.exit(1);
}
} else {
console.log('the ndjson action [' + action + '] was not valid');
process.exit(1);
}
} else {
console.log('ERROR: the index name [' + indexName + '] was not recognized');
process.exit(1);
}
} else {
console.log('the ndjson action [' + action + '] was not valid');
process.exit(1);
console.log('mode [' + mode + '] is not supported');
}
}
// Must make array empty here for next pass
jsonArr.length = 0;
} // if (jsonArr.length > 100000 || i == xzFiles.length)
}
if (mode == 'getinfo') {
return info;
} else {
console.log('mode [' + mode + '] is not supported');
}
}

Expand All @@ -155,58 +227,78 @@ async function main() {
}

getInstancesInfo(instances);

// Always use the last instance (last --host) provided by the user
var instance = instances[instances.length - 1];
const instance = instances[instances.length - 1];
if (program.dir) {
var allDocTypes = ['run', 'iteration', 'sample', 'period', 'param', 'tag', 'metric_desc', 'metric_data'];
var info = await processDir(instance, program.dir, 'getinfo');
const allDocTypes = ['run', 'iteration', 'sample', 'period', 'param', 'tag', 'metric_desc', 'metric_data'];
const info = await processDir(instance, program.dir, 'getinfo');
// The cdmVer going forward must be set based on the data found in the ndjson files.
// We cannot rely on automatic detection of cdmVer based on what is present in the instance
// because the instance may contain only cdmVer that is different from the *new* data we are adding.
//
// Also, the cdm version is embedded in this new data that will be indexed. It is not possible
// (without significant effort) to index data already in one cdm version to another [directly].
if (Object.keys(info['indices']).length == 1) {
const cdmVer = Object.keys(info['indices'])[0];
if (!cdm.supportedCdmVersions.includes(cdmVer)) {
console.log(
'ERROR: the CDM version found in the documents to be indexed [' +
cdmver +
'] is not included in the list of supported CDM versions [' +
cdm.supportedCdmVersions +
']'
);
runIds = Object.keys(info['runIds']);
for (var runIdx = 0; runIdx < runIds.length; runIdx++) {
const runId = runIds[runIdx];
var cdmVer;
if (Object.keys(info['runIds'][runId]['indices']).length == 1) {
cdmVer = Object.keys(info['runIds'][runId]['indices'])[0];
if (!cdm.supportedCdmVersions.includes(cdmVer)) {
console.log(
'ERROR: the CDM version found in the documents to be indexed [' +
cdmver +
'] is not included in the list of supported CDM versions [' +
cdm.supportedCdmVersions +
']'
);
process.exit(1);
}
instance['ver'] = cdmVer;
//if (!Object.keys(instance['indices']).includes(cdmVer)) {
//instance['indices'][cdmVer] = [];
//}
} else {
console.log('ERROR: there was not exactly one CDM version found in the data to be indexed:\n');
console.log(Object.keys(info['indices']));
console.log('info\n' + JSON.stringify(info['indices'], null, 2));
process.exit(1);
}
instance['cdmVer'] = cdmVer;
if (!Object.keys(instance['indices']).includes(cdmVer)) {
instance['indices'][cdmVer] = [];
// For cdmv9 and newer, any time a document is to be indexed, it is imperitive
// that a check for the existence of the index is done, and if not found, create
// the index with the *correct* mappings and settings. If this is not done, an index
// may be auto-created with the *incorrect* mappings and settings, and documents can
// be indexed, but not properly, and subsequent queries will *NOT* work.
debuglog(JSON.stringify(info['runIds'][runId]['indices'][cdmVer], null, 2));
for (var i = 0; i < info['runIds'][runId]['indices'][cdmVer].length; i++) {
debuglog('checking for index ' + info['runIds'][runId]['indices'][cdmVer][i]);
cdm.checkCreateIndex(instance, info['runIds'][runId]['indices'][cdmVer][i]);
}
} else {
console.log('ERROR: there was not exactly one CDM version found in the data to be indexed:\n');
console.log(Object.keys(info['indices']));
console.log('info\n' + JSON.stringify(info['indices'], null, 2));
process.exit(1);
}
for (i = 0; i < info['runIds'].length; i++) {
console.log('Deleting any existing documents for runId ' + info['runIds'][i]);
var runId = info['runIds'][i];
var q = { query: { bool: { filter: [{ term: { 'run.run-uuid': runId } }] } } };
cdm.deleteDocs(instance, allDocTypes, q);
var numDocTypes = await cdm.waitForDeletedDocs(instances[instances.length - 1], runId, allDocTypes);
// Before indexing any documents, we must check for any existing ones. Having duplicate
// documents is really bad
console.log('Deleting any existing documents for runId ' + runId);
const q = { query: { bool: { filter: [{ term: { 'run.run-uuid': runId } }] } } };
cdm.deleteDocs(instance, allDocTypes, q, info['runIds'][runId]['yearDotMonth']);
const numDocTypes = await cdm.waitForDeletedDocs(
instances[instances.length - 1],
runId,
allDocTypes,
info['runIds'][runId]['yearDotMonth']
);
Comment thread
atheurer marked this conversation as resolved.
if (numDocTypes > 0) {
console.log('Warning: could not delete all documents for ' + docTypes + ' with ' + numAttempts);
console.log(
'These documents may continue to be deleted in the background. To check on the status, run this utility again'
);
process.exit(1);
}
const begin = Date.now() / 1000;
console.log('Indexing documents');
await processDir(instance, program.dir, 'index');
const end = Date.now() / 1000;
console.log('Time (seconds) to submit all documents for indexing: ' + (end - begin));
}
for (i = 0; i < info['indices'].length; i++) {
cdm.checkCreateIndex(instance, info['indices'][i]);
}
console.log('Indexing documents');
await processDir(instance, program.dir, 'index');
} else {
console.log('You must provide a --dir <directory with ndjsons>');
process.exit(1);
Expand Down
Loading
Loading