Overview of MongoDB Sharded Clusters

This guide outlines the process of deploying a MongoDB 4.0 sharded cluster on CentOS 7, leveraging a Primary-Secondary-Arbiter (PSA) replica set configuration for high availability and data distribution. A sharded cluster enhances scalability by distributing data across multiple servers (shards), enabling horizontal scaling and handling larger datasets and higher throughput than a single replica set.

Cluster Architecture

The proposed sharded cluster setup consists of:

• Config Servers (Config SVR): A replica set storing the cluster's metadata, including chunk ranges and shard locations. Three instances ensure high availability of metadata.
• Shard Replica Sets (Shards): Individual replica sets (each with a primary, secondary, and arbiter) that store the actual data. For this deployment, we'll configure two shard replica sets (Shard1 and Shard2).
• Query Routers (Mongos): Client-facing processes that route read and write operations to the appropriate shards. Multiple mongos instances provide load balancing and fault tolerance.

Component Distribution Across Hosts

We will use five CentOS 7 hosts, generically named host1 through host5, with specific roles and port assignments:

Hostname	Role(s)
host1	Shard1 Arbiter (Port 27026), Shard2 Arbiter (Port 27027)
host2	mongos (Port 20000), Config Server (Port 27000), Shard1 Secondary (Port 27001)
host3	mongos (Port 20000), Config Server (Port 27000), Shard2 Secondary (Port 27001)
host4	mongos (Port 20000), Config Server (Port 27000), Shard1 Primary (Port 27001)
host5	mongos (Port 20000), Shard2 Primary (Port 27001)

Port Assignments

Component	Port
mongos	20000
Config Server	27000
Shard Member	27001
Arbiter 1	27026
Arbiter 2	27027

Initial System Configuration on All Hosts (host1-host5)

1. Configure MongoDB Yum Repository

Create the repository file /etc/yum.repos.d/mongodb-org-4.0.repo:

sudo tee /etc/yum.repos.d/mongodb-org-4.0.repo << EOF
[mongodb-org-4.0]
name=MongoDB Repository
baseurl=https://repo.mongodb.org/yum/redhat/7/mongodb-org/4.0/x86_64/
gpgcheck=1
enabled=1
gpgkey=https://www.mongodb.org/static/pgp/server-4.0.asc
EOF

2. Install MongoDB 4.0 Packages

sudo yum install -y mongodb-org-4.0.9 mongodb-org-server-4.0.9 mongodb-org-shell-4.0.9 mongodb-org-mongos-4.0.9 mongodb-org-tools-4.0.9

To prevent unintended upgrades, exclude MongoDB packages from future yum updates:

echo 'exclude=mongodb-org,mongodb-org-server,mongodb-org-shell,mongodb-org-mongos,mongodb-org-tools' | sudo tee -a /etc/yum.conf

3. Create Data and Log Directories, Disable SELinux

Create necessary directories and adjust permissions. The mongod user will own these directories:

sudo mkdir -p /var/lib/mongo-cluster/config
sudo mkdir -p /var/lib/mongo-cluster/mongos
sudo mkdir -p /var/lib/mongo-cluster/shards/shard1
sudo mkdir -p /var/lib/mongo-cluster/shards/shard2
sudo mkdir -p /var/lib/mongo-cluster/arbiters/arbiter1
sudo mkdir -p /var/lib/mongo-cluster/arbiters/arbiter2

sudo chown -R mongod:mongod /var/lib/mongo-cluster

Disable SELinux by editing /etc/sysconfig/selinux and setting SELINUX=disabled. Then enforce the change immediately:

sudo sed -i 's/SELINUX=enforcing/SELINUX=disabled/' /etc/sysconfig/selinux
sudo setenforce 0

4. Adjust File Open Limits

Increase the maximum number of open files for the mongod process:

# For current session
ulimit -n 1000000

# For persistent changes
sudo tee /etc/security/limits.d/99-mongodb.conf << EOF
* soft nofile 1000000
* hard nofile 1000000
* soft fsize unlimited
* hard fsize unlimited
* soft cpu unlimited
* hard cpu unlimited
* soft nproc 500000
* hard nproc 500000
EOF

