Compare commits

...

2 Commits

Author SHA1 Message Date
Alexander Chan 3da3727cb7 test 2018-06-28 11:35:45 -07:00
Alexander Chan 2bdd443628 rf: use mongodb aggregate method for item count 2018-06-28 10:27:56 -07:00
3 changed files with 233 additions and 215 deletions

View File

@ -850,6 +850,10 @@ class MongoClientInterface {
{ error: err.message }); { error: err.message });
return cb(errors.InternalError); return cb(errors.InternalError);
} }
if (isUserBucket(c.s.name) && result.value) {
const objVal = result.value;
this.dataCount.delObject(objVal.value, DEL_VER);
}
return cb(null); return cb(null);
}); });
} }
@ -886,15 +890,7 @@ class MongoClientInterface {
}); });
} }
return this.deleteObjectVerNotMaster(c, bucketName, objName, return this.deleteObjectVerNotMaster(c, bucketName, objName,
params, log, err => { params, log, cb);
if (err) {
return cb(err);
}
if (isUserBucket(c.s.name)) {
this.dataCount.delObject(mst.value, DEL_VER);
}
return cb();
});
}); });
} }
@ -1118,12 +1114,11 @@ class MongoClientInterface {
if (err) { if (err) {
return next(errors.InternalError); return next(errors.InternalError);
} }
if (results.bucketList && results.counter && if (results.bucketList && results.dataManaged) {
results.counter.dataManaged) {
res.bucketList.push(results.bucketList); res.bucketList.push(results.bucketList);
res.objects += results.counter.objects; res.objects += results.objects;
res.versions += results.counter.versions; res.versions += results.versions;
consolidateData(results.counter.dataManaged); consolidateData(results.dataManaged);
} }
return next(); return next();
}); });
@ -1139,168 +1134,152 @@ class MongoClientInterface {
return undefined; return undefined;
} }
getDataManaged(c, query, log, cb) { _handleResults(res, isVersioned) {
const dataManaged = { const total = { curr: 0, prev: 0 };
total: 0, const locations = {};
locations: {},
};
const addLocation = (site, size) => { Object.keys(res.nullData).forEach(loc => {
dataManaged.total += size; const bytes = res.nullData[loc];
if (!dataManaged.locations[site]) { const locName = loc === 'mem' || loc === 'file' ?
dataManaged.locations[site] = 0; 'us-east-1' : loc;
if (!locations[locName]) {
locations[locName] = { curr: 0, prev: 0 };
} }
dataManaged.locations[site] += size; total.curr += bytes;
locations[locName].curr += bytes;
});
if (isVersioned) {
Object.keys(res.versionData).forEach(loc => {
const bytes = res.versionData[loc];
const locName = loc === 'mem' || loc === 'file' ?
'us-east-1' : loc;
if (!locations[locName]) {
locations[locName] = { curr: 0, prev: 0 };
}
total.prev += bytes;
locations[locName].prev += bytes;
});
}
Object.keys(res.masterData).forEach(loc => {
const bytes = res.masterData[loc];
const locName = loc === 'mem' || loc === 'file' ?
'us-east-1' : loc;
if (!locations[locName]) {
locations[locName] = { curr: 0, prev: 0 };
}
total.curr += bytes;
locations[locName].curr += bytes;
if (isVersioned) {
total.prev -= bytes;
locations[locName].prev -= bytes;
}
});
return {
versions: isVersioned ?
res.versionCount - res.masterCount : 0,
objects: res.masterCount + res.nullCount,
dataManaged: {
total,
locations,
},
};
}
_handleCount(entry) {
return entry && entry.count > 0 ? entry.count : 0;
}
_handleEntries(entries) {
const results = {};
if (entries) {
entries.forEach(entry => {
results[entry._id] = entry.bytes;
});
}
return results;
}
_handleMongo(c, filter, log, cb) {
const reducedFields = {
'_id': 1,
'value.versionId': 1,
'value.replicationInfo.backends': 1,
'value.content-length': 1,
'value.dataStoreName': 1,
}; };
c.find(query).toArray((err, entries) => { const aggCount = [
{ $project: { _id: 1 } },
{ $group: { _id: null, count: { $sum: 1 } } },
];
const aggData = [
{ $project: {
'value.dataStoreName': 1,
'value.content-length': 1,
} },
{ $group: {
_id: '$value.dataStoreName',
bytes: { $sum: '$value.content-length' },
} },
];
const aggRepData = [
{ $project: {
'value.replicationInfo.backends': 1,
'value.content-length': 1,
} },
{ $unwind: '$value.replicationInfo.backends' },
{ $match: {
'value.replicationInfo.backends.status': { $eq: 'COMPLETED' },
} },
{ $group: {
_id: '$value.replicationInfo.backends.site',
bytes: { $sum: '$value.content-length' },
} },
];
return c.aggregate([
{ $project: reducedFields },
{ $match: filter },
{ $facet: {
count: aggCount,
data: aggData,
repData: aggRepData,
} },
]).toArray((err, res) => {
if (err) { if (err) {
log.error('error occured in mongo client', { log.error('Error when processing mongo entries', {
method: 'getDataManaged', method: '_handleMongo',
error: err, error: err,
}); });
return cb(err); return cb(err);
} }
if (!res || res.length < 1) {
// change to use replicatio log.debug('aggregate returned empty results', {
entries.forEach(entry => { method: '_handleMongo',
const { backends } = entry.value.replicationInfo || {}; });
const { dataStoreName, 'content-length': size } = entry.value; return cb(null, {});
addLocation(dataStoreName, size);
if (backends && Array.isArray(backends)) {
backends.forEach(loc => {
const { site, status } = loc;
if (status === 'COMPLETED') {
addLocation(site, size);
} }
const agg = res[0];
if (!agg || !agg.count || !agg.data || !agg.repData) {
log.debug('missing field in aggregate results', {
method: '_handleMongo',
}); });
return cb(null, {});
} }
}); const retResult = {
return cb(null, { count: this._handleCount(agg.count[0] || {}),
objCount: entries.length, data: this._handleEntries(agg.data),
dataManaged,
});
});
}
getDataInfoNoVer(c, log, callback) {
return this.getDataManaged(c, {}, log, (err, res) => {
if (err) {
log.error('error occured in mongo client', {
method: 'getDataInfoNoVer',
error: err,
});
return callback(err);
}
const dataManaged = {
total: { curr: 0, prev: 0 },
locations: {},
}; };
const data = res.dataManaged.locations; const repDataEntries = this._handleEntries(agg.repData);
// add total Object.keys(repDataEntries).forEach(site => {
dataManaged.total.curr += res.dataManaged.total; if (!retResult.data[site]) {
Object.keys(data).forEach(loc => { retResult.data[site] = 0;
if (!dataManaged.locations[loc]) {
dataManaged.locations[loc] = { prev: 0, curr: 0 };
} }
dataManaged.locations[loc].curr = data[loc]; retResult.data[site] += repDataEntries[site];
});
return callback(null, {
objects: res.objCount,
versions: 0,
dataManaged,
});
});
}
getDataInfoVer(c, mstVerIds, log, callback) {
// query for removing placeholder entries
/* eslint-disable quote-props */
const queryFilter = {
'$or': [
{ '_id': { '$regex': /\0.*$/g } },
{ 'value.isNull': { '$exists': true } },
{
'value.nullVersionId': { '$exists': false },
'value.versionId': { '$exists': false },
},
],
};
/* eslint-enable quote-props */
return async.series({
master: done => {
// query for getting only master version entries
const mstFilter = [];
/* eslint-disable quote-props */
mstVerIds.forEach(id => {
if (id) {
mstFilter.push({ 'value.versionId': { '$eq': id } });
}
});
const query = {
'$and': [
queryFilter,
{ '$or': (mstFilter.length ? mstFilter : [{}]) },
],
};
/* eslint-enable quote-props */
return this.getDataManaged(c, query, log, done);
},
archived: done => {
// query for getting only versioned entries
const mstFilter = [];
/* eslint-disable quote-props */
mstVerIds.forEach(id => {
if (id) {
mstFilter.push({ 'value.versionId': { '$ne': id } });
}
});
const query = {
'$and': [
queryFilter,
{ '$and': (mstFilter.length ? mstFilter : [{}]) },
],
};
/* eslint-enable quote-props */
return this.getDataManaged(c, query, log, done);
},
}, (err, res) => {
if (err) {
log.error('error occured in mongo client', {
method: 'getDataInfoVer',
error: err,
});
return callback(err);
}
const dataManaged = {
total: { curr: 0, prev: 0 },
locations: {},
};
const mstData = res.master.dataManaged.locations;
const verData = res.archived.dataManaged.locations;
// add total
dataManaged.total.curr += res.master.dataManaged.total;
dataManaged.total.prev += res.archived.dataManaged.total;
Object.keys(mstData).forEach(loc => {
if (!dataManaged.locations[loc]) {
dataManaged.locations[loc] = { prev: 0, curr: 0 };
}
dataManaged.locations[loc].curr = mstData[loc];
});
Object.keys(verData).forEach(loc => {
if (!dataManaged.locations[loc]) {
dataManaged.locations[loc] = { prev: 0, curr: 0 };
}
dataManaged.locations[loc].prev = verData[loc];
});
return callback(null, {
objects: res.master.objCount,
versions: res.archived.objCount,
dataManaged,
}); });
return cb(null, retResult);
}); });
} }
@ -1310,60 +1289,40 @@ class MongoClientInterface {
name: bucketName, name: bucketName,
location: bucketInfo.getLocationConstraint(), location: bucketInfo.getLocationConstraint(),
}; };
if (bucketInfo && bucketInfo._versioningConfiguration &&
(bucketInfo._versioningConfiguration.Status === 'Suspended' || const mstFilter = {
bucketInfo._versioningConfiguration.Status === 'Enabled')) { '_id': { $regex: /^[^\0]+$/ },
// if versioning is enabled 'value.versionId': { $exists: true },
c.distinct('_id').then(keys => { };
const trimmedKeys = keys.map(key => key.replace(/\0.*$/g, '')); const verFilter = { _id: { $regex: /\0/ } };
const uniqKeys = trimmedKeys.filter( const nullFilter = {
(key, index, self) => self.indexOf(key) === index); '_id': { $regex: /^[^\0]+$/ },
// for each uniqKey get master version id 'value.versionId': { $exists: false },
return async.map(uniqKeys, (key, done) => { };
this.getLatestVersion(c, key, log, (err, mst) => {
if (err) { async.parallel({
if (err.NoSuchKey) { version: done => this._handleMongo(c, verFilter, log, done),
log.debug('NoSuchKey master info', { null: done => this._handleMongo(c, nullFilter, log, done),
method: 'getObjectMDStats', master: done => this._handleMongo(c, mstFilter, log, done),
error: err, }, (err, res) => {
});
return done();
}
log.error('unable to retrieve master info', {
method: 'getObjectMDStats',
error: err,
});
return done(err);
}
return done(null, mst.versionId);
});
}, (err, mstVerIds) => {
if (err) { if (err) {
return callback(err); return callback(err);
} }
return this.getDataInfoVer(c, mstVerIds, log, const resObj = {
(err, res) => { masterCount: res.master.count || 0,
if (err) { masterData: res.master.data || {},
return callback(err); nullCount: res.null.count || 0,
} nullData: res.null.data || {},
return callback(null, { versionCount: res.version.count || 0,
bucketList: retBucketInfo, versionData: res.version.data || {},
counter: res, };
const bucketStatus = bucketInfo.getVersioningConfiguration();
const isVer = (bucketStatus && (bucketStatus.Status === 'Enabled' ||
bucketStatus.Status === 'Suspended'));
const retResult = this._handleResults(resObj, isVer);
retResult.bucketList = retBucketInfo;
return callback(null, retResult);
}); });
});
});
}).catch(callback);
} else {
this.getDataInfoNoVer(c, log, (err, res) => {
if (err) {
return callback(err);
}
return callback(null, {
bucketList: retBucketInfo,
counter: res,
});
});
}
} }
} }

