openstack
/
octavia
Code Issues Proposed changes

Amphora agent refactor to classes 

This patch is a prep work needed for Id99948aec64656a0532afc68e146f0610bff1378
which comes to Fix the amphora-agent support for RH based Linux flavors.

This is a pure refactor. Functions were gathered under classes (making
them methods) so state, such as the operating system flavor, can be preserved
throughout the entire amphora agent process lifecycle.

Related-Bug: #1548070

Change-Id: Ic149211dba8ea78e08cb06b6e1f65da00a6571c7
This commit is contained in:
Nir Magnezi 2016-08-05 09:45:02 +03:00
parent c6557c9412
commit 12d2a0f01b
10 changed files with 852 additions and 836 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

122
octavia/amphorae/backends/agent/api_server/keepalived.py
View File

@ -35,74 +35,76 @@ template = j2_env.get_template(consts.KEEPALIVED_CONF)
check_script_template = j2_env.get_template(consts.CHECK_SCRIPT_CONF)
def upload_keepalived_config():
stream = listener.Wrapped(flask.request.stream)
class Keepalived(object):
if not os.path.exists(util.keepalived_dir()):
os.makedirs(util.keepalived_dir())
os.makedirs(util.keepalived_check_scripts_dir())
def upload_keepalived_config(self):
stream = listener.Wrapped(flask.request.stream)
conf_file = util.keepalived_cfg_path()
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00644
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
with os.fdopen(os.open(conf_file, flags, mode), 'w') as f:
b = stream.read(BUFFER)
while b:
f.write(b)
if not os.path.exists(util.keepalived_dir()):
os.makedirs(util.keepalived_dir())
os.makedirs(util.keepalived_check_scripts_dir())
conf_file = util.keepalived_cfg_path()
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00644
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
with os.fdopen(os.open(conf_file, flags, mode), 'w') as f:
b = stream.read(BUFFER)
while b:
f.write(b)
b = stream.read(BUFFER)
file_path = util.keepalived_init_path()
# mode 00755
mode = (stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP |
stat.S_IROTH | stat.S_IXOTH)
if not os.path.exists(file_path):
with os.fdopen(os.open(file_path, flags, mode), 'w') as text_file:
text = template.render(
keepalived_pid=util.keepalived_pid_path(),
keepalived_cmd=consts.KEEPALIVED_CMD,
keepalived_cfg=util.keepalived_cfg_path(),
keepalived_log=util.keepalived_log_path(),
amphora_nsname=consts.AMPHORA_NAMESPACE
)
text_file.write(text)
file_path = util.keepalived_init_path()
# mode 00755
mode = (stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP |
stat.S_IROTH | stat.S_IXOTH)
if not os.path.exists(file_path):
with os.fdopen(os.open(file_path, flags, mode), 'w') as text_file:
text = template.render(
keepalived_pid=util.keepalived_pid_path(),
keepalived_cmd=consts.KEEPALIVED_CMD,
keepalived_cfg=util.keepalived_cfg_path(),
keepalived_log=util.keepalived_log_path(),
amphora_nsname=consts.AMPHORA_NAMESPACE
)
text_file.write(text)
# Renders the Keepalived check script
keepalived_path = util.keepalived_check_script_path()
open_obj = os.open(keepalived_path, flags, mode)
with os.fdopen(open_obj, 'w') as text_file:
text = check_script_template.render(
check_scripts_dir=util.keepalived_check_scripts_dir()
)
text_file.write(text)
# Renders the Keepalived check script
keepalived_path = util.keepalived_check_script_path()
open_obj = os.open(keepalived_path, flags, mode)
with os.fdopen(open_obj, 'w') as text_file:
text = check_script_template.render(
check_scripts_dir=util.keepalived_check_scripts_dir()
)
text_file.write(text)
res = flask.make_response(flask.jsonify({
'message': 'OK'}), 200)
res.headers['ETag'] = stream.get_md5()
res = flask.make_response(flask.jsonify({
'message': 'OK'}), 200)
res.headers['ETag'] = stream.get_md5()
return res
return res
def manager_keepalived_service(self, action):
action = action.lower()
if action not in [consts.AMP_ACTION_START,
consts.AMP_ACTION_STOP,
consts.AMP_ACTION_RELOAD]:
return flask.make_response(flask.jsonify(dict(
message='Invalid Request',
details="Unknown action: {0}".format(action))), 400)
def manager_keepalived_service(action):
action = action.lower()
if action not in [consts.AMP_ACTION_START,
consts.AMP_ACTION_STOP,
consts.AMP_ACTION_RELOAD]:
return flask.make_response(flask.jsonify(dict(
message='Invalid Request',
details="Unknown action: {0}".format(action))), 400)
cmd = ("/usr/sbin/service octavia-keepalived {action}".format(
action=action))
cmd = ("/usr/sbin/service octavia-keepalived {action}".format(
action=action))
try:
subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
LOG.debug("Failed to {0} keepalived service: {1}".format(action,
e))
return flask.make_response(flask.jsonify(dict(
message="Failed to {0} keepalived service".format(action),
details=e.output)), 500)
try:
subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
LOG.debug("Failed to {0} keepalived service: {1}".format(action, e))
return flask.make_response(flask.jsonify(dict(
message="Failed to {0} keepalived service".format(action),
details=e.output)), 500)
return flask.make_response(flask.jsonify(
dict(message='OK',
details='keepalived {action}ed'.format(action=action))), 202)
return flask.make_response(flask.jsonify(
dict(message='OK',
details='keepalived {action}ed'.format(action=action))), 202)

680
octavia/amphorae/backends/agent/api_server/listener.py
View File

