Adds keepalived based VRRPIPManager

This adds a new IP manager driver for configuring addresses
and routes via keepalived instead of directly.  It used when
the logical resource is configured to be highly-available,
according to configuration pushed by the orchestrator.

We rely on a 'ha_resource' flag attached to the main config
dict to enable it, and use specific HA config about peers and
cluster priority contained in the 'ha_config' section of the
main config.

The resulting keepalived cluster contains a VRRP instance for
each interface, with the exception of the management interface.

Partially-implements: blueprint appliance-ha

Change-Id: I5ababa41d65642b00f6b808197af9b2a59ebc67a
This commit is contained in:
Adam Gandelman 2016-02-09 17:37:58 -08:00 committed by mark mcclain
parent f02b1f48ce
commit 02383adf64
16 changed files with 585 additions and 32 deletions

View File

@ -11,6 +11,8 @@
- ntp
- tcpdump
- vim
- keepalived
- conntrackd
- name: latest bash (CVE-2014-6271)
apt: name=bash state=latest install_recommends=no

View File

@ -44,5 +44,6 @@ def main():
# app.config.from_object('astara_router.config.Default')
# manager.state_path = app.config['STATE_PATH']
app.run(host=manager.management_address(ensure_configuration=True),
addr = str(manager.ip_mgr.get_interfaces()[0].addresses[0])
app.run(host=addr,
port=5000)

View File

@ -151,4 +151,9 @@ class ARPManager(base.Manager):
:type ip: str
:param ip: IP address to search for in the ARP table.
"""
self.sudo('-d', ip)
try:
self.sudo('-d', ip)
except:
# We may be attempting to delete from ARP for interfaces which
# are managed by keepalived and do not yet have addresses
pass

View File

@ -22,7 +22,7 @@ import re
import netaddr
from astara_router import models
from astara_router.drivers import base
from astara_router.drivers import base, keepalived
from astara_router import utils
LOG = logging.getLogger(__name__)
@ -555,3 +555,54 @@ def _parse_lladdr(line):
"""
tokens = line.split()
return tokens[1]
class VRRPIPManager(IPManager):
def __init__(self, root_helper='sudo astara-rootwrap /etc/rootwrap.conf'):
super(VRRPIPManager, self).__init__(root_helper)
self.keepalived = keepalived.KeepalivedManager(root_helper)
self.ensure_mapping()
def set_peers(self, peers):
self.keepalived.peers = peers
def set_priority(self, priority):
self.keepalived.set_priority(priority)
def update_interfaces(self, interfaces):
for interface in interfaces:
if interface.management:
# the mgt interface is not managed as a vip, but
# it used for keepalived mcast cluster comms
self.update_interface(interface)
self.keepalived.set_management_address(
address=interface.first_v4 or interface.first_v6)
else:
self.up(interface)
self.keepalived.add_vrrp_instance(
interface=self.generic_to_host(interface.ifname),
addresses=interface.all_addresses)
def _set_default_gateway(self, gateway_ip, ifname):
"""
Sets the default gateway.
:param gateway_ip: the IP address to set as the default gateway_ip
:type gateway_ip: netaddr.IPAddress
:param ifname: the interface name (in our case, of the external
network)
:type ifname: str
"""
version = 4
if gateway_ip.version == 6:
version = 6
self.keepalived.set_default_gateway(
ip_version=version, gateway_ip=gateway_ip,
interface=self.generic_to_host(ifname))
def update_host_routes(self, config, cache):
# XXX TODO
return
def reload(self):
self.keepalived.reload()

View File

@ -0,0 +1,35 @@
{%- for instance in vrrp_instances %}
vrrp_instance {{ instance.name }} {
native_ipv6
state {{ instance.state }}
interface {{ instance.interface }}
virtual_router_id {{ instance.vrrp_id }}
priority {{ priority }}
garp_master_delay {{ instance.garp_master_delay }}
unicast_src_ip {{ instance.unicast_src_ip }}
unicast_peer {
{%- for peer in peers %}
{{ peer }}
{%- endfor %}
}
{%- if instance.vips %}
virtual_ipaddress {
{{ instance.vips[0].address }} dev {{ instance.vips[0].interface }}
}
virtual_ipaddress_excluded {
{%- for vip in instance.vips[1:] %}
{{ vip.address }} dev {{ vip.interface }}
{%- endfor %}
}
{%- endif %}
{%- if instance.routes %}
virtual_routes {
{%- for route in instance.routes %}
{{ route.destination }} via {{ route.gateway }} dev {{ instance.interface }}
{%- endfor %}
}
{%- endif %}
}
{%- endfor %}

View File

@ -0,0 +1,156 @@
# Copyright (c) 2016 Akanda, Inc. All Rights Reserved.
#
# 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
from astara_router.drivers import base
from astara_router import utils
class KeepalivedVipAddress(object):
"""A virtual address entry of a keepalived configuration."""
def __init__(self, address, interface):
self.address = address
self.interface = interface
def __eq__(self, other):
return (isinstance(other, KeepalivedVipAddress) and
self.address.ip == other.address.ip)
class KeepalivedRoute(object):
"""A virtual route entry in keepalived instance configuration"""
def __init__(self, destination, gateway):
self.destination = destination
self.gateway = gateway
def __eq__(self, other):
return (
isinstance(other, KeepalivedRoute) and
(self.destination, self.gateway) ==
(other.destination, other.gateway)
)
class KeepalivedInstance(object):
def __init__(self, interface, unicast_src_ip, vrrp_id, state='BACKUP',
garp_master_delay=60):
self.interface = interface
self.vrrp_id = vrrp_id
self.unicast_src_ip = unicast_src_ip
self.name = 'astara_vrrp_' + interface
self.state = state
self.garp_master_delay = 60
self.vips = []
self.routes = []
def add_vip(self, address):
vip = KeepalivedVipAddress(address, self.interface)
if vip not in self.vips:
self.vips.append(vip)
def add_route(self, destination, gateway):
route = KeepalivedRoute(destination, gateway)
if route not in self.routes:
self.routes.append(route)
class KeepalivedManager(base.Manager):
CONFIG_FILE_TEMPLATE = os.path.join(
os.path.dirname(__file__), 'keepalived.conf.template')
# Debian defaults
CONFIG_FILE = '/etc/keepalived/keepalived.conf'
PID_FILE = '/var/run/keepalived.pid'
EXECUTABLE = 'service'
def __init__(self, root_helper='sudo astara-rootwrap /etc/rootwrap.conf'):
super(KeepalivedManager, self).__init__(root_helper)
self.instances = {}
self.unicast_src_ip = None
self.config_tmpl = utils.load_template(self.CONFIG_FILE_TEMPLATE)
self.peers = []
self.priority = 0
self._last_config_hash = None
def set_management_address(self, address):
"""Specify the address used for keepalived cluster communication"""
self.unicast_src_ip = address
for instance in self.instances.values():
instance.unicast_src_ip = address
def _get_instance(self, interface):
if interface in self.instances:
return self.instances[interface]
vrrp_id = len(self.instances) + 1
self.instances[interface] = KeepalivedInstance(
interface, self.unicast_src_ip, vrrp_id=vrrp_id)
return self.instances[interface]
def _is_running(self):
if not os.path.isfile(self.PID_FILE):
return False
pid = open(self.PID_FILE).read().strip()
proc_cmd = os.path.join('/proc', pid, 'cmdline')
if not os.path.isfile(proc_cmd):
return False
if 'keepalived' not in open(proc_cmd).read():
return False
return True
def add_vrrp_instance(self, interface, addresses):
instance = self._get_instance(interface)
[instance.add_vip(addr) for addr in addresses]
def config(self):
return self.config_tmpl.render(
priority=self.priority,
peers=self.peers,
vrrp_instances=self.instances.values())
def reload(self):
try:
last_config_hash = utils.hash_file(self.CONFIG_FILE)
except IOError:
last_config_hash = None
utils.replace_file('/tmp/keepalived.conf', self.config())
utils.execute(
['mv', '/tmp/keepalived.conf', '/etc/keepalived/keepalived.conf'],
self.root_helper)
if utils.hash_file(self.CONFIG_FILE) == last_config_hash:
return
if self._is_running():
self.sudo('keepalived', 'reload')
else:
self.sudo('keepalived', 'restart')
def set_default_gateway(self, ip_version, gateway_ip, interface):
instance = self._get_instance(interface)
if ip_version == 6:
default = 'default6'
else:
default = 'default'
instance.add_route(default, gateway_ip)
def set_priority(self, priority):
self.priority = priority

View File

@ -12,18 +12,11 @@
# License for the specific language governing permissions and limitations
# under the License.
import os
import jinja2
from astara_router.drivers import base
from astara_router.utils import execute
class NginxTemplateNotFound(Exception):
# TODO(adam_g): These should return 50x errors and not logged
# exceptions.
pass
from astara_router.utils import execute, load_template
class NginxLB(base.Manager):
@ -35,25 +28,15 @@ class NginxLB(base.Manager):
def __init__(self, root_helper='sudo astara-rootwrap /etc/rootwrap.conf'):
"""
Initializes DHCPManager class.
Initializes NginxLB class.
:type root_helper: str
:param root_helper: System utility used to gain escalate privileges.
"""
super(NginxLB, self).__init__(root_helper)
self._load_template()
def _load_template(self):
if not os.path.exists(self.CONFIG_FILE_TEMPLATE):
raise NginxTemplateNotFound(
'NGINX Config template not found @ %s' %
self.CONFIG_FILE_TEMPLATE
)
self.config_tmpl = jinja2.Template(
open(self.CONFIG_FILE_TEMPLATE).read())
self.config_tmpl = load_template(self.CONFIG_FILE_TEMPLATE)
def _render_config_template(self, path, config):
self._load_template()
with open(path, 'w') as out:
out.write(
self.config_tmpl.render(loadbalancer=config)

View File

@ -28,8 +28,35 @@ class ServiceManagerBase(object):
def __init__(self, state_path='.'):
self._config = None
self.state_path = os.path.abspath(state_path)
self.ip_mgr = ip.IPManager()
self.ip_mgr.ensure_mapping()
self._vrrp_ip_mgr = None
self._reload_callbacks = []
@property
def ip_mgr(self):
ip_mgr = ip.IPManager()
ip_mgr.ensure_mapping()
if not self._config:
# we do not yet have config, so use standard ip manager for
# ensuring initial intrefaces
return ip_mgr
if self._config and self._config.ha:
if not self._vrrp_ip_mgr:
self._vrrp_ip_mgr = ip.VRRPIPManager()
self._reload_callbacks.append(self._vrrp_ip_mgr.reload)
# peers and prio can change and be updated via config, need to
# ensure the vrrp manager is up to date every access.
self._vrrp_ip_mgr.set_peers(
self._config.ha_config.get('peers', []))
self._vrrp_ip_mgr.set_priority(
self._config.ha_config.get('priority', 0))
return self._vrrp_ip_mgr
else:
# we may not yet have config, so use standard ip manager for
# ensuring initial interfaces
return ip_mgr
@property
def config(self):
@ -48,8 +75,17 @@ class ServiceManagerBase(object):
return
for network in self._config.networks:
self.ip_mgr.disable_duplicate_address_detection(network)
self.ip_mgr.update_interfaces(self._config.interfaces)
def reload_config(self):
"""Calls any post-config reload callbacks to reload services
Required for things like keepalived, which gets its config built
by multiple drivers, in order to avoid unncessary restarts.
"""
[cb() for cb in self._reload_callbacks]
class SystemManager(ServiceManagerBase):
def __init__(self, state_path='.'):
@ -77,6 +113,7 @@ class RouterManager(ServiceManagerBase):
self.update_firewall()
self.update_routes(cache)
self.update_arp()
self.reload_config()
def update_dhcp(self):
mgr = dnsmasq.DHCPManager()
@ -106,9 +143,8 @@ class RouterManager(ServiceManagerBase):
mgr.restart()
def update_routes(self, cache):
mgr = ip.IPManager()
mgr.update_default_gateway(self._config)
mgr.update_host_routes(self._config, cache)
self.ip_mgr.update_default_gateway(self._config)
self.ip_mgr.update_host_routes(self._config, cache)
def update_arp(self):
mgr = arp.ARPManager()

View File

@ -38,7 +38,7 @@ class Interface(ModelBase):
"""
def __init__(self, ifname=None, addresses=[], groups=None, flags=None,
lladdr=None, mtu=None, media=None,
description=None, **extra_params):
description=None, management=False, **extra_params):
self.ifname = ifname
self.description = description
self.addresses = addresses
@ -49,6 +49,7 @@ class Interface(ModelBase):
self.media = media
self.extra_params = extra_params
self._aliases = []
self.management = management
def __repr__(self):
return '<Interface: %s %s>' % (self.ifname,
@ -375,7 +376,7 @@ class Network(ModelBase):
v4_conf_service=SERVICE_STATIC,
v6_conf_service=SERVICE_STATIC,
address_allocations=None,
subnets=None):
subnets=None, ha=False):
self.id = id_
self.interface = interface
self.name = name
@ -385,6 +386,7 @@ class Network(ModelBase):
self.address_allocations = address_allocations or []
self.subnets = subnets or []
self.floating_ips = []
self.ha = ha
@property
def is_tenant_network(self):
@ -462,6 +464,11 @@ class Network(ModelBase):
if missing:
raise ValueError('Missing required data: %s.' % missing)
if d.get('network_type') == cls.TYPE_MANAGEMENT:
d['interface']['management'] = True
else:
d['interface']['management'] = False
return cls(
d['network_id'],
interface=Interface.from_dict(d['interface']),
@ -471,7 +478,8 @@ class Network(ModelBase):
v4_conf_service=d.get('v4_conf_service', cls.SERVICE_STATIC),
address_allocations=[
Allocation.from_dict(a) for a in d.get('allocations', [])],
subnets=[Subnet.from_dict(s) for s in d.get('subnets', [])])
subnets=[Subnet.from_dict(s) for s in d.get('subnets', [])],
ha=d.get('ha', False))
class LoadBalancer(ModelBase):
@ -663,6 +671,8 @@ class SystemConfiguration(ModelBase):
self.hostname = conf_dict.get('hostname')
self.networks = [
Network.from_dict(n) for n in conf_dict.get('networks', [])]
self.ha = conf_dict.get('ha_resource', False)
self.ha_config = conf_dict.get('ha_config', {})
def validate(self):
# TODO: Improve this interface, it currently sucks.

View File

@ -15,6 +15,7 @@
# under the License.
import functools
import hashlib
import json
import os
import shlex
@ -22,6 +23,7 @@ import subprocess
import tempfile
import flask
import jinja2
import netaddr
from astara_router import models
@ -30,6 +32,10 @@ DEFAULT_ENABLED_SERVICES = ['router']
VALID_SERVICES = ['router', 'loadbalancer']
class TemplateNotFound(Exception):
pass
def execute(args, root_helper=None):
if root_helper:
cmd = shlex.split(root_helper) + args
@ -104,3 +110,17 @@ def blueprint_factory(name):
blueprint_name = "_".join(name_parts)
url_prefix = "/" + "/".join(name_parts)
return flask.Blueprint(blueprint_name, name, url_prefix=url_prefix)
def load_template(template_file):
if not os.path.exists(template_file):
raise TemplateNotFound(
'Config template not found @ %s' % template_file)
return jinja2.Template(open(template_file).read())
def hash_file(path):
h = hashlib.md5()
with open(path, 'rb') as _in:
h.update(_in.read())
return h.hexdigest()

View File

@ -23,6 +23,9 @@ ip: IpFilter, ip, root
sysctl: CommandFilter, sysctl, root
conntrack: CommandFilter, conntrack, root
# astara_router/drivers/keepalived.py:
mv_keepalived: RegExpFilter, mv, root, mv, /tmp/keepalived\.conf, /etc/keepalived/keepalived\.conf
# astara_router/drivers/ping.py:
ping: CommandFilter, ping, root
ping6: CommandFilter, ping6, root

View File

@ -0,0 +1,5 @@
---
features:
- The appliance is now built with keepalived installed and supports receiving
cluster configuration from ``astara-orchestartor``, which allows pairs of
appliance VMs to cluster among themselves, providing HA routers.

View File

@ -8,3 +8,4 @@ eventlet!=0.18.3,>=0.18.2 # MIT
requests!=2.9.0,>=2.8.1 # Apache-2.0
greenlet>=0.3.2 # MIT
oslo.rootwrap>=2.0.0 # Apache-2.0
Jinja2>=2.8 # BSD License (3 clause)

View File

@ -22,8 +22,11 @@ from unittest2 import TestCase
import mock
import netaddr
from test.unit import fakes
from astara_router import models
from astara_router.drivers import ip
from astara_router.drivers.keepalived import KeepalivedManager
SAMPLE_OUTPUT = """1: lo: <LOOPBACK,UP,LOWER_UP> mtu 16436 qdisc noqueue state UNKNOWN
link/loopback 00:00:00:00:00:00 brd 00:00:00:00:00:00
@ -511,3 +514,81 @@ class ParseTestCase(TestCase):
str(retval),
str(netaddr.IPNetwork('fe80::f816:3eff:fe7a:d864/64'))
)
class TestVRRPIPManager(TestCase):
def setUp(self):
super(TestVRRPIPManager, self).setUp()
self.fake_keepalived = mock.Mock(
spec=KeepalivedManager)
p = 'astara_router.drivers.keepalived.KeepalivedManager'
with mock.patch(p) as ka:
ka.return_value = self.fake_keepalived
self.mgr = ip.VRRPIPManager()
@mock.patch.object(ip.VRRPIPManager, 'ensure_mapping')
@mock.patch('astara_router.drivers.keepalived.KeepalivedManager')
def test_init(self, fake_keepalived, fake_ensure_map):
mgr = ip.VRRPIPManager()
self.assertTrue(fake_ensure_map.called)
fake_keepalived.return_value = 'fake_keepalived'
self.assertTrue(fake_keepalived.called)
mgr.keepalived = 'fake_keepalived'
def test_set_peers(self):
self.mgr.set_peers(['foo', 'bar'])
self.assertEqual(
self.mgr.keepalived.peers, ['foo', 'bar'])
def test_set_prio(self):
self.mgr.set_priority(100)
self.mgr.keepalived.set_priority.assert_called_with(100)
@mock.patch.object(ip.VRRPIPManager, 'generic_to_host')
@mock.patch.object(ip.VRRPIPManager, 'update_interface')
@mock.patch.object(ip.VRRPIPManager, 'up')
def test_update_interfaces(self, fake_up, fake_update_int, fake_gth):
interface = fakes.fake_interface()
mgt_interface = fakes.fake_mgt_interface()
fake_gth.return_value = 'eth1'
self.mgr.update_interfaces(
interfaces=[interface, mgt_interface],
)
self.assertEqual(len(fake_update_int.call_args_list), 1)
fake_update_int.assert_called_with(mgt_interface)
self.mgr.keepalived.set_management_address.assert_called_with(
address=netaddr.IPAddress(mgt_interface.addresses[0]))
self.assertEqual(len(fake_up.call_args_list), 1)
fake_up.assert_called_with(interface)
self.mgr.keepalived.add_vrrp_instance.assert_called_with(
addresses=[netaddr.IPNetwork(interface.addresses[0])],
interface='eth1')
def test_reload(self):
self.mgr.reload()
self.assertTrue(self.mgr.keepalived.reload.called)
@mock.patch.object(ip.VRRPIPManager, 'generic_to_host')
def test__set_default_gateway_v4(self, fake_gth):
fake_gth.return_value = 'eth0'
ip = netaddr.IPAddress('10.0.0.1')
self.mgr._set_default_gateway(gateway_ip=ip, ifname='ge0')
self.mgr.keepalived.set_default_gateway.assert_called_with(
ip_version=4,
gateway_ip=netaddr.IPAddress('10.0.0.1'),
interface='eth0',
)
@mock.patch.object(ip.VRRPIPManager, 'generic_to_host')
def test__set_default_gateway_v6(self, fake_gth):
fake_gth.return_value = 'eth0'
v6_ip = 'fdca:3ba5:a17a:acda:f816:3eff:fe5d:84'
ip = netaddr.IPAddress(v6_ip)
self.mgr._set_default_gateway(gateway_ip=ip, ifname='ge0')
self.mgr.keepalived.set_default_gateway.assert_called_with(
ip_version=6,
gateway_ip=netaddr.IPAddress(v6_ip),
interface='eth0',
)

View File

@ -0,0 +1,141 @@
# Copyright 2016 Akanda, Inc.
#
# Author: Akanda, Inc.
#
# 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.
from unittest2 import TestCase
import mock
import netaddr
from astara_router.drivers import keepalived
class KeepalivedVipAddressTestCase(TestCase):
def test_vip_address(self):
addr = netaddr.IPNetwork('10.0.0.1/32')
vip = keepalived.KeepalivedVipAddress(
address=addr, interface='eth0')
self.assertEqual(vip.address, addr)
self.assertEqual(vip.interface, 'eth0')
def test_vip_address_equal(self):
addr = netaddr.IPNetwork('10.0.0.1/32')
vip1 = keepalived.KeepalivedVipAddress(
address=addr, interface='eth0')
addr = netaddr.IPNetwork('10.0.0.1/32')
vip2 = keepalived.KeepalivedVipAddress(
address=addr, interface='eth0')
self.assertTrue(vip1 == vip2)
def test_vip_address_not_equal(self):
addr = netaddr.IPNetwork('10.0.0.1/32')
vip1 = keepalived.KeepalivedVipAddress(
address=addr, interface='eth0')
addr = netaddr.IPNetwork('10.0.0.21/32')
vip2 = keepalived.KeepalivedVipAddress(
address=addr, interface='eth0')
self.assertFalse(vip1 == vip2)
class KeepalivedRouteTestCase(TestCase):
def test_keepalived_route(self):
route = keepalived.KeepalivedRoute(
destination='10.0.0.0/24',
gateway='10.0.0.1')
self.assertEqual(route.destination, '10.0.0.0/24')
self.assertEqual(route.gateway, '10.0.0.1')
def test_keepalived_route_equal(self):
route1 = keepalived.KeepalivedRoute(
destination='10.0.0.0/24',
gateway='10.0.0.1')
route2 = keepalived.KeepalivedRoute(
destination='10.0.0.0/24',
gateway='10.0.0.1')
self.assertTrue(route1 == route2)
def test_keepalived_route_not_equal(self):
route1 = keepalived.KeepalivedRoute(
destination='10.0.0.0/24',
gateway='10.0.0.1')
route2 = keepalived.KeepalivedRoute(
destination='10.0.0.0/24',
gateway='10.0.0.2')
self.assertFalse(route1 == route2)
class KeepalivedInstanceTestCase(TestCase):
def setUp(self):
self.instance = keepalived.KeepalivedInstance(
interface='eth0',
vrrp_id=1,
unicast_src_ip='10.0.0.1')
def test_init(self):
self.assertEqual(self.instance.interface, 'eth0')
self.assertEqual(self.instance.vrrp_id, 1)
self.assertEqual(self.instance.unicast_src_ip, '10.0.0.1')
self.assertEqual(self.instance.name, 'astara_vrrp_eth0')
@mock.patch.object(keepalived, 'KeepalivedVipAddress')
def test_add_vip(self, fake_vip):
addr = netaddr.IPNetwork('10.0.0.1/32')
fake_vip.return_value = 'fake_vip'
self.instance.add_vip(addr)
self.assertIn('fake_vip', self.instance.vips)
fake_vip.assert_called_with(addr, self.instance.interface)
@mock.patch.object(keepalived, 'KeepalivedRoute')
def test_add_route(self, fake_route):
fake_route.return_value = 'fake_route'
self.instance.add_route('10.0.0.0/24', '10.0.0.1')
self.assertIn('fake_route', self.instance.routes)
fake_route.assert_called_with('10.0.0.0/24', '10.0.0.1')
class KeepalivedManagerTestCase(TestCase):
def setUp(self):
super(KeepalivedManagerTestCase, self).setUp()
self.fake_instance = mock.Mock(
spec=keepalived.KeepalivedInstance, name='fake_instance')
self.get_instance_p = mock.patch.object(
keepalived.KeepalivedManager, '_get_instance')
self.fake_get_instance = self.get_instance_p.start()
self.fake_get_instance.return_value = self.fake_instance
self.addCleanup(self.get_instance_p.stop)
self.mgr = keepalived.KeepalivedManager()
self.mgr.instances = {
'eth0': self.fake_instance
}
def test_set_management_address(self):
self.mgr.set_management_address('10.0.0.1')
self.assertEqual(self.fake_instance.unicast_src_ip, '10.0.0.1')
def test_set_default_gateway(self):
self.mgr.set_default_gateway(
ip_version=4, gateway_ip='10.0.0.1', interface='eth0')
self.fake_instance.add_route.assert_called_with(
'default', '10.0.0.1')
def test_set_default_gateway_v6(self):
ip = 'fdca:3ba5:a17a:acda:f816:3eff:fe5d:84'
self.mgr.set_default_gateway(
ip_version=6, gateway_ip=ip, interface='eth0')
self.fake_instance.add_route.assert_called_with(
'default6', ip)
def test_set_priority(self):
self.mgr.set_priority(60)
self.assertEqual(self.mgr.priority, 60)

View File

@ -1,4 +1,7 @@
from copy import copy
import netaddr
from astara_router import models
FAKE_SYSTEM_DICT = {
@ -135,3 +138,23 @@ def fake_loadbalancer_dict(listener=False, pool=False, members=False):
lb_dict['listeners'][0]['default_pool']['members'] = \
[copy(FAKE_MEMBER_DICT)]
return lb_dict
def _fake_interface(ifname, addresses=None, management=False):
addresses = addresses or ['10.0.0.1']
return models.Interface(
ifname=ifname,
description='fake_interface',
addresses=[netaddr.IPAddress(addr) for addr in addresses],
management=management,
)
def fake_interface(ifname='ge1', addresses=None):
return _fake_interface(
ifname=ifname, addresses=(addresses or ['10.0.0.1']), management=False)
def fake_mgt_interface(ifname='ge0', addresses=None):
return _fake_interface(
ifname=ifname, addresses=(addresses or ['11.0.0.1']), management=True)