#!/bin/env python3
"""This script takes care of NGCP database versioning.
It can initialise, upgrade and downgrade an NGCP database
using .up and .down scripts
For more info you can run check the docstring descriptions or
run the script with --help
"""
import argparse
import configparser
import json
import os
import re
import signal
import sys
import tempfile
import textwrap
import time
from io import BufferedRandom
from pathlib import Path
from subprocess import PIPE, Popen
from typing import Any, Dict, List, Optional
import pymysql.cursors
from prettytable import PrettyTable
from pymysql import Connection
Revisions = Dict[str, Dict[int, str]]
DBSchema = Dict[int, List[str]]
ConfigDict = Dict[str, str | int | bool]
class RevisionApplyError(Exception):
"""Exception for revision apply error."""
pass
class Config:
"""Config class.
Contains all the script collected configuration, states, etc.
It's a global variable and available throughout the script but
it is instantiated in main()
Attributes:
db_socket_user (str): database user when connected via socket
db_socket (str): database socket
db_ro_socket (str): database socket of R/O instance
db_defaults_conf_file (str): database defaults conf file to use when
socket is not available
db_connect_method (str): 'socket' or 'config' (default: socket)
db_schema_table (str): database schema table name
node_name_file (str): node name file
roles_file (str): node roles file
db_file (str): node db config file
ngcp_sync_db_script (str): ngcp sync db script
ngcp_version_file: (str): file containing ngcp version string
automated (bool): wether the script is run in the automated mode
that automatically init the db schema if it's empty.
this option is taken from either
AUTOMATED_INSTALL_MODE os.environ
(backward compatibility) or from the args
skip_ro_db_sync (bool): skip db synchronisation (if applicable)
this option is taken from either SKIP_SYNC_DB
os.environ (backward compatibility) or from the args
debug (bool): debug/verbose mode
mode (str): 'up' mode or 'down' mode
to_revision (int): processes revisions depending on the mode:
- mode=up: applies all revisions including to_revision
- mode=down: removes all revisions excluding to_revision
set_release (str): specify release version that is set
to all applied .up scripts instead of fetching it from
/etc/ngcp_version. Only works with mode=up
downgrade_release (str): downgrades the specified release
mutually exclusive with 'mode'
force (bool): run this script on the active node
batch_mode (bool): instead of applying scripts one by one, collects
them all and builds a single sql file that is applied in one go
db_scripts_dir (str): directory containing .up/.down scripts
get_release_info (bool): instead of applying revisionins, it returns
a list of applied releases
as_json (bool): return result from get-release-info as json rather
than a table (and potentially other return data in the future)
ngcp_upgrade (bool): defines if the script is run under 'ngcp-upgrade'
this option automatically uses NGCP_UPGRADE os.environ,
unless provided
supported_args (List[str]): a list of the attributes described above
that can be also passed args, to automate the code
_args (Any): parsed argspare result (usually Namespace())
_subproc (Popen): Popen handler to an opened subprocess, if any
_prompt_is_shown (bool): if the prompt is shown,
a helper to handle SIGINT
_interrupted (bool): if the script was interrupted with SIGINT
_not_replicated_was_applied (bool): if at least one not_replicated
script was applied/removed
_temp_sql_file (BufferedRandom): used to accumulate the sql scripts
data when batch mode is enabled
_run_cmd_mysql_options (List[str]) = extra options that are passed
to every invokation of run_cmd() that calls the 'mysql' console
command. needed to use same connection (socket/config/user, etc.)
as the pymysql db connection
_scripts_dir_order (List[str]): order to apply scripts
from the subdirectories
_post_set_release_for_revisions (List[int]): set release in the
db_schema table for all revisions that were applied when
the `release` column was not available
node_name (str): current node name
node_roles (ConfigDict): current node roles
node_dbconf (ConfigDict): current node db config
node_state (str): current node state (active/inactive,etc.)
ngcp_version (str): current NGCP version
db_conn (Connection): database connection handler
db_schema (DBSchema): database schema revisions state
revisions (Revisions): revisions that are read from
the scripts directory
sync_db_databases (str): a string containing a space separated list
of databases that ngcp-sync-db will synchronise
sync_db_ignore_tables (str): a string containing a space separated list
of tables that ngcp-sync-db should ignore
"""
# db options
db_socket_user: str = 'root'
db_socket: str = '/run/mysqld/mysqld.sock'
db_ro_socket: str = '/run/mysqld/mysqld2.sock'
db_defaults_conf_file: str = '/etc/mysql/sipwise_extra.cnf'
db_connect_method: str = 'socket'
# required config files location options
db_schema_table: str = 'ngcp.db_schema'
node_name_file: str = '/etc/ngcp_nodename'
roles_file: str = '/etc/default/ngcp-roles'
db_file: str = '/etc/default/ngcp-db'
ngcp_sync_db_script: str = '/usr/sbin/ngcp-sync-db'
ngcp_version_file: str = '/etc/ngcp_version'
# default config options
automated: bool = False
skip_ro_db_sync: bool
force_ro_db_sync: bool
debug: bool = False
mode: str = 'up'
to_revision: int
set_release: str
downgrade_release: str
force: bool = False
batch_mode: bool = False
db_scripts_dir: str = '/usr/share/ngcp-db-schema/db_scripts'
get_release_info: bool = False
as_json: bool = False
ngcp_upgrade: bool = False
supported_args: List[str] = ['automated', 'batch_mode',
'db_socket', 'db_connect_method',
'db_defaults_conf_file',
'db_scripts_dir',
'debug', 'force',
'mode',
'skip_ro_db_sync',
'force_ro_db_sync',
'to_revision',
'downgrade_release',
'set_release',
'get_release_info',
'as_json',
'ngcp_upgrade']
# internal attributes
_args: Any = None
_subproc: Optional[Popen[Any]] = None
_prompt_is_shown: bool = False
_interrupted: bool = False
_script_was_applied: bool = False
_not_replicated_was_applied: bool = False
_temp_sql_file: BufferedRandom = None # type: ignore
_run_cmd_mysql_options: List[str] = []
_scripts_dir_order: List[str] = ['init', 'base', 'diff']
_post_set_release_for_revisions: List[int] = []
# attributes that are initialised in __init__()
node_name: str
node_roles: ConfigDict
node_dbconf: ConfigDict
node_state: str
ngcp_version: str
db_conn: Connection # type: ignore
db_schema: DBSchema
revisions: Revisions
# ngcp sync db related options
sync_db_databases = 'ngcp billing carrier kamailio freeswitch ' + \
'provisioning prosody'
sync_db_ignore_tables = 'kamailio.voicemail_spool ' + \
'provisioning.autoprov_firmwares_data ' + \
'provisioning.voip_fax_data ' + \
'billing.journals'
def __init__(self) -> None:
"""Config class constructor.
Takes over SIGINT handling
"""
def handle_signal(signum: int, _: Any) -> None:
if self._prompt_is_shown:
log('aborted.')
shutdown(0)
if self._subproc:
log('waiting for the subprocess to finish...')
self._subproc.wait()
self._interrupted = True
signal.signal(signal.SIGINT, handle_signal)
def setup_config(self) -> None:
"""Prepares info that is used by the script."""
self.node_name = get_node_name()
self.node_state = get_node_state()
self.node_roles = get_node_roles()
self.node_dbconf = get_node_dbconf()
self.ngcp_version = get_ngcp_version()
def setup_db(self) -> None:
"""Prepares database related setup."""
self.db_conn = connect_db()
self.db_schema = get_db_schema()
class MyRawTextHelpFormatter(argparse.RawDescriptionHelpFormatter):
"""Class that provides with better argparse help formatting."""
def __add_whitespace(self, idx: int, w_space: int, text: str) -> Any:
"""Adds whitespace formatting."""
if idx == 0:
return text
return (' ' * w_space) + text
def _split_lines(self, text: Any, width: int) -> Any:
"""Handles wrapping and bullet points formatting."""
text = text.splitlines()
for idx, line in enumerate(text):
if not re.match(r'^\s*$', line):
line = line.strip() + '\n'
search = re.search(r'\s*[0-9\-]{0,}\.?\s*', line)
if line.strip() == '':
text[idx] = ''
elif search:
l_space = search.end()
lines = [self.__add_whitespace(i, l_space, x)
for i, x in enumerate(textwrap.wrap(line, width))]
text[idx] = lines
return [item for sublist in text for item in sublist]
"""Instantiated in main()."""
config: Config
def _logit(l_str: str, indent: int, end: str, s_char: str = '->') -> None:
"""Formats and prints the log string.
Used by public functions such as log(), debug(), error()
Args:
l_str (str): log string
indent (int): indentation level
end (str): line ending string
s_char: line start string
Returns:
None
"""
print('%s%s %s' % (' ' * 4 * indent, s_char, l_str), end=end)
def log(l_str: str, indent: int = 0, end: str = '\n') -> None:
"""Prints the log string.
Public function used in the code to always print a log string
Args:
l_str (str): log string
indent (int): indentation level
end (str): line ending string
Returns:
None
"""
_logit(l_str, indent, end)
def debug(l_str: str, indent: int = 0, end: str = '\n') -> None:
"""Print the debug string.
Public function used in the code to print debug info,
only if debug is enabled
Args:
l_str (str): log string
indent (int): indentation level
end (str): line ending string
Returns:
None
"""
if config.debug:
_logit(l_str, indent, end, '=>')
def error(l_str: str, indent: int = 0, end: str = '\n') -> None:
"""Print the error string.
Public function used in the code to print error
Args:
l_str (str): log string
indent (int): indentation level
end (str): line ending string
Returns:
None
"""
_logit(f'Error: {l_str}', indent, end, '!>')
def str_to_bool(in_str: str) -> bool:
"""Converts a string into its boolean representation.
'true', 'yes', '1' - boolean strings
Args:
in_str (str): input string
Returns:
bool: True if the string has boolean representation,
otherwise False
"""
return in_str.lower() in ('true', 'yes', '1')
def c_str(in_str: str) -> str:
"""Makes a string compact by trimming excess whitespace.
Args:
in_str (str): input string
Returns:
str: compact string
"""
out_str = in_str.strip()
return re.sub(r'\s{2,}|\n', ' ', out_str, flags=re.MULTILINE)
def connect_db() -> Connection: # type: ignore
"""Connects to the database to perform versioning on.
Args:
None
Returns:
Connection: database connection handler
"""
db_host = config.node_dbconf['pair_dbhost']
db_port = int(config.node_dbconf['pair_dbport'])
db_user = config.db_socket_user
db_socket = config.db_socket
db_defaults_conf_file = config.db_defaults_conf_file
db_connect_method = config.db_connect_method
socket_available = os.access(str(db_socket), os.R_OK)
if db_connect_method == 'socket' and not socket_available:
debug(c_str(f"""
socket file {db_socket} is not readable,
using {db_defaults_conf_file}
"""))
db_connect_method == 'config'
if db_connect_method == 'socket':
config._run_cmd_mysql_options = [
'-h', str(db_host), '-P', str(db_port),
'-u', db_user, '-S', db_socket
]
try:
db_conn: Connection = pymysql.connect( # type: ignore
host=db_host,
port=db_port,
user=db_user,
unix_socket=db_socket,
autocommit=0,
)
debug(
'connected to database %s:%d via socket %s as %s' %
(db_host, db_port, db_socket, db_user)
)
except pymysql.Error as e:
error(
'could not connect to database %s:%d via socket %s as %s: %s' %
(db_host, db_port, db_socket, db_user, e)
)
shutdown(1)
elif db_connect_method == 'config':
config._run_cmd_mysql_options = [
f'--defaults-extra-file={db_defaults_conf_file}',
'-h', str(db_host), '-P', str(db_port),
]
try:
with open(db_defaults_conf_file) as file:
for line in file.readlines():
m = re.match(r'^user\s+=\s+(\S+)', line)
if m:
db_user = m.group(1)
break
except FileNotFoundError as e:
error(f'cannot access {db_defaults_conf_file}: {e}')
shutdown(1)
try:
db_conn: Connection = pymysql.connect( # type: ignore
host=db_host,
port=db_port,
user=db_user,
read_default_file=db_defaults_conf_file,
autocommit=0,
)
log(
'connected to database %s:%d via config %s as %s' %
(db_host, db_port, db_defaults_conf_file, db_user)
)
except pymysql.Error as e:
error(
'could not connect to database %s:%d via config %s as %s: %s' %
(db_host, db_port, db_defaults_conf_file, db_user, e)
)
shutdown(1)
else:
error('unknown database connection scenario')
shutdown(1)
db_conn.query('SET SESSION wait_timeout = 600;')
return db_conn
def parse_args() -> None:
"""Parses command line arguments.
Stores them in config.args upon success as well as
Update the corresponding config attributes provided in
config.supported_args
Args:
None
Returns:
None
"""
parser = argparse.ArgumentParser(
description=c_str("""
NGCP database schema versioning tool that upgrades
and downgrades the database schema state.
"""),
formatter_class=MyRawTextHelpFormatter,
allow_abbrev=False,
argument_default=argparse.SUPPRESS,
)
ro_db_sync_group = parser.add_mutually_exclusive_group()
downgrade_group = parser.add_mutually_exclusive_group()
# Define arguments
parser.add_argument(
'--verbose', '-v',
dest='debug',
action='store_true',
help="""
enable verbose output (debug mode).
""")
parser.add_argument(
'--force', '-f',
dest='force',
action='store_true',
help="""
force the database schema to be applied even if the node is inactive.
""")
downgrade_group.add_argument(
'--mode', '-m',
dest='mode',
type=str,
help="""
schema upgrade mode:
- 'up' (default): upgrades the database
- 'down': downgrades the database.
mutually exclusive with --downgrade-release
""")
parser.add_argument(
'--to-revision', '-t',
dest='to_revision',
type=int,
help="""
max revision to upgrade/downgrade the schema depending on the mode:
- mode=up: applies all revisions including --to-revision
- mode=down: removes all revisions excluding --to-revision
""")
parser.add_argument(
'--set-release',
dest='set_release',
type=str,
help="""
set release to a specific version instead of fetching it from
/etc/ngcp_version.
only works with mode=up
""")
downgrade_group.add_argument(
'--downgrade-release',
dest='downgrade_release',
type=str,
help="""
downgrade all applied revisions of the specified release
mutually exclusive with --mode
""")
parser.add_argument(
'--automated', '-a',
dest='automated',
action='store_true',
help="""
automated mode, does not prompt about an empty db_schema and
automatically initialises it if empty '(including the revision).
""")
parser.add_argument(
'--batch-mode', '-b',
dest='batch_mode',
action='store_true',
help="""
(Experimental batch mode)
Instead of applying scripts and sql statements one by one,
they are accumulated and applied all at once.
""")
ro_db_sync_group.add_argument(
'--skip-ro-db-sync',
dest='skip_ro_db_sync',
action='store_true',
help="""
(Carrier only): skip automatic ngcp-sync-db invokation on proxy
nodes if at least one not_replicated scripts was applied.
""")
ro_db_sync_group.add_argument(
'--force-ro-db-sync',
dest='force_ro_db_sync',
action='store_true',
help="""
(Carrier only): force automatic ngcp-sync-db invokation on proxy
nodes even if no not_replicated scripts were applied.
""")
parser.add_argument(
'--db-socket',
dest='db_socket',
type=str,
help=f"""
Database socket file to use instead of the default one
({config.db_socket})
""")
parser.add_argument(
'--db-defaults-conf-file',
dest='db_defaults_conf_file',
type=str,
help=f"""
Database defaults config file to use insted of the default one
({config.db_defaults_conf_file})
""")
parser.add_argument(
'--db-connect-method',
dest='db_connect_method',
choices=['socket', 'config'],
type=str,
help=f"""
Force database connect method instead of the default one
({config.db_connect_method})
""")
parser.add_argument(
'--db-scripts-dir',
dest='db_scripts_dir',
type=str,
help=f"""
Override database .up/.down scripts directory instead of the
default one ({config.db_scripts_dir})
""")
parser.add_argument(
'--get-release-info',
dest='get_release_info',
action='store_true',
help="""
Fetch a list of releases and min/max revision per release.
""")
parser.add_argument(
'--as-json',
dest='as_json',
action='store_true',
help="""
Return data is represented as JSON
(only works for --get-release-info)
""")
parser.add_argument(
'--ngcp-upgrade', '-u',
dest='ngcp_upgrade',
action='store_true',
help="""
ngcp upgrade mode, affects certain aspects of applying up scripts
and sync R/O db
""")
args = parser.parse_args()
for arg in config.supported_args:
if hasattr(args, arg):
config.__dict__[arg] = getattr(args, arg)
elif (arg == 'automated'):
config.automated = str_to_bool(
os.getenv('AUTOMATED_INSTALL_MODE', '')
)
elif (arg == 'skip_ro_db_sync'):
config.skip_ro_db_sync = str_to_bool(
os.getenv('SKIP_SYNC_DB', '')
)
elif (arg == 'db_socket'):
config.db_socket = os.getenv('MYSQL_SOCKET', config.db_socket)
elif (arg == 'ngcp_upgrade'):
config.ngcp_upgrade = str_to_bool(
os.getenv('NGCP_UPGRADE', '')
)
if hasattr(args, 'downgrade_release'):
config.mode = 'down'
if hasattr(config, 'set_release') and config.mode == 'down':
parser.print_help(sys.stderr)
print(
sys.argv[0] +
' ' +
'error: argument --set-release: requires --mode=up',
file=sys.stderr
)
sys.exit(1)
config._args = args
def get_node_name() -> str:
"""Fetches node name.
Args:
None
Returns:
str: node name
"""
with open(config.node_name_file, 'r', encoding='utf8') as file:
return file.readline().strip()
def run_cmd(*cmd: Any, **kwargs: Any) -> tuple[bytes, bytes]:
"""Runs system command.
Args:
cmd (tuple[str]): command and arguments
Returns:
(bytes, bytes): stdout, stderr
"""
stdin = None
if 'stdin' in kwargs:
stdin = kwargs['stdin']
config._subproc = Popen(cmd, stdin=PIPE, stdout=PIPE, stderr=PIPE)
out, err = config._subproc.communicate(input=stdin)
config._subproc = None
return out, err
def read_revision_file(filename: str) -> bytes:
"""Reads revision file.
Currently not used due to issues with the existing scripts
not being correctly SQL syntax compatible and only "mysql console"
compatible
Args:
filename (str): filename to read
Returns:
bytes: sql data
"""
raise NotImplementedError(c_str(
f'{sys._getframe().f_code.co_name}() must not be used'
))
sql: bytes = bytes()
default_delimiter = ';'
delimiter_set = ''
rx_delimiter_declare = re.compile(r'^\s*delimiter\s+(\S+)', re.IGNORECASE)
rx_delimiter_char: re.Pattern[str] = re.Pattern()
rx_delimiter_end: re.Pattern[str] = re.Pattern()
rx_use_stmt_no_end = re.compile(r'^use\s*\S+[^;]\s*$', re.IGNORECASE)
with open(filename, mode='rb') as file:
for b_line in file.readlines():
line = new_line = b_line.decode()
modified = False
m_delim = rx_delimiter_declare.search(line)
if m_delim:
delim = m_delim.group(1)
if delimiter_set:
if delim == default_delimiter:
delimiter_set = ''
elif delimiter_set != delim and \
delimiter_set != default_delimiter:
error(c_str(f"""
Found an unexpected delimiter combo,
active delimiter={delimiter_set},
new delimiter={delim}
"""))
shutdown(1)
else:
delimiter_set = delim
else:
delimiter_set = delim
rx_delimiter_char = re.compile(rf'\s*\{delimiter_set}\s*$')
rx_delimiter_end = re.compile(rf'\w+\s*\{delimiter_set}\s*$')
new_line = rx_delimiter_declare.sub('', line)
if new_line != line:
modified = True
elif delimiter_set:
m_char = rx_delimiter_char.search(line)
if m_char:
m_end = rx_delimiter_end.search(line)
new_line = line.replace(
f'{delimiter_set}',
m_end and f'{default_delimiter}' or '',
1
)
if line != new_line:
modified = True
m_use_no_end = rx_use_stmt_no_end.search(line.strip())
if m_use_no_end:
new_line = f'{line.strip()}{default_delimiter}\n'
if line != new_line:
modified = True
sql += new_line.encode() if modified else b_line
return sql
def get_ngcp_version() -> str:
"""Fetches NGCP version.
Args:
None
Returns
str: the version string
"""
with open(config.ngcp_version_file, 'r', encoding='utf8') as file:
return file.readline().strip()
def get_node_state() -> str:
"""Fetches node state.
state: ['active','standby','unknown','transition']
Args:
None
Returns
str: the state string
"""
stdout, stderr = run_cmd(
'/usr/sbin/ngcp-check-active', '-v', 'root'
)
node_state = 'unknown'
if stderr:
node_state = 'unknown'
else:
node_state = stdout.decode().strip()
return node_state
def get_node_roles() -> ConfigDict:
"""Fetches node roles.
Args:
None
Returns
ConfigDict: config Dict containing node roles
as well as its type
"""
roles_config = configparser.ConfigParser()
default_section = roles_config.default_section
node_roles: ConfigDict = {}
with open(config.roles_file) as file:
roles_config.read_string(
f'[{default_section}]\n{file.read()}'
)
for k, v in roles_config[default_section].items():
m = re.search('ngcp_is_(\\w+)', k)
if m:
val = True if v == '"yes"' else False
node_roles[m.group(1)] = val
elif k == 'ngcp_type':
node_roles[k] = v.replace('"', '')
return node_roles
def get_node_dbconf() -> ConfigDict:
"""Fetches node db config.
Db config represents hosts, ports and other info
Args:
None
Returns
ConfigDict: config Dict containing node db config
"""
db_config = configparser.ConfigParser()
default_section = db_config.default_section
node_dbconf: ConfigDict = {}
with open(config.db_file) as file:
db_config.read_string(
f'[{default_section}]\n{file.read()}'
)
for k, v in db_config[default_section].items():
node_dbconf[k.lower()] = v
return node_dbconf
def get_db_schema() -> DBSchema:
"""Fetches db schema.
Db schema contains already applied revisions
Args:
None
Returns
DbSchema: revisions applied to nodes, the return
value is an empty Dict if the db schema
is not yet initialised
"""
db_schema: DBSchema = {}
cursor: pymysql.cursors.Cursor = config.db_conn.cursor()
cursor.execute("""
SELECT table_schema, table_name
FROM information_schema.tables
WHERE concat(table_schema ,'.', table_name) = %s
""", config.db_schema_table)
if cursor.rowcount == 0:
debug(f'{config.db_schema_table} table does not exist')
return db_schema
debug(f'found existing {config.db_schema_table} table')
if getattr(config, 'downgrade_release', False):
if not check_release_column_exists():
error('Cannot use downgrade release on the old database version')
shutdown(1)
cursor.execute(f"""
SELECT revision, node
FROM {config.db_schema_table}
WHERE `release` = %s
""", config.downgrade_release)
else:
cursor.execute(f"""
SELECT revision, node
FROM {config.db_schema_table}
""")
for row in cursor:
revision = row[0]
node = row[1]
if revision not in db_schema:
db_schema[revision] = []
db_schema[revision].append(node)
cursor.close()
config.db_conn.commit()
if config.mode == 'down' and not db_schema:
if config.downgrade_release:
log(c_str(f"""
could not find entries in db_schema
table matching release '{config.downgrade_release}'
"""))
else:
log('could not find entries in db_schema')
shutdown(0)
return db_schema
def check_release_column_exists() -> bool:
"""Checks existence of ngcp.db_schema release colimn.
Args:
None
Returns
bool: True if exists, False otherwise
"""
cursor: pymysql.cursors.Cursor = config.db_conn.cursor()
cursor.execute("""
SELECT COUNT(column_name)
FROM information_schema.columns
WHERE concat(table_schema ,'.', table_name) = %s
AND column_name = 'release'
""", config.db_schema_table)
column_exists = 0
row = cursor.fetchone()
if row:
column_exists = row[0]
cursor.close()
config.db_conn.commit()
return bool(column_exists)
def get_revisions() -> Revisions:
"""Fetches revisions.
The revisions are read from the scripts directory
Args:
None
Returns
Revision: a Dict structure with revisions
and script names
"""
debug('fetching db scripts...')
mode = config.mode
revisions: Revisions = {}
for spec in config._scripts_dir_order:
rev_dir = f'{config.db_scripts_dir}/{spec}'
if not os.access(rev_dir, os.R_OK):
error(f'non-existing or empty scripts dir {rev_dir}')
shutdown(1)
revisions[spec] = {}
for path in Path(f'{config.db_scripts_dir}/{spec}').glob(f'*.{mode}'):
revision = int(path.stem.split('_')[0])
revisions[spec][revision] = path.name
return revisions
def apply_revisions(revisions: Revisions) -> None:
"""Applies revisions.
Processes through fetched revisions and depending
on the mode calls either apply_up_scipt() or apply_down_script()
If the batch mode is enabled it also invokes run_cmd() with
the accumulated config_temp_sql_file
Args:
revisions (Revisions): revisions structure
Returns
None
"""
mode = config.mode
dir_specs = config._scripts_dir_order
max_schema_revision = 0
if config.db_schema:
del dir_specs[0]
max_schema_revision = max(config.db_schema)
if mode == 'down':
dir_specs.reverse()
if config.batch_mode:
config._temp_sql_file = tempfile.NamedTemporaryFile() # type: ignore
config._temp_sql_file.write(
str(
'SET SESSION wait_timeout = 600;\n' +
'SET SESSION net_read_timeout = 600;\n' +
'SET SESSION net_write_timeout = 600;\n'
).encode()
)
try:
for spec in dir_specs:
sorted_revisions = sorted(revisions[spec].keys())
if mode == 'down':
sorted_revisions.reverse()
for revision in sorted_revisions:
script = revisions[spec][revision]
script_name = f'{spec}/{script}'
ret = 0
if mode == 'up':
ret = apply_up_script(script_name, revision)
else:
# silently skip all revisionns that are greater
# than max revision in the db_schema as there is
# nothing to remove
if max_schema_revision and revision > max_schema_revision:
continue
ret = apply_down_script(script_name, revision)
if not ret:
break
if config._interrupted:
log('stopped.')
return
else:
continue
break
if config.batch_mode and config._temp_sql_file.tell():
debug(f'using accumulated sql {config._temp_sql_file.name}')
log('applying schema changes')
config._temp_sql_file.flush()
_, stderr = run_cmd(
'/usr/bin/mysql', *config._run_cmd_mysql_options,
'-e', f'source {config._temp_sql_file.name}'
)
if stderr:
error(f'could not apply schema changes: {stderr.decode()}')
shutdown(1)
config.db_conn.commit()
config._script_was_applied = True
post_set_release_for_applied_revisions()
except RevisionApplyError:
post_set_release_for_applied_revisions()
shutdown(1)
if not config._script_was_applied:
log('nothing to apply')
def apply_up_script(script_name: str, revision: int) -> int:
"""Applies .up script.
Performs various checks to determins if the .up script
should be applied or skipped.
The script is skipped if:
- it is 'replicated' and already present in db_schema[revision]
- it is 'not_replicated' and already present in
db_schema[revision][node_name]
The script and further scripts are stopped if:
- config.to_revision > 0 and the current revision number
is greater than config.to_revision
Calls run_cmd() to apply the script file using 'mysql' consoel
If batch mode is enabled, instead of calling run_cmd() and insert
into the db_schema table applied revisions, all the data is accumulated
into config._temp_sql_file, to be then applied all at once in
apply_revisions() all at once
Upon success, insert a new entry into ngcp.db_schema
Args:
script_name (str): script name (excluding the base dir)
revision (int): revision number
Returns
int: 1 - if successfully applied or skipped,
0 - if should be stopped and future revisions will not
be applied
"""
node_name = config.node_name
db_schema = config.db_schema
not_replicated = 'not_replicated' in script_name
node_in_db_schema = False
if revision in db_schema:
node_in_db_schema = node_name in db_schema[revision]
to_revision = getattr(config, 'to_revision', 0)
if to_revision > 0 and revision > config.to_revision:
log(f'stopping as next revision {revision} ' +
f'is greater than {config.to_revision}')
return 0
if revision in db_schema:
if not not_replicated or node_in_db_schema:
debug(f'skip already applied revision {script_name}')
return 1
script_filename = f'{config.db_scripts_dir}/{script_name}'
if not config.batch_mode:
log(f'applying revision {script_name}')
_, stderr = run_cmd(
'/usr/bin/mysql', *config._run_cmd_mysql_options,
'-e', f'source {script_filename}'
)
if stderr:
error(f'could not apply revision: {stderr.decode()}')
shutdown(1)
else:
log(f'adding revision {script_name}')
with open(script_filename, 'rb') as file:
config._temp_sql_file.write(file.read())
try:
config.db_conn.ping(reconnect=True)
if not config.batch_mode:
with config.db_conn.cursor() as cursor:
cursor.execute(f"""
INSERT INTO {config.db_schema_table}
(revision, node) VALUES (%s, %s)
""", (revision, node_name))
config.db_conn.commit()
config._script_was_applied = True
else:
sql = c_str(f"""
INSERT INTO {config.db_schema_table}
(revision, node)
VALUES
({revision}, '{node_name}')
""")
config._temp_sql_file.write(
str(f'\n; {sql}; COMMIT;\n').encode()
)
config._post_set_release_for_revisions.append(revision)
if revision in db_schema:
db_schema[revision].append(node_name)
else:
db_schema[revision] = [node_name]
if not_replicated:
config._not_replicated_was_applied = True
except pymysql.Error as e:
error(f'could not apply revision: {e}')
raise RevisionApplyError(e)
return 1
def apply_down_script(script_name: str, revision: int) -> int:
"""Applies .down script.
Performs various checks to determins if the .down script
should be applied or skipped.
The script is skipped if:
- it is 'replicated' and not present in db_schema[revision]
- it is 'not_replicaed' and not present in
db_schema[revision][node_name]
The script and further scripts are stopped if:
- config.to_revision > 0 and to_revision <= config.to_revision
Calls run_cmd() to apply the script file using 'mysql' consoel
Upon success, removes the corresponding entry from ngcp.db_schema
If batch mode is enabled, instead of calling run_cmd() and insert
into the db_schema table applied revisions, all the data is accumulated
into config._temp_sql_file, to be then applied all at once in
apply_revisions() all at once
Note: down script 15800 is automatically skipped as it contains
addition of the 'release' table, which is crucial for downgrades.
Args:
script_name (str): script name (excluding the base dir)
revision (int): revision number
Returns
int: 1 - if successfully applied or skipped,
0 - if should be stopped and future revisions will not
be applied
"""
node_name = config.node_name
db_schema = config.db_schema
not_replicated = 'not_replicated' in script_name
node_in_db_schema = False
if revision in db_schema:
node_in_db_schema = node_name in db_schema[revision]
to_revision = getattr(config, 'to_revision', 0)
if to_revision > 0 and revision <= config.to_revision:
log(f'stopping as next revision {revision} ' +
f'is lesser or equal to {config.to_revision}')
return 0
if revision == 15800:
log(f'skip non-removable revision {script_name}')
return 1
if revision not in db_schema or \
not_replicated and not node_in_db_schema:
debug(f'skip already removed revision {script_name}')
return 1
script_filename = f'{config.db_scripts_dir}/{script_name}'
if not config.batch_mode:
log(f'Removing revision {script_name}')
_, stderr = run_cmd(
'/usr/bin/mysql', *config._run_cmd_mysql_options,
'-e', f'source {script_filename}'
)
if stderr:
error(
f'could not remove revision {script_name}: {stderr.decode()}'
)
shutdown(1)
else:
log(f'adding revision {script_name}')
with open(script_filename, 'rb') as file:
config._temp_sql_file.write(file.read())
try:
config.db_conn.ping(reconnect=True)
if not config.batch_mode:
with config.db_conn.cursor() as cursor:
rows = 0
if not_replicated:
rows = cursor.execute(f"""
DELETE FROM {config.db_schema_table}
WHERE revision = %s
AND node = %s
""", (revision, node_name))
else:
rows = cursor.execute(f"""
DELETE FROM {config.db_schema_table}
WHERE revision = %s
""", revision)
if rows <= 0:
raise ValueError(script_name)
config.db_conn.commit()
config._script_was_applied = True
else:
sql = c_str(f"""
DELETE FROM {config.db_schema_table}
WHERE revision = {revision}
""")
if not_replicated:
sql += f" AND node = '{node_name}'"
config._temp_sql_file.write(
str(f'\n; {sql}; COMMIT;\n').encode()
)
if revision in db_schema:
if not_replicated:
db_schema[revision].remove(node_name)
else:
del db_schema[revision]
if not_replicated:
config._not_replicated_was_applied = True
except pymysql.Error as e:
error(f'could not apply revision {script_name}: {e}')
raise RevisionApplyError(e)
except ValueError as e:
error(f'could not remove revision {e}')
raise RevisionApplyError(e)
return 1
def post_set_release_for_applied_revisions() -> None:
"""Sets release column for applied revisions.
Sets release column for applied revisions
Args:
None
Returns
None
"""
if config.mode != 'up':
return
if not config._post_set_release_for_revisions:
return
if not check_release_column_exists():
return
release = getattr(config, 'set_release', config.ngcp_version)
cursor: pymysql.cursors.Cursor = config.db_conn.cursor()
cursor.execute(f"""
UPDATE {config.db_schema_table}
SET `release` = '{release}',
applied_at = applied_at
WHERE revision IN ({
','.join(str(x) for x in config._post_set_release_for_revisions)
})
""")
cursor.close()
config.db_conn.commit()
def sync_ro_db() -> None:
"""Synchronises R/O database.
This function is called after successful database upgrade/downgrade
operations with the .up/.down scripts
It calls ngcp-sync-db script with specific options to re-establish
master-slave replication from the central db* nodes
It is only called on:
- carrier platform
- proxy nodes
Args:
None
Returns
None
"""
ngcp_type = config.node_roles['ngcp_type']
if ngcp_type != 'carrier':
debug('skip sync R/O database on non "carrier" nodes')
return
if 'proxy' not in config.node_roles or config.node_roles['proxy'] is False:
debug(c_str("""
skip R/O database sync because this node
does not have the 'proxy' role.
"""))
return
if getattr(config, 'skip_ro_db_sync', False):
debug(c_str("""
skip sync R/O database preparation as
'skip_ro_db_sync' option is provided
"""))
return
if getattr(config, 'force_ro_db_sync', False):
debug(c_str("""
force sync R/O database preparation as
'force_ro_db_sync' option is provided
"""))
elif config._not_replicated_was_applied:
debug(c_str("""
at least one not_replicated script was applied,
preparing to sync R/O database.'
"""))
else:
debug(c_str("""
skip sync R/O database preparation as
'not_replicated' scripts were not applied
"""))
return
log('preparing RO db sync...')
sync_dbs_opt = config.sync_db_databases
ignore_tables_opt = config.sync_db_ignore_tables
db_ro_socket = config.db_ro_socket
db_defaults_conf_file = config.db_defaults_conf_file
extra_sync_opt = []
if config.automated:
extra_sync_opt += ['--set-local-grants']
db_user = ''
db_pass = ''
try:
with open(db_defaults_conf_file) as file:
for line in file.readlines():
m = re.match(r'^user\s+=\s+(\S+)', line)
if m:
db_user = m.group(1)
m = re.match(r'^password\s+=\s+(\S+)', line)
if m:
db_pass = m.group(1)
except FileNotFoundError as e:
error(f'cannot access {db_defaults_conf_file}: {e}')
shutdown(1)
socket_available = os.access(str(db_ro_socket), os.R_OK)
if not socket_available:
error(f'socket {db_ro_socket} is not available')
shutdown(1)
if not (db_user and db_pass):
error(f'could not fetch credentials from {db_defaults_conf_file}')
shutdown(1)
debug(f'set privileges for user={db_user} socket={db_ro_socket}')
try:
run_cmd_mysql_options = ['-u', 'root', '-S', db_ro_socket]
_, stderr = run_cmd(
'/usr/bin/mysql', *run_cmd_mysql_options,
'-e', c_str(f"""
grant all privileges
on *.*
to '{db_user}'@'localhost' identified by '{db_pass}'
with grant option;
flush privileges;
""")
)
if stderr:
raise ChildProcessError(stderr.decode())
except ChildProcessError as e:
error(f'could not set privileges for user={db_user}: {e}')
shutdown(1)
master_node_pref = ''
if config.ngcp_upgrade:
central_dbhost = config.node_dbconf['central_dbhost']
if central_dbhost == 'db01':
node_side = ''
is_a_side = re.search(r'\d{2}a$', config.node_name)
is_b_side = re.search(r'\d{2}b$', config.node_name)
if is_a_side or config.node_name == 'sp1':
node_side = 'a'
elif is_b_side or config.node_name == 'sp2':
node_side = 'b'
else:
error('could not determine valid node side during upgrade')
shutdown(1)
master_node_pref = f'{central_dbhost}{node_side}'
elif central_dbhost in ['db01a', 'db01b']:
master_node_pref = str(central_dbhost)
if master_node_pref:
extra_sync_opt += ['--master-host', master_node_pref]
else:
error(c_str("""
could not determine valid master db node during upgrade
expected: db01, db01a, db01b
"""))
shutdown(1)
local_dbhost = config.node_dbconf['local_dbhost']
local_dbport = config.node_dbconf['local_dbport']
cmd: List[str | int] = [config.ngcp_sync_db_script,
'--verbose',
'--force',
'--use-central-db',
'--repl-mode', 'master-slave',
'--databases', sync_dbs_opt,
'--ignore-tables', ignore_tables_opt,
'--ssh-tunnel', '33125',
*extra_sync_opt,
'--local-host', local_dbhost,
'--local-port', local_dbport]
try:
debug(f'run_cmd as: {cmd}')
log('running ngcp-sync-db...')
stdout, stderr = run_cmd(*tuple(cmd))
log(stdout.decode())
if stderr:
raise ChildProcessError(stderr.decode())
except ChildProcessError as e:
error(f'could not complete ngcp-sync-db: {e}')
shutdown(1)
def check_active_node() -> None:
"""Checks if the current node is active.
Stops the script if the current node is active and the --force
argument is not provided
This function always succeeds on CE platforms
Args:
None
Returns
None
"""
ngcp_type = config.node_roles['ngcp_type']
if ngcp_type == 'spce':
return
node_state = get_node_state()
if node_state != 'standby':
if not config.force:
log(c_str(f"""
stopping because node {config.node_name} state is {node_state}
"""))
shutdown(1)
else:
debug(c_str(
f"""current node {config.node_name} state is {node_state}
but force mode is enabled."""
))
def check_and_show_prompt() -> None:
"""Checks and shows the db init agreement prompt.
If the db_schema is not yet initialised and --automated mode
is not set, a prompt is shown that expects 'agree' confirmation
from the user, otherwise the script is aborted
Args:
None
Returns
None
"""
if config.mode == 'down' and not config.db_schema:
shutdown(1)
if not config.db_schema and not config.automated:
prompt = """
=================================================================
Warning: the db_schema table of the ngcp database is empty.
Are you sure you want to proceed with applying db-schema changes ?
This will DROP and then re-initialize your existing database.
Please type 'agree' and press Enter to really continue: """
prompt = prompt.strip()
prompt = '\n'.join([line.strip() for line in prompt.splitlines()])
prompt += ' '
config._prompt_is_shown = True
response = input(prompt)
if response != 'agree':
log('Aborted.')
shutdown(0)
def show_release_info() -> None:
"""Shows releases info.
Args:
None
Returns
None
"""
if not check_release_column_exists():
error('Cannot use get release info on the old database version')
shutdown(1)
cursor: pymysql.cursors.Cursor = config.db_conn.cursor()
cursor.execute(f"""
SELECT `release`, min(revision), max(revision), COUNT(revision)
FROM {config.db_schema_table}
GROUP BY 1
""")
if config.as_json:
data: Dict[str, Dict[str, int]] = {}
for row in cursor:
release = row[0]
min_rev = row[1]
max_rev = row[2]
count = row[3]
data[release] = {
'min_revision': min_rev,
'max_revision': max_rev,
'count': count
}
print(json.dumps(data, sort_keys=True))
else:
table = PrettyTable(['release',
'min_revision', 'max_revision',
'count'])
table.align = 'r'
for row in cursor:
table.add_row(list(row))
print(table)
def show_start_summary() -> None:
"""Print options that the script is invoked with.
Args:
None
Returns
None
"""
log(f'{os.path.basename(sys.argv[0])} started with the following options:')
log('-----------------------------------')
for arg in config.supported_args:
if hasattr(config, arg):
v = getattr(config, arg)
log(f'{arg}: {v}')
log('-----------------------------------')
log('node_name: ' + config.node_name)
log('node_state: ' + config.node_state)
log('ngcp_version: ' + config.ngcp_version)
log('-----------------------------------')
def shutdown(retcode: int) -> None:
"""Called when the script needs to be manually stopped.
Args:
ret (int): return code, 0 - ok
1 - error
Returns
None
"""
sys.exit(retcode)
def main() -> None:
"""Main entry function.
Args:
None
Returns
None
"""
try:
started_at = time.time()
global config
config = Config()
parse_args()
config.setup_config()
if config.get_release_info:
config.setup_db()
show_release_info()
else:
show_start_summary()
config.setup_db()
check_active_node()
check_and_show_prompt()
revisions = get_revisions()
apply_revisions(revisions)
sync_ro_db()
log('done in %.2fs.' % (time.time()-started_at))
shutdown(0)
except Exception as err: # noqa: B902
error(str(err))
shutdown(1)
if __name__ == '__main__':
main()