@ -69,404 +69,386 @@ class Wrapped(object):
return getattr(self.stream, attr)
"""Gets the haproxy config
class Listener(object):
:param listenerid: the id of the listener
"""
def get_haproxy_config(self, listener_id):
"""Gets the haproxy config
:param listener_id: the id of the listener
"""
self._check_listener_exists(listener_id)
with open(util.config_path(listener_id), 'r') as file:
cfg = file.read()
resp = flask.Response(cfg, mimetype='text/plain', )
resp.headers['ETag'] = hashlib.md5(six.b(cfg)).hexdigest() # nosec
return resp
def get_haproxy_config(listener_id):
_check_listener_exists(listener_id)
with open(util.config_path(listener_id), 'r') as file:
cfg = file.read()
resp = flask.Response(cfg, mimetype='text/plain', )
resp.headers['ETag'] = hashlib.md5(six.b(cfg)).hexdigest() # nosec
return resp
def upload_haproxy_config(self, amphora_id, listener_id):
"""Upload the haproxy config
:param amphora_id: The id of the amphora to update
:param listener_id: The id of the listener
"""
stream = Wrapped(flask.request.stream)
# We have to hash here because HAProxy has a string length limitation
# in the configuration file "peer <peername>" lines
peer_name = octavia_utils.base64_sha1_string(amphora_id).rstrip('=')
if not os.path.exists(util.haproxy_dir(listener_id)):
os.makedirs(util.haproxy_dir(listener_id))
"""Upload the haproxy config
:param amphora_id: The id of the amphora to update
:param listener_id: The id of the listener
"""
def upload_haproxy_config(amphora_id, listener_id):
stream = Wrapped(flask.request.stream)
# We have to hash here because HAProxy has a string length limitation
# in the configuration file "peer <peername>" lines
peer_name = octavia_utils.base64_sha1_string(amphora_id).rstrip('=')
if not os.path.exists(util.haproxy_dir(listener_id)):
os.makedirs(util.haproxy_dir(listener_id))
name = os.path.join(util.haproxy_dir(listener_id), 'haproxy.cfg.new')
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00600
mode = stat.S_IRUSR | stat.S_IWUSR
with os.fdopen(os.open(name, flags, mode), 'w') as file:
b = stream.read(BUFFER)
while (b):
file.write(b)
name = os.path.join(util.haproxy_dir(listener_id), 'haproxy.cfg.new')
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00600
mode = stat.S_IRUSR | stat.S_IWUSR
with os.fdopen(os.open(name, flags, mode), 'w') as file:
b = stream.read(BUFFER)
while (b):
file.write(b)
b = stream.read(BUFFER)
# use haproxy to check the config
cmd = "haproxy -c -L {peer} -f {config_file}".format(config_file=name,
peer=peer_name)
# use haproxy to check the config
cmd = "haproxy -c -L {peer} -f {config_file}".format(config_file=name,
peer=peer_name)
try:
subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
LOG.debug("Failed to verify haproxy file: %s", e)
os.remove(name) # delete file
return flask.make_response(flask.jsonify(dict(
message="Invalid request",
details=e.output)), 400)
# file ok - move it
os.rename(name, util.config_path(listener_id))
use_upstart = util.CONF.haproxy_amphora.use_upstart
file = util.init_path(listener_id)
# mode 00755
mode = (stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP |
stat.S_IROTH | stat.S_IXOTH)
if not os.path.exists(file):
with os.fdopen(os.open(file, flags, mode), 'w') as text_file:
template = UPSTART_TEMPLATE if use_upstart else SYSVINIT_TEMPLATE
text = template.render(
peer_name=peer_name,
haproxy_pid=util.pid_path(listener_id),
haproxy_cmd=util.CONF.haproxy_amphora.haproxy_cmd,
haproxy_cfg=util.config_path(listener_id),
respawn_count=util.CONF.haproxy_amphora.respawn_count,
respawn_interval=util.CONF.haproxy_amphora.respawn_interval,
amphora_nsname=consts.AMPHORA_NAMESPACE
)
text_file.write(text)
if not use_upstart:
insrvcmd = ("insserv {file}".format(file=file))
try:
subprocess.check_output(insrvcmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
LOG.debug("Failed to make %(file)s executable: %(err)s",
{'file': file, 'err': e})
return flask.make_response(flask.jsonify(dict(
message="Error making file {0} executable".format(file),
details=e.output)), 500)
res = flask.make_response(flask.jsonify({
'message': 'OK'}), 202)
res.headers['ETag'] = stream.get_md5()
return res
def start_stop_listener(listener_id, action):
action = action.lower()
if action not in [consts.AMP_ACTION_START,
consts.AMP_ACTION_STOP,
consts.AMP_ACTION_RELOAD]:
return flask.make_response(flask.jsonify(dict(
message='Invalid Request',
details="Unknown action: {0}".format(action))), 400)
_check_listener_exists(listener_id)
# Since this script should be created at LB create time
# we can check for this path to see if VRRP is enabled
# on this amphora and not write the file if VRRP is not in use
if os.path.exists(util.keepalived_check_script_path()):
vrrp_check_script_update(listener_id, action)
# HAProxy does not start the process when given a reload
# so start it if haproxy is not already running
if action == consts.AMP_ACTION_RELOAD:
if consts.OFFLINE == _check_haproxy_status(listener_id):
action = consts.AMP_ACTION_START
cmd = ("/usr/sbin/service haproxy-{listener_id} {action}".format(
listener_id=listener_id, action=action))
try:
subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
if 'Job is already running' not in e.output:
LOG.debug("Failed to %(action)s HAProxy service: %(err)s",
{'action': action, 'err': e})
return flask.make_response(flask.jsonify(dict(
message="Error {0}ing haproxy".format(action),
details=e.output)), 500)
if action in [consts.AMP_ACTION_STOP,
consts.AMP_ACTION_RELOAD]:
return flask.make_response(flask.jsonify(
dict(message='OK',
details='Listener {listener_id} {action}ed'.format(
listener_id=listener_id, action=action))), 202)
details = (
'Configuration file is valid\nhaproxy daemon for {0} '.format(
listener_id) + 'started')
return flask.make_response(flask.jsonify(
dict(message='OK',
details=details)), 202)
def delete_listener(listener_id):
_check_listener_exists(listener_id)
# check if that haproxy is still running and if stop it
if os.path.exists(util.pid_path(listener_id)) and os.path.exists(
os.path.join('/proc', util.get_haproxy_pid(listener_id))):
cmd = "/usr/sbin/service haproxy-{0} stop".format(listener_id)
try:
subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
LOG.debug("Failed to stop HAProxy service: %s", e)
LOG.debug("Failed to verify haproxy file: %s", e)
os.remove(name) # delete file
return flask.make_response(flask.jsonify(dict(
message="Error stopping haproxy",
details=e.output)), 500)
message="Invalid request",
details=e.output)), 400)
# parse config and delete stats socket
try:
cfg = _parse_haproxy_file(listener_id)
os.remove(cfg['stats_socket'])
except Exception:
pass
# file ok - move it
os.rename(name, util.config_path(listener_id))
# delete the ssl files
try:
shutil.rmtree(_cert_dir(listener_id))
except Exception:
pass
use_upstart = util.CONF.haproxy_amphora.use_upstart
file = util.init_path(listener_id)
# mode 00755
mode = (stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP |
stat.S_IROTH | stat.S_IXOTH)
if not os.path.exists(file):
with os.fdopen(os.open(file, flags, mode), 'w') as text_file:
template = (UPSTART_TEMPLATE if use_upstart
else SYSVINIT_TEMPLATE)
text = template.render(
peer_name=peer_name,
haproxy_pid=util.pid_path(listener_id),
haproxy_cmd=util.CONF.haproxy_amphora.haproxy_cmd,
haproxy_cfg=util.config_path(listener_id),
respawn_count=util.CONF.haproxy_amphora.respawn_count,
respawn_interval=(util.CONF.haproxy_amphora.
respawn_interval),
amphora_nsname=consts.AMPHORA_NAMESPACE
)
text_file.write(text)
# delete the directory + init script for that listener
shutil.rmtree(util.haproxy_dir(listener_id))
if os.path.exists(util.init_path(listener_id)):
os.remove(util.init_path(listener_id))
if not use_upstart:
insrvcmd = ("insserv {file}".format(file=file))
return flask.jsonify({'message': 'OK'})
try:
subprocess.check_output(insrvcmd.split(),
stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
LOG.debug("Failed to make %(file)s executable: %(err)s",
{'file': file, 'err': e})
return flask.make_response(flask.jsonify(dict(
message="Error making file {0} executable".format(file),
details=e.output)), 500)
res = flask.make_response(flask.jsonify({
'message': 'OK'}), 202)
res.headers['ETag'] = stream.get_md5()
return res
"""Gets the status of all listeners
def start_stop_listener(self, listener_id, action):
action = action.lower()
if action not in [consts.AMP_ACTION_START,
consts.AMP_ACTION_STOP,
consts.AMP_ACTION_RELOAD]:
return flask.make_response(flask.jsonify(dict(
message='Invalid Request',
details="Unknown action: {0}".format(action))), 400)
This method will not consult the stats socket
so a listener might show as ACTIVE but still be
in ERROR
self._check_listener_exists(listener_id)
Currently type==SSL is also not detected
"""
# Since this script should be created at LB create time
# we can check for this path to see if VRRP is enabled
# on this amphora and not write the file if VRRP is not in use
if os.path.exists(util.keepalived_check_script_path()):
self.vrrp_check_script_update(listener_id, action)
# HAProxy does not start the process when given a reload
# so start it if haproxy is not already running
if action == consts.AMP_ACTION_RELOAD:
if consts.OFFLINE == self._check_haproxy_status(listener_id):
action = consts.AMP_ACTION_START
def get_all_listeners_status():
listeners = list()
cmd = ("/usr/sbin/service haproxy-{listener_id} {action}".format(
listener_id=listener_id, action=action))
for listener in util.get_listeners():
status = _check_listener_status(listener)
listener_type = ''
try:
subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
if 'Job is already running' not in e.output:
LOG.debug("Failed to %(action)s HAProxy service: %(err)s",
{'action': action, 'err': e})
return flask.make_response(flask.jsonify(dict(
message="Error {0}ing haproxy".format(action),
details=e.output)), 500)
if action in [consts.AMP_ACTION_STOP,
consts.AMP_ACTION_RELOAD]:
return flask.make_response(flask.jsonify(
dict(message='OK',
details='Listener {listener_id} {action}ed'.format(
listener_id=listener_id, action=action))), 202)
if status == consts.ACTIVE:
listener_type = _parse_haproxy_file(listener)['mode']
details = (
'Configuration file is valid\nhaproxy daemon for {0} '.format(
listener_id) + 'started')
listeners.append({
'status': status,
'uuid': listener,
'type': listener_type,
})
return flask.make_response(flask.jsonify(
dict(message='OK',
details=details)), 202)
# Can't use jsonify since lists are not supported
# for security reason: http://stackoverflow.com/
# questions/12435297/how-do-i-jsonify-a-list-in-flask
return flask.Response(json.dumps(listeners),
mimetype='application/json')
def delete_listener(self, listener_id):
self._check_listener_exists(listener_id)
# check if that haproxy is still running and if stop it
if os.path.exists(util.pid_path(listener_id)) and os.path.exists(
os.path.join('/proc', util.get_haproxy_pid(listener_id))):
cmd = "/usr/sbin/service haproxy-{0} stop".format(listener_id)
try:
subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
LOG.debug("Failed to stop HAProxy service: %s", e)
return flask.make_response(flask.jsonify(dict(
message="Error stopping haproxy",
details=e.output)), 500)
"""Gets the status of a listener
# parse config and delete stats socket
try:
cfg = self._parse_haproxy_file(listener_id)
os.remove(cfg['stats_socket'])
except Exception:
pass
This method will consult the stats socket
so calling this method will interfere with
the health daemon with the risk of the amphora
shut down
# delete the ssl files
try:
shutil.rmtree(self._cert_dir(listener_id))
except Exception:
pass
Currently type==SSL is not detected
"""
# delete the directory + init script for that listener
shutil.rmtree(util.haproxy_dir(listener_id))
if os.path.exists(util.init_path(listener_id)):
os.remove(util.init_path(listener_id))
return flask.jsonify({'message': 'OK'})
def get_listener_status(listener_id):
_check_listener_exists(listener_id)
def get_all_listeners_status(self):
"""Gets the status of all listeners
status = _check_listener_status(listener_id)
This method will not consult the stats socket
so a listener might show as ACTIVE but still be
in ERROR
if status != consts.ACTIVE:
Currently type==SSL is also not detected
"""
listeners = list()
for listener in util.get_listeners():
status = self._check_listener_status(listener)
listener_type = ''
if status == consts.ACTIVE:
listener_type = self._parse_haproxy_file(listener)['mode']
listeners.append({
'status': status,
'uuid': listener,
'type': listener_type,
})
# Can't use jsonify since lists are not supported
# for security reason: http://stackoverflow.com/
# questions/12435297/how-do-i-jsonify-a-list-in-flask
return flask.Response(json.dumps(listeners),
mimetype='application/json')
def get_listener_status(self, listener_id):
"""Gets the status of a listener
This method will consult the stats socket
so calling this method will interfere with
the health daemon with the risk of the amphora
shut down
Currently type==SSL is not detected
:param listener_id: The id of the listener
"""
self._check_listener_exists(listener_id)
status = self._check_listener_status(listener_id)
if status != consts.ACTIVE:
stats = dict(
status=status,
uuid=listener_id,
type=''
)
return flask.jsonify(stats)
cfg = self._parse_haproxy_file(listener_id)
stats = dict(
status=status,
uuid=listener_id,
type=''
type=cfg['mode']
)
# read stats socket
q = query.HAProxyQuery(cfg['stats_socket'])
servers = q.get_pool_status()
stats['pools'] = list(servers.values())
return flask.jsonify(stats)
cfg = _parse_haproxy_file(listener_id)
stats = dict(
status=status,
uuid=listener_id,
type=cfg['mode']
)
def upload_certificate(self, listener_id, filename):
self._check_ssl_filename_format(filename)
# read stats socket
q = query.HAProxyQuery(cfg['stats_socket'])
servers = q.get_pool_status()
stats['pools'] = list(servers.values())
return flask.jsonify(stats)
# create directory if not already there
if not os.path.exists(self._cert_dir(listener_id)):
os.makedirs(self._cert_dir(listener_id))
def upload_certificate(listener_id, filename):
_check_ssl_filename_format(filename)
# create directory if not already there
if not os.path.exists(_cert_dir(listener_id)):
os.makedirs(_cert_dir(listener_id))
stream = Wrapped(flask.request.stream)
file = _cert_file_path(listener_id, filename)
flags = os.O_WRONLY | os.O_CREAT
# mode 00600
mode = stat.S_IRUSR | stat.S_IWUSR
with os.fdopen(os.open(file, flags, mode), 'w') as crt_file:
b = stream.read(BUFFER)
while (b):
crt_file.write(b)
stream = Wrapped(flask.request.stream)
file = self._cert_file_path(listener_id, filename)
flags = os.O_WRONLY | os.O_CREAT
# mode 00600
mode = stat.S_IRUSR | stat.S_IWUSR
with os.fdopen(os.open(file, flags, mode), 'w') as crt_file:
b = stream.read(BUFFER)
while (b):
crt_file.write(b)
b = stream.read(BUFFER)
resp = flask.jsonify(dict(message='OK'))
resp.headers['ETag'] = stream.get_md5()
return resp
def get_certificate_md5(listener_id, filename):
_check_ssl_filename_format(filename)
cert_path = _cert_file_path(listener_id, filename)
path_exists = os.path.exists(cert_path)
if not path_exists:
return flask.make_response(flask.jsonify(dict(
message='Certificate Not Found',
details="No certificate with filename: {f}".format(
f=filename))), 404)
with open(cert_path, 'r') as crt_file:
cert = crt_file.read()
md5 = hashlib.md5(six.b(cert)).hexdigest() # nosec
resp = flask.jsonify(dict(md5sum=md5))
resp.headers['ETag'] = md5
resp = flask.jsonify(dict(message='OK'))
resp.headers['ETag'] = stream.get_md5()
return resp
def get_certificate_md5(self, listener_id, filename):
self._check_ssl_filename_format(filename)
def delete_certificate(listener_id, filename):
_check_ssl_filename_format(filename)
if not os.path.exists(_cert_file_path(listener_id, filename)):
return flask.make_response(flask.jsonify(dict(
message='Certificate Not Found',
details="No certificate with filename: {f}".format(
f=filename))), 404)
cert_path = self._cert_file_path(listener_id, filename)
path_exists = os.path.exists(cert_path)
if not path_exists:
return flask.make_response(flask.jsonify(dict(
message='Certificate Not Found',
details="No certificate with filename: {f}".format(
f=filename))), 404)
os.remove(_cert_file_path(listener_id, filename))
return flask.jsonify(dict(message='OK'))
with open(cert_path, 'r') as crt_file:
cert = crt_file.read()
md5 = hashlib.md5(six.b(cert)).hexdigest() # nosec
resp = flask.jsonify(dict(md5sum=md5))
resp.headers['ETag'] = md5
return resp
def delete_certificate(self, listener_id, filename):
self._check_ssl_filename_format(filename)
if not os.path.exists(self._cert_file_path(listener_id, filename)):
return flask.make_response(flask.jsonify(dict(
message='Certificate Not Found',
details="No certificate with filename: {f}".format(
f=filename))), 404)
def _check_listener_status(listener_id):
if os.path.exists(util.pid_path(listener_id)):
if os.path.exists(
os.path.join('/proc', util.get_haproxy_pid(listener_id))):
# Check if the listener is disabled
with open(util.config_path(listener_id), 'r') as file:
cfg = file.read()
m = re.search('frontend {}'.format(listener_id), cfg)
if m:
return consts.ACTIVE
else:
return consts.OFFLINE
else: # pid file but no process...
return consts.ERROR
else:
return consts.OFFLINE
os.remove(self._cert_file_path(listener_id, filename))
return flask.jsonify(dict(message='OK'))
def _parse_haproxy_file(listener_id):
with open(util.config_path(listener_id), 'r') as file:
cfg = file.read()
m = re.search('mode\s+(http|tcp)', cfg)
if not m:
raise ParsingError()
mode = m.group(1).upper()
m = re.search('stats socket\s+(\S+)', cfg)
if not m:
raise ParsingError()
stats_socket = m.group(1)
m = re.search('ssl crt\s+(\S+)', cfg)
ssl_crt = None
if m:
ssl_crt = m.group(1)
mode = 'TERMINATED_HTTPS'
return dict(mode=mode,
stats_socket=stats_socket,
ssl_crt=ssl_crt)
def _check_listener_exists(listener_id):
# check if we know about that listener
if not os.path.exists(util.config_path(listener_id)):
raise exceptions.HTTPException(
response=flask.make_response(flask.jsonify(dict(
message='Listener Not Found',
details="No listener with UUID: {0}".format(
listener_id))), 404))
def _check_ssl_filename_format(filename):
# check if the format is (xxx.)*xxx.pem
if not re.search('(\w.)+pem', filename):
raise exceptions.HTTPException(
response=flask.make_response(flask.jsonify(dict(
message='Filename has wrong format')), 400))
def _cert_dir(listener_id):
return os.path.join(util.CONF.haproxy_amphora.base_cert_dir,
listener_id)
def _cert_file_path(listener_id, filename):
return os.path.join(_cert_dir(listener_id), filename)
def vrrp_check_script_update(listener_id, action):
listener_ids = util.get_listeners()
if action == consts.AMP_ACTION_STOP:
listener_ids.remove(listener_id)
args = []
for listener_id in listener_ids:
args.append(util.haproxy_sock_path(listener_id))
if not os.path.exists(util.keepalived_dir()):
os.makedirs(util.keepalived_dir())
os.makedirs(util.keepalived_check_scripts_dir())
cmd = 'haproxy-vrrp-check {args}; exit $?'.format(args=' '.join(args))
with open(util.haproxy_check_script_path(), 'w') as text_file:
text_file.write(cmd)
def _check_haproxy_status(listener_id):
if os.path.exists(util.pid_path(listener_id)):
if os.path.exists(
os.path.join('/proc', util.get_haproxy_pid(listener_id))):
return consts.ACTIVE
else: # pid file but no process...
def _check_listener_status(self, listener_id):
if os.path.exists(util.pid_path(listener_id)):
if os.path.exists(
os.path.join('/proc', util.get_haproxy_pid(listener_id))):
# Check if the listener is disabled
with open(util.config_path(listener_id), 'r') as file:
cfg = file.read()
m = re.search('frontend {}'.format(listener_id), cfg)
if m:
return consts.ACTIVE
else:
return consts.OFFLINE
else: # pid file but no process...
return consts.ERROR
else:
return consts.OFFLINE
def _parse_haproxy_file(self, listener_id):
with open(util.config_path(listener_id), 'r') as file:
cfg = file.read()
m = re.search('mode\s+(http|tcp)', cfg)
if not m:
raise ParsingError()
mode = m.group(1).upper()
m = re.search('stats socket\s+(\S+)', cfg)
if not m:
raise ParsingError()
stats_socket = m.group(1)
m = re.search('ssl crt\s+(\S+)', cfg)
ssl_crt = None
if m:
ssl_crt = m.group(1)
mode = 'TERMINATED_HTTPS'
return dict(mode=mode,
stats_socket=stats_socket,
ssl_crt=ssl_crt)
def _check_listener_exists(self, listener_id):
# check if we know about that listener
if not os.path.exists(util.config_path(listener_id)):
raise exceptions.HTTPException(
response=flask.make_response(flask.jsonify(dict(
message='Listener Not Found',
details="No listener with UUID: {0}".format(
listener_id))), 404))
def _check_ssl_filename_format(self, filename):
# check if the format is (xxx.)*xxx.pem
if not re.search('(\w.)+pem', filename):
raise exceptions.HTTPException(
response=flask.make_response(flask.jsonify(dict(
message='Filename has wrong format')), 400))
def _cert_dir(self, listener_id):
return os.path.join(util.CONF.haproxy_amphora.base_cert_dir,
listener_id)
def _cert_file_path(self, listener_id, filename):
return os.path.join(self._cert_dir(listener_id), filename)
def vrrp_check_script_update(self, listener_id, action):
listener_ids = util.get_listeners()
if action == consts.AMP_ACTION_STOP:
listener_ids.remove(listener_id)
args = []
for listener_id in listener_ids:
args.append(util.haproxy_sock_path(listener_id))
if not os.path.exists(util.keepalived_dir()):
os.makedirs(util.keepalived_dir())
os.makedirs(util.keepalived_check_scripts_dir())
cmd = 'haproxy-vrrp-check {args}; exit $?'.format(args=' '.join(args))
with open(util.haproxy_check_script_path(), 'w') as text_file:
text_file.write(cmd)
def _check_haproxy_status(self, listener_id):
if os.path.exists(util.pid_path(listener_id)):
if os.path.exists(
os.path.join('/proc', util.get_haproxy_pid(listener_id))):
return consts.ACTIVE
else: # pid file but no process...
return consts.OFFLINE
else:
return consts.OFFLINE
else:
return consts.OFFLINE

555
octavia/amphorae/backends/agent/api_server/plug.py
View File

@ -48,306 +48,307 @@ template_port = j2_env.get_template(ETH_X_PORT_CONF)
template_vip = j2_env.get_template(ETH_X_VIP_CONF)
def plug_vip(vip, subnet_cidr, gateway,
mac_address, vrrp_ip=None, host_routes=None):
# Validate vip and subnet_cidr, calculate broadcast address and netmask
try:
render_host_routes = []
ip = ipaddress.ip_address(
vip if six.text_type == type(vip) else six.u(vip))
network = ipaddress.ip_network(
subnet_cidr if six.text_type == type(subnet_cidr)
else six.u(subnet_cidr))
vip = ip.exploded
broadcast = network.broadcast_address.exploded
netmask = (network.prefixlen if ip.version is 6
else network.netmask.exploded)
vrrp_version = None
if vrrp_ip:
vrrp_ip_obj = ipaddress.ip_address(
vrrp_ip if six.text_type == type(vrrp_ip) else six.u(vrrp_ip)
)
vrrp_version = vrrp_ip_obj.version
if host_routes:
for hr in host_routes:
network = ipaddress.ip_network(
hr['destination'] if isinstance(
hr['destination'], six.text_type) else
six.u(hr['destination']))
render_host_routes.append({'network': network,
'gw': hr['nexthop']})
except ValueError:
return flask.make_response(flask.jsonify(dict(
message="Invalid VIP")), 400)
class Plug(object):
# Check if the interface is already in the network namespace
# Do not attempt to re-plug the VIP if it is already in the
# network namespace
if _netns_interface_exists(mac_address):
return flask.make_response(flask.jsonify(dict(
message="Interface already exists")), 409)
# This is the interface prior to moving into the netns
default_netns_interface = _interface_by_mac(mac_address)
# Always put the VIP interface as eth1
primary_interface = consts.NETNS_PRIMARY_INTERFACE
secondary_interface = "{interface}:0".format(interface=primary_interface)
# We need to setup the netns network directory so that the ifup
# commands used here and in the startup scripts "sees" the right
# interfaces and scripts.
interface_file_path = util.get_network_interface_file(primary_interface)
os.makedirs('/etc/netns/' + consts.AMPHORA_NAMESPACE)
shutil.copytree('/etc/network',
'/etc/netns/{}/network'.format(consts.AMPHORA_NAMESPACE),
symlinks=True,
ignore=shutil.ignore_patterns('eth0*', 'openssh*'))
name = '/etc/netns/{}/network/interfaces'.format(consts.AMPHORA_NAMESPACE)
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00644
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
with os.fdopen(os.open(name, flags, mode), 'w') as int_file:
int_file.write('auto lo\n')
int_file.write('iface lo inet loopback\n')
if not CONF.amphora_agent.agent_server_network_file:
int_file.write('source /etc/netns/{}/network/'
'interfaces.d/*.cfg\n'.format(
consts.AMPHORA_NAMESPACE))
# write interface file
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
# If we are using a consolidated interfaces file, just append
# otherwise clear the per interface file as we are rewriting it
# TODO(johnsom): We need a way to clean out old interfaces records
if CONF.amphora_agent.agent_server_network_file:
flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND
else:
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
with os.fdopen(os.open(interface_file_path, flags, mode),
'w') as text_file:
text = template_vip.render(
interface=primary_interface,
vip=vip,
vip_ipv6=ip.version is 6,
broadcast=broadcast,
netmask=netmask,
gateway=gateway,
vrrp_ip=vrrp_ip,
vrrp_ipv6=vrrp_version is 6,
host_routes=render_host_routes,
)
text_file.write(text)
# Update the list of interfaces to add to the namespace
# This is used in the amphora reboot case to re-establish the namespace
_update_plugged_interfaces_file(primary_interface, mac_address)
# Create the namespace
netns = pyroute2.NetNS(consts.AMPHORA_NAMESPACE, flags=os.O_CREAT)
netns.close()
with pyroute2.IPRoute() as ipr:
# Move the interfaces into the namespace
idx = ipr.link_lookup(ifname=default_netns_interface)[0]
ipr.link('set', index=idx, net_ns_fd=consts.AMPHORA_NAMESPACE,
IFLA_IFNAME=primary_interface)
# bring interfaces up
_bring_if_down(primary_interface)
_bring_if_down(secondary_interface)
_bring_if_up(primary_interface, 'VIP')
_bring_if_up(secondary_interface, 'VIP')
return flask.make_response(flask.jsonify(dict(
message="OK",
details="VIP {vip} plugged on interface {interface}".format(
vip=vip, interface=primary_interface))), 202)
def _generate_network_file_text(netns_interface, fixed_ips):
text = ''
if fixed_ips is None:
text = template_port.render(interface=netns_interface)
else:
for index, fixed_ip in enumerate(fixed_ips, -1):
if index == -1:
netns_ip_interface = netns_interface
else:
netns_ip_interface = "{int}:{ip}".format(
int=netns_interface, ip=index)
try:
ip_addr = fixed_ip['ip_address']
cidr = fixed_ip['subnet_cidr']
ip = ipaddress.ip_address(
ip_addr if six.text_type == type(
ip_addr) else six.u(ip_addr))
network = ipaddress.ip_network(
cidr if six.text_type == type(
cidr) else six.u(cidr))
broadcast = network.broadcast_address.exploded
netmask = (network.prefixlen if ip.version is 6
else network.netmask.exploded)
host_routes = []
for hr in fixed_ip.get('host_routes', []):
def plug_vip(self, vip, subnet_cidr, gateway,
mac_address, vrrp_ip=None, host_routes=None):
# Validate vip and subnet_cidr, calculate broadcast address and netmask
try:
render_host_routes = []
ip = ipaddress.ip_address(
vip if six.text_type == type(vip) else six.u(vip))
network = ipaddress.ip_network(
subnet_cidr if six.text_type == type(subnet_cidr)
else six.u(subnet_cidr))
vip = ip.exploded
broadcast = network.broadcast_address.exploded
netmask = (network.prefixlen if ip.version is 6
else network.netmask.exploded)
vrrp_version = None
if vrrp_ip:
vrrp_ip_obj = ipaddress.ip_address(
vrrp_ip if six.text_type == type(vrrp_ip)
else six.u(vrrp_ip)
)
vrrp_version = vrrp_ip_obj.version
if host_routes:
for hr in host_routes:
network = ipaddress.ip_network(
hr['destination'] if isinstance(
hr['destination'], six.text_type) else
six.u(hr['destination']))
host_routes.append({'network': network,
'gw': hr['nexthop']})
except ValueError:
return flask.make_response(flask.jsonify(dict(
message="Invalid network IP")), 400)
new_text = template_port.render(interface=netns_ip_interface,
ipv6=ip.version is 6,
ip_address=ip.exploded,
broadcast=broadcast,
netmask=netmask,
host_routes=host_routes)
text = '\n'.join([text, new_text])
return text
render_host_routes.append({'network': network,
'gw': hr['nexthop']})
except ValueError:
return flask.make_response(flask.jsonify(dict(
message="Invalid VIP")), 400)
# Check if the interface is already in the network namespace
# Do not attempt to re-plug the VIP if it is already in the
# network namespace
if self._netns_interface_exists(mac_address):
return flask.make_response(flask.jsonify(dict(
message="Interface already exists")), 409)
def _check_ip_addresses(fixed_ips):
if fixed_ips:
for ip in fixed_ips:
try:
socket.inet_pton(socket.AF_INET, ip.get('ip_address'))
except socket.error:
socket.inet_pton(socket.AF_INET6, ip.get('ip_address'))
# This is the interface prior to moving into the netns
default_netns_interface = self._interface_by_mac(mac_address)
# Always put the VIP interface as eth1
primary_interface = consts.NETNS_PRIMARY_INTERFACE
secondary_interface = "{interface}:0".format(
interface=primary_interface)
def plug_network(mac_address, fixed_ips):
# Check if the interface is already in the network namespace
# Do not attempt to re-plug the network if it is already in the
# network namespace
if _netns_interface_exists(mac_address):
return flask.make_response(flask.jsonify(dict(
message="Interface already exists")), 409)
# This is the interface as it was initially plugged into the
# default network namespace, this will likely always be eth1
try:
_check_ip_addresses(fixed_ips=fixed_ips)
except socket.error:
return flask.make_response(flask.jsonify(dict(
message="Invalid network port")), 400)
default_netns_interface = _interface_by_mac(mac_address)
# We need to determine the interface name when inside the namespace
# to avoid name conflicts
with pyroute2.NetNS(consts.AMPHORA_NAMESPACE, flags=os.O_CREAT) as netns:
# 1 means just loopback, but we should already have a VIP
# This works for the add/delete/add case as we don't delete interfaces
# Note, eth0 is skipped because that is the VIP interface
netns_interface = 'eth{0}'.format(len(netns.get_links()))
LOG.info(_LI('Plugged interface {0} will become {1} in the '
'namespace {2}').format(default_netns_interface,
netns_interface,
consts.AMPHORA_NAMESPACE))
interface_file_path = util.get_network_interface_file(netns_interface)
# write interface file
# If we are using a consolidated interfaces file, just append
# otherwise clear the per interface file as we are rewriting it
# TODO(johnsom): We need a way to clean out old interfaces records
if CONF.amphora_agent.agent_server_network_file:
flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND
else:
# We need to setup the netns network directory so that the ifup
# commands used here and in the startup scripts "sees" the right
# interfaces and scripts.
interface_file_path = util.get_network_interface_file(
primary_interface)
os.makedirs('/etc/netns/' + consts.AMPHORA_NAMESPACE)
shutil.copytree(
'/etc/network',
'/etc/netns/{}/network'.format(consts.AMPHORA_NAMESPACE),
symlinks=True,
ignore=shutil.ignore_patterns('eth0*', 'openssh*'))
name = '/etc/netns/{}/network/interfaces'.format(
consts.AMPHORA_NAMESPACE)
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00644
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
with os.fdopen(os.open(name, flags, mode), 'w') as int_file:
int_file.write('auto lo\n')
int_file.write('iface lo inet loopback\n')
if not CONF.amphora_agent.agent_server_network_file:
int_file.write('source /etc/netns/{}/network/'
'interfaces.d/*.cfg\n'.format(
consts.AMPHORA_NAMESPACE))
# mode 00644
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
# write interface file
with os.fdopen(os.open(interface_file_path, flags, mode),
'w') as text_file:
text = _generate_network_file_text(netns_interface, fixed_ips)
text_file.write(text)
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
# Update the list of interfaces to add to the namespace
_update_plugged_interfaces_file(netns_interface, mac_address)
# If we are using a consolidated interfaces file, just append
# otherwise clear the per interface file as we are rewriting it
# TODO(johnsom): We need a way to clean out old interfaces records
if CONF.amphora_agent.agent_server_network_file:
flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND
else:
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
with pyroute2.IPRoute() as ipr:
# Move the interfaces into the namespace
idx = ipr.link_lookup(ifname=default_netns_interface)[0]
ipr.link('set', index=idx,
net_ns_fd=consts.AMPHORA_NAMESPACE,
IFLA_IFNAME=netns_interface)
with os.fdopen(os.open(interface_file_path, flags, mode),
'w') as text_file:
text = template_vip.render(
interface=primary_interface,
vip=vip,
vip_ipv6=ip.version is 6,
broadcast=broadcast,
netmask=netmask,
gateway=gateway,
vrrp_ip=vrrp_ip,
vrrp_ipv6=vrrp_version is 6,
host_routes=render_host_routes,
)
text_file.write(text)
_bring_if_down(netns_interface)
_bring_if_up(netns_interface, 'network')
# Update the list of interfaces to add to the namespace
# This is used in the amphora reboot case to re-establish the namespace
self._update_plugged_interfaces_file(primary_interface, mac_address)
return flask.make_response(flask.jsonify(dict(
message="OK",
details="Plugged on interface {interface}".format(
interface=netns_interface))), 202)
# Create the namespace
netns = pyroute2.NetNS(consts.AMPHORA_NAMESPACE, flags=os.O_CREAT)
netns.close()
with pyroute2.IPRoute() as ipr:
# Move the interfaces into the namespace
idx = ipr.link_lookup(ifname=default_netns_interface)[0]
ipr.link('set', index=idx, net_ns_fd=consts.AMPHORA_NAMESPACE,
IFLA_IFNAME=primary_interface)
def _interface_by_mac(mac):
for interface in netifaces.interfaces():
if netifaces.AF_LINK in netifaces.ifaddresses(interface):
for link in netifaces.ifaddresses(interface)[netifaces.AF_LINK]:
if link.get('addr', '').lower() == mac.lower():
return interface
raise exceptions.HTTPException(
response=flask.make_response(flask.jsonify(dict(
details="No suitable network interface found")), 404))
# bring interfaces up
self._bring_if_down(primary_interface)
self._bring_if_down(secondary_interface)
self._bring_if_up(primary_interface, 'VIP')
self._bring_if_up(secondary_interface, 'VIP')
return flask.make_response(flask.jsonify(dict(
message="OK",
details="VIP {vip} plugged on interface {interface}".format(
vip=vip, interface=primary_interface))), 202)
def _bring_if_up(interface, what):
# Note, we are not using pyroute2 for this as it is not /etc/netns
# aware.
cmd = ("ip netns exec {ns} ifup {params}".format(
ns=consts.AMPHORA_NAMESPACE, params=interface))
try:
subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
LOG.error(_LE('Failed to if up {0} due to '
'error: {1}').format(interface, str(e)))
def _generate_network_file_text(self, netns_interface, fixed_ips):
text = ''
if fixed_ips is None:
text = template_port.render(interface=netns_interface)
else:
for index, fixed_ip in enumerate(fixed_ips, -1):
if index == -1:
netns_ip_interface = netns_interface
else:
netns_ip_interface = "{int}:{ip}".format(
int=netns_interface, ip=index)
try:
ip_addr = fixed_ip['ip_address']
cidr = fixed_ip['subnet_cidr']
ip = ipaddress.ip_address(
ip_addr if six.text_type == type(
ip_addr) else six.u(ip_addr))
network = ipaddress.ip_network(
cidr if six.text_type == type(
cidr) else six.u(cidr))
broadcast = network.broadcast_address.exploded
netmask = (network.prefixlen if ip.version is 6
else network.netmask.exploded)
host_routes = []
for hr in fixed_ip.get('host_routes', []):
network = ipaddress.ip_network(
hr['destination'] if isinstance(
hr['destination'], six.text_type) else
six.u(hr['destination']))
host_routes.append({'network': network,
'gw': hr['nexthop']})
except ValueError:
return flask.make_response(flask.jsonify(dict(
message="Invalid network IP")), 400)
new_text = template_port.render(interface=netns_ip_interface,
ipv6=ip.version is 6,
ip_address=ip.exploded,
broadcast=broadcast,
netmask=netmask,
host_routes=host_routes)
text = '\n'.join([text, new_text])
return text
def _check_ip_addresses(self, fixed_ips):
if fixed_ips:
for ip in fixed_ips:
try:
socket.inet_pton(socket.AF_INET, ip.get('ip_address'))
except socket.error:
socket.inet_pton(socket.AF_INET6, ip.get('ip_address'))
def plug_network(self, mac_address, fixed_ips):
# Check if the interface is already in the network namespace
# Do not attempt to re-plug the network if it is already in the
# network namespace
if self._netns_interface_exists(mac_address):
return flask.make_response(flask.jsonify(dict(
message="Interface already exists")), 409)
# This is the interface as it was initially plugged into the
# default network namespace, this will likely always be eth1
try:
self._check_ip_addresses(fixed_ips=fixed_ips)
except socket.error:
return flask.make_response(flask.jsonify(dict(
message="Invalid network port")), 400)
default_netns_interface = self._interface_by_mac(mac_address)
# We need to determine the interface name when inside the namespace
# to avoid name conflicts
with pyroute2.NetNS(consts.AMPHORA_NAMESPACE,
flags=os.O_CREAT) as netns:
# 1 means just loopback, but we should already have a VIP. This
# works for the add/delete/add case as we don't delete interfaces
# Note, eth0 is skipped because that is the VIP interface
netns_interface = 'eth{0}'.format(len(netns.get_links()))
LOG.info(_LI('Plugged interface {0} will become {1} in the '
'namespace {2}').format(default_netns_interface,
netns_interface,
consts.AMPHORA_NAMESPACE))
interface_file_path = util.get_network_interface_file(netns_interface)
# write interface file
# If we are using a consolidated interfaces file, just append
# otherwise clear the per interface file as we are rewriting it
# TODO(johnsom): We need a way to clean out old interfaces records
if CONF.amphora_agent.agent_server_network_file:
flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND
else:
flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC
# mode 00644
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
with os.fdopen(os.open(interface_file_path, flags, mode),
'w') as text_file:
text = self._generate_network_file_text(netns_interface, fixed_ips)
text_file.write(text)
# Update the list of interfaces to add to the namespace
self._update_plugged_interfaces_file(netns_interface, mac_address)
with pyroute2.IPRoute() as ipr:
# Move the interfaces into the namespace
idx = ipr.link_lookup(ifname=default_netns_interface)[0]
ipr.link('set', index=idx,
net_ns_fd=consts.AMPHORA_NAMESPACE,
IFLA_IFNAME=netns_interface)
self._bring_if_down(netns_interface)
self._bring_if_up(netns_interface, 'network')
return flask.make_response(flask.jsonify(dict(
message="OK",
details="Plugged on interface {interface}".format(
interface=netns_interface))), 202)
def _interface_by_mac(self, mac):
for interface in netifaces.interfaces():
if netifaces.AF_LINK in netifaces.ifaddresses(interface):
for link in netifaces.ifaddresses(
interface)[netifaces.AF_LINK]:
if link.get('addr', '').lower() == mac.lower():
return interface
raise exceptions.HTTPException(
response=flask.make_response(flask.jsonify(dict(
message='Error plugging {0}'.format(what),
details=e.output)), 500))
details="No suitable network interface found")), 404))
def _bring_if_up(self, interface, what):
# Note, we are not using pyroute2 for this as it is not /etc/netns
# aware.
cmd = ("ip netns exec {ns} ifup {params}".format(
ns=consts.AMPHORA_NAMESPACE, params=interface))
try:
subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError as e:
LOG.error(_LE('Failed to if up {0} due to '
'error: {1}').format(interface, str(e)))
raise exceptions.HTTPException(
response=flask.make_response(flask.jsonify(dict(
message='Error plugging {0}'.format(what),
details=e.output)), 500))
def _bring_if_down(interface):
# Note, we are not using pyroute2 for this as it is not /etc/netns
# aware.
cmd = ("ip netns exec {ns} ifdown {params}".format(
ns=consts.AMPHORA_NAMESPACE, params=interface))
try:
subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError:
pass
def _bring_if_down(self, interface):
# Note, we are not using pyroute2 for this as it is not /etc/netns
# aware.
cmd = ("ip netns exec {ns} ifdown {params}".format(
ns=consts.AMPHORA_NAMESPACE, params=interface))
try:
subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
except subprocess.CalledProcessError:
pass
def _update_plugged_interfaces_file(self, interface, mac_address):
# write interfaces to plugged_interfaces file and prevent duplicates
plug_inf_file = consts.PLUGGED_INTERFACES
flags = os.O_RDWR | os.O_CREAT
# mode 0644
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
with os.fdopen(os.open(plug_inf_file, flags, mode), 'r+') as text_file:
inf_list = [inf.split()[0].rstrip() for inf in text_file]
if mac_address not in inf_list:
text_file.write("{mac_address} {interface}\n".format(
mac_address=mac_address, interface=interface))
def _update_plugged_interfaces_file(interface, mac_address):
# write interfaces to plugged_interfaces file and prevent duplicates
plug_inf_file = consts.PLUGGED_INTERFACES
flags = os.O_RDWR | os.O_CREAT
# mode 0644
mode = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH
with os.fdopen(os.open(plug_inf_file, flags, mode), 'r+') as text_file:
inf_list = [inf.split()[0].rstrip() for inf in text_file]
if mac_address not in inf_list:
text_file.write("{mac_address} {interface}\n".format(
mac_address=mac_address, interface=interface))
def _netns_interface_exists(mac_address):
with pyroute2.NetNS(consts.AMPHORA_NAMESPACE, flags=os.O_CREAT) as netns:
for link in netns.get_links():
for attr in link['attrs']:
if attr[0] == 'IFLA_ADDRESS' and attr[1] == mac_address:
return True
return False
def _netns_interface_exists(self, mac_address):
with pyroute2.NetNS(consts.AMPHORA_NAMESPACE,
flags=os.O_CREAT) as netns:
for link in netns.get_links():
for attr in link['attrs']:
if attr[0] == 'IFLA_ADDRESS' and attr[1] == mac_address:
return True
return False