5. Install NUMA Utilities and Disable Transparent Huge Pages

Install numactl:

sudo yum install -y numactl

Disable Transparent Huge Pages (THP) as recommended for MongoDB performance:

echo "never" | sudo tee /sys/kernel/mm/transparent_hugepage/enabled
echo "never" | sudo tee /sys/kernel/mm/transparent_hugepage/defrag

Also, disable zone reclaim mode:

sudo sysctl -w vm.zone_reclaim_mode=0

Configuring and Starting Cluster Components

1. Config Server Replica Set Setup (host2, host3, host4)

On each of host2, host3, and host4, create the config server configuration file /var/lib/mongo-cluster/config/mongodb.conf:

processManagement:
   fork: true
net:
   bindIp: 0.0.0.0
   port: 27000
storage:
   dbPath: /var/lib/mongo-cluster/config/data
replication:
   replSetName: cfgReplSet
sharding:
   clusterRole: configsvr
systemLog:
   destination: file
   path: "/var/lib/mongo-cluster/config/log/config.log"
   logAppend: true
   timeStampFormat: ctime

Create data and log directories for the config servers on these hosts:

sudo mkdir -p /var/lib/mongo-cluster/config/data
sudo mkdir -p /var/lib/mongo-cluster/config/log
sudo chown -R mongod:mongod /var/lib/mongo-cluster/config

Start the config server processes on host2, host3, and host4:

numactl --interleave=all mongod -f /var/lib/mongo-cluster/config/mongodb.conf

Initialize the config server replica set. Connect to any one of the config servers (e.g., host2):

mongo --port 27000

Then execute:

rs.initiate(
  {
    _id: "cfgReplSet",
    configsvr: true,
    members: [
      { _id : 0, host : "host2:27000" },
      { _id : 1, host : "host3:27000" },
      { _id : 2, host : "host4:27000" }
    ]
  }
)

Verify the replica set status with rs.status().

2. Shard Replica Sets Setup (Shard1 and Shard2)

Shard Member Configuration

On host2 (Shard1 Secondary), host3 (Shard2 Secondary), host4 (Shard1 Primary), and host5 (Shard2 Primary), create the shard member configuration file /var/lib/mongo-cluster/shards/shard_member.conf. Note that replSetName will differ for Shard1 and Shard2.

For Shard1 Members (host2, host4): Use replSetName: shard1

processManagement:
   fork: true
   pidFilePath: /var/lib/mongo-cluster/shards/shard1/log/mongod.pid
net:
   bindIp: 0.0.0.0
   port: 27001
   serviceExecutor: adaptive
storage:
   dbPath: /var/lib/mongo-cluster/shards/shard1/data
   journal:
      enabled: true
      commitIntervalMs: 200
replication:
   replSetName: shard1
sharding:
   clusterRole: shardsvr
systemLog:
   destination: file
   path: "/var/lib/mongo-cluster/shards/shard1/log/shard1.log"
   logAppend: true
   timeStampFormat: ctime

For Shard2 Members (host3, host5): Use replSetName: shard2

processManagement:
   fork: true
   pidFilePath: /var/lib/mongo-cluster/shards/shard2/log/mongod.pid
net:
   bindIp: 0.0.0.0
   port: 27001
   serviceExecutor: adaptive
storage:
   dbPath: /var/lib/mongo-cluster/shards/shard2/data
   journal:
      enabled: true
      commitIntervalMs: 200
replication:
   replSetName: shard2
sharding:
   clusterRole: shardsvr
systemLog:
   destination: file
   path: "/var/lib/mongo-cluster/shards/shard2/log/shard2.log"
   logAppend: true
   timeStampFormat: ctime

Create data and log directories for shard members:

• On host2 and host4:

sudo mkdir -p /var/lib/mongo-cluster/shards/shard1/data
sudo mkdir -p /var/lib/mongo-cluster/shards/shard1/log
sudo chown -R mongod:mongod /var/lib/mongo-cluster/shards/shard1

57. On host3 and host5:

