MongoDB

MongoDBReader reads data from the Replica Set Oplog, so to use it you must be running a replica set (see Deploy a Replica Set or Convert a Standalone to a Replica Set, or your MongoDB cloud provider's documentation) on each shard to be read. For more information, see Replication and Replica Set Data Synchronization.

For all versions of MongoDB, the user specified in MongoDBReader's Username property must have read access to the config database (see MongoDB Manual > Config Database

For MongoDB Atlas, for both InitialLoad and Incremental mode, the user specified in MongoDBReader's Username property must have the readAnyDatabase role (see Built-in Roles).

For other versions of MongoDB, In InitialLoad mode, the user specified in MongoDBReader's Username property must have read access to all databases containing the specified collections..

In Incremental mode, for MongoDB 4.2 and later, the user specified in MongoDBReader's Username property must have changeStream and find privileges on all the collections of the cluster. You may want to create a role with these two privileges (see MongoDB Manual > User-Defined Roles).

In Incremental mode, for MongoDB 4.1 and earlier, the user specified in MongoDBReader's Username property must have read access to the local database and the oplog.rs collection. The oplog is a capped collection, which means that the oldest data is automatically removed to keep it within the specified size. To support recovery, the oplog must be large enough to retain all data that may need to be recovered. See Oplog Size and Change Oplog Size for more information.

To support recovery (see Recovering applications), for all versions of MongoDB, the replica set's oplog must be large enough to retain all the events generated while Striim is offline.

If you have an on-premise MongoDB deployment with your own certificate authority, set the Security Config properties as discussed below.

To set up SSL in MongoDB, see MongoDB Manual > Configure mongod and mongos for TLS/SSL.

To set up Kerberos in MongoDB, see MongoDB Manual > Kerberos Authentication.

To set up X.509 in MongoDB, see MongoDB Manual > Use x.509 Certificates to Authenticate Clients.

To secure your authentication parameters, store the entire Security Config string in a vault (see use Using vaults). For example, assuming your Kerberos realm is MYREALM.COM, its Key Distribution Center (KDC) is kerberos.realm.com, the path to the SSL trust store is /cacerts, the path to the SSL keystore file is /client.pkcs12, and the password for both stores is MyPassword, the Striim console commands to store the Security Config string with the key SSLKerberos in a vault named MongoDBVault would be:

CREATE VAULT MongoDBVault;
WRITE INTO MongoDBVault (
  vaultKey: "SSLKerberos",
  vaultValue: "RealmName:MYREALM.COM;
    KDC:kerberos.myrealm.com;
    KeyStore:/keystore.pkcs12;
    TrustStore:/cacerts;
    trustStorePassword:MyPassword;
    KeyStorePassword:MyPassword"
);

Enter READ ALL FROM MongoDBVault; to verify the contents.

In TQL or the Flow Designer, you would then specify the Security Config as [[MongoDBVault.SSLKerberos]].

The following are examples for each authentication option.

Kerberos authentication

Without SSL:

CREATE VAULT MongoDBVault;
WRITE INTO MongoDBVault (
  vaultKey: "Kerberos",
  vaultValue : "RealmName:MYREALM.COM;
  KDC:kdc.myrealm.com"
);

With SSL:

CREATE VAULT MongoDBVault;
WRITE INTO MongoDBVault (
  vaultKey: "KerberosSSL",
  vaultValue : "RealmName:MYREALM.COM;
  KDC:kdc.myrealm.com;
  KeyStore:UploadedFiles/keystore.ks;
  KeyStorePassword:MyPassword;
  TrustStore:Platform/UploadedFiles/truststore.ks;
  TrustStorePassword:MyPassword"
);

If required, also specify:

SSL

If you are using MongoDB on premise with SSL, use Security Config to configure key store and trust stores.

CREATE VAULT MongoDBVault;
WRITE INTO MongoDBVault (
  vaultKey: "SSL",
  vaultValue : "KeyStore:UploadedFiles/keystore.ks;
  TrustStore:UploadedFiles/truststore.ks;
  TrustStorePassword:MyPassword;
  KeyStorePassword:MyPassword"
);

If required, also specify:

X.509 authentication

CREATE VAULT MongoDBVault;
WRITE INTO MongoDBVault (
  vaultKey: "X509",
  vaultValue : "KeyStore:UploadedFiles/keystore.ks;
  KeyStorePassword:MyPassword"
);

If required, also specify:

Striim provides support for connecting to MongoDB Atlas clusters using private endpoints or network peering.

There are no Striim-specific configuration steps required; the configuration steps you need to perform are in MongoDB Atlas.

See the following MongoDB Atlas doc topics:

The MongoDB driver is bundled with Striim, so no installation is necessary.

The adapter properties are:

You can use the MongoDB Config property to define a subset of documents to be selected.

{
  "QueryClause": {
    "<database name>.<collection name>": {
      "Filter": {
        "<field name>": "<field value>"
      }
    }
  }
}

{
  "QueryClause": {
    "MyDB.MyCollection": {
      "Filter": {
        "city": "Bangalore"
      }
    }
  }
}

{
  "QueryClause": {
    "MyDB.MyCollection": {
      "Filter": {
        "OR": [
          {
            "operator": {
              "city": "Bangalore"
            }
          },
          {
            "operator": {
              "city": "Bangalore"
            }
          }
        ]
      }
    }
  }
}

{
  "QueryClause": {
    "mongodb.employee": {
      "Filter": {
        "City": "Chennai"
      }
    },
    "mongodb.dep$": {
      "Filter": {
        "and": [
          {
            "!=": {
              "Name": "Accounts"
            }
          },
          {
            "State": "Tamil Nadu"
          }
        ]
      }
    },
    "mongodb.payroll": {
      "Filter": {
        "or": [
          {
            "!=": {
              "Source": "F"
            }
          },
          {
            "and": [
              {
                ">": {
                  "Age": 30
                }
              },
              {
                "City": "Bangalore"
              }
            ]
          }
        ]
      }
    }
  }
}

The output type for MongoDBReader is JSONNodeEvent. The fields are:

data: contains the field names and values of a document, for example:

data: {"_id":2441,"company":"Striim","city":"Palo Alto"}

Updates include only the modified values. Deletes include only the document ID.

removedfields: contains the names of any fields deleted by the $unset function. If no fields were deleted, the value of removedfields is null. For example:

removedfields: {"myField":true}

Or if no fields have been removed:

removedfields: null

metadata: contains the following elements:

For example:

metadata: {"CollectionName":"employee","OperationName":"SELECT","DatabaseName":"test",
  "DocumentKey":1.0,"NameSpace":"test.employee","TimeStamp":1537433999609}

The following Striim application will write change data for the specified collection to SysOut. To run this yourself, replace striim and ****** with the user name and password for the MongoDB user account discussed in MongoDB setup, specify the correct connection URL for your instance, and replace mydb with the name of your database.

CREATE APPLICATION MongoDBTest;

CREATE SOURCE MongoDBIn USING MongoDBReader (
  Username:'striim',
  Password:'******',
  ConnectionURL:'192.168.1.10:27107',
  Collections:'mydb.employee'
) 
OUTPUT TO MongoDBStream;

CREATE TARGET MongoDBCOut
USING SysOut(name:MongoDB)
INPUT FROM MongoDBStream;

END APPLICATION MongoDBTest;

With the above application running, the following MongoDB shell commands:

use mydb;
db.employee.insertOne({_id:1,"firstname":"Larry","lastname":"Talbot","age":10,"comment":"new record"});
db.employee.updateOne({_id:1},{$set:{ "age":40, "comment":"partial update"}});
db.employee.deleteOne({_id:1});

would produce output similar to the following:

data: {"_id":1,"firstname":"Larry","lastname":"Talbot","age":10,"comment":"new record"}
metadata: {"CollectionName":"employee","OperationName":"INSERT","DatabaseName":"mydb","DocumentKey":1,
"NameSpace":"mydb.employee","TimeStamp":1537250474, "Partitioned":false,"FullDocumentReceived":true,
"PartitionKeys":{}}
...
data: {"_id":1.0,"age":40,"comment":"partial update"}
metadata: {"CollectionName":"employee","OperationName":"UPDATE","DatabaseName":"mydb","DocumentKey":1,"
NameSpace":"mydb.employee","TimeStamp":1537250474, "Partitioned":false,"FullDocumentReceived":false,
"PartitionKeys":{}}
...
data: {"_id":1}
metadata: {"CollectionName":"employee","OperationName":"DELETE","DatabaseName":"mydb","DocumentKey":1,
"NameSpace":"mydb.employee","TimeStamp":1537250477, "Partitioned":false,"FullDocumentReceived":false,
"PartitionKeys":{}}

Note that output for the "partial" update and delete operations includes only the fields specified in the shell commands. See Replicating MongoDB data to Azure CosmosDB for discussion of the issues this can cause when writing to targets and how to work around those issues.

To replicate one or many MongoDB collections to Cosmos DB, specify multiple collections in the Collections properties of MongoDBReader and CosmosDBWriter. You may use wildcards ($ for MongoDB, % for Cosmos DB) to replicate all collections in a database, as in the example below, or specify multiple collections manually, as described in the notes for Cosmos DB Writer's Collections property.

You must create the target collections in Cosmos DB manually. The partition key names must match one of the fields in the MongoDB documents.

Data will be read only from collections that exist when the source starts. Additional collections added later will be ignored until the source is restarted. When the target collection is in a fixed container (see Partition and scale in Azure Cosmos DB), inserts, updates, and deletes are handled automatically. When the target collection is in an unlimited container, updates require special handling and deletes must be done manually, as discussed below.

If you wish to run the examples, adjust the MongoDB Reader properties and Cosmos DB Writer properties to reflect your own environment.

When the target collection is in an unlimited container

When a Cosmos DB collection is in an unlimited container, it must have a partition key, which must be specified when you create the collection.

• When MongoDB save operations create new documents, all fields are included in MongoDBReader's output, so CosmosDBWriter can write to the correct partition.

• When MongoDB save operations update existing documents, all fields are included in MongoDBReader's output, so CosmosDBWriter can use the partition key and document ID to update the correct target document.

• MongoDB update operations do not include all fields, so the partition key may be missing from MongoDBReader's output. In those cases, the PartialRecordPolicy open processor retrieves the missing fields from MongoDB and adds them before passing the data to CosmosDBWriter.

• MongoDB remove operations include only the document ID, so the partition key is missing from MongoDBReader's output. Since CosmosDBWriter would be unable to determine the correct partition, the application writes the database name, collection name, and document key to a DeleteOps collection in CosmosDB.

CREATE APPLICATION Mongo2CosmosIncrementalUnlimitedContainer; 
 
CREATE SOURCE MongoDBIn USING MongoDBReader (
  Username:'striim',
  Password:'******',
  ConnectionURL:'<MongoDB connection string>',
  authType: 'NoAuth',
  Mode:'Incremental',
  startTimestamp: '<timestamp>',
  Collections:'mydb.$'
 )
OUTPUT TO MongoDBStream;

CREATE STREAM FilteredMongoDBStream OF Global.JsonNodeEvent;

CREATE CQ ExcludeDeleteOperations
INSERT INTO FilteredMongoDBStream
SELECT META(MongoDBStream,"DatabaseName"),
  META(MongoDBStream,"CollectionName"),
  META(MongoDBStream,"DocumentKey")
FROM MongoDBStream
WHERE META(MongoDBStream,"OperationName").toString() != "DELETE";

CREATE STREAM FullDocstream OF Global.JsonNodeEvent;

CREATE OPEN PROCESSOR CompletePartialDocs USING MongoPartialRecordPolicy ( 
  ConnectionURL:'<MongoDB connection string>', 
  authType:'NoAuth',
  OnMissingDocument: 'Process'
)
INSERT INTO FullDocstream
FROM FilteredMongoDBStream;

CREATE TARGET WriteToCosmos USING CosmosDBWriter (
  ServiceEndpoint:'<Cosmos DB connection string>',
  AccessKey:'<Cosmos DB account read-write key>',
  Collections:'mydb.$,mydb.%',
  IgnorableExceptionCode:'PARTITION_KEY_NOT_FOUND'
)
INPUT FROM FullDocstream;

CREATE CQ SelectDeleteOperations
INSERT INTO DeleteOpsStream
SELECT TO_STRING(META(MongoDBStream,"DatabaseName")) AS DatabaseName,
  TO_STRING(META(MongoDBStream,"CollectionName")) AS CollectionName,
  TO_STRING(META(MongoDBStream,"DocumentKey")) AS DocumentKey
FROM MongoDBStream
WHERE META(MongoDBStream,"OperationName").toString() = "DELETE";

CREATE TARGET WriteDeleteOpsToCosmos USING CosmosDBWriter (
  ServiceEndpoint:'<Cosmos DB connection string>',
  AccessKey:'<Cosmos DB account read-write key>',
  Collections:'mydb.DeleteOps'
)
INPUT FROM DeleteOpsStream;
 
END APPLICATION Mongo2CosmosIncrementalUnlimitedContainer;