226
octavia/amphorae/backends/agent/api_server/server.py
View File

@ -24,8 +24,7 @@ from octavia.amphorae.backends.agent.api_server import keepalived
from octavia.amphorae.backends.agent.api_server import listener
from octavia.amphorae.backends.agent.api_server import plug
app = flask.Flask(__name__)
PATH_PREFIX = '/' + api_server.VERSION
# make the error pages all json
@ -36,124 +35,147 @@ def make_json_error(ex):
return response
for code in six.iterkeys(exceptions.default_exceptions):
app.register_error_handler(code, make_json_error)
def register_app_error_handler(app):
for code in six.iterkeys(exceptions.default_exceptions):
app.register_error_handler(code, make_json_error)
@app.route('/' + api_server.VERSION +
'/listeners/<amphora_id>/<listener_id>/haproxy',
methods=['PUT'])
def upload_haproxy_config(amphora_id, listener_id):
return listener.upload_haproxy_config(amphora_id, listener_id)
class Server(object):
def __init__(self):
self.app = flask.Flask(__name__)
self._keepalived = keepalived.Keepalived()
self._listener = listener.Listener()
self._plug = plug.Plug()
register_app_error_handler(self.app)
@app.route('/' + api_server.VERSION + '/listeners/<listener_id>/haproxy',
methods=['GET'])
def get_haproxy_config(listener_id):
return listener.get_haproxy_config(listener_id)
self.app.add_url_rule(rule=PATH_PREFIX +
'/listeners/<amphora_id>/<listener_id>/haproxy',
view_func=self.upload_haproxy_config,
methods=['PUT'])
self.app.add_url_rule(rule=PATH_PREFIX +
'/listeners/<listener_id>/haproxy',
view_func=self.get_haproxy_config,
methods=['GET'])
self.app.add_url_rule(rule=PATH_PREFIX +
'/listeners/<listener_id>/<action>',
view_func=self.start_stop_listener,
methods=['PUT'])
self.app.add_url_rule(rule=PATH_PREFIX + '/listeners/<listener_id>',
view_func=self.delete_listener,
methods=['DELETE'])
self.app.add_url_rule(rule=PATH_PREFIX + '/details',
view_func=self.get_details,
methods=['GET'])
self.app.add_url_rule(rule=PATH_PREFIX + '/info',
view_func=self.get_info,
methods=['GET'])
self.app.add_url_rule(rule=PATH_PREFIX + '/listeners',
view_func=self.get_all_listeners_status,
methods=['GET'])
self.app.add_url_rule(rule=PATH_PREFIX + '/listeners/<listener_id>',
view_func=self.get_listener_status,
methods=['GET'])
self.app.add_url_rule(rule=PATH_PREFIX + '/listeners/<listener_id>'
'/certificates/<filename>',
view_func=self.upload_certificate,
methods=['PUT'])
self.app.add_url_rule(rule=PATH_PREFIX + '/listeners/<listener_id>'
'/certificates/<filename>',
view_func=self.get_certificate_md5,
methods=['GET'])
self.app.add_url_rule(rule=PATH_PREFIX + '/listeners/<listener_id>'
'/certificates/<filename>',
view_func=self.delete_certificate,
methods=['DELETE'])
self.app.add_url_rule(rule=PATH_PREFIX + '/plug/vip/<vip>',
view_func=self.plug_vip,
methods=['POST'])
self.app.add_url_rule(rule=PATH_PREFIX + '/plug/network',
view_func=self.plug_network,
methods=['POST'])
self.app.add_url_rule(rule=PATH_PREFIX + '/certificate',
view_func=self.upload_cert, methods=['PUT'])
self.app.add_url_rule(rule=PATH_PREFIX + '/vrrp/upload',
view_func=self.upload_vrrp_config,
methods=['PUT'])
self.app.add_url_rule(rule=PATH_PREFIX + '/vrrp/<action>',
view_func=self.manage_service_vrrp,
methods=['PUT'])
self.app.add_url_rule(rule='/' + api_server.VERSION +
'/interface/<ip_addr>',
view_func=self.get_interface,
methods=['GET'])
def upload_haproxy_config(self, amphora_id, listener_id):
return self._listener.upload_haproxy_config(amphora_id, listener_id)
@app.route('/' + api_server.VERSION +
'/listeners/<listener_id>/<action>',
methods=['PUT'])
def start_stop_listener(listener_id, action):
return listener.start_stop_listener(listener_id, action)
def get_haproxy_config(self, listener_id):
return self._listener.get_haproxy_config(listener_id)
def start_stop_listener(self, listener_id, action):
return self._listener.start_stop_listener(listener_id, action)
@app.route('/' + api_server.VERSION +
'/listeners/<listener_id>', methods=['DELETE'])
def delete_listener(listener_id):
return listener.delete_listener(listener_id)
def delete_listener(self, listener_id):
return self._listener.delete_listener(listener_id)
def get_details(self):
return amphora_info.compile_amphora_details()
@app.route('/' + api_server.VERSION + '/details',
methods=['GET'])
def get_details():
return amphora_info.compile_amphora_details()
def get_info(self):
return amphora_info.compile_amphora_info()
def get_all_listeners_status(self):
return self._listener.get_all_listeners_status()
@app.route('/' + api_server.VERSION + '/info',
methods=['GET'])
def get_info():
return amphora_info.compile_amphora_info()
def get_listener_status(self, listener_id):
return self._listener.get_listener_status(listener_id)
def upload_certificate(self, listener_id, filename):
return self._listener.upload_certificate(listener_id, filename)
@app.route('/' + api_server.VERSION + '/listeners',
methods=['GET'])
def get_all_listeners_status():
return listener.get_all_listeners_status()
def get_certificate_md5(self, listener_id, filename):
return self._listener.get_certificate_md5(listener_id, filename)
def delete_certificate(self, listener_id, filename):
return self._listener.delete_certificate(listener_id, filename)
@app.route('/' + api_server.VERSION + '/listeners/<listener_id>',
methods=['GET'])
def get_listener_status(listener_id):
return listener.get_listener_status(listener_id)
def plug_vip(self, vip):
# Catch any issues with the subnet info json
try:
net_info = flask.request.get_json()
assert type(net_info) is dict
assert 'subnet_cidr' in net_info
assert 'gateway' in net_info
assert 'mac_address' in net_info
except Exception:
raise exceptions.BadRequest(
description='Invalid subnet information')
return self._plug.plug_vip(vip,
net_info['subnet_cidr'],
net_info['gateway'],
net_info['mac_address'],
net_info.get('vrrp_ip'),
net_info.get('host_routes'))
def plug_network(self):
try:
port_info = flask.request.get_json()
assert type(port_info) is dict
assert 'mac_address' in port_info
except Exception:
raise exceptions.BadRequest(description='Invalid port information')
return self._plug.plug_network(port_info['mac_address'],
port_info.get('fixed_ips'))
@app.route('/' + api_server.VERSION + '/listeners/<listener_id>/certificates'
+ '/<filename>', methods=['PUT'])
def upload_certificate(listener_id, filename):
return listener.upload_certificate(listener_id, filename)
def upload_cert(self):
return certificate_update.upload_server_cert()
def upload_vrrp_config(self):
return self._keepalived.upload_keepalived_config()
@app.route('/' + api_server.VERSION + '/listeners/<listener_id>/certificates'
+ '/<filename>', methods=['GET'])
def get_certificate_md5(listener_id, filename):
return listener.get_certificate_md5(listener_id, filename)
def manage_service_vrrp(self, action):
return self._keepalived.manager_keepalived_service(action)
@app.route('/' + api_server.VERSION + '/listeners/<listener_id>/certificates'
+ '/<filename>', methods=['DELETE'])
def delete_certificate(listener_id, filename):
return listener.delete_certificate(listener_id, filename)
@app.route('/' + api_server.VERSION + '/plug/vip/<vip>', methods=['POST'])
def plug_vip(vip):
# Catch any issues with the subnet info json
try:
net_info = flask.request.get_json()
assert type(net_info) is dict
assert 'subnet_cidr' in net_info
assert 'gateway' in net_info
assert 'mac_address' in net_info
except Exception:
raise exceptions.BadRequest(description='Invalid subnet information')
return plug.plug_vip(vip,
net_info['subnet_cidr'],
net_info['gateway'],
net_info['mac_address'],
net_info.get('vrrp_ip'),
net_info.get('host_routes'))
@app.route('/' + api_server.VERSION + '/plug/network', methods=['POST'])
def plug_network():
try:
port_info = flask.request.get_json()
assert type(port_info) is dict
assert 'mac_address' in port_info
except Exception:
raise exceptions.BadRequest(description='Invalid port information')
return plug.plug_network(port_info['mac_address'],
port_info.get('fixed_ips'))
@app.route('/' + api_server.VERSION + '/certificate', methods=['PUT'])
def upload_cert():
return certificate_update.upload_server_cert()
@app.route('/' + api_server.VERSION + '/vrrp/upload', methods=['PUT'])
def upload_vrrp_config():
return keepalived.upload_keepalived_config()
@app.route('/' + api_server.VERSION + '/vrrp/<action>', methods=['PUT'])
def manage_service_vrrp(action):
return keepalived.manager_keepalived_service(action)
@app.route('/' + api_server.VERSION + '/interface/<ip_addr>', methods=['GET'])
def get_interface(ip_addr):
return amphora_info.get_interface(ip_addr)
def get_interface(self, ip_addr):
return amphora_info.get_interface(ip_addr)

