Merge pull request #1 from singer-io/master

bringing in updates from the source

Author: bbaltagi-dtsl

.circleci/config.yml

@@ -13,7 +13,6 @@ jobs:
         command: [mongod, --replSet, rs0]
     steps:
       - checkout
-      - add_ssh_keys
       - run:
           name: 'Install Dockerize'
           command: wget https://github.com/jwilder/dockerize/releases/download/$DOCKERIZE_VERSION/dockerize-linux-amd64-$DOCKERIZE_VERSION.tar.gz && tar -C /usr/local/bin -xzvf dockerize-linux-amd64-$DOCKERIZE_VERSION.tar.gz && rm dockerize-linux-amd64-$DOCKERIZE_VERSION.tar.gz
@@ -37,7 +36,6 @@ jobs:
       - run:
           name: 'Setup virtual env'
           command: |
-            aws s3 cp s3://com-stitchdata-dev-deployment-assets/environments/tap-tester/sandbox tap-tester.env
             pyenv local 3.5.6
             python3 -mvenv /usr/local/share/virtualenvs/tap-mongodb
             source /usr/local/share/virtualenvs/tap-mongodb/bin/activate
@@ -52,32 +50,41 @@ jobs:
           name: "Unit Tests"
           command: |
             source /usr/local/share/virtualenvs/tap-mongodb/bin/activate
-            pip install pymongo==3.8.0
+            pip install pymongo==3.12.3
             nosetests tests/unittests/
       - run:
           name: 'Integration Tests'
           command: |
             source tap-tester.env
+            mkdir /tmp/${CIRCLE_PROJECT_REPONAME}
+            export STITCH_CONFIG_DIR=/tmp/${CIRCLE_PROJECT_REPONAME}
             source /usr/local/share/virtualenvs/tap-tester/bin/activate
-            pip install pymongo==3.8.0
+            pip install pymongo==3.12.3
             run-test --tap=tap-mongodb tests
+      - run:
+          name: 'Get Curl'
+          command: |
+            apt update
+            apt install -y curl
       - slack/notify-on-failure:
           only_for_branches: master
+      - store_artifacts:
+          path: /tmp/tap-mongodb
 
 workflows:
   version: 2
-  commit:
+  commit: &commit_jobs
     jobs:
       - build:
-          context: circleci-user
+          context:
+            - circleci-user
+            - tier-1-tap-user
   build_daily:
+    <<: *commit_jobs
     triggers:
       - schedule:
-          cron: "0 6 * * *"
+          cron: "0 1 * * *"
           filters:
             branches:
               only:
                 - master
-    jobs:
-      - build:
-          context: circleci-user

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changelog
 