sudo mkdir -p /var/lib/mongo-cluster/shards/shard2/data
sudo mkdir -p /var/lib/mongo-cluster/shards/shard2/log
sudo chown -R mongod:mongod /var/lib/mongo-cluster/shards/shard2

Arbiter Node Configuration (host1)

On host1, create two arbiter configuration files:

For Shard1 Arbiter: /var/lib/mongo-cluster/arbiters/arbiter1/arbiter.conf

processManagement:
   fork: true
   pidFilePath: /var/lib/mongo-cluster/arbiters/arbiter1/log/arbiter.pid
net:
   bindIp: 0.0.0.0
   port: 27026
   serviceExecutor: adaptive
storage:
   dbPath: /var/lib/mongo-cluster/arbiters/arbiter1/data
   journal:
      enabled: true
      commitIntervalMs: 200
replication:
   replSetName: shard1
sharding:
   clusterRole: shardsvr
systemLog:
   destination: file
   path: "/var/lib/mongo-cluster/arbiters/arbiter1/log/arbiter.log"
   logAppend: true
   timeStampFormat: ctime

For Shard2 Arbiter: /var/lib/mongo-cluster/arbiters/arbiter2/arbiter.conf

processManagement:
   fork: true
   pidFilePath: /var/lib/mongo-cluster/arbiters/arbiter2/log/arbiter.pid
net:
   bindIp: 0.0.0.0
   port: 27027
   serviceExecutor: adaptive
storage:
   dbPath: /var/lib/mongo-cluster/arbiters/arbiter2/data
   journal:
      enabled: true
      commitIntervalMs: 200
replication:
   replSetName: shard2
sharding:
   clusterRole: shardsvr
systemLog:
   destination: file
   path: "/var/lib/mongo-cluster/arbiters/arbiter2/log/arbiter.log"
   logAppend: true
   timeStampFormat: ctime

Create data and log directories for arbiters on host1:

sudo mkdir -p /var/lib/mongo-cluster/arbiters/arbiter1/data
sudo mkdir -p /var/lib/mongo-cluster/arbiters/arbiter1/log
sudo mkdir -p /var/lib/mongo-cluster/arbiters/arbiter2/data
sudo mkdir -p /var/lib/mongo-cluster/arbiters/arbiter2/log
sudo chown -R mongod:mongod /var/lib/mongo-cluster/arbiters

Start Shard Members and Arbiters

• Start shard members on host2, host3, host4, host5 (adjust config file path based on shard):

numactl --interleave=all mongod -f /var/lib/mongo-cluster/shards/shard1/shard_member.conf # For host2, host4
numactl --interleave=all mongod -f /var/lib/mongo-cluster/shards/shard2/shard_member.conf # For host3, host5

70. Start arbiters on host1:

numactl --interleave=all mongod -f /var/lib/mongo-cluster/arbiters/arbiter1/arbiter.conf
numactl --interleave=all mongod -f /var/lib/mongo-cluster/arbiters/arbiter2/arbiter.conf

Initialize Shard Replica Sets

Connect to a primary or secondary of each shard replica set (e.g., host4:27001 for Shard1, host5:27001 for Shard2) and initialize:

mongo --port 27001

For Shard1 (e.g., connected to host4:27001):

rs.initiate(
  {
    _id: "shard1",
    members: [
      { _id: 0, host: "host4:27001", priority: 2 }, // Primary
      { _id: 1, host: "host2:27001", priority: 1 }, // Secondary
      { _id: 2, host: "host1:27026", arbiterOnly: true } // Arbiter
    ]
  }
)

For Shard2 (e.g., connected to host5:27001):

rs.initiate(
  {
    _id: "shard2",
    members: [
      { _id: 0, host: "host5:27001", priority: 2 }, // Primary
      { _id: 1, host: "host3:27001", priority: 1 }, // Secondary
      { _id: 2, host: "host1:27027", arbiterOnly: true } // Arbiter
    ]
  }
)

Verify each replica set status with rs.status().

3. Mongos Query Router Setup (host2, host3, host4, host5)