5
octavia/cmd/agent.py
View File

@ -81,11 +81,14 @@ def main():
ctx.load_cert_chain(CONF.amphora_agent.agent_server_cert,
ca=CONF.amphora_agent.agent_server_ca)
# Initiate server class
server_instance = server.Server()
# This will trigger a reload if any files change and
# in particular the certificate file
serving.run_simple(hostname=CONF.haproxy_amphora.bind_host,
port=CONF.haproxy_amphora.bind_port,
application=server.app,
application=server_instance.app,
use_debugger=CONF.debug,
ssl_context=ctx,
use_reloader=True,

26
octavia/tests/functional/amphorae/backend/agent/api_server/test_server.py
View File

@ -25,7 +25,6 @@ import six
from octavia.amphorae.backends.agent import api_server
from octavia.amphorae.backends.agent.api_server import certificate_update
from octavia.amphorae.backends.agent.api_server import listener
from octavia.amphorae.backends.agent.api_server import server
from octavia.amphorae.backends.agent.api_server import util
from octavia.common import constants as consts
@ -42,7 +41,8 @@ class TestServerTestCase(base.TestCase):
app = None
def setUp(self):
self.app = server.app.test_client()
self.test_server = server.Server()
self.app = self.test_server.app.test_client()
super(TestServerTestCase, self).setUp()
@mock.patch('os.path.exists')
@ -140,7 +140,7 @@ class TestServerTestCase(base.TestCase):
mock_remove.assert_called_once_with(file_name)
@mock.patch('os.path.exists')
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'vrrp_check_script_update')
@mock.patch('subprocess.check_output')
def test_start(self, mock_subprocess, mock_vrrp, mock_exists):
@ -185,9 +185,9 @@ class TestServerTestCase(base.TestCase):
['/usr/sbin/service', 'haproxy-123', 'start'], stderr=-2)
@mock.patch('os.path.exists')
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'vrrp_check_script_update')
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'_check_haproxy_status')
@mock.patch('subprocess.check_output')
def test_reload(self, mock_subprocess, mock_haproxy_status,
@ -320,9 +320,9 @@ class TestServerTestCase(base.TestCase):
@mock.patch('octavia.amphorae.backends.agent.api_server.util.'
'get_listeners')
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'_check_listener_status')
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'_parse_haproxy_file')
def test_get_all_listeners(self, mock_parse, mock_status, mock_listener):
# no listeners
@ -355,9 +355,9 @@ class TestServerTestCase(base.TestCase):
{'status': consts.ERROR, 'type': '', 'uuid': '456'}],
json.loads(rv.data.decode('utf-8')))
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'_check_listener_status')
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'_parse_haproxy_file')
@mock.patch('octavia.amphorae.backends.utils.haproxy_query.HAProxyQuery')
@mock.patch('os.path.exists')
@ -462,7 +462,7 @@ class TestServerTestCase(base.TestCase):
mock_exists.return_value = True
mock_exists.side_effect = None
path = listener._cert_file_path('123', 'test.pem')
path = self.test_server._listener._cert_file_path('123', 'test.pem')
self.useFixture(test_utils.OpenFixture(path, CONTENT))
rv = self.app.get('/' + api_server.VERSION +
@ -482,7 +482,7 @@ class TestServerTestCase(base.TestCase):
self.assertEqual(400, rv.status_code)
mock_exists.return_value = True
path = listener._cert_file_path('123', 'test.pem')
path = self.test_server._listener._cert_file_path('123', 'test.pem')
m = self.useFixture(test_utils.OpenFixture(path)).mock_open
with mock.patch('os.open'), mock.patch.object(os, 'fdopen', m):
@ -527,7 +527,7 @@ class TestServerTestCase(base.TestCase):
@mock.patch('pyroute2.NetNS')
@mock.patch('subprocess.check_output')
@mock.patch('octavia.amphorae.backends.agent.api_server.'
'plug._netns_interface_exists')
'plug.Plug._netns_interface_exists')
def test_plug_network(self, mock_int_exists, mock_check_output, mock_netns,
mock_pyroute2, mock_ifaddress, mock_interfaces):
port_info = {'mac_address': '123'}
@ -802,7 +802,7 @@ class TestServerTestCase(base.TestCase):
'ifup', consts.NETNS_PRIMARY_INTERFACE], stderr=-2)
@mock.patch('octavia.amphorae.backends.agent.api_server.'
'plug._netns_interface_exists')
'plug.Plug._netns_interface_exists')
@mock.patch('netifaces.interfaces')
@mock.patch('netifaces.ifaddresses')
@mock.patch('pyroute2.IPRoute')

