Multi Node Cluster support for move-node

bin/get-shard-mapping.js

@@ -0,0 +1,13 @@
+#!/usr/bin/env node
+
+const { getShardMapping } = require('../src/utils');
+
+(async () => {
+  try {
+    const shardMapping = await getShardMapping();
+    console.log(JSON.stringify(shardMapping));
+  } catch (err) {
+    console.error('An unexpected error occurred', err);
+    process.exit(1);
+  }
+})();

bin/move-node.js

@@ -1,16 +1,70 @@
 #!/usr/bin/env node
+const [,, toNode] = process.argv;
 
 const { moveNode, syncShards } = require('../src/move-node');
 const { removeNode } = require('../src/remove-node');
 
+const parseNodeMapping = (input) => {
+  if (!input) {
+    // Single node migration with unspecified node
+    return;
+  }
+  try {
+    // Multi-node migration with specified node mapping
+    // Example input: '{"oldNode":"newNode"}'
+    return JSON.parse(input);
+  } catch (err) {
+    // Single node migration with specified node
+    console.warn('Failed to parse node mapping JSON. Defaulting to single node mapping');
+    return input;
+  }
+};
+
+const getStdin = () => {
+  return new Promise((resolve) => {
+    if (process.stdin.isTTY) {
+      return resolve('');
+    }
+    let data = '';
+    process.stdin.setEncoding('utf8');
+    process.stdin.on('data', chunk => data += chunk);
+    process.stdin.on('end', () => resolve(data));
+  });
+};
+
+const parseShardMapJsonInput = async () => {
+  if (process.env.SHARD_MAPPING) {
+    try {
+      return JSON.parse(process.env.SHARD_MAPPING);
+    } catch (err) {
+      console.warn('Failed to parse SHARD_MAPPING environment variable');
+    }
+  }
+
+  try {
+    const input = await getStdin();
+    if (input.trim()) {
+      return JSON.parse(input.trim());
+    }
+  } catch (err) {
+    console.warn('Failed to parse shard mapping JSON from stdin');
+  }
+  return;
+};
+
 (async () => {
   try {
-    const removedNodes = await moveNode();
-    for (const node of removedNodes) {
+    const parsedToNode = parseNodeMapping(toNode);
+    const shardMapJson = await parseShardMapJsonInput();
+
+    const movedNodes = await moveNode(parsedToNode, shardMapJson);
+    console.log('Node moved successfully');
+
+    for (const node of movedNodes) {
       await removeNode(node);
     }
     await syncShards();
-    console.log('Node moved successfully');
+    console.log('Removed nodes:', movedNodes);
   } catch (err) {
     console.error('An unexpected error occurred', err);
     process.exit(1);

docker-compose.yml

@@ -2,7 +2,7 @@ version: '3.9'
 
 services:
   couch-migration:
-    image: public.ecr.aws/medic/couchdb-migration:1.0.3
+    image: public.ecr.aws/medic/couchdb-migration:1.1.0
     networks:
       - cht-net
     environment:

package.json

@@ -1,6 +1,6 @@
 {
   "name": "couchdb-migration",
-  "version": "1.0.3",
+  "version": "1.1.0",
   "description": "Tool that interfaces with CouchDB to help migrating data within a cluster.",
   "scripts": {
     "eslint": "eslint --color --cache ./src ./test",
@@ -22,7 +22,8 @@
     "remove-node": "bin/remove-node.js",
     "check-couchdb-up": "bin/check-couchdb-up.js",
     "pre-index-views": "bin/pre-index-views.js",
-    "verify": "bin/verify.js"
+    "verify": "bin/verify.js",
+    "get-shard-mapping": "bin/get-shard-mapping.js"
   },
   "repository": {
     "type": "git",

src/check-couch-up.js

@@ -1,5 +1,5 @@
 const utils = require('./utils');
-const NUM_RETRY = 200;
+const NUM_RETRY = 300;
 const TIMEOUT_RETRY = 1000; // 1 second
 
 const isCouchUp = async () => {
@@ -33,14 +33,14 @@ const repeatRetry = async (promiseFn) => {
 const checkCouchUp = async () => {
   const isUp = await repeatRetry(isCouchUp);
   if (!isUp) {
-    throw new Error('CouchDb is not up after 100 seconds.');
+    throw new Error('CouchDb is not up after 300 seconds.');
   }
 };
 
 const checkClusterReady = async (nbrNodes) => {
   const isReady = await repeatRetry(isClusterReady.bind({}, nbrNodes));
   if (!isReady) {
-    throw new Error('CouchDb Cluster is not ready after 100 seconds.');
+    throw new Error('CouchDb Cluster is not ready after 300 seconds.');
   }
 };
 

src/move-node.js

@@ -1,13 +1,13 @@
 const utils = require('./utils');
 const moveShard = require('./move-shard');
 
-const moveNode = async (toNode) => {
+const replaceForSingleNode = async (toNode) => {
   const removedNodes = [];
 
   if (!toNode) {
     const nodes = await utils.getNodes();
     if (nodes.length > 1) {
-      throw new Error('More than one node found.');
+      throw new Error('More than one node found. Please specify a node mapping in the format {"oldNode":"newNode"}.');
     }
     toNode = nodes[0];
   }
@@ -20,10 +20,75 @@ const moveNode = async (toNode) => {
   return [...new Set(removedNodes)];
 };
 
+const replaceInCluster = async (nodeMap, shardMapJson) => {
+  const removedNodes = [];
+  if (!shardMapJson) {
+    throw new Error('Shard map JSON is required for multi-node migration');
+  }
+
+  const [oldNode, newNode] = Object.entries(nodeMap)[0];
+  console.log(`Migrating from ${oldNode} to ${newNode}`);
+
+  // Use the provided shard map
+  const currentDistribution = shardMapJson;
+
+  // For each shard range in the current distribution
+  console.log('Current distribution:', currentDistribution);
+  for (const [shardRange, dbNodes] of Object.entries(currentDistribution)) {
+    console.log('Shard range:', shardRange);
+    console.log('DB nodes:', dbNodes);
+    // dbNodes is an object mapping db names to node names
+    for (const [dbName, nodeName] of Object.entries(dbNodes)) {
+      console.log('DB name:', dbName);
+      console.log('Node name:', nodeName);
+      if (nodeName === oldNode) {
+        console.log(
+          `Moving shard ${shardRange} for db ${dbName} from ${oldNode} to ${newNode}`
+        );
+        const oldNodes = await moveShard.moveShard(shardRange, newNode, dbName);
+        removedNodes.push(...oldNodes);
+      }
+    }
+  }
+  if (!removedNodes.includes(oldNode)) {
+    removedNodes.push(oldNode);
+  }
+  return [...new Set(removedNodes)];
+};
+
+const moveNode = async (nodeMap, shardMapJson) => {
+  if (typeof nodeMap === 'object') {
+    // Multi-node migration - replace node in clustered couch
+    return await replaceInCluster(nodeMap, shardMapJson);
+  }
+
+  // Single node migration - replace node in a single node couch
+  return await replaceForSingleNode(nodeMap);
+};
+
 const syncShards = async () => {
-  const allDbs = await utils.getDbs();
-  for (const db of allDbs) {
-    await utils.syncShards(db);
+  try {
+    const membership = await utils.getMembership();
+    const { all_nodes, cluster_nodes } = membership;
+
+    const clusterComplete =
+      all_nodes.length === cluster_nodes.length &&
+      all_nodes.every((node) => cluster_nodes.includes(node));
+
+    if (clusterComplete) {
+      const allDbs = await utils.getDbs();
+      for (const db of allDbs) {
+        await utils.syncShards(db);
+      }
+      console.log('Shards synchronized.');
+    } else {
+      console.log(
+        'Multi-node migration detected. Shard synchronization will be run once all nodes have been migrated.'
+      );
+    }
+  } catch (err) {
+    console.error('Error during shard synchronization:', err);
+    throw err;
   }
 };
 

src/move-shard.js

@@ -40,11 +40,11 @@ const updateDbMetadata = (metadata, shard, toNode) => {
   return fromNodes;
 };
 
-const moveShard = async (shard, toNode) => {
+const moveShard = async (shard, toNode, dbName = null) => {
   await validateCall(shard, toNode);
   const removedNodes = [];
 
-  const dbs = await utils.getDbs();
+  const dbs = dbName ? [dbName] : await utils.getDbs();
   for (const dbName of dbs) {
     const metadata = await utils.getDbMetadata(dbName);
     const oldNodes = updateDbMetadata(metadata, shard, toNode);

src/utils.js

@@ -221,6 +221,43 @@ const getShards = async () => {
   }
 };
 
+/*
+This function creates a mapping of shard ranges to nodes and databases
+*/
+const getShardMapping = async () => {
+  try {
+    const allDbs = await getDbs();
+    const shardMap = {};
+
+    for (const db of allDbs) {
+      const url = await getUrl(`${db}/_shards`);
+      const shardInfo = await request({ url });
+
+      for (const [shardRange, nodeList] of Object.entries(shardInfo.shards)) {
+        // In n=1 setup, there should be only one node per shard range.
+        // We will have to revisit this if we ever support n>1.
+        if (nodeList.length !== 1) {
+          console.warn(`Unexpected number of nodes for range ${shardRange} in db ${db}: ${nodeList.length}`);
+        }
+        const node = nodeList[0];
+
+        // Initialize the shard range in the shardMap if not present
+        if (!shardMap[shardRange]) {
+          shardMap[shardRange] = {};
+        }
+
+        // Map the database to the node for the shard range
+        shardMap[shardRange][db] = node;
+      }
+    }
+
+    return shardMap;
+  } catch (err) {
+    console.error('Error getting shard mapping:', err);
+    throw new Error('Failed to get shard mapping');
+  }
+};
+
 const getNodes = async () => {
   const membership = await getMembership();
   return membership.all_nodes;
@@ -257,6 +294,7 @@ module.exports = {
   deleteNode,
   syncShards,
   getShards,
+  getShardMapping,
   getNodes,
   getConfig,
   getCouchUrl,

src/verify.js

@@ -1,4 +1,5 @@
 const utils = require('./utils');
+const DBS_TO_IGNORE = ['_global_changes', '_replicator'];
 
 const verifyViews = async (dbName, numDocs) => {
   if (!numDocs) {
@@ -29,6 +30,9 @@ const verifyViews = async (dbName, numDocs) => {
 };
 
 const verifyDb = async (dbName) => {
+  if (DBS_TO_IGNORE.includes(dbName)) {
+    return;
+  }
   console.info(`Verifying ${dbName}`);
   await utils.syncShards(dbName);
 

test/docker-compose-test.yml

@@ -11,3 +11,4 @@ services:
 networks:
   cht-net:
     name: ${CHT_NETWORK:-cht-net}
+    external: true

test/e2e/multi-node-3.sh

@@ -29,37 +29,37 @@ docker rm -f -v scripts-couchdb-1.local-1 scripts-couchdb-2.local-1 scripts-couc
 # create docker network
 docker network create $CHT_NETWORK || true
 # build service image
-docker-compose -f ../docker-compose-test.yml up --build
+docker compose -f ../docker-compose-test.yml up --build
 
 # launch vanilla couch, populate with some data
-docker-compose -f ./scripts/couchdb-vanilla.yml up -d
-docker-compose -f ../docker-compose-test.yml run couch-migration check-couchdb-up
+docker compose -f ./scripts/couchdb-vanilla.yml up -d
+docker compose -f ../docker-compose-test.yml run couch-migration check-couchdb-up
 node ./scripts/generate-documents $jsondataddir
 # pre-index 4.0.1 views
-docker-compose -f ../docker-compose-test.yml run couch-migration pre-index-views 4.4.0
+docker compose -f ../docker-compose-test.yml run couch-migration pre-index-views 4.4.0
 sleep 5 # this is needed, CouchDb runs fsync with a 5 second delay
 # export env for cht 4.x couch
-export $(docker-compose -f ../docker-compose-test.yml run couch-migration get-env | xargs)
-docker-compose -f ./scripts/couchdb-vanilla.yml down --remove-orphans --volumes
+export $(docker compose -f ../docker-compose-test.yml run couch-migration get-env | xargs)
+docker compose -f ./scripts/couchdb-vanilla.yml down --remove-orphans --volumes
 
 # launch cht 4.x CouchDb cluster
-docker-compose -f ./scripts/couchdb3-cluster.yml up -d
-docker-compose -f ../docker-compose-test.yml run couch-migration check-couchdb-up 3
+docker compose -f ./scripts/couchdb3-cluster.yml up -d
+docker compose -f ../docker-compose-test.yml run couch-migration check-couchdb-up 3
 
 # generate shard matrix
 # this is an object that assigns every shard to one of the nodes
-shard_matrix=$(docker-compose -f ../docker-compose-test.yml run couch-migration generate-shard-distribution-matrix)
+shard_matrix=$(docker compose -f ../docker-compose-test.yml run couch-migration generate-shard-distribution-matrix)
 file_matrix="{\"[email protected]\":\"$couch1dir\",\"[email protected]\":\"$couch2dir\",\"[email protected]\":\"$couch3dir\"}"
 echo $shard_matrix
 echo $file_matrix
 # moves shard data files to their corresponding nodes, according to the matrix
-docker-compose -f ../docker-compose-test.yml run couch-migration shard-move-instructions $shard_matrix
+docker compose -f ../docker-compose-test.yml run couch-migration shard-move-instructions $shard_matrix
 node ./scripts/distribute-shards.js $shard_matrix $file_matrix
 # change database metadata to match the shard physical locations
-docker-compose -f ../docker-compose-test.yml run couch-migration move-shards $shard_matrix
-docker-compose -f ../docker-compose-test.yml run couch-migration verify
+docker compose -f ../docker-compose-test.yml run couch-migration move-shards $shard_matrix
+docker compose -f ../docker-compose-test.yml run couch-migration verify
 # test that data exists, database shard maps are correct and view indexes are preserved
 node ./scripts/assert-dbs.js $jsondataddir $shard_matrix
 
-docker-compose -f ./scripts/couchdb-cluster.yml down --remove-orphans --volumes
+docker compose -f ./scripts/couchdb-cluster.yml down --remove-orphans --volumes