rename_extract.py
rename_extract.py
#!/usr/bin/env python3
"""Rename a GoldenGate integrated Extract with the REST API.
This automates the extract-side of a rename: it reads the oldest unprocessed
transaction of the old extract, stops it, then creates the new extract
registered at the data dictionary build SCN and started at the oldest
unprocessed SCN, so no transaction is missed.
The data dictionary build is a database operation (DBMS_CAPTURE_ADM.BUILD)
and cannot be done through the GoldenGate REST API. Pass its SCN directly
with --dictionary-scn, or let the script look it up itself with --db-dsn /
--db-user (and --db-password, prompted if omitted), using the oracledb
package when it is installed, or shelling out to sqlplus (whichever is on
PATH) when it is not, so this also runs on a host with no `pip install
oracledb`. If no existing build predates the start SCN, the script builds
a fresh one and waits for the still-running old extract's checkpoint to
move past it before continuing.
Full post:
https://juliendelattre.com/blog/rename-goldengate-extract-rest-api/
adminclient version (manual steps, same logic):
https://juliendelattre.com/blog/rename-goldengate-extract/
Steps:
- Get the SCN of the oldest unprocessed transaction (the start SCN).
- Find the dictionary build to register the new extract (the dictionary
SCN), building a fresh one and waiting for the checkpoint to pass it if
none predates the start SCN yet.
- Stop the extract to rename.
- Create the new extract, registered and started at those two SCNs, reusing
the old extract's parameter file with only the name and trail changed.
Uses oggrestapi.py, the GoldenGate REST client:
https://github.com/juliendlttr/ogg
"""
import argparse
import csv
import getpass
import io
import logging
import re
import shutil
import subprocess
import sys
import time
from oggrestapi import OGGRestAPI
logging.basicConfig(level=logging.INFO, format='%(message)s')
try:
import oracledb
HAVE_ORACLEDB = True
except ImportError:
oracledb = None
HAVE_ORACLEDB = False
DICTIONARY_SCN_SQL = """
SELECT first_change#
FROM v$archived_log
WHERE dictionary_begin = 'YES'
AND standby_dest = 'NO'
AND name IS NOT NULL
AND status = 'A'
AND first_change# < {start_scn}
ORDER BY first_change# DESC
FETCH FIRST 1 ROWS ONLY
"""
def get_oldest_unprocessed_scn(client, extract, connection):
"""Find the SCN of the oldest transaction the extract has not yet processed.
Args:
client: Connected OGGRestAPI client.
extract: Extract name.
connection: Connection name to check for active transactions.
Returns:
The smallest SCN among the extract's recovery checkpoint and any
active transactions on the connection.
"""
# The recovery checkpoint holds the position of the oldest unprocessed
# transaction. We also look at the active transactions and keep the
# smallest SCN, so a long running transaction is never left behind.
checkpoints = client.get_extract_checkpoint(extract)
recovery_scn = checkpoints['current']['input'][0]['recovery']['csn']
candidates = [recovery_scn] if recovery_scn else []
active = client.get_active_transactions(connection)
txn_scns = [t['txnStartScn'] for t in active.get('activeTransactions', [])]
candidates.extend(txn_scns)
if not candidates:
msg = f'Could not determine an oldest unprocessed SCN for {extract}'
raise RuntimeError(msg)
return min(candidates)
def build_config(old_config, new_extract, trail):
"""Build the new extract's parameter file from the old extract's, renaming EXTRACT and EXTTRAIL.
Args:
old_config: Old extract's parameter file lines.
new_extract: New extract's name.
trail: New extract's trail.
Returns:
The new extract's parameter file lines.
"""
# Parameter file of the new extract, copied from the old one: only the
# EXTRACT name and the EXTTRAIL line change, everything else (USERIDALIAS,
# SOURCECATALOG, TABLE, TRANLOGOPTIONS, ...) carries over untouched.
config = []
for line in old_config:
if line.startswith('EXTRACT '):
config.append(f'EXTRACT {new_extract}')
elif line.startswith('EXTTRAIL '):
config.append(f'EXTTRAIL {trail}')
else:
config.append(line)
return config
def split_trail(trail):
"""Split an EXTTRAIL value into the trail name and its path, for the REST API's "targets" field.
Args:
trail: Trail value as used in EXTTRAIL, e.g. "pdb1/bb" or "bb".
Returns:
A (name, path) tuple; path is None when trail has no "/".
"""
# The EXTTRAIL parameter takes a path/name form (e.g. pdb1/bb), but the
# REST API's "targets" field, which is what actually creates the trail on
# disk, wants the two-character name and the path split apart. Without
# "targets", create_extract only writes the EXTTRAIL line into the
# parameter file and never creates the trail itself, so the new extract
# abends on start with OGG-02454 (trail not found in checkpoint file).
if '/' in trail:
path, name = trail.rsplit('/', 1)
return name, path
return trail, None
def pick_driver(requested):
"""Resolve which driver to use for the dictionary SCN database calls.
Args:
requested: "oracledb", "sqlplus", or "auto".
Returns:
The resolved driver name, "oracledb" or "sqlplus".
"""
# "auto" prefers oracledb when it is installed, falling back to sqlplus.
if requested == 'oracledb':
if not HAVE_ORACLEDB:
msg = '--driver oracledb requested but the oracledb package is not installed'
raise SystemExit(msg)
return 'oracledb'
if requested == 'sqlplus':
if not shutil.which('sqlplus'):
msg = '--driver sqlplus requested but sqlplus is not on PATH'
raise SystemExit(msg)
return 'sqlplus'
if HAVE_ORACLEDB:
return 'oracledb'
if shutil.which('sqlplus'):
return 'sqlplus'
msg = 'neither the oracledb package nor sqlplus is available - install one, or pass --dictionary-scn directly'
raise SystemExit(msg)
def get_dictionary_scn_oracledb(dsn, user, password, start_scn):
"""Look up an existing dictionary build SCN before start_scn, via the oracledb driver.
Args:
dsn: Source database DSN.
user: Source database user.
password: Source database password.
start_scn: Only consider a build with a first_change# below this SCN.
Returns:
The build's first_change#, or None if no matching build exists.
"""
sql = DICTIONARY_SCN_SQL.format(start_scn=start_scn)
with oracledb.connect(dsn=dsn, user=user, password=password) as conn, conn.cursor() as cur:
cur.execute(sql)
row = cur.fetchone()
return row[0] if row else None
def get_dictionary_scn_sqlplus(dsn, user, password, start_scn):
"""Look up an existing dictionary build SCN before start_scn, by shelling out to sqlplus.
Args:
dsn: Source database DSN.
user: Source database user.
password: Source database password.
start_scn: Only consider a build with a first_change# below this SCN.
Returns:
The build's first_change#, or None if no matching build exists.
"""
# Unlike oracledb, sqlplus needs a trailing ";" to actually run the
# statement - without it, the query just sits unterminated and EXIT
# closes the session with no output at all.
sql = DICTIONARY_SCN_SQL.format(start_scn=start_scn).rstrip() + ';'
script = (
'WHENEVER SQLERROR EXIT SQL.SQLCODE\n'
'SET MARKUP CSV ON\n'
'SET FEEDBACK OFF\n'
'SET ECHO OFF\n'
'SET VERIFY OFF\n'
'SET HEADING OFF\n'
'SET PAGESIZE 0\n'
'SET TERMOUT OFF\n'
'SET TRIMSPOOL ON\n'
f'CONNECT {user}/{password}@{dsn}\n'
'SET TERMOUT ON\n'
f'{sql}\n'
'EXIT\n'
)
proc = subprocess.run(
['sqlplus', '-s', '/nolog'],
input=script,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
timeout=60,
)
if proc.returncode != 0 or 'ORA-' in proc.stdout:
msg = f'sqlplus dictionary SCN query against {dsn} failed:\n{proc.stdout}\n{proc.stderr}'
raise RuntimeError(msg)
rows = [row for row in csv.reader(io.StringIO(proc.stdout)) if row]
return int(rows[0][0]) if rows else None
def get_dictionary_scn(dsn, user, password, start_scn, driver):
"""Look up an existing dictionary build SCN before start_scn, via the resolved driver.
Args:
dsn: Source database DSN.
user: Source database user.
password: Source database password.
start_scn: Only consider a build with a first_change# below this SCN.
driver: "oracledb", "sqlplus", or "auto".
Returns:
The build's first_change#, or None if no matching build exists.
"""
driver = pick_driver(driver)
if driver == 'oracledb':
return get_dictionary_scn_oracledb(dsn, user, password, start_scn)
return get_dictionary_scn_sqlplus(dsn, user, password, start_scn)
def build_dictionary_oracledb(dsn, user, password):
"""Trigger a fresh data dictionary build via the oracledb driver.
Args:
dsn: Source database DSN.
user: Source database user.
password: Source database password.
Returns:
The new build's first_change# SCN.
"""
with oracledb.connect(dsn=dsn, user=user, password=password) as conn, conn.cursor() as cur:
first_scn = cur.var(int)
cur.execute('BEGIN DBMS_CAPTURE_ADM.BUILD(first_scn => :first_scn); END;', first_scn=first_scn)
return int(first_scn.getvalue())
def build_dictionary_sqlplus(dsn, user, password):
"""Trigger a fresh data dictionary build by shelling out to sqlplus.
Args:
dsn: Source database DSN.
user: Source database user.
password: Source database password.
Returns:
The new build's first_change# SCN.
"""
script = (
'WHENEVER SQLERROR EXIT SQL.SQLCODE\n'
'SET FEEDBACK OFF\n'
'SET ECHO OFF\n'
'SET VERIFY OFF\n'
'SET SERVEROUTPUT ON\n'
'SET TERMOUT OFF\n'
f'CONNECT {user}/{password}@{dsn}\n'
'SET TERMOUT ON\n'
'DECLARE\n'
' scn NUMBER;\n'
'BEGIN\n'
' DBMS_CAPTURE_ADM.BUILD(first_scn => scn);\n'
" DBMS_OUTPUT.PUT_LINE('DICTIONARY_BUILD_SCN:' || scn);\n"
'END;\n'
'/\n'
'EXIT\n'
)
proc = subprocess.run(
['sqlplus', '-s', '/nolog'],
input=script,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=True,
timeout=300,
)
if proc.returncode != 0 or 'ORA-' in proc.stdout:
msg = f'sqlplus dictionary build against {dsn} failed:\n{proc.stdout}\n{proc.stderr}'
raise RuntimeError(msg)
match = re.search(r'DICTIONARY_BUILD_SCN:(\d+)', proc.stdout)
if not match:
msg = f'Could not find the build SCN in sqlplus output:\n{proc.stdout}'
raise RuntimeError(msg)
return int(match.group(1))
def build_dictionary(dsn, user, password, driver):
"""Trigger a fresh data dictionary build via the resolved driver.
Args:
dsn: Source database DSN.
user: Source database user.
password: Source database password.
driver: "oracledb", "sqlplus", or "auto".
Returns:
The new build's first_change# SCN.
"""
driver = pick_driver(driver)
if driver == 'oracledb':
return build_dictionary_oracledb(dsn, user, password)
return build_dictionary_sqlplus(dsn, user, password)
def wait_for_dictionary_scn(client, extract, connection, dictionary_scn, poll_seconds=30, timeout_seconds=600):
"""Poll the extract's checkpoint until it moves past a freshly built dictionary SCN.
Args:
client: Connected OGGRestAPI client.
extract: Extract name.
connection: Connection name to check for active transactions.
dictionary_scn: The dictionary build SCN to wait past.
poll_seconds: Seconds to sleep between checkpoint checks.
timeout_seconds: Give up and raise after this many seconds.
Returns:
The extract's oldest unprocessed SCN, once it exceeds dictionary_scn.
"""
# A fresh dictionary build's SCN is later than the start SCN we already
# computed (it was built after that point), so it cannot register the
# extract yet. Waiting lets the still-running old extract's checkpoint
# advance past it, the same wait the adminclient version needs by hand.
deadline = time.monotonic() + timeout_seconds
while True:
start_scn = get_oldest_unprocessed_scn(client, extract, connection)
if start_scn > dictionary_scn:
return start_scn
if time.monotonic() >= deadline:
msg = (
f"{extract}'s checkpoint ({start_scn}) did not move past the new dictionary "
f'SCN ({dictionary_scn}) within {timeout_seconds}s. Rerun once it has.'
)
raise RuntimeError(msg)
logging.info(
" %s's checkpoint (%s) has not passed the dictionary SCN (%s) yet, waiting...",
extract,
start_scn,
dictionary_scn,
)
time.sleep(poll_seconds)
def get_source_catalog(config):
"""Find the SOURCECATALOG value in a parameter file.
Args:
config: Parameter file lines.
Returns:
The catalog (PDB) name from the SOURCECATALOG line.
Raises:
RuntimeError: No SOURCECATALOG line is present.
"""
for line in config:
if line.startswith('SOURCECATALOG '):
return line.split(None, 1)[1].rstrip(';')
msg = "No SOURCECATALOG line in the old extract's config"
raise RuntimeError(msg)
def rename_extract(client, args):
"""Rename an integrated Extract: stop the old one and create the new one at the right SCNs.
Args:
client: Connected OGGRestAPI client.
args: Parsed command-line arguments.
"""
connection = args.connection
old = args.old_extract
new = args.new_extract
old_extract = client.get_extract(old)
old_config = old_extract['config']
credentials = old_extract['credentials']
catalog = get_source_catalog(old_config)
logging.info('Oldest unprocessed SCN of %s...', old)
start_scn = get_oldest_unprocessed_scn(client, old, connection)
logging.info(' start SCN (oldest unprocessed): %s', start_scn)
# The dictionary lookup (and, if needed, a fresh build plus the wait for
# it) happens before stopping the old extract: a fresh build's SCN is
# later than this start SCN, so registering the new extract at it would
# miss the transactions in between. The old extract has to keep running
# so its checkpoint can move past the new build first.
dictionary_scn = args.dictionary_scn
if dictionary_scn is None:
logging.info('Looking up the dictionary build SCN through %s...', args.driver)
dictionary_scn = get_dictionary_scn(args.db_dsn, args.db_user, args.db_password, start_scn, args.driver)
if dictionary_scn is None:
logging.info('No dictionary build predates start SCN %s, building a fresh one...', start_scn)
dictionary_scn = build_dictionary(args.db_dsn, args.db_user, args.db_password, args.driver)
logging.info(' new dictionary SCN: %s', dictionary_scn)
logging.info("Waiting for %s's checkpoint to move past it...", old)
start_scn = wait_for_dictionary_scn(client, old, connection, dictionary_scn)
logging.info(' start SCN (oldest unprocessed): %s', start_scn)
else:
logging.info(' dictionary SCN: %s', dictionary_scn)
logging.info('Stopping extract %s...', old)
client.stop_extract(old)
# Re-read the checkpoint once stopped, the recovery position is final now.
start_scn = get_oldest_unprocessed_scn(client, old, connection)
logging.info(' start SCN after stop: %s', start_scn)
config = build_config(old_config, new, args.trail)
trail_name, trail_path = split_trail(args.trail)
target = {'name': trail_name}
if trail_path:
target['path'] = trail_path
logging.info('Creating extract %s (register at %s, begin at %s)...', new, dictionary_scn, start_scn)
client.create_extract(
extract=new,
begin={'at': {'csn': start_scn}},
registration={'containers': [catalog], 'csn': dictionary_scn, 'replace': True},
source='tranlogs',
config=config,
credentials=credentials,
description=f'Renamed from {old}',
targets=[target],
)
logging.info('Starting extract %s...', new)
client.start_extract(new)
logging.info('Done.')
def parse_args():
"""Parse and validate command-line arguments, prompting for the DB password if needed.
Returns:
The parsed argument namespace.
"""
parser = argparse.ArgumentParser(description='Rename a GoldenGate integrated Extract via the REST API')
parser.add_argument('--url', required=True, help='OGG REST API URL, e.g. https://vmogg or https://nginx-host')
parser.add_argument('--user', required=True, help='OGG REST API username')
parser.add_argument('--deployment', help='Deployment name (required with --reverse-proxy)')
parser.add_argument('--reverse-proxy', action='store_true', help='Connect through an NGINX reverse proxy')
parser.add_argument('--ca-cert', help='Path to a trusted CA certificate for self-signed setups')
parser.add_argument('--old-extract', required=True, help='Name of the extract to rename')
parser.add_argument('--new-extract', required=True, help='New extract name')
parser.add_argument('--connection', required=True, help='Connection name, e.g. OracleGoldenGate.source_cdb')
parser.add_argument(
'--dictionary-scn',
type=int,
help=(
'SCN of the data dictionary build (DBMS_CAPTURE_ADM.BUILD). If omitted, looked up with --db-dsn/--db-user'
),
)
parser.add_argument(
'--db-dsn',
help=(
'Source database DSN for the dictionary SCN lookup, e.g. localhost:1521/CDB01 '
'(required if --dictionary-scn is omitted)'
),
)
parser.add_argument(
'--db-user',
help=(
'Source database user for the dictionary SCN lookup, e.g. c##oggadmin '
'(required if --dictionary-scn is omitted)'
),
)
parser.add_argument('--db-password', help='Source database password (prompted if omitted and --db-user is set)')
parser.add_argument(
'--driver',
choices=['auto', 'oracledb', 'sqlplus'],
default='auto',
help=('How to run the dictionary SCN lookup: oracledb, sqlplus, or auto (oracledb if installed, else sqlplus)'),
)
parser.add_argument(
'--trail',
required=True,
help="Trail for the new extract, e.g. pdb1/bb (must differ from the old extract's trail)",
)
args = parser.parse_args()
if args.dictionary_scn is None:
if not args.db_dsn or not args.db_user:
parser.error('--dictionary-scn is required, or both --db-dsn and --db-user to look it up')
if not args.db_password:
args.db_password = getpass.getpass(f'Password for {args.db_user}@{args.db_dsn}: ')
return args
def main():
"""Parse args, connect to the deployment, and run the rename."""
args = parse_args()
if args.reverse_proxy and not args.deployment:
sys.exit('--deployment is required when --reverse-proxy is set')
client = OGGRestAPI(
url=args.url,
username=args.user,
deployment=args.deployment,
reverse_proxy=args.reverse_proxy,
ca_cert=args.ca_cert,
)
rename_extract(client, args)
if __name__ == '__main__':
main() Referenced from: