Merge pull request #14 from mjsqu/feat/add_metadata_to_record_count_m…

…essages feat: Add tags to record_count messages to identify table
s7clarke10 · Jun 6, 2023 · affc49a · affc49a
2 parents 3c27f01 + 41e5643
commit affc49a
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 1 deletion.
diff --git a/tap_oracle/sync_strategies/full_table.py b/tap_oracle/sync_strategies/full_table.py
@@ -58,6 +58,10 @@ def sync_view(conn_config, stream, state, desired_columns):
       singer.write_message(activate_version_message)
 
    with metrics.record_counter(None) as counter:
+
+      counter.tags["schema"] = escaped_schema
+      counter.tags["table"] = escaped_table
+
       select_sql      = 'SELECT {} FROM {}.{}'.format(','.join(escaped_columns),
                                                       escaped_schema,
                                                       escaped_table)
@@ -121,6 +125,10 @@ def sync_table(conn_config, stream, state, desired_columns):
       singer.write_message(activate_version_message)
 
    with metrics.record_counter(None) as counter:
+
+      counter.tags["schema"] = escaped_schema
+      counter.tags["table"] = escaped_table
+
       ora_rowscn = singer.get_bookmark(state, stream.tap_stream_id, 'ORA_ROWSCN')
       if not USE_ORA_ROWSCN:
          # Warning there is not restart recovery if the ORA_ROWSCN is ignored.

diff --git a/tap_oracle/sync_strategies/incremental.py b/tap_oracle/sync_strategies/incremental.py
@@ -68,6 +68,10 @@ def sync_table(conn_config, stream, state, desired_columns):
       typed_offset_value = f"INTERVAL '{OFFSET_VALUE}' SECOND"
 
    with metrics.record_counter(None) as counter:
+
+      counter.tags["schema"] = escaped_schema
+      counter.tags["table"] = escaped_table
+
       if replication_key_value:
          LOGGER.info(f"Resuming Incremental replication from {replication_key} = {replication_key_value} + {typed_offset_value}")
          casted_where_clause_arg = common.prepare_where_clause_arg(replication_key_value, replication_key_sql_datatype)

diff --git a/tap_oracle/sync_strategies/log_miner.py b/tap_oracle/sync_strategies/log_miner.py
@@ -208,7 +208,11 @@ def sync_tables_logminer(cur, streams, state, start_scn, end_scn):
       rows_saved = 0
       columns_for_record = desired_columns + ['scn', '_sdc_deleted_at']
       with metrics.record_counter(None) as counter:
-         LOGGER.info("Examing log for table %s", stream.tap_stream_id)
+
+         counter.tags["schema"] = schema_name
+         counter.tags["table"] = stream.table
+
+         LOGGER.info("Examining log for table %s", stream.tap_stream_id)
          common.send_schema_message(stream, ['lsn'])
          LOGGER.info("mine_sql=%s", mine_sql)
          for op, redo, scn, cscn, commit_ts, *col_vals in cur.execute(mine_sql, binds):