1
0
Fork 0

Fixed new discrepancies between the etcd agency and the high level javascript communication, based on Shards in Collections in the Current state

This commit is contained in:
Michael Hackstein 2014-01-21 14:16:24 +01:00
parent 41b35c82da
commit 7cd01e816c
2 changed files with 94 additions and 14 deletions

View File

@ -39,7 +39,10 @@ exports.Communication = function() {
storeServersInCache, storeServersInCache,
Target, Target,
mapCollectionIDsToNames, mapCollectionIDsToNames,
updateCollectionRouteForName,
updateDatabaseRoutes,
difference, difference,
self = this,
_ = require("underscore"); _ = require("underscore");
splitServerName = function(route) { splitServerName = function(route) {
@ -135,17 +138,17 @@ exports.Communication = function() {
var target = addLevel(this, "target", "Target"); var target = addLevel(this, "target", "Target");
addLevel(target, "dbServers", "DBServers", ["get", "set", "remove", "checkVersion"]); addLevel(target, "dbServers", "DBServers", ["get", "set", "remove", "checkVersion"]);
addLevel(target, "db", "Collections", ["list"]); addLevel(target, "db", "Collections", ["list"]);
addLevelsForDBs(target.db, true); //addLevelsForDBs(target.db, true);
addLevel(target, "coordinators", "Coordinators", ["list", "set", "remove", "checkVersion"]); addLevel(target, "coordinators", "Coordinators", ["list", "set", "remove", "checkVersion"]);
var plan = addLevel(this, "plan", "Plan"); var plan = addLevel(this, "plan", "Plan");
addLevel(plan, "dbServers", "DBServers", ["get", "checkVersion"]); addLevel(plan, "dbServers", "DBServers", ["get", "checkVersion"]);
addLevel(plan, "db", "Collections", ["list"]); addLevel(plan, "db", "Collections", ["list"]);
addLevelsForDBs(plan.db); //addLevelsForDBs(plan.db);
addLevel(plan, "coordinators", "Coordinators", ["list", "checkVersion"]); addLevel(plan, "coordinators", "Coordinators", ["list", "checkVersion"]);
var current = addLevel(this, "current", "Current"); var current = addLevel(this, "current", "Current");
addLevel(current, "dbServers", "DBServers", ["get", "checkVersion"]); addLevel(current, "dbServers", "DBServers", ["get", "checkVersion"]);
addLevel(current, "db", "Collections", ["list"]); addLevel(current, "db", "Collections", ["list"]);
addLevelsForDBs(current.db); //addLevelsForDBs(current.db);
addLevel(current, "coordinators", "Coordinators", ["list", "checkVersion"]); addLevel(current, "coordinators", "Coordinators", ["list", "checkVersion"]);
addLevel(current, "registered", "ServersRegistered", ["get", "checkVersion"]); addLevel(current, "registered", "ServersRegistered", ["get", "checkVersion"]);
@ -153,6 +156,7 @@ exports.Communication = function() {
addLevel(sync, "beat", "ServerStates", ["get"]); addLevel(sync, "beat", "ServerStates", ["get"]);
addLevel(sync, "interval", "HeartbeatIntervalMs", ["get"]); addLevel(sync, "interval", "HeartbeatIntervalMs", ["get"]);
this.addLevel = addLevel;
}; };
agency = new AgencyWrapper(); agency = new AgencyWrapper();
@ -162,6 +166,33 @@ exports.Communication = function() {
// --SECTION-- Helper Functions // --SECTION-- Helper Functions
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
updateDatabaseRoutes = function(base, writeAccess) {
var list = self.plan.Databases().getList();
_.each(_.keys(base), function(k) {
if (k !== "route" && k !== "list") {
delete base[k];
}
});
_.each(list, function(d) {
agency.addLevel(base, d, d, ["get", "checkVersion"]);
});
};
updateCollectionRouteForName = function(route, db, name, writeAccess) {
var list = self.plan.Databases().select(db).getCollectionObjects();
var cId = null;
_.each(list, function(v, k) {
if (v.name === name) {
cId = splitServerName(k);
}
});
var acts = ["get"];
if (writeAccess) {
acts.push("set");
}
agency.addLevel(route, name, cId, acts);
};
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
/// @brief Stores database servers in cache /// @brief Stores database servers in cache
/// ///
@ -318,18 +349,27 @@ exports.Communication = function() {
/// It allos to get a list of collections and to select one of them for /// It allos to get a list of collections and to select one of them for
/// further information. /// further information.
//////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////
var DBObject = function(route, writeAccess) { var DBObject = function(route, db, writeAccess) {
var cache; var cache;
var getList = function() { var getRaw = function() {
if (!cache || !route.checkVersion()) { if (!cache || !route.checkVersion()) {
cache = _.keys(mapCollectionIDsToNames(route.get(true))).sort(); cache = route.get(true);
} }
return cache; return cache;
}; };
var getList = function() {
return _.keys(mapCollectionIDsToNames(
self.plan.Databases().select(db).getCollectionObjects()
)).sort();
};
this.getCollectionObjects = function() {
return getRaw();
};
this.getCollections = function() { this.getCollections = function() {
return getList(); return getList();
}; };
this.collection = function(name) { this.collection = function(name) {
updateCollectionRouteForName(route, db, name, writeAccess);
var colroute = route[name]; var colroute = route[name];
if (!colroute) { if (!colroute) {
return false; return false;
@ -350,11 +390,12 @@ exports.Communication = function() {
return route.list(); return route.list();
}; };
this.select = function(name) { this.select = function(name) {
updateDatabaseRoutes(route, writeAccess);
var subroute = route[name]; var subroute = route[name];
if (!subroute) { if (!subroute) {
return false; return false;
} }
return new DBObject(subroute, writeAccess); return new DBObject(subroute, name, writeAccess);
}; };
}; };

View File

@ -101,6 +101,42 @@
"11235": {name: "s"}, "11235": {name: "s"},
"6512": {name: "a"}, "6512": {name: "a"},
"123": {name: "d"} "123": {name: "d"}
},
current: {
_system: {
"98213": {
"sg1": {},
"sg2": {},
"sg3": {}
},
"87123": {
"sv1": {},
"sv2": {},
"sv3": {}
},
"89123": {
"se1": {},
"se2": {},
"se3": {}
}
},
a_db: {
"11235": {
"s01": {},
"s02": {},
"s03": {}
},
"6512": {
"s11": {},
"s12": {},
"s13": {}
},
"123": {
"s21": {},
"s22": {},
"s23": {}
}
}
} }
}; };
var ips = { var ips = {
@ -162,8 +198,8 @@
dummy.current.coordinators = createResult([agencyRoutes.current, agencyRoutes.sub.coords], coordinators); dummy.current.coordinators = createResult([agencyRoutes.current, agencyRoutes.sub.coords], coordinators);
dummy.current.registered = createResult([agencyRoutes.current, agencyRoutes.sub.registered], ips); dummy.current.registered = createResult([agencyRoutes.current, agencyRoutes.sub.registered], ips);
dummy.current.databases = databases; dummy.current.databases = databases;
dummy.current.syscollections = createResult([agencyRoutes.current, agencyRoutes.sub.databases, agencyRoutes.sub.colls, "_system"], collections._system); dummy.current.syscollections = createResult([agencyRoutes.current, agencyRoutes.sub.databases, agencyRoutes.sub.colls, "_system"], collections.current._system);
dummy.current.acollections = createResult([agencyRoutes.current, agencyRoutes.sub.databases, agencyRoutes.sub.colls, "a_db"], collections.a_db); dummy.current.acollections = createResult([agencyRoutes.current, agencyRoutes.sub.databases, agencyRoutes.sub.colls, "a_db"], collections.current.a_db);
dummy.current.vInfo = vInfo; dummy.current.vInfo = vInfo;
dummy.sync = {}; dummy.sync = {};
@ -240,9 +276,9 @@
} }
break; break;
default: default:
fail(); fail("Requested route: GET " + route);
} }
fail(); fail("Requested route: GET " + route);
}, },
list: function(route, recursive, flat) { list: function(route, recursive, flat) {
var parts = route.split("/"); var parts = route.split("/");
@ -280,7 +316,7 @@
} }
break; break;
default: default:
fail(); fail("Requested route: LIST " + route);
} }
} }
}; };
@ -562,7 +598,7 @@
].sort(); ].sort();
assertEqual(dbs.getList(), list); assertEqual(dbs.getList(), list);
}, },
testGetCollectionListForDatabase: function() { testGetCollectionListForDatabase: function() {
var syslist = [ var syslist = [
"_graphs", "_graphs",
@ -631,6 +667,7 @@
assertTrue(wasCalled, "Agency has not been informed to move shard.."); assertTrue(wasCalled, "Agency has not been informed to move shard..");
assertEqual(colV.getServerForShard(shard), target); assertEqual(colV.getServerForShard(shard), target);
} }
}; };
}; };
@ -777,6 +814,7 @@
assertEqual(colV.getServerForShard("v1"), "pavel"); assertEqual(colV.getServerForShard("v1"), "pavel");
assertEqual(colV.getServerForShard("v2"), "paul"); assertEqual(colV.getServerForShard("v2"), "paul");
} }
}; };
}; };
@ -877,7 +915,7 @@
].sort(); ].sort();
assertEqual(dbs.getList(), list); assertEqual(dbs.getList(), list);
}, },
testGetCollectionListForDatabase: function() { testGetCollectionListForDatabase: function() {
var syslist = [ var syslist = [
"_graphs", "_graphs",
@ -927,6 +965,7 @@
assertEqual(colV.getServerForShard("v1"), "pavel"); assertEqual(colV.getServerForShard("v1"), "pavel");
assertEqual(colV.getServerForShard("v2"), "paul"); assertEqual(colV.getServerForShard("v2"), "paul");
} }
}; };
}; };