On each of host2, host3, host4, and host5, create the mongos configuration file /var/lib/mongo-cluster/mongos/mongos.conf:

processManagement:
   fork: true
   pidFilePath: /var/lib/mongo-cluster/mongos/log/mongos.pid
net:
   bindIp: 0.0.0.0
   port: 20000
   serviceExecutor: adaptive
sharding:
   configDB: cfgReplSet/host2:27000,host3:27000,host4:27000
systemLog:
   destination: file
   path: "/var/lib/mongo-cluster/mongos/log/mongos.log"
   logAppend: true
   timeStampFormat: ctime

Create log directories for mongos instances on these hosts:

sudo mkdir -p /var/lib/mongo-cluster/mongos/log
sudo chown -R mongod:mongod /var/lib/mongo-cluster/mongos

Start the mongos processes on host2, host3, host4, and host5:

numactl --interleave=all mongos -f /var/lib/mongo-cluster/mongos/mongos.conf

4. Configure the Sharded Cluster

Connect to any mongos instance (e.g., host2:20000) to configure the cluster:

mongo --port 20000

Add the shard replica sets to the cluster:

sh.addShard("shard1/host4:27001,host2:27001,host1:27026")
sh.addShard("shard2/host5:27001,host3:27001,host1:27027")

Verify the cluster status:

sh.status()

Ensure the balancer is running:

sh.getBalancerState() // Check state
sh.startBalancer()    // Start if not running

Securing the Cluster with Keyfile Authentication

Keyfile authentication is crucial for inter-component communication within the sharded cluster. Each mongod and mongos instance uses the keyfile as a shared password.

1. Create an Administrator User

Connect to a mongos instance and create an administrative user with root privileges:

mongo --port 20000

use admin
db.createUser(
  {
    user: "clusterAdmin",
    pwd: "yourStrongPassword",
    roles: [ { role: "root", db: "admin" } ]
  }
)

2. Generate a Keyfile

On each of host1 through host5, generate a random keyfile and set appropriate permissions (read-only for the owner):

sudo openssl rand -base64 756 > /var/lib/mongo-cluster/mongodb-keyfile
sudo chmod 600 /var/lib/mongo-cluster/mongodb-keyfile
sudo chown mongod:mongod /var/lib/mongo-cluster/mongodb-keyfile

Ensure this keyfile is identical across all hosts that run mongod or mongos processes.

3. Update Configuration Files

Edit the configuration files for all Config Servers, Shard Members, Arbiters, and Mongos instances to enable authentication:

security:
   authorization: enabled
   keyFile: /var/lib/mongo-cluster/mongodb-keyfile

Add these lines under the security section of the respective .conf files (/var/lib/mongo-cluster/config/mongodb.conf, /var/lib/mongo-cluster/shards/shard1/shard_member.conf, /var/lib/mongo-cluster/shards/shard2/shard_member.conf, /var/lib/mongo-cluster/arbiters/arbiter1/arbiter.conf, /var/lib/mongo-cluster/arbiters/arbiter2/arbiter.conf, /var/lib/mongo-cluster/mongos/mongos.conf).

4. Restart the Cluster with Authentication

Stop the Cluster

First, stop the balancer via a mongos instance:

mongo --port 20000 --authenticationDatabase "admin" -u "clusterAdmin" -p "yourStrongPassword"
sh.stopBalancer()

Then, shut down all mongos, then shard members (secondary first, then primary), and finally config servers:

# On each mongos host (host2, host3, host4, host5)
mongo --port 20000 --authenticationDatabase "admin" -u "clusterAdmin" -p "yourStrongPassword"
use admin
db.shutdownServer()
exit

# On each shard member (host1, host2, host3, host4, host5), for secondary/arbiter first
# For shard members:
mongo --port 27001 --authenticationDatabase "admin" -u "clusterAdmin" -p "yourStrongPassword"
db.getSiblingDB("admin").shutdownServer()
# For arbiters:
mongo --port 27026 --authenticationDatabase "admin" -u "clusterAdmin" -p "yourStrongPassword"
db.getSiblingDB("admin").shutdownServer()
mongo --port 27027 --authenticationDatabase "admin" -u "clusterAdmin" -p "yourStrongPassword"
db.getSiblingDB("admin").shutdownServer()

