stackforge/libra
Code Issues Proposed changes

Merge "[WORKER] Support new GALERA protocol"

Browse Source
This commit is contained in:
Jenkins
2013-09-25 21:27:40 +00:00
committed by Gerrit Code Review
parent db1aec3c9c 684cb24611
commit db4508677e
4 changed files with 124 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
libra/tests/test_haproxy_driver.py
View File

@@ -27,22 +27,50 @@ class TestHAProxyDriver(testtools.TestCase):
self.assertEqual(self.driver._config[proto]['bind_address'], '0.0.0.0') self.assertEqual(self.driver._config[proto]['bind_address'], '0.0.0.0')
self.assertEqual(self.driver._config[proto]['bind_port'], 443) self.assertEqual(self.driver._config[proto]['bind_port'], 443)
proto = 'galera'
self.driver.add_protocol(proto, 3306)
self.assertIn(proto, self.driver._config)
self.assertEqual(self.driver._config[proto]['bind_address'], '0.0.0.0')
self.assertEqual(self.driver._config[proto]['bind_port'], 3306)
proto = 'tnetennba'
e = self.assertRaises(Exception, self.driver.add_protocol, proto, 99)
self.assertEqual("Unsupported protocol: %s" % proto, e.message)
def testAddGaleraRequiresPort(self):
e = self.assertRaises(Exception, self.driver.add_protocol, 'galera', None)
self.assertEqual("Port is required for this protocol.", e.message)
def testAddTCPRequiresPort(self): def testAddTCPRequiresPort(self):
e = self.assertRaises(Exception, self.driver.add_protocol, 'tcp', None) e = self.assertRaises(Exception, self.driver.add_protocol, 'tcp', None)
self.assertEqual("Port is required for TCP protocol.", e.message) self.assertEqual("Port is required for this protocol.", e.message)
def testAddServer(self): def testAddServer(self):
""" Test the HAProxy add_server() method """ """ Test the HAProxy add_server() method """
proto = 'http' proto = 'http'
self.driver.add_protocol(proto, None) self.driver.add_protocol(proto, None)
self.driver.add_server(proto, 100, '1.2.3.4', 7777) self.driver.add_server(proto, 100, '1.2.3.4', 7777)
self.driver.add_server(proto, 101, '5.6.7.8', 8888) self.driver.add_server(proto, 101, '5.6.7.8', 8888, 1, True)
self.driver.add_server(proto, 102, '2.3.4.5', 9999,
weight=2, backup=True)
self.assertIn(proto, self.driver._config) self.assertIn(proto, self.driver._config)
self.assertIn('servers', self.driver._config[proto]) self.assertIn('servers', self.driver._config[proto])
servers = self.driver._config[proto]['servers'] servers = self.driver._config[proto]['servers']
self.assertEqual(len(servers), 2) self.assertEqual(len(servers), 3)
self.assertEqual(servers[0], (100, '1.2.3.4', 7777, 1)) self.assertEqual(servers[0], (100, '1.2.3.4', 7777, 1, False))
self.assertEqual(servers[1], (101, '5.6.7.8', 8888, 1)) self.assertEqual(servers[1], (101, '5.6.7.8', 8888, 1, True))
self.assertEqual(servers[2], (102, '2.3.4.5', 9999, 2, True))
def testAddServerMultipleGaleraPrimaries(self):
proto = 'galera'
self.driver.add_protocol(proto, 33306)
self.driver.add_server(proto, 100, '1.2.3.4', 3306, backup=False)
self.driver.add_server(proto, 101, '1.2.3.5', 3306, backup=True)
e = self.assertRaises(Exception, self.driver.add_server,
proto, 101, '1.2.3.6', 3306, backup=False)
self.assertEqual(
"Galera protocol does not accept more than one non-backup node",
e.message)
def testSetAlgorithm(self): def testSetAlgorithm(self):
""" Test the HAProxy set_algorithm() method """ """ Test the HAProxy set_algorithm() method """
@@ -64,7 +92,7 @@ class TestHAProxyDriver(testtools.TestCase):
self.driver.add_server(proto, 100, '1.2.3.4', 7777, 10) self.driver.add_server(proto, 100, '1.2.3.4', 7777, 10)
servers = self.driver._config[proto]['servers'] servers = self.driver._config[proto]['servers']
self.assertEqual(len(servers), 1) self.assertEqual(len(servers), 1)
self.assertEqual(servers[0], (100, '1.2.3.4', 7777, 10)) self.assertEqual(servers[0], (100, '1.2.3.4', 7777, 10, False))
def testServerWeightStr(self): def testServerWeightStr(self):
""" Test setting string server weights """ """ Test setting string server weights """
@@ -73,7 +101,7 @@ class TestHAProxyDriver(testtools.TestCase):
self.driver.add_server(proto, 100, '1.2.3.4', 7777, "20") self.driver.add_server(proto, 100, '1.2.3.4', 7777, "20")
servers = self.driver._config[proto]['servers'] servers = self.driver._config[proto]['servers']
self.assertEqual(len(servers), 1) self.assertEqual(len(servers), 1)
self.assertEqual(servers[0], (100, '1.2.3.4', 7777, 20)) self.assertEqual(servers[0], (100, '1.2.3.4', 7777, 20, False))
def testServerWeightInvalid(self): def testServerWeightInvalid(self):
""" Test setting string server weights """ """ Test setting string server weights """