View File

@ -3,7 +3,7 @@
"engines": { "engines": {
"node": ">=6.9.5" "node": ">=6.9.5"
}, },
"version": "8.0.0", "version": "8.0.0-603.6",
"description": "Common utilities for the S3 project components", "description": "Common utilities for the S3 project components",
"main": "index.js", "main": "index.js",
"repository": { "repository": {

View File

@ -328,3 +328,62 @@ describe('MongoClientInterface::dataCount', () => {
}); });
}); });
}); });
describe('MongoClientInterface::_handleResults', () => {
it('should return zero-result', () => {
const testInput = {
masterCount: 0, masterData: {},
nullCount: 0, nullData: {},
versionCount: 0, versionData: {},
};
const testResults = mongoTestClient._handleResults(testInput, true);
const expectedRes = {
versions: 0, objects: 0,
dataManaged: {
total: { curr: 0, prev: 0 },
locations: {},
},
};
assert.deepStrictEqual(testResults, expectedRes);
});
it('should return correct value if isVer is false', () => {
const testInput = {
masterCount: 2, masterData: { test1: 10, test2: 10 },
nullCount: 2, nullData: { test1: 10, test2: 10 },
versionCount: 2, versionData: { test1: 20, test2: 20 },
};
const testResults = mongoTestClient._handleResults(testInput, false);
const expectedRes = {
versions: 0, objects: 4,
dataManaged: {
total: { curr: 40, prev: 0 },
locations: {
test1: { curr: 20, prev: 0 },
test2: { curr: 20, prev: 0 },
},
},
};
assert.deepStrictEqual(testResults, expectedRes);
});
it('should return correct value if isVer is true', () => {
const testInput = {
masterCount: 2, masterData: { test1: 10, test2: 10 },
nullCount: 2, nullData: { test1: 10, test2: 10 },
versionCount: 4, versionData: { test1: 20, test2: 20 },
};
const testResults = mongoTestClient._handleResults(testInput, true);
const expectedRes = {
versions: 2, objects: 4,
dataManaged: {
total: { curr: 40, prev: 20 },
locations: {
test1: { curr: 20, prev: 10 },
test2: { curr: 20, prev: 10 },
},
},
};
assert.deepStrictEqual(testResults, expectedRes);
});
});