20
octavia/tests/functional/amphorae/backend/agent/api_server/test_server_sysvinit.py
View File

@ -26,7 +26,6 @@ import six
from octavia.amphorae.backends.agent import api_server
from octavia.amphorae.backends.agent.api_server import certificate_update
from octavia.amphorae.backends.agent.api_server import listener
from octavia.amphorae.backends.agent.api_server import server
from octavia.amphorae.backends.agent.api_server import util
from octavia.common import constants as consts
@ -43,7 +42,8 @@ class ServerTestCase(base.TestCase):
def setUp(self):
cfg.CONF.set_override('use_upstart', False, group='haproxy_amphora')
self.app = server.app.test_client()
self.test_server = server.Server()
self.app = self.test_server.app.test_client()
super(ServerTestCase, self).setUp()
@mock.patch('os.path.exists')
@ -147,7 +147,7 @@ class ServerTestCase(base.TestCase):
mock_remove.assert_called_once_with(file_name)
@mock.patch('os.path.exists')
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'vrrp_check_script_update')
@mock.patch('subprocess.check_output')
def test_start(self, mock_subprocess, mock_vrrp, mock_exists):
@ -292,9 +292,9 @@ class ServerTestCase(base.TestCase):
@mock.patch('octavia.amphorae.backends.agent.api_server.util.'
'get_listeners')
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'_check_listener_status')
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'_parse_haproxy_file')
def test_get_all_listeners(self, mock_parse, mock_status, mock_listener):
# no listeners
@ -327,9 +327,9 @@ class ServerTestCase(base.TestCase):
{'status': consts.ERROR, 'type': '', 'uuid': '456'}],
json.loads(rv.data.decode('utf-8')))
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'_check_listener_status')
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.'
@mock.patch('octavia.amphorae.backends.agent.api_server.listener.Listener.'
'_parse_haproxy_file')
@mock.patch('octavia.amphorae.backends.utils.haproxy_query.HAProxyQuery')
@mock.patch('os.path.exists')
@ -434,7 +434,7 @@ class ServerTestCase(base.TestCase):
mock_exists.return_value = True
mock_exists.side_effect = None
path = listener._cert_file_path('123', 'test.pem')
path = self.test_server._listener._cert_file_path('123', 'test.pem')
self.useFixture(test_utils.OpenFixture(path, CONTENT))
rv = self.app.get('/' + api_server.VERSION +
'/listeners/123/certificates/test.pem')
@ -452,7 +452,7 @@ class ServerTestCase(base.TestCase):
self.assertEqual(400, rv.status_code)
mock_exists.return_value = True
path = listener._cert_file_path('123', 'test.pem')
path = self.test_server._listener._cert_file_path('123', 'test.pem')
m = self.useFixture(test_utils.OpenFixture(path)).mock_open
with mock.patch('os.open'), mock.patch.object(os, 'fdopen', m):
@ -497,7 +497,7 @@ class ServerTestCase(base.TestCase):
@mock.patch('pyroute2.NetNS')
@mock.patch('subprocess.check_output')
@mock.patch('octavia.amphorae.backends.agent.api_server.'
'plug._netns_interface_exists')
'plug.Plug._netns_interface_exists')
def test_plug_network(self, mock_int_exists, mock_check_output, mock_netns,
mock_pyroute2, mock_ifaddress, mock_interfaces):
port_info = {'mac_address': '123'}

7
octavia/tests/unit/amphorae/backends/agent/api_server/test_keepalived.py
View File

@ -28,21 +28,22 @@ class KeepalivedTestCase(base.TestCase):
self.client = self.app.test_client()
self._ctx = self.app.test_request_context()
self._ctx.push()
self.test_keepalived = keepalived.Keepalived()
@mock.patch('subprocess.check_output')
def test_manager_keepalived_service(self, mock_check_output):
res = keepalived.manager_keepalived_service('start')
res = self.test_keepalived.manager_keepalived_service('start')
cmd = ("/usr/sbin/service octavia-keepalived {action}".format(
action='start'))
mock_check_output.assert_called_once_with(cmd.split(),
stderr=subprocess.STDOUT)
self.assertEqual(202, res.status_code)
res = keepalived.manager_keepalived_service('restart')
res = self.test_keepalived.manager_keepalived_service('restart')
self.assertEqual(400, res.status_code)
mock_check_output.side_effect = subprocess.CalledProcessError(1,
'blah!')
res = keepalived.manager_keepalived_service('start')
res = self.test_keepalived.manager_keepalived_service('start')
self.assertEqual(500, res.status_code)

27
octavia/tests/unit/amphorae/backends/agent/api_server/test_listener.py
View File

@ -34,6 +34,7 @@ class ListenerTestCase(base.TestCase):
self.jinja_cfg = jinja_cfg.JinjaTemplater(
base_amp_path=BASE_AMP_PATH,
base_crt_dir=BASE_CRT_PATH)
self.test_listener = listener.Listener()
def test_parse_haproxy_config(self):
# template_tls
@ -49,7 +50,7 @@ class ListenerTestCase(base.TestCase):
path = agent_util.config_path(LISTENER_ID1)
self.useFixture(test_utils.OpenFixture(path, rendered_obj))
res = listener._parse_haproxy_file(LISTENER_ID1)
res = self.test_listener._parse_haproxy_file(LISTENER_ID1)
self.assertEqual('TERMINATED_HTTPS', res['mode'])
self.assertEqual('/var/lib/octavia/sample_listener_id_1.sock',
res['stats_socket'])
@ -69,7 +70,7 @@ class ListenerTestCase(base.TestCase):
self.useFixture(test_utils.OpenFixture(path, rendered_obj))
res = listener._parse_haproxy_file(LISTENER_ID1)
res = self.test_listener._parse_haproxy_file(LISTENER_ID1)
self.assertEqual('TERMINATED_HTTPS', res['mode'])
self.assertEqual(BASE_AMP_PATH + '/sample_listener_id_1.sock',
res['stats_socket'])
@ -84,7 +85,7 @@ class ListenerTestCase(base.TestCase):
self.useFixture(test_utils.OpenFixture(path, rendered_obj))
res = listener._parse_haproxy_file(LISTENER_ID1)
res = self.test_listener._parse_haproxy_file(LISTENER_ID1)
self.assertEqual('HTTP', res['mode'])
self.assertEqual(BASE_AMP_PATH + '/sample_listener_id_1.sock',
res['stats_socket'])
@ -96,7 +97,7 @@ class ListenerTestCase(base.TestCase):
sample_configs.sample_listener_tuple(proto='HTTPS'))
self.useFixture(test_utils.OpenFixture(path, rendered_obj))
res = listener._parse_haproxy_file(LISTENER_ID1)
res = self.test_listener._parse_haproxy_file(LISTENER_ID1)
self.assertEqual('TCP', res['mode'])
self.assertEqual(BASE_AMP_PATH + '/sample_listener_id_1.sock',
res['stats_socket'])
@ -105,7 +106,7 @@ class ListenerTestCase(base.TestCase):
# Bogus format
self.useFixture(test_utils.OpenFixture(path, 'Bogus'))
try:
res = listener._parse_haproxy_file(LISTENER_ID1)
res = self.test_listener._parse_haproxy_file(LISTENER_ID1)
self.fail("No Exception?")
except listener.ParsingError:
pass
@ -121,17 +122,17 @@ class ListenerTestCase(base.TestCase):
self.useFixture(test_utils.OpenFixture(config_path, file_contents))
self.assertEqual(
consts.ACTIVE,
listener._check_listener_status(LISTENER_ID1))
self.test_listener._check_listener_status(LISTENER_ID1))
mock_exists.side_effect = [True, False]
self.assertEqual(
consts.ERROR,
listener._check_listener_status(LISTENER_ID1))
self.test_listener._check_listener_status(LISTENER_ID1))
mock_exists.side_effect = [False]
self.assertEqual(
consts.OFFLINE,
listener._check_listener_status(LISTENER_ID1))
self.test_listener._check_listener_status(LISTENER_ID1))
@mock.patch('os.makedirs')
@mock.patch('os.path.exists')
@ -152,7 +153,7 @@ class ListenerTestCase(base.TestCase):
path = agent_util.keepalived_dir()
m = self.useFixture(test_utils.OpenFixture(path)).mock_open
listener.vrrp_check_script_update(LISTENER_ID1, 'stop')
self.test_listener.vrrp_check_script_update(LISTENER_ID1, 'stop')
handle = m()
handle.write.assert_called_once_with(cmd)
@ -162,7 +163,7 @@ class ListenerTestCase(base.TestCase):
'$?')
m = self.useFixture(test_utils.OpenFixture(path)).mock_open
listener.vrrp_check_script_update(LISTENER_ID1, 'start')
self.test_listener.vrrp_check_script_update(LISTENER_ID1, 'start')
handle = m()
handle.write.assert_called_once_with(cmd)
@ -174,14 +175,14 @@ class ListenerTestCase(base.TestCase):
mock_exists.side_effect = [True, True]
self.assertEqual(
consts.ACTIVE,
listener._check_haproxy_status(LISTENER_ID1))
self.test_listener._check_haproxy_status(LISTENER_ID1))
mock_exists.side_effect = [True, False]
self.assertEqual(
consts.OFFLINE,
listener._check_haproxy_status(LISTENER_ID1))
self.test_listener._check_haproxy_status(LISTENER_ID1))
mock_exists.side_effect = [False]
self.assertEqual(
consts.OFFLINE,
listener._check_haproxy_status(LISTENER_ID1))
self.test_listener._check_haproxy_status(LISTENER_ID1))