27
libra/worker/controller.py
View File

@@ -202,7 +202,11 @@ class LBaaSController(object):
return self.msg return self.msg
for lb_node in current_lb['nodes']: for lb_node in current_lb['nodes']:
port, address, node_id, weight = None, None, None, None port = None
address = None
node_id = None
weight = None
backup = False
if 'port' in lb_node: if 'port' in lb_node:
port = lb_node['port'] port = lb_node['port']
@@ -222,21 +226,30 @@ class LBaaSController(object):
if 'weight' in lb_node: if 'weight' in lb_node:
weight = lb_node['weight'] weight = lb_node['weight']
if 'backup' in lb_node and lb_node['backup'].lower() == 'true':
backup = True
try: try:
self.driver.add_server(current_lb['protocol'], self.driver.add_server(current_lb['protocol'],
node_id, node_id,
address, address,
port, port,
weight) weight,
backup)
except NotImplementedError: except NotImplementedError:
self.logger.error(
"Selected driver does not support adding a server."
)
lb_node['condition'] = self.NODE_ERR lb_node['condition'] = self.NODE_ERR
error = "Selected driver does not support adding a server"
self.logger.error(error)
self.msg[self.ERROR_FIELD] = error
self.msg[self.RESPONSE_FIELD] = self.RESPONSE_FAILURE
return self.msg
except Exception as e: except Exception as e:
self.logger.error("Failure trying adding server: %s, %s" %
(e.__class__, e))
lb_node['condition'] = self.NODE_ERR lb_node['condition'] = self.NODE_ERR
error = "Failure adding server %s: %s" % (node_id, e)
self.logger.error(error)
self.msg[self.ERROR_FIELD] = error
self.msg[self.RESPONSE_FIELD] = self.RESPONSE_FAILURE
return self.msg
else: else:
self.logger.debug("Added server: %s:%s" % (address, port)) self.logger.debug("Added server: %s:%s" % (address, port))
lb_node['condition'] = self.NODE_OK lb_node['condition'] = self.NODE_OK

2
libra/worker/drivers/base.py
View File

@@ -48,7 +48,7 @@ class LoadBalancerDriver(object):
""" Add a supported protocol and listening port for the instance. """ """ Add a supported protocol and listening port for the instance. """
raise NotImplementedError() raise NotImplementedError()
def add_server(self, protocol, host, port, weight): def add_server(self, protocol, host, port, weight, backup):
""" Add a server for the protocol for which we will proxy. """ """ Add a server for the protocol for which we will proxy. """
raise NotImplementedError() raise NotImplementedError()

95
libra/worker/drivers/haproxy/driver.py
View File

