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
# Not a real error, but 'ORA-' still appears in stdout on CONNECT - would otherwise be misread as a
# failed query, or (for CSV output) parsed as a spurious extra row.
BENIGN_SQLPLUS_WARNING_RE = re.compile(r'(?m)^ORA-28002:.*\n?')
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):
"""Return the SCN of the oldest transaction the extract has not yet fully processed.
The recovery checkpoint holds the position of the oldest unprocessed transaction.
Also looks at the active transactions and keeps 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 one, with only the name and trail changed.
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 path/name (e.g. "pdb1/bb") into its two-character name and path.
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 "auto"/"oracledb"/"sqlplus" to the driver actually available.
"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 the latest dictionary build SCN predating start_scn, via the oracledb driver."""
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 the latest dictionary build SCN predating start_scn, by shelling out to sqlplus.
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'
'SET LINESIZE 32767\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,
)
stdout = BENIGN_SQLPLUS_WARNING_RE.sub('', proc.stdout)
if proc.returncode != 0 or 'ORA-' in 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(stdout)) if row]
return int(rows[0][0]) if rows else None
def get_dictionary_scn(dsn, user, password, start_scn, driver):
"""Look up the dictionary build SCN using whichever driver pick_driver() resolves to."""
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):
"""Run DBMS_CAPTURE_ADM.BUILD via the oracledb driver and return the new dictionary's 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):
"""Run DBMS_CAPTURE_ADM.BUILD by shelling out to sqlplus and return the new dictionary's 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 BENIGN_SQLPLUS_WARNING_RE.sub('', 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):
"""Build a fresh data dictionary and return its SCN, using whichever driver pick_driver() resolves to."""
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 until the extract's checkpoint moves past a freshly built dictionary's SCN.
A fresh dictionary build's SCN is later than the start SCN 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):
"""Return the SOURCECATALOG value from an extract's parameter file lines."""
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):
"""Stop the old extract and create the renamed replacement registered at the right SCNs."""
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 CLI args, prompting for the DB password if a lookup is needed and none was given."""
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():
"""Connect to the OGG REST API 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: