694 lines
25 KiB
Smarty
694 lines
25 KiB
Smarty
#!/usr/bin/python
|
|
# -*- coding: utf-8 -*-
|
|
# Copyright 2018 The Openstack-Helm Authors.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
import os
|
|
import sys
|
|
import re
|
|
import logging
|
|
import select
|
|
import signal
|
|
import subprocess
|
|
import socket
|
|
import tempfile
|
|
import time
|
|
import threading
|
|
from datetime import datetime, timedelta
|
|
|
|
import configparser
|
|
import iso8601
|
|
import kubernetes.client
|
|
import kubernetes.config
|
|
|
|
# Create logger, console handler and formatter
|
|
logger = logging.getLogger('OpenStack-Helm Mariadb')
|
|
logger.setLevel(logging.INFO)
|
|
ch = logging.StreamHandler()
|
|
ch.setLevel(logging.INFO)
|
|
formatter = logging.Formatter(
|
|
'%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
|
|
|
# Set the formatter and add the handler
|
|
ch.setFormatter(formatter)
|
|
logger.addHandler(ch)
|
|
|
|
# Get the local hostname
|
|
local_hostname = socket.gethostname()
|
|
logger.info("This instance hostname: {0}".format(local_hostname))
|
|
|
|
# Setup k8s client credentials and check api version
|
|
kubernetes.config.load_incluster_config()
|
|
kubernetes_version = kubernetes.client.VersionApi().get_code().git_version
|
|
logger.info("Kubernetes API Version: {0}".format(kubernetes_version))
|
|
k8s_api_instance = kubernetes.client.CoreV1Api()
|
|
|
|
|
|
def check_env_var(env_var):
|
|
"""Check if an env var exists.
|
|
|
|
Keyword arguments:
|
|
env_var -- the env var to check for the existance of
|
|
"""
|
|
if env_var in os.environ:
|
|
return True
|
|
else:
|
|
logger.critical("environment variable \"{0}\" not set".format(env_var))
|
|
sys.exit(1)
|
|
|
|
|
|
# Set some variables from env vars injected into the container
|
|
if check_env_var("STATE_CONFIGMAP"):
|
|
state_configmap_name = os.environ['STATE_CONFIGMAP']
|
|
logger.info("Will use \"{0}\" configmap for cluster state info".format(
|
|
state_configmap_name))
|
|
if check_env_var("POD_NAMESPACE"):
|
|
pod_namespace = os.environ['POD_NAMESPACE']
|
|
if check_env_var("DIRECT_SVC_NAME"):
|
|
direct_svc_name = os.environ['DIRECT_SVC_NAME']
|
|
if check_env_var("MARIADB_REPLICAS"):
|
|
mariadb_replicas = os.environ['MARIADB_REPLICAS']
|
|
if check_env_var("POD_NAME_PREFIX"):
|
|
pod_name_prefix = os.environ['POD_NAME_PREFIX']
|
|
if check_env_var("DISCOVERY_DOMAIN"):
|
|
discovery_domain = os.environ['DISCOVERY_DOMAIN']
|
|
if check_env_var("WSREP_PORT"):
|
|
wsrep_port = os.environ['WSREP_PORT']
|
|
if check_env_var("MYSQL_ROOT_PASSWORD"):
|
|
mysql_root_password = os.environ['MYSQL_ROOT_PASSWORD']
|
|
|
|
# Set some variables for tuneables
|
|
cluster_leader_ttl = 120
|
|
state_configmap_update_period = 10
|
|
default_sleep = 20
|
|
|
|
|
|
def ensure_state_configmap(pod_namespace, configmap_name, configmap_body):
|
|
"""Ensure the state configmap exists.
|
|
|
|
Keyword arguments:
|
|
pod_namespace -- the namespace to house the configmap
|
|
configmap_name -- the configmap name
|
|
configmap_body -- the configmap body
|
|
"""
|
|
try:
|
|
k8s_api_instance.read_namespaced_config_map(
|
|
name=configmap_name, namespace=pod_namespace)
|
|
return True
|
|
except:
|
|
k8s_api_instance.create_namespaced_config_map(
|
|
namespace=pod_namespace, body=configmap_body)
|
|
return False
|
|
|
|
|
|
def run_cmd_with_logging(popenargs,
|
|
logger,
|
|
stdout_log_level=logging.INFO,
|
|
stderr_log_level=logging.INFO,
|
|
**kwargs):
|
|
"""Run subprocesses and stream output to logger."""
|
|
child = subprocess.Popen(
|
|
popenargs, stdout=subprocess.PIPE, stderr=subprocess.PIPE, **kwargs)
|
|
log_level = {
|
|
child.stdout: stdout_log_level,
|
|
child.stderr: stderr_log_level
|
|
}
|
|
|
|
def check_io():
|
|
ready_to_read = select.select([child.stdout, child.stderr], [], [],
|
|
1000)[0]
|
|
for io in ready_to_read:
|
|
line = io.readline()
|
|
logger.log(log_level[io], line[:-1])
|
|
|
|
while child.poll(
|
|
) is None: # keep checking stdout/stderr until the child exits
|
|
check_io()
|
|
check_io() # check again to catch anything after the process exits
|
|
return child.wait()
|
|
|
|
|
|
def stop_mysqld():
|
|
"""Stop mysqld, assuming pid file in default location."""
|
|
logger.info("Shutting down any mysqld instance if required")
|
|
mysqld_pidfile_path = "/var/lib/mysql/{0}.pid".format(local_hostname)
|
|
|
|
def is_pid_running(pid):
|
|
if os.path.isdir('/proc/{0}'.format(pid)):
|
|
return True
|
|
return False
|
|
|
|
def is_pid_mysqld(pid):
|
|
with open('/proc/{0}/comm'.format(pid), "r") as mysqld_pidfile:
|
|
comm = mysqld_pidfile.readlines()[0].rstrip('\n')
|
|
if comm.startswith('mysqld'):
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
if os.path.isfile(mysqld_pidfile_path):
|
|
logger.info(
|
|
"Previous pid file found for mysqld, attempting to shut it down")
|
|
with open(mysqld_pidfile_path, "r") as mysqld_pidfile:
|
|
mysqld_pid = int(mysqld_pidfile.readlines()[0].rstrip('\n'))
|
|
if is_pid_running(mysqld_pid):
|
|
if is_pid_mysqld(mysqld_pid):
|
|
logger.info("pid from pidfile is mysqld")
|
|
os.kill(mysqld_pid, 15)
|
|
pid, status = os.waitpid(mysqld_pid, 0)
|
|
logger.info("Mysqld stopped: pid = {0}, "
|
|
"exit status = {1}".format(pid, status))
|
|
else:
|
|
logger.error(
|
|
"pidfile process is not mysqld, removing pidfile and panic"
|
|
)
|
|
os.remove(mysqld_pidfile_path)
|
|
sys.exit(1)
|
|
else:
|
|
logger.info(
|
|
"Mysqld was not running with pid {0}, going to remove stale "
|
|
"file".format(mysqld_pid))
|
|
os.remove(mysqld_pidfile_path)
|
|
else:
|
|
logger.debug("No previous pid file found for mysqld")
|
|
|
|
|
|
def mysqld_write_cluster_conf(mode='run'):
|
|
"""Write out dynamic cluster config.
|
|
|
|
Keyword arguments:
|
|
mode -- whether we are writing the cluster config for the cluster to 'run'
|
|
or 'bootstrap' (default 'run')
|
|
"""
|
|
logger.info("Setting up cluster config")
|
|
cluster_config = configparser.ConfigParser()
|
|
cluster_config['mysqld'] = {}
|
|
cluster_config_params = cluster_config['mysqld']
|
|
wsrep_cluster_members = []
|
|
for node in range(int(mariadb_replicas)):
|
|
node_hostname = "{0}-{1}".format(pod_name_prefix, node)
|
|
if local_hostname == node_hostname:
|
|
wsrep_node_address = "{0}.{1}:{2}".format(
|
|
node_hostname, discovery_domain, wsrep_port)
|
|
cluster_config_params['wsrep_node_address'] = wsrep_node_address
|
|
wsrep_node_name = "{0}.{1}".format(node_hostname, discovery_domain)
|
|
cluster_config_params['wsrep_node_name'] = wsrep_node_name
|
|
else:
|
|
addr = "{0}.{1}:{2}".format(node_hostname, discovery_domain,
|
|
wsrep_port)
|
|
wsrep_cluster_members.append(addr)
|
|
if wsrep_cluster_members and mode == 'run':
|
|
cluster_config_params['wsrep_cluster_address'] = "gcomm://{0}".format(
|
|
",".join(wsrep_cluster_members))
|
|
else:
|
|
cluster_config_params['wsrep_cluster_address'] = "gcomm://"
|
|
cluster_config_file = '/etc/mysql/conf.d/10-cluster-config.cnf'
|
|
logger.info(
|
|
"Writing out cluster config to: {0}".format(cluster_config_file))
|
|
with open(cluster_config_file, 'w') as configfile:
|
|
cluster_config.write(configfile)
|
|
|
|
|
|
# Function to setup mysqld
|
|
def mysqld_bootstrap():
|
|
"""Boostrap the db if no data found in the 'bootstrap_test_dir'"""
|
|
logger.info("Boostrapping Mariadb")
|
|
mysql_data_dir = '/var/lib/mysql'
|
|
bootstrap_test_dir = "{0}/mysql".format(mysql_data_dir)
|
|
if not os.path.isdir(bootstrap_test_dir):
|
|
stop_mysqld()
|
|
mysqld_write_cluster_conf(mode='bootstrap')
|
|
run_cmd_with_logging([
|
|
'mysql_install_db', '--user=mysql',
|
|
"--datadir={0}".format(mysql_data_dir)
|
|
], logger)
|
|
template = (
|
|
"DELETE FROM mysql.user ;\n"
|
|
"CREATE OR REPLACE USER 'root'@'%' IDENTIFIED BY \'{0}\' ;\n"
|
|
"GRANT ALL ON *.* TO 'root'@'%' WITH GRANT OPTION ;\n"
|
|
"DROP DATABASE IF EXISTS test ;\n"
|
|
"FLUSH PRIVILEGES ;\n"
|
|
"SHUTDOWN ;".format(mysql_root_password))
|
|
bootstrap_sql_file = tempfile.NamedTemporaryFile(suffix='.sql').name
|
|
with open(bootstrap_sql_file, 'w') as f:
|
|
f.write(template)
|
|
f.close()
|
|
run_cmd_with_logging([
|
|
'mysqld', '--bind-address=127.0.0.1',
|
|
"--init-file={0}".format(bootstrap_sql_file)
|
|
], logger)
|
|
os.remove(bootstrap_sql_file)
|
|
else:
|
|
logger.info("Skipping bootstrap as {0} directory is present".format(
|
|
bootstrap_test_dir))
|
|
|
|
|
|
def safe_update_configmap(configmap_dict, configmap_patch):
|
|
"""Update a configmap with locking.
|
|
|
|
Keyword arguments:
|
|
configmap_dict -- a dict representing the configmap to be patched
|
|
configmap_patch -- a dict containign the patch
|
|
"""
|
|
logger.debug("Safe Patching configmap")
|
|
#NOTE(portdirect): Explictly set the resource version we are patching to
|
|
# ensure nothing else has modified the confimap since we read it.
|
|
configmap_patch['metadata']['resourceVersion'] = configmap_dict[
|
|
'metadata']['resource_version']
|
|
try:
|
|
api_response = k8s_api_instance.patch_namespaced_config_map(
|
|
name=state_configmap_name,
|
|
namespace=pod_namespace,
|
|
body=configmap_patch)
|
|
return True
|
|
except kubernetes.client.rest.ApiException as error:
|
|
logger.error("Failed to set configmap: {0}".format(error))
|
|
return error
|
|
|
|
|
|
def set_configmap_annotation(key, value):
|
|
"""Update a configmap's annotations via patching.
|
|
|
|
Keyword arguments:
|
|
key -- the key to be patched
|
|
value -- the value to give the key
|
|
"""
|
|
logger.debug("Setting configmap annotation key={0} value={1}".format(
|
|
key, value))
|
|
configmap_dict = k8s_api_instance.read_namespaced_config_map(
|
|
name=state_configmap_name, namespace=pod_namespace).to_dict()
|
|
configmap_patch = {'metadata': {'annotations': {}}}
|
|
configmap_patch['metadata']['annotations'][key] = value
|
|
return safe_update_configmap(
|
|
configmap_dict=configmap_dict, configmap_patch=configmap_patch)
|
|
|
|
|
|
def set_configmap_data(key, value):
|
|
"""Update a configmap's data via patching.
|
|
|
|
Keyword arguments:
|
|
key -- the key to be patched
|
|
value -- the value to give the key
|
|
"""
|
|
logger.debug("Setting configmap data key={0} value={1}".format(key, value))
|
|
configmap_dict = k8s_api_instance.read_namespaced_config_map(
|
|
name=state_configmap_name, namespace=pod_namespace).to_dict()
|
|
configmap_patch = {'data': {}, 'metadata': {}}
|
|
configmap_patch['data'][key] = value
|
|
return safe_update_configmap(
|
|
configmap_dict=configmap_dict, configmap_patch=configmap_patch)
|
|
|
|
|
|
def get_configmap_value(key, type='data'):
|
|
"""Get a configmap's key's value.
|
|
|
|
Keyword arguments:
|
|
key -- the key to retrive the data from
|
|
type -- the type of data to retrive from the configmap, can either be 'data'
|
|
or an 'annotation'. (default data)
|
|
"""
|
|
state_configmap = k8s_api_instance.read_namespaced_config_map(
|
|
name=state_configmap_name, namespace=pod_namespace)
|
|
state_configmap_dict = state_configmap.to_dict()
|
|
if type == 'data':
|
|
state_configmap_data = state_configmap_dict['data']
|
|
elif type == 'annotation':
|
|
state_configmap_data = state_configmap_dict['metadata']['annotations']
|
|
else:
|
|
logger.error(
|
|
"Unknown data type \"{0}\" reqested for retrival".format(type))
|
|
return False
|
|
if state_configmap_data and key in state_configmap_data:
|
|
return state_configmap_data[key]
|
|
else:
|
|
return None
|
|
|
|
|
|
def get_cluster_state():
|
|
"""Get the current cluster state from a configmap, creating the configmap
|
|
if it does not already exist.
|
|
"""
|
|
logger.info("Getting cluster state")
|
|
state = None
|
|
while state is None:
|
|
try:
|
|
state = get_configmap_value(
|
|
type='annotation',
|
|
key='openstackhelm.openstack.org/cluster.state')
|
|
logger.info(
|
|
"The cluster is currently in \"{0}\" state.".format(state))
|
|
except:
|
|
logger.info("The cluster configmap \"{0}\" does not exist.".format(
|
|
state_configmap_name))
|
|
initial_configmap_body = {
|
|
"apiVersion": "v1",
|
|
"kind": "ConfigMap",
|
|
"metadata": {
|
|
"name": state_configmap_name,
|
|
"annotations": {
|
|
"openstackhelm.openstack.org/cluster.state": "new"
|
|
}
|
|
},
|
|
"data": {}
|
|
}
|
|
ensure_state_configmap(
|
|
pod_namespace=pod_namespace,
|
|
configmap_name=state_configmap_name,
|
|
configmap_body=initial_configmap_body)
|
|
return state
|
|
|
|
|
|
def declare_myself_cluser_leader(ttl):
|
|
"""Declare the current pod as the cluster leader.
|
|
|
|
Keyword arguments:
|
|
ttl -- the ttl for the leader period
|
|
"""
|
|
logger.info("Declaring myself current cluster leader")
|
|
leader_expiry_raw = datetime.utcnow() + timedelta(seconds=120)
|
|
leader_expiry = "{0}Z".format(leader_expiry_raw.isoformat("T"))
|
|
set_configmap_annotation(
|
|
key='openstackhelm.openstack.org/leader.node', value=local_hostname)
|
|
set_configmap_annotation(
|
|
key='openstackhelm.openstack.org/leader.expiry', value=leader_expiry)
|
|
|
|
|
|
def deadmans_leader_election():
|
|
"""Run a simplisic deadmans leader election."""
|
|
leader_node = get_configmap_value(
|
|
type='annotation', key='openstackhelm.openstack.org/leader.node')
|
|
leader_expiry = get_configmap_value(
|
|
type='annotation', key='openstackhelm.openstack.org/leader.expiry')
|
|
if iso8601.parse_date(leader_expiry).replace(
|
|
tzinfo=None) < datetime.utcnow().replace(tzinfo=None):
|
|
logger.info("Current cluster leader has expired")
|
|
declare_myself_cluser_leader(ttl=cluster_leader_ttl)
|
|
elif local_hostname == leader_node:
|
|
logger.info("Renewing cluster leader lease")
|
|
declare_myself_cluser_leader(ttl=cluster_leader_ttl)
|
|
|
|
|
|
def get_grastate_val(key):
|
|
"""Extract data from grastate.dat.
|
|
|
|
Keyword arguments:
|
|
key -- the key to extract the value of
|
|
"""
|
|
logger.debug("Reading grastate.dat key={0}".format(key))
|
|
with open("/var/lib/mysql/grastate.dat", "r") as myfile:
|
|
grastate_raw = map(lambda s: s.strip(), myfile.readlines())
|
|
return [i for i in grastate_raw
|
|
if i.startswith("{0}:".format(key))][0].split(':')[1].strip()
|
|
|
|
|
|
def set_grastate_val(key, value):
|
|
"""Set values in grastate.dat.
|
|
|
|
Keyword arguments:
|
|
key -- the key to set the value of
|
|
value -- the value to set the key to
|
|
"""
|
|
logger.debug("Updating grastate.dat key={0} value={1}".format(key, value))
|
|
with open("/var/lib/mysql/grastate.dat", "r") as sources:
|
|
lines = sources.readlines()
|
|
for line_num, line_content in enumerate(lines):
|
|
if line_content.startswith("{0}:".format(key)):
|
|
line_content = "{0}: {1}\n".format(key, value)
|
|
lines[line_num] = line_content
|
|
with open("/var/lib/mysql/grastate.dat", "w") as sources:
|
|
for line in lines:
|
|
sources.write(line)
|
|
|
|
|
|
def update_grastate_configmap():
|
|
"""Update state configmap with grastate.dat info."""
|
|
while not os.path.exists('/var/lib/mysql/grastate.dat'):
|
|
time.sleep(1)
|
|
logger.info("Updating grastate configmap")
|
|
grastate = dict()
|
|
grastate['version'] = get_grastate_val(key='version')
|
|
grastate['uuid'] = get_grastate_val(key='uuid')
|
|
grastate['seqno'] = get_grastate_val(key='seqno')
|
|
grastate['safe_to_bootstrap'] = get_grastate_val(key='safe_to_bootstrap')
|
|
grastate['sample_time'] = "{0}Z".format(datetime.utcnow().isoformat("T"))
|
|
for grastate_key, grastate_value in grastate.iteritems():
|
|
configmap_key = "{0}.{1}".format(grastate_key, local_hostname)
|
|
if get_configmap_value(
|
|
type='data', key=configmap_key) != grastate_value:
|
|
set_configmap_data(key=configmap_key, value=grastate_value)
|
|
|
|
|
|
def update_grastate_on_restart():
|
|
"""Update the grastate.dat on node restart."""
|
|
logger.info("Updating grastate info for node")
|
|
if get_grastate_val(key='seqno') == '-1':
|
|
logger.info(
|
|
"Node shutdown was not clean, getting position via wsrep-recover")
|
|
|
|
def recover_wsrep_position():
|
|
"""Extract recoved wsrep position from uncleanly exited node."""
|
|
wsrep_recover = subprocess.Popen(
|
|
[
|
|
'mysqld', '--bind-address=127.0.0.1',
|
|
'--wsrep_cluster_address=gcomm://', '--wsrep-recover'
|
|
],
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE)
|
|
out, err = wsrep_recover.communicate()
|
|
for item in err.split("\n"):
|
|
if "WSREP: Recovered position:" in item:
|
|
line = item.strip().split()
|
|
wsrep_rec_pos = line[-1].split(':')[-1]
|
|
return wsrep_rec_pos
|
|
|
|
set_grastate_val(key='seqno', value=recover_wsrep_position())
|
|
else:
|
|
logger.info("Node shutdown was clean, using grastate.dat")
|
|
|
|
update_grastate_configmap()
|
|
|
|
|
|
def check_for_active_nodes(endpoints_name=direct_svc_name,
|
|
namespace=pod_namespace):
|
|
"""Check K8s endpoints to see if there are active Mariadb Instances.
|
|
|
|
Keyword arguments:
|
|
endpoints_name -- endpoints to check for active backends
|
|
(default direct_svc_name)
|
|
namespace -- namespace to check for endpoints (default pod_namespace)
|
|
"""
|
|
logger.info("Checking for active nodes")
|
|
endpoints = k8s_api_instance.read_namespaced_endpoints(
|
|
name=endpoints_name, namespace=pod_namespace)
|
|
endpoints_dict = endpoints.to_dict()
|
|
addresses_index = [
|
|
i for i, s in enumerate(endpoints_dict['subsets']) if 'addresses' in s
|
|
][0]
|
|
active_endpoints = endpoints_dict['subsets'][addresses_index]['addresses']
|
|
if active_endpoints and len(active_endpoints) >= 1:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
|
|
def check_if_cluster_data_is_fresh():
|
|
"""Check if the state_configmap is both current and reasonably stable."""
|
|
logger.info("Checking to see if cluster data is fresh")
|
|
state_configmap = k8s_api_instance.read_namespaced_config_map(
|
|
name=state_configmap_name, namespace=pod_namespace)
|
|
state_configmap_dict = state_configmap.to_dict()
|
|
sample_times = dict()
|
|
for key, value in state_configmap_dict['data'].iteritems():
|
|
keyitems = key.split('.')
|
|
key = keyitems[0]
|
|
node = keyitems[1]
|
|
if key == 'sample_time':
|
|
sample_times[node] = value
|
|
sample_time_ok = True
|
|
for key, value in sample_times.iteritems():
|
|
sample_time = iso8601.parse_date(value).replace(tzinfo=None)
|
|
sample_cutoff_time = datetime.utcnow().replace(
|
|
tzinfo=None) - timedelta(seconds=20)
|
|
if not sample_time >= sample_cutoff_time:
|
|
logger.info(
|
|
"The data we have from the cluster is too old to make a "
|
|
"decision for node {0}".format(key))
|
|
sample_time_ok = False
|
|
else:
|
|
logger.info(
|
|
"The data we have from the cluster is ok for node {0}".format(
|
|
key))
|
|
return sample_time_ok
|
|
|
|
|
|
def check_if_i_lead():
|
|
"""Check on full restart of cluster if this node should lead the cluster
|
|
reformation."""
|
|
logger.info("Checking to see if I lead the cluster for reboot")
|
|
# as we sample on the update period - we sample for a full cluster
|
|
# leader election period as a simplistic way of ensureing nodes are
|
|
# reliably checking in following full restart of cluster.
|
|
count = cluster_leader_ttl / state_configmap_update_period
|
|
counter = 0
|
|
while counter <= count:
|
|
if check_if_cluster_data_is_fresh():
|
|
counter += 1
|
|
else:
|
|
counter = 0
|
|
time.sleep(state_configmap_update_period)
|
|
logger.info(
|
|
"Cluster info has been uptodate {0} times out of the required "
|
|
"{1}".format(counter, count))
|
|
state_configmap = k8s_api_instance.read_namespaced_config_map(
|
|
name=state_configmap_name, namespace=pod_namespace)
|
|
state_configmap_dict = state_configmap.to_dict()
|
|
seqnos = dict()
|
|
for key, value in state_configmap_dict['data'].iteritems():
|
|
keyitems = key.split('.')
|
|
key = keyitems[0]
|
|
node = keyitems[1]
|
|
if key == 'seqno':
|
|
seqnos[node] = value
|
|
max_seqno = max(seqnos.values())
|
|
max_seqno_node = sorted(
|
|
[k for k, v in seqnos.items() if v == max_seqno])[0]
|
|
if local_hostname == max_seqno_node:
|
|
logger.info("I lead the cluster")
|
|
return True
|
|
else:
|
|
logger.info("{0} leads the cluster".format(max_seqno_node))
|
|
return False
|
|
|
|
|
|
def monitor_cluster():
|
|
"""Function to kick off grastate configmap updating thread"""
|
|
while True:
|
|
update_grastate_configmap()
|
|
time.sleep(state_configmap_update_period)
|
|
|
|
|
|
# Setup the thread for the cluster monitor
|
|
monitor_cluster_thread = threading.Thread(target=monitor_cluster, args=())
|
|
monitor_cluster_thread.daemon = True
|
|
|
|
|
|
def launch_cluster_monitor():
|
|
"""Launch grastate configmap updating thread"""
|
|
if not monitor_cluster_thread.isAlive():
|
|
monitor_cluster_thread.start()
|
|
|
|
|
|
def leader_election():
|
|
"""Function to kick off leader election thread"""
|
|
while True:
|
|
deadmans_leader_election()
|
|
time.sleep(cluster_leader_ttl / 2)
|
|
|
|
|
|
# Setup the thread for the leader election
|
|
leader_election_thread = threading.Thread(target=leader_election, args=())
|
|
leader_election_thread.daemon = True
|
|
|
|
|
|
def launch_leader_election():
|
|
"""Launch leader election thread"""
|
|
if not leader_election_thread.isAlive():
|
|
leader_election_thread.start()
|
|
|
|
|
|
def run_mysqld(cluster='existing'):
|
|
"""Launch the mysqld instance for the pod.
|
|
|
|
Keyword arguments:
|
|
cluster -- whether we going to form a cluster 'new' or joining an existing
|
|
cluster 'existing' (default 'existing')
|
|
"""
|
|
stop_mysqld()
|
|
mysqld_write_cluster_conf(mode='run')
|
|
launch_leader_election()
|
|
launch_cluster_monitor()
|
|
mysqld_cmd = ['mysqld']
|
|
if cluster == 'new':
|
|
mysqld_cmd.append('--wsrep-new-cluster')
|
|
run_cmd_with_logging(mysqld_cmd, logger)
|
|
|
|
|
|
def mysqld_reboot():
|
|
"""Reboot a mysqld cluster."""
|
|
declare_myself_cluser_leader(ttl=cluster_leader_ttl)
|
|
set_grastate_val(key='safe_to_bootstrap', value='1')
|
|
run_mysqld(cluster='new')
|
|
|
|
|
|
def sigterm_shutdown(x, y):
|
|
"""Shutdown the instnace of mysqld on shutdown signal."""
|
|
logger.info("Got a sigterm from the container runtime, time to go.")
|
|
stop_mysqld()
|
|
|
|
|
|
# Register the signal to the handler
|
|
signal.signal(signal.SIGTERM, sigterm_shutdown)
|
|
|
|
# Main logic loop
|
|
if get_cluster_state() == 'new':
|
|
set_configmap_annotation(
|
|
key='openstackhelm.openstack.org/cluster.state', value='init')
|
|
declare_myself_cluser_leader(ttl=cluster_leader_ttl)
|
|
launch_leader_election()
|
|
mysqld_bootstrap()
|
|
update_grastate_configmap()
|
|
set_configmap_annotation(
|
|
key='openstackhelm.openstack.org/cluster.state', value='live')
|
|
run_mysqld(cluster='new')
|
|
elif get_cluster_state() == 'init':
|
|
logger.info("Waiting for cluster to start running")
|
|
while not get_cluster_state() == 'live':
|
|
time.sleep(default_sleep)
|
|
while not check_for_active_nodes():
|
|
time.sleep(default_sleep)
|
|
launch_leader_election()
|
|
run_mysqld()
|
|
elif get_cluster_state() == 'live':
|
|
logger.info("Cluster has been running starting restore/rejoin")
|
|
if not mariadb_replicas > 1:
|
|
logger.info(
|
|
"There is only a single node in this cluster, we are good to go")
|
|
update_grastate_on_restart()
|
|
mysqld_reboot()
|
|
else:
|
|
if check_for_active_nodes():
|
|
logger.info(
|
|
"There are currently running nodes in the cluster, we can "
|
|
"join them")
|
|
run_mysqld()
|
|
else:
|
|
logger.info("This cluster has lost all running nodes, we need to "
|
|
"determine the new lead node")
|
|
update_grastate_on_restart()
|
|
launch_leader_election()
|
|
launch_cluster_monitor()
|
|
if check_if_i_lead():
|
|
logger.info("I won the ability to reboot the cluster")
|
|
mysqld_reboot()
|
|
else:
|
|
logger.info(
|
|
"Waiting for the lead node to come online before joining "
|
|
"it")
|
|
while not check_for_active_nodes():
|
|
time.sleep(default_sleep)
|
|
run_mysqld()
|
|
else:
|
|
logger.critical("Dont understand cluster state, exiting with error status")
|
|
sys.exit(1)
|