@@ -82,36 +82,39 @@ class HAProxyDriver(LoadBalancerDriver):
for proto in self._config: for proto in self._config:
protocfg = self._config[proto] protocfg = self._config[proto]
real_proto = proto
if proto == 'galera':
real_proto = 'tcp'
#------------------------ #------------------------
# Frontend configuration # Frontend configuration
#------------------------ #------------------------
output.append('frontend %s-in' % proto) output.append('frontend %s-in' % real_proto)
output.append(' mode %s' % proto) output.append(' mode %s' % real_proto)
output.append(' bind %s:%s' % (protocfg['bind_address'], output.append(' bind %s:%s' % (protocfg['bind_address'],
protocfg['bind_port'])) protocfg['bind_port']))
output.append(' default_backend %s-servers' % proto) output.append(' default_backend %s-servers' % real_proto)
# HTTP specific options for the frontend # HTTP specific options for the frontend
if proto == 'http': if real_proto == 'http':
output.append(' option httplog') output.append(' option httplog')
# TCP specific options for the frontend # TCP specific options for the frontend
elif proto == 'tcp': elif real_proto == 'tcp':
output.append(' option tcplog') output.append(' option tcplog')
#------------------------ #------------------------
# Backend configuration # Backend configuration
#------------------------ #------------------------
output.append('backend %s-servers' % proto) output.append('backend %s-servers' % real_proto)
output.append(' mode %s' % proto) output.append(' mode %s' % real_proto)
output.append(' balance %s' % protocfg['algorithm']) output.append(' balance %s' % protocfg['algorithm'])
# default healthcheck if none specified # default healthcheck if none specified
monitor = 'check inter 30s' monitor = 'check inter 30s'
# HTTP specific options for the backend # HTTP specific options for the backend
if proto == 'http': if real_proto == 'http':
output.append(' cookie SERVERID insert indirect') output.append(' cookie SERVERID insert indirect')
output.append(' option httpclose') output.append(' option httpclose')
output.append(' option forwardfor') output.append(' option forwardfor')
@@ -127,19 +130,40 @@ class HAProxyDriver(LoadBalancerDriver):
monitor = "check inter %ds rise %d fall %d" % ( monitor = "check inter %ds rise %d fall %d" % (
mon['delay'], mon['attempts'], mon['attempts']) mon['delay'], mon['attempts'], mon['attempts'])
for (node_id, addr, port, weight) in protocfg['servers']: for (node_id, addr, port, wt, bkup) in protocfg['servers']:
output.append(' server id-%s %s:%s cookie id-%s ' if bkup:
'weight %d %s' % output.append(
(node_id, addr, port, node_id, ' server id-%s %s:%s backup cookie id-%s'
weight, monitor)) ' weight %d %s' %
(node_id, addr, port, node_id, wt, monitor)
)
else:
output.append(
' server id-%s %s:%s cookie id-%s'
' weight %d %s' %
(node_id, addr, port, node_id, wt, monitor)
)
# TCP or Galera specific options for the backend
#
# The Galera protocol is a convenience option that lets us set
# our TCP options specifically for load balancing between Galera
# database nodes in a manner that helps avoid deadlocks. A main
# node is chosen which will act as the 'write' node, sending all
# updates to this one node.
# TCP specific options for the backend
else: else:
# Allow session stickiness for TCP connections. The 'size'
# value affects memory usage (about 50 bytes per entry). # No stick table for Galera protocol since we want to return to
output.append(' stick-table type ip size 200k expire 30m') # the main backend node once it is available after being down.
output.append(' stick store-request src') if proto == 'tcp':
output.append(' stick match src') # Allow session stickiness for TCP connections. The 'size'
# value affects memory usage (about 50 bytes per entry).
output.append(
' stick-table type ip size 200k expire 30m'
)
output.append(' stick store-request src')
output.append(' stick match src')
if 'monitor' in self._config[proto]: if 'monitor' in self._config[proto]:
mon = self._config[proto]['monitor'] mon = self._config[proto]['monitor']
@@ -152,9 +176,17 @@ class HAProxyDriver(LoadBalancerDriver):
monitor = "check inter %ds rise %d fall %d" % ( monitor = "check inter %ds rise %d fall %d" % (
mon['delay'], mon['attempts'], mon['attempts']) mon['delay'], mon['attempts'], mon['attempts'])
for (node_id, addr, port, weight) in protocfg['servers']: for (node_id, addr, port, wt, bkup) in protocfg['servers']:
output.append(' server id-%s %s:%s weight %d %s' % if bkup:
(node_id, addr, port, weight, monitor)) output.append(
' server id-%s %s:%s backup weight %d %s' %
(node_id, addr, port, wt, monitor)
)
else:
output.append(
' server id-%s %s:%s weight %d %s' %
(node_id, addr, port, wt, monitor)
)
return '\n'.join(output) + '\n' return '\n'.join(output) + '\n'
@@ -251,7 +283,7 @@ class HAProxyDriver(LoadBalancerDriver):
def add_protocol(self, protocol, port=None): def add_protocol(self, protocol, port=None):
proto = protocol.lower() proto = protocol.lower()
if proto not in ('tcp', 'http', 'health'): if proto not in ('tcp', 'http', 'galera'):
raise Exception("Unsupported protocol: %s" % protocol) raise Exception("Unsupported protocol: %s" % protocol)
if proto in self._config: if proto in self._config:
raise Exception("Protocol '%s' is already defined." % protocol) raise Exception("Protocol '%s' is already defined." % protocol)
@@ -259,14 +291,15 @@ class HAProxyDriver(LoadBalancerDriver):
self._config[proto] = dict() self._config[proto] = dict()
if port is None: if port is None:
if proto == 'tcp': if proto in ('tcp', 'galera'):
raise Exception('Port is required for TCP protocol.') raise Exception('Port is required for this protocol.')
elif proto == 'http': elif proto == 'http':
self._bind(proto, '0.0.0.0', 80) self._bind(proto, '0.0.0.0', 80)
else: else:
self._bind(proto, '0.0.0.0', port) self._bind(proto, '0.0.0.0', port)
def add_server(self, protocol, node_id, host, port, weight=1): def add_server(self, protocol, node_id, host, port,
weight=1, backup=False):
proto = protocol.lower() proto = protocol.lower()
if weight is None: if weight is None:
weight = 1 weight = 1
@@ -281,7 +314,15 @@ class HAProxyDriver(LoadBalancerDriver):
if 'servers' not in self._config[proto]: if 'servers' not in self._config[proto]:
self._config[proto]['servers'] = [] self._config[proto]['servers'] = []
self._config[proto]['servers'].append((node_id, host, port, weight))
if proto == 'galera':
for (n, h, p, w, b) in self._config[proto]['servers']:
if b is False and backup is False:
raise Exception("Galera protocol does not accept more"
" than one non-backup node")
self._config[proto]['servers'].append((node_id, host, port,
weight, backup))
def set_algorithm(self, protocol, algo): def set_algorithm(self, protocol, algo):
proto = protocol.lower() proto = protocol.lower()