# On each config server (host2, host3, host4), secondary first
mongo --port 27000 --authenticationDatabase "admin" -u "clusterAdmin" -p "yourStrongPassword"
db.getSiblingDB("admin").shutdownServer()

Start the Cluster (with authentication enabled)

Start the components in the following order, ensuring to use numactl:

1. Config Servers: On host2, host3, host4:

numactl --interleave=all mongod -f /var/lib/mongo-cluster/config/mongodb.conf

3. Shard Members & Arbiters:

   • On host2, host4: Shard1 members
   • On host3, host5: Shard2 members
   • On host1: Arbiters for Shard1 and Shard2

   numactl --interleave=all mongod -f /var/lib/mongo-cluster/shards/shard1/shard_member.conf
   numactl --interleave=all mongod -f /var/lib/mongo-cluster/shards/shard2/shard_member.conf
   numactl --interleave=all mongod -f /var/lib/mongo-cluster/arbiters/arbiter1/arbiter.conf
   numactl --interleave=all mongod -f /var/lib/mongo-cluster/arbiters/arbiter2/arbiter.conf
   
   

4. Mongos Routers: On host2, host3, host4, host5:

numactl --interleave=all mongos -f /var/lib/mongo-cluster/mongos/mongos.conf

6. Re-enable Balancer: Connect to any mongos instance using the admin credentials:

mongo --port 20000 --authenticationDatabase "admin" -u "clusterAdmin" -p "yourStrongPassword"
sh.startBalancer()
sh.getBalancerState() // Verify balancer status

5. Connection Test with Authentication

Test connections to various cluster components using the created admin user:

mongo --port 20000 --authenticationDatabase "admin" -u "clusterAdmin" -p "yourStrongPassword"
mongo --port 27000 --authenticationDatabase "admin" -u "clusterAdmin" -p "yourStrongPassword"
mongo --port 27001 --authenticationDatabase "admin" -u "clusterAdmin" -p "yourStrongPassword"

Data Flow in a Sharded Cluster

Query Data Flow

1. Client Request: A client sends a query to a mongos router.
2. Metadata Lookup: The mongos queries the Config Servers to determine which shards hold the relevant data, based on the query's shard key.
3. Query Routing: mongos routes the query to the appropriate shard(s). If the query involves data across multiple shards, mongos might send concurrent queries.
4. Shard Execution: Each target mongod on the shard processes its portion of the query. This involves parsing the query, checking indexes, and retrieving data.
5. Index Optimization: If an index covers the query, the mongod can rapidly locate data without scanning entire collections. Data might be in memory (cache) or on disk.
6. Data Retrieval: If data is on disk, the storage engine loads it into memory.
7. Filtering and Processing: Each mongod filters, sorts, and aggregates its local data according to the query criteria.
8. Result Consolidation: Each mongod sends its partial results back to the mongos router.
9. Final Result Assembly: The mongos aggregates the results from all shards, performs any final sorting or merging, and returns the complete result set to the client.

Insert Data Flow

1. Client Request: A client sends an insert operation with document data to a mongos router.
2. Shard Key Determination: mongos extracts the shard key value from the document to be inserted.
3. Chunk Identification: mongos consults the Config Servers to determine which chunk range the shard key falls into and, consequently, which shard owns that chunk.
4. Target Shard Selection: Based on the shard key and chunk distribution, mongos selects the appropriate shard to handle the insert.
5. Data Insertion: mongos forwards the insert request to the primary member of the target shard's replica set.
6. Local Write Operation: The primary mongod on the target shard performs the data insertion locally. If the insertion causes a chunk to grow beyond its maximum size, a chunk split operation might be triggered asynchronously by the balancer.
7. Replication: The primary replicates the write operation to its secondary members.
8. Acknowledgement: Once the write is confirmed (according to the write concern), the target shard's primary sends an acknowledgment back to the mongos.
9. Client Confirmation: mongos forwards the success or failure of the insert operation back to the client.