+## 2.1.2
+  * Update pymongo to v3.12.3 [#81](https://github.com/singer-io/tap-mongodb/pull/81)
+
+## 2.1.1
+  * Fix bug in oplog bookmarking where the bookmark would not advance due to fencepost querying finding a single record [#80](https://github.com/singer-io/tap-mongodb/pull/80)
+
+## 2.1.0
+  * Optimize oplog extractions to only query for the selected tables [#78](https://github.com/singer-io/tap-mongodb/pull/78)
+
 ## 2.0.1
   * Modify `get_databases` function to return a unique list of databases [#58](https://github.com/singer-io/tap-mongodb/pull/58)
 
@@ -78,7 +87,7 @@
 
 ## 0.1.0
  * Added key-based incremental sync [commit](https://github.com/singer-io/tap-mongodb/commit/b618b11d91e111680f70b402c6e94c9bf40c7b8f)
- 
+
 ## 0.0.5
  * Fixed bug in oplog projections [commit](https://github.com/singer-io/tap-mongodb/commit/b400836678440499d4a15fb7d5b0a40a13e3342e)
 

bin/test-db

@@ -34,7 +34,7 @@ def start_container(name):
 
     ip_addr = get_ip_addr(name)
     CONFIGURE_COMMAND = """
-    docker run --rm mongo mongo --host {} \test -u {} -p {} --authenticationDatabase admin --eval {}
+    docker run --rm mongo mongo --host {} test -u {} -p {} --authenticationDatabase admin --eval {}
     """.format(ip_addr,
                os.getenv('TAP_MONGODB_USER'),
                os.getenv('TAP_MONGODB_PASSWORD'),

setup.py

@@ -3,15 +3,15 @@
 from setuptools import setup
 
 setup(name='tap-mongodb',
-      version='2.0.1',
+      version='2.1.2',
       description='Singer.io tap for extracting data from MongoDB',
       author='Stitch',
       url='https://singer.io',
       classifiers=['Programming Language :: Python :: 3 :: Only'],
       py_modules=['tap_mongodb'],
       install_requires=[
           'singer-python==5.8.0',
-          'pymongo==3.8.0',
+          'pymongo==3.12.3',
           'tzlocal==2.0.0',
           'terminaltables==3.1.0',
       ],

tap_mongodb/__init__.py

@@ -146,6 +146,8 @@ def sync_stream(client, stream, state):
                 LOGGER.info("Clearing state because Oplog has aged out")
                 state.get('bookmarks', {}).pop(tap_stream_id)
 
+            collection_oplog_ts = oplog.get_latest_ts(client)
+
             # make sure initial full table sync has been completed
             if not singer.get_bookmark(state, tap_stream_id, 'initial_full_table_complete'):
                 msg = 'Must complete full table sync before starting oplog replication for %s'
@@ -154,12 +156,11 @@ def sync_stream(client, stream, state):
                 # only mark current ts in oplog on first sync so tap has a
                 # starting point after the full table sync
                 if singer.get_bookmark(state, tap_stream_id, 'version') is None:
-                    collection_oplog_ts = oplog.get_latest_ts(client)
                     oplog.update_bookmarks(state, tap_stream_id, collection_oplog_ts)
 
                 full_table.sync_collection(client, stream, state, stream_projection)
 
-            oplog.sync_collection(client, stream, state, stream_projection)
+            oplog.sync_collection(client, stream, state, stream_projection, collection_oplog_ts)
 
         elif replication_method == 'FULL_TABLE':
             full_table.sync_collection(client, stream, state, stream_projection)

tap_mongodb/sync_strategies/full_table.py

@@ -157,4 +157,4 @@ def sync_collection(client, stream, state, projection):
 
     singer.write_message(activate_version_message)
 
-    LOGGER.info('Syncd {} records for {}'.format(rows_saved, tap_stream_id))
+    LOGGER.info('Synced {} records for {}'.format(rows_saved, tap_stream_id))

tap_mongodb/sync_strategies/incremental.py

@@ -127,4 +127,4 @@ def sync_collection(client, stream, state, projection):
 
     singer.write_message(activate_version_message)
 
-    LOGGER.info('Syncd %s records for %s', rows_saved, tap_stream_id)
+    LOGGER.info('Synced %s records for %s', rows_saved, tap_stream_id)

tap_mongodb/sync_strategies/oplog.py

@@ -1,5 +1,4 @@
 #!/usr/bin/env python3
-import copy
 import time
 
 import pymongo
@@ -105,7 +104,7 @@ def flush_buffer(client, update_buffer, stream_projection, db_name, collection_n
 
 
 # pylint: disable=too-many-locals, too-many-branches, too-many-statements
-def sync_collection(client, stream, state, stream_projection):
+def sync_collection(client, stream, state, stream_projection, max_oplog_ts=None):
     tap_stream_id = stream['tap_stream_id']
     LOGGER.info('Starting oplog sync for %s', tap_stream_id)
 
@@ -130,7 +129,8 @@ def sync_collection(client, stream, state, stream_projection):
     start_time = time.time()
 
     oplog_query = {
-        'ts': {'$gte': oplog_ts}
+        'ts': {'$gte': oplog_ts},
+        'ns': {'$eq' : '{}.{}'.format(database_name, collection_name)}
     }
 
     projection = transform_projection(stream_projection)
@@ -160,13 +160,6 @@ def sync_collection(client, stream, state, stream_projection):
                 raise common.MongoAssertionException(
                     "Mongo is not honoring the sort ascending param")
 
-            if row.get('ns') != '{}.{}'.format(database_name, collection_name):
-                if row.get('ts'):
-                    state = update_bookmarks(state,
-                                             tap_stream_id,
-                                             row['ts'])
-                continue
-
             row_op = row['op']
 
             if row_op == 'i':
@@ -240,7 +233,7 @@ def sync_collection(client, stream, state, stream_projection):
                 update_buffer = set()
 
                 # write state
-                singer.write_message(singer.StateMessage(value=copy.deepcopy(state)))
+                singer.write_message(singer.StateMessage(value=state))
 
         # flush buffer if finished with oplog
         for buffered_row in flush_buffer(client,
@@ -257,6 +250,18 @@ def sync_collection(client, stream, state, stream_projection):
             singer.write_message(record_message)
             rows_saved += 1
 
+
+    # Compare the current bookmark with the max_oplog_ts and write the max
+    bookmarked_ts = timestamp.Timestamp(state.get('bookmarks', {}).get(tap_stream_id, {}).get('oplog_ts_time'),
+                                        state.get('bookmarks', {}).get(tap_stream_id, {}).get('oplog_ts_inc'))
+
+    actual_max_ts = max(bookmarked_ts, max_oplog_ts)
+
+    state = update_bookmarks(state,
+                             tap_stream_id,
+                             actual_max_ts)
+    singer.write_message(singer.StateMessage(value=state))
+
     common.COUNTS[tap_stream_id] += rows_saved
     common.TIMES[tap_stream_id] += time.time()-start_time
-    LOGGER.info('Syncd %s records for %s', rows_saved, tap_stream_id)
+    LOGGER.info('Synced %s records for %s', rows_saved, tap_stream_id)

tests/mongodb_common.py

@@ -1,3 +1,30 @@
+import os
+import pymongo
+
+
+def ensure_environment_variables_set():
+    missing_envs = [x for x in ['TAP_MONGODB_HOST',
+                                'TAP_MONGODB_USER',
+                                'TAP_MONGODB_PASSWORD',
+                                'TAP_MONGODB_PORT',
+                                'TAP_MONGODB_DBNAME'] if os.getenv(x) is None]
+    if len(missing_envs) != 0:
+        raise Exception(f"set environment variables: {missing_envs}")
+
+##########################################################################
+### Database Interactions
+##########################################################################
+
+def get_test_connection():
+    username = os.getenv('TAP_MONGODB_USER')
+    password = os.getenv('TAP_MONGODB_PASSWORD')
+    host= os.getenv('TAP_MONGODB_HOST')
+    auth_source = os.getenv('TAP_MONGODB_DBNAME')
+    port = os.getenv('TAP_MONGODB_PORT')
+    ssl = False
+    conn = pymongo.MongoClient(host=host, username=username, password=password, port=27017, authSource=auth_source, ssl=ssl)
+    return conn
+
 def drop_all_collections(client):
     ############# Drop all dbs/collections #############
     for db_name in client.list_database_names():