20
octavia/tests/unit/amphorae/backends/agent/api_server/test_plug.py
View File

@ -34,8 +34,8 @@ FAKE_INTERFACE = 'eth0'
class TestPlug(base.TestCase):
def setUp(self):
super(TestPlug, self).setUp()
self.mock_netifaces = mock.patch.object(plug, "netifaces").start()
self.test_plug = plug.Plug()
self.addCleanup(self.mock_netifaces.stop)
# Set up our fake interface
@ -48,7 +48,7 @@ class TestPlug(base.TestCase):
}
def test__interface_by_mac_case_insensitive(self):
interface = plug._interface_by_mac(FAKE_MAC_ADDRESS.upper())
interface = self.test_plug._interface_by_mac(FAKE_MAC_ADDRESS.upper())
self.assertEqual(FAKE_INTERFACE, interface)
@mock.patch.object(plug, "flask")
@ -63,7 +63,7 @@ class TestPlug(base.TestCase):
mock_pyroute2, mock_flask):
m = mock.mock_open()
with mock.patch('os.open'), mock.patch.object(os, 'fdopen', m):
plug.plug_vip(
self.test_plug.plug_vip(
vip=FAKE_IP_IPV4,
subnet_cidr=FAKE_CIDR_IPV4,
gateway=FAKE_GATEWAY_IPV4,
@ -87,7 +87,7 @@ class TestPlug(base.TestCase):
mock_pyroute2, mock_flask):
m = mock.mock_open()
with mock.patch('os.open'), mock.patch.object(os, 'fdopen', m):
plug.plug_vip(
self.test_plug.plug_vip(
vip=FAKE_IP_IPV6,
subnet_cidr=FAKE_CIDR_IPV6,
gateway=FAKE_GATEWAY_IPV6,
@ -111,7 +111,7 @@ class TestPlug(base.TestCase):
mock_pyroute2, mock_flask):
m = mock.mock_open()
with mock.patch('os.open'), mock.patch.object(os, 'fdopen', m):
plug.plug_vip(
self.test_plug.plug_vip(
vip="error",
subnet_cidr=FAKE_CIDR_IPV4,
gateway=FAKE_GATEWAY_IPV4,
@ -128,13 +128,16 @@ class TestPlug(base.TestCase):
'attrs': [['IFLA_ADDRESS', '123']]}]
# Interface is found in netns
self.assertTrue(plug._netns_interface_exists('123'))
self.assertTrue(self.test_plug._netns_interface_exists('123'))
# Interface is not found in netns
self.assertFalse(plug._netns_interface_exists('321'))
self.assertFalse(self.test_plug._netns_interface_exists('321'))
class TestPlugNetwork(base.TestCase):
def setUp(self):
super(TestPlugNetwork, self).setUp()
self.test_plug = plug.Plug()
def test__generate_network_file_text_static_ip(self):
netns_interface = 'eth1234'
@ -151,7 +154,8 @@ class TestPlugNetwork(base.TestCase):
{'destination': DEST1, 'nexthop': NEXTHOP},
{'destination': DEST2, 'nexthop': NEXTHOP}
]}]
text = plug._generate_network_file_text(netns_interface, fixed_ips)
text = self.test_plug._generate_network_file_text(netns_interface,
fixed_ips)
expected_text = (
'\n\n# Generated by Octavia agent\n'
'auto ' + netns_interface + '\n'