Implement UDP heartbeat sender and receiver

Used binary compressed encoding of json dumped object. To reduce
the size needed to send heart beats incase some stats objects
start getting sent later on. Also used sha256 instead of sha1
with hmac.

Co-Authored-By: Michael Johnson <johnsomor@gmail.com>
Co-Authored-By: German Eichberger <german.eichbeger@hp.com>
Co-Authored-By: Carlos Garza <carlos.garza@rackspace.com>
Partially implements: health-manager
Change-Id: I932c693101b94c9132e1741291610508876eab43

bin/create_certificates.sh

@@ -13,7 +13,7 @@
 #
 #Example for client use:
 #
-#curl -k -v --key client.key --cacert ca_01.pem --cert client.pem https://0.0.0.0:8443/
+#curl -k -v --key client.key --cacert ca_01.pem --cert client.pem https://0.0.0.0:9443/
 #
 #
 #Notes:

devstack/plugin.sh

@@ -67,6 +67,8 @@ function octavia_configure {
     iniset $OCTAVIA_CONF controller_worker compute_driver compute_nova_driver
     iniset $OCTAVIA_CONF controller_worker network_driver allowed_address_pairs_driver
 
+    iniset $OCTAVIA_CONF health_manager heartbeat_key ${OCTAVIA_HEALTH_KEY}
+
     iniset $OCTAVIA_CONF DEFAULT api_handler queue_producer
 
     iniset $OCTAVIA_CONF oslo_messaging_rabbit rabbit_port 5672
@@ -118,7 +120,7 @@ function build_mgmt_network {
     neutron security-group-create lb-mgmt-sec-grp
     neutron security-group-rule-create --protocol icmp lb-mgmt-sec-grp
     neutron security-group-rule-create --protocol tcp --port-range-min 22 --port-range-max 22 lb-mgmt-sec-grp
-    neutron security-group-rule-create --protocol tcp --port-range-min 8443 --port-range-max 8443 lb-mgmt-sec-grp
+    neutron security-group-rule-create --protocol tcp --port-range-min 9443 --port-range-max 9443 lb-mgmt-sec-grp
 
     OCTAVIA_MGMT_SEC_GRP_ID=$(nova secgroup-list | awk ' / lb-mgmt-sec-grp / {print $2}')
     iniset ${OCTAVIA_CONF} controller_worker amp_secgroup_list ${OCTAVIA_MGMT_SEC_GRP_ID}

devstack/settings

@@ -34,6 +34,8 @@ OCTAVIA_AMP_ACTIVE_RETRIES=${OCTAVIA_AMP_ACTIVE_RETRIES:-"500"}
 OCTAVIA_AMP_IMAGE_NAME=${OCTAVIA_AMP_IMAGE_NAME:-"amphora-x64-haproxy"}
 OCTAVIA_AMP_IMAGE_FILE=${OCTAVIA_AMP_IMAGE_FILE:-${OCTAVIA_DIR}/diskimage-create/${OCTAVIA_AMP_IMAGE_NAME}.qcow2}
 
+OCTAVIA_HEALTH_KEY=${OCTAVIA_HEALTH_KEY:-"insecure"}
+
 OCTAVIA_API_BINARY=${OCTAVIA_API_BINARY:-${OCTAVIA_BIN_DIR}/octavia-api}
 OCTAVIA_CONSUMER_BINARY=${OCTAVIA_CONSUMER_BINARY:-${OCTAVIA_BIN_DIR}/octavia-worker}
 OCTAVIA_HOUSEKEEPER_BINARY=${OCTAVIA_HOUSEKEEPER_BINARY:-${OCTAVIA_BIN_DIR}/octavia-housekeeping}

elements/amphora-agent/install.d/75-run_setup_install

@@ -5,9 +5,8 @@ install-packages libffi-dev libssl-dev
 cd /opt/amphora-agent/
 pip install -r requirements.txt
 python setup.py install
-cp etc/init/octavia-agent.conf /etc/init/
+cp etc/init/amphora-agent.conf /etc/init/
 mkdir /etc/octavia
-cp etc/octavia.conf /etc/octavia
 # we assume certs, etc will come in through the config drive
 mkdir /etc/octavia/certs
 mkdir /var/lib/octavia

etc/init/amphora-agent.conf

@@ -5,4 +5,4 @@ start on startup
 respawn
 respawn limit 2 2
 
-exec amphora-agent --config-file /etc/octavia/octavia.conf
+exec amphora-agent --config-file /etc/octavia/amphora-agent.conf

etc/octavia.conf

@@ -34,9 +34,17 @@
 # configuration file.
 
 [health_manager]
+# bind_ip = 0.0.0.0
+# bind_port = 5555
+# controller_ip_port_list example: 127.0.0.1:5555, 127.0.0.1:5555
+# controller_ip_port_list =
 # failover_threads = 10
-# interval = 3
-# heartbeat_timeout = 10
+# status_update_threads = 50
+# heartbeat_interval = 10
+# heartbeat_key =
+# heartbeat_timeout = 60
+# health_check_interval = 3
+# sock_rlimit = 0
 
 [keystone_authtoken]
 # auth_uri = https://localhost:5000/v3
@@ -89,8 +97,6 @@
 # respawn_interval = 2
 # Change for production to a ram drive
 # haproxy_cert_dir = /tmp
-# agent_server_cert = /etc/octavia/certs/server.pem
-# agent_server_ca = /etc/octavia/certs/client_ca.pem
 
 [controller_worker]
 # amp_active_retries = 10

octavia/amphorae/backends/agent/agent_jinja_cfg.py

@@ -0,0 +1,57 @@
+# Copyright 2015 Hewlett-Packard Development Company, L.P.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import os
+
+import jinja2
+
+from octavia.common.config import cfg
+from octavia.common import constants
+
+CONF = cfg.CONF
+CONF.import_group('amphora_agent', 'octavia.common.config')
+CONF.import_group('haproxy_amphora', 'octavia.common.config')
+CONF.import_group('health_manager', 'octavia.common.config')
+
+TEMPLATES_DIR = (os.path.dirname(os.path.realpath(__file__)) +
+                 constants.AGENT_API_TEMPLATES + '/')
+
+
+class AgentJinjaTemplater(object):
+
+    def __init__(self):
+        template_loader = jinja2.FileSystemLoader(searchpath=os.path.dirname(
+            TEMPLATES_DIR))
+        jinja_env = jinja2.Environment(loader=template_loader)
+        self.agent_template = jinja_env.get_template(
+            constants.AGENT_CONF_TEMPLATE)
+
+    def build_agent_config(self, amphora_id):
+        return self.agent_template.render(
+            {'agent_server_ca': CONF.amphora_agent.agent_server_ca,
+             'agent_server_cert': CONF.amphora_agent.agent_server_cert,
+             'agent_server_network_dir':
+                 CONF.amphora_agent.agent_server_network_dir,
+             'amphora_id': amphora_id,
+             'base_cert_dir': CONF.haproxy_amphora.base_cert_dir,
+             'base_path': CONF.haproxy_amphora.base_path,
+             'bind_host': CONF.haproxy_amphora.bind_host,
+             'bind_port': CONF.haproxy_amphora.bind_port,
+             'controller_list': CONF.health_manager.controller_ip_port_list,
+             'debug': CONF.debug,
+             'haproxy_cmd': CONF.haproxy_amphora.haproxy_cmd,
+             'heartbeat_interval': CONF.health_manager.heartbeat_interval,
+             'heartbeat_key': CONF.health_manager.heartbeat_key,
+             'respawn_count': CONF.haproxy_amphora.respawn_count,
+             'respawn_interval': CONF.haproxy_amphora.respawn_interval})

octavia/amphorae/backends/agent/api_server/certificate_update.py

@@ -22,17 +22,17 @@ MODE_OWNER = 0o600
 BUFFER = 1024
 
 CONF = cfg.CONF
-CONF.import_group('haproxy_amphora', 'octavia.common.config')
+CONF.import_group('amphora_agent', 'octavia.common.config')
 
 
 def upload_server_cert():
     stream = flask.request.stream
-    with open(CONF.haproxy_amphora.agent_server_cert, 'w') as crt_file:
+    with open(CONF.amphora_agent.agent_server_cert, 'w') as crt_file:
         b = stream.read(BUFFER)
         while b:
             crt_file.write(b)
             b = stream.read(BUFFER)
         os.fchmod(crt_file.fileno(), MODE_OWNER)  # only accessible by owner
 
-    return  flask.make_response(flask.jsonify({
-        'message': 'OK'}), 202)
+    return flask.make_response(flask.jsonify({
+        'message': 'OK'}), 202)

octavia/amphorae/backends/agent/api_server/server.py

@@ -43,7 +43,7 @@ for code in six.iterkeys(exceptions.default_exceptions):
 
 
 # Tested with curl -k -XPUT --data-binary @/tmp/test.txt
-# https://127.0.0.1:8443/0.5/listeners/123/haproxy
+# https://127.0.0.1:9443/0.5/listeners/123/haproxy
 @app.route('/' + api_server.VERSION + '/listeners/<listener_id>/haproxy',
            methods=['PUT'])
 def upload_haproxy_config(listener_id):

octavia/amphorae/backends/agent/api_server/util.py

@@ -18,6 +18,7 @@ import os
 from oslo_config import cfg
 
 CONF = cfg.CONF
+CONF.import_group('amphora_agent', 'octavia.common.config')
 CONF.import_group('haproxy_amphora', 'octavia.common.config')
 UPSTART_DIR = '/etc/init'
 
@@ -31,7 +32,7 @@ def haproxy_dir(listener_id):
 
 
 def pid_path(listener_id):
-    return os.path.join(haproxy_dir(listener_id), 'haproxy.pid')
+    return os.path.join(haproxy_dir(listener_id), listener_id + '.pid')
 
 
 def config_path(listener_id):
@@ -63,5 +64,5 @@ def is_listener_running(listener_id):
 
 
 def get_network_interface_file(interface):
-    return os.path.join(CONF.haproxy_amphora.agent_server_network_dir,
-                        interface + '.cfg')
+    return os.path.join(CONF.amphora_agent.agent_server_network_dir,
+                        interface + '.cfg')

octavia/amphorae/backends/agent/templates/amphora_agent_conf.template

@@ -0,0 +1,36 @@
+{# Copyright 2015 Hewlett-Packard Development Company, L.P.
+#
+# 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.
+#}
+[DEFAULT]
+debug = {{ debug }}
+
+[haproxy_amphora]
+base_cert_dir = {{ base_cert_dir }}
+base_path = {{ base_path }}
+bind_host = {{ bind_host }}
+bind_port = {{ bind_port }}
+haproxy_cmd = {{ haproxy_cmd }}
+respawn_count = {{ respawn_count }}
+respawn_interval = {{ respawn_interval }}
+
+[health_manager]
+controller_ip_port_list = {{ controller_list|join(', ') }}
+heartbeat_interval = {{ heartbeat_interval }}
+heartbeat_key = {{ heartbeat_key }}
+
+[amphora_agent]
+agent_server_ca = {{ agent_server_ca }}
+agent_server_cert = {{ agent_server_cert }}
+agent_server_network_dir = {{ agent_server_network_dir }}
+amphora_id = {{ amphora_id }}

octavia/amphorae/backends/health_daemon/config.py

@@ -1,76 +0,0 @@
-#    Copyright 2014 Hewlett-Packard Development Company, L.P.
-#
-#    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 collections
-import json
-
-import octavia.amphorae.backends.health_daemon.singleton as singleton
-
-
-@singleton.singleton
-class JSONFileConfig(collections.Mapping):
-    def __init__(self):
-        self.filename = None
-        self.conf = {}
-        self.observers = set()
-
-    """ Set the config filename and perform the first read
-
-    :param filename: a JSON file that contains the config
-    """
-    def set_filename(self, filename):
-        self.filename = filename
-        self.read_config()
-
-    def __iter__(self):
-        return iter(self.conf)
-
-    def __getitem__(self, k):
-        return self.conf[k]
-
-    def __len__(self):
-        return len(self.conf)
-
-    """ Add a callable to be notified of config changes
-
-    :param obs: a callable to receive change events
-    """
-    def add_observer(self, obs):
-        self.observers.add(obs)
-
-    """ Remove a callable to be notified of config changes
-
-    By design if the callable passed doesn't exist then just return
-
-    :param obs: a callable to attempt to remove
-    """
-    def remove_observer(self, obs):
-        self.observers.discard(obs)
-
-    """ Force a reread of the config file and inform all observers
-    """
-    def check_update(self):
-        self.read_config()
-        self.confirm_update()
-
-    def confirm_update(self):
-        for observer in self.observers:
-            observer()
-
-    def read_config(self):
-        if self.filename is None:
-            return
-
-        self.cfile = open(self.filename, 'r')
-        self.conf = json.load(self.cfile)

octavia/amphorae/backends/health_daemon/health_daemon.py

@@ -14,52 +14,93 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-import argparse
-import sys
+import os
 import time
 
-import config
-import health_sender
+from oslo_config import cfg
+from oslo_log import log as logging
+import six
+
+from octavia.amphorae.backends.agent.api_server import util
+from octavia.amphorae.backends.health_daemon import health_sender
+from octavia.amphorae.backends.utils import haproxy_query
+from octavia.i18n import _LI
+
+if six.PY2:
+    import Queue as queue
+else:
+    import queue
+
+CONF = cfg.CONF
+CONF.import_group('amphora_agent', 'octavia.common.config')
+CONF.import_group('haproxy_amphora', 'octavia.common.config')
+CONF.import_group('health_manager', 'octavia.common.config')
+LOG = logging.getLogger(__name__)
+SEQ = 0
 
 
-def run_sender():
+def list_sock_stat_files(hadir=None):
+    stat_sock_files = {}
+    if hadir is None:
+        hadir = CONF.haproxy_amphora.base_path
+    listener_ids = util.get_listeners()
+    for listener_id in listener_ids:
+        sock_file = listener_id + ".sock"
+        stat_sock_files[listener_id] = os.path.join(hadir, sock_file)
+    return stat_sock_files
+
+
+def run_sender(cmd_queue):
+    LOG.info(_LI('Health Manager Sender starting.'))
     sender = health_sender.UDPStatusSender()
-    cfg = config.JSONFileConfig()
-
-    sighup_received = False
-    seq = 0
     while True:
-        if sighup_received:
-            print('re-reading config file')
-            sighup_received = False
-            cfg.check_update()
-
-        message = {'not the answer': 43,
-                   'id': cfg['id'],
-                   'seq': seq}
-        seq = seq + 1
+        message = build_stats_message()
         sender.dosend(message)
-        time.sleep(cfg['delay'])
+        try:
+            cmd = cmd_queue.get_nowait()
+            if cmd is 'reload':
+                LOG.info(_LI('Reloading configuration'))
+                CONF.reload_config_files()
+            elif cmd is 'shutdown':
+                LOG.info(_LI('Health Manager Sender shutting down.'))
+                break
+        except queue.Empty:
+            pass
+        time.sleep(CONF.health_manager.heartbeat_interval)
 
 
-def parse_args():
-    parser = argparse.ArgumentParser(description='Health Sender Daemon')
-    parser.add_argument('-c', '--config', type=str, required=False,
-                        help='config file path',
-                        default='/etc/amphora/status_sender.json')
-    args = parser.parse_args()
-    return vars(args)
+def get_stats(stat_sock_file):
+    stats_query = haproxy_query.HAProxyQuery(stat_sock_file)
+    stats = stats_query.show_stat()
+    pool_status = stats_query.get_pool_status()
+    return stats, pool_status
 
 
-if __name__ == '__main__':
-    args = parse_args()
-    cfg = config.JSONFileConfig()
-    try:
-        cfg.set_filename(args['config'])
-    except IOError as exception:
-        print(exception)
-        sys.exit(1)
-
-    # Now start up the sender loop
-    run_sender()
-    sys.exit(0)
+def build_stats_message():
+    global SEQ
+    msg = {'id': CONF.amphora_agent.amphora_id,
+           'seq': SEQ, "listeners": {}}
+    SEQ += 1
+    stat_sock_files = list_sock_stat_files()
+    for listener_id, stat_sock_file in six.iteritems(stat_sock_files):
+        listener_dict = {'pools': {}, 'status': 'DOWN',
+                                      'stats': {'tx': 0, 'rx': 0,
+                                                'conns': 0, 'totconns': 0}}
+        msg['listeners'][listener_id] = listener_dict
+        if util.is_listener_running(listener_id):
+            (stats, pool_status) = get_stats(stat_sock_file)
+            listener_dict = msg['listeners'][listener_id]
+            for row in stats:
+                if row['svname'] == 'FRONTEND':
+                    listener_dict['stats']['tx'] = int(row['bout'])
+                    listener_dict['stats']['rx'] = int(row['bin'])
+                    listener_dict['stats']['conns'] = int(row['scur'])
+                    listener_dict['stats']['totconns'] = int(row['stot'])
+                    listener_dict['status'] = row['status']
+            for oid, pool in six.iteritems(pool_status):
+                if oid != listener_id:
+                    pool_id = oid
+                    pools = listener_dict['pools']
+                    pools[pool_id] = {"status": pool['status'],
+                                      "members": pool['members']}
+    return msg

octavia/amphorae/backends/health_daemon/health_sender.py

@@ -12,43 +12,63 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-# TODO(barclaac) Need to decide how this hooks into rest of system,
-# e.g. daemon, subprocess, thread etc.
-
 import socket
 
-import config
-import status_message
+from oslo_config import cfg
+from oslo_log import log as logging
+
+from octavia.amphorae.backends.health_daemon import status_message
+from octavia.i18n import _LE
+
+CONF = cfg.CONF
+CONF.import_group('health_manager', 'octavia.common.config')
+LOG = logging.getLogger(__name__)
 
 
-class UDPStatusSender:
+def round_robin_addr(addrinfo_list):
+    if len(addrinfo_list) <= 0:
+        return None
+    addrinfo = addrinfo_list.pop(0)
+    addrinfo_list.append(addrinfo)
+    return addrinfo
+
+
+class UDPStatusSender(object):
     def __init__(self):
-        self.cfg = config.JSONFileConfig()
-        self.dests = {}
-        self.update(self.cfg['destination'], self.cfg['port'])
+        self.dests = []
+        for ipport in CONF.health_manager.controller_ip_port_list:
+            parts = ipport.split(':')
+            self.update(parts[0], parts[1])
         self.v4sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self.v6sock = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
-        self.key = str(self.cfg['key'])
-        self.cfg.add_observer(self.config_change)
+        self.key = str(CONF.health_manager.heartbeat_key)
 
-    # TODO(barclaac) Still need to reread the address list if it gets changed
-    def config_change(self):
-        pass
+    def update(self, dest, port):
+        addrlist = socket.getaddrinfo(dest, port, 0, socket.SOCK_DGRAM)
+        # addrlist = [(family, socktype, proto, canonname, sockaddr) ...]
+        # e.g. 4 = sockaddr - what we actually need
+        for addr in addrlist:
+            self.dests.append(addr)  # Just grab the first match
+            break
 
-    def update(self, dest_list, port):
-        for dest in dest_list:
-            addrlist = socket.getaddrinfo(dest, port, 0, socket.SOCK_DGRAM)
-            # addrlist = [(family, socktype, proto, canonname, sockaddr) ...]
-            # e.g. 4 = sockaddr - what we actually need
-            for addr in addrlist:
-                self.dests[addr[4]] = addr
-
-    def dosend(self, envelope):
-        envelope_str = status_message.encode(envelope, self.key)
-        for dest in self.dests.itervalues():
-            # addrlist = [(family, socktype, proto, canonname, sockaddr) ...]
-            # e.g. 0 = sock family, 4 = sockaddr - what we actually need
-            if dest[0] == socket.AF_INET:
-                self.v4sock.sendto(envelope_str, dest[4])
-            elif dest[0] == socket.AF_INET6:
-                self.v6sock.sendto(envelope_str, dest[4])
+    def dosend(self, obj):
+        envelope_str = status_message.wrap_envelope(obj, self.key)
+        addrinfo = round_robin_addr(self.dests)
+        # dest = (family, socktype, proto, canonname, sockaddr)
+        # e.g. 0 = sock family, 4 = sockaddr - what we actually need
+        if addrinfo is None:
+            LOG.error(_LE('No controller address found. '
+                          'Unable to send heartbeat.'))
+            return
+        try:
+            if addrinfo[0] == socket.AF_INET:
+                self.v4sock.sendto(envelope_str, addrinfo[4])
+            elif addrinfo[0] == socket.AF_INET6:
+                self.v6sock.sendto(envelope_str, addrinfo[4])
+        except socket.error:
+            # Pass here as on amp boot it will get one or more
+            # error: [Errno 101] Network is unreachable
+            # while the networks are coming up
+            # No harm in trying to send as it will still failover
+            # if the message isn't received
+            pass

octavia/amphorae/backends/health_daemon/sample_config.json

@@ -1,7 +0,0 @@
-{
-    "key": "asamplekey",
-    "delay": 2.5,
-    "destination": [ "::1", "127.1" ],
-    "port": 12345,
-    "id": "0dc47eda-872b-11e4-920b-000c294b76ae"
-}

octavia/amphorae/backends/health_daemon/singleton.py

@@ -1,26 +0,0 @@
-#    Copyright 2014 Hewlett-Packard Development Company, L.P.
-#
-#    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.
-
-# TODO(barclaac) Someone needs to move this to be a library function for
-# all of Octavia (Oslo even?)
-
-
-def singleton(cls):
-    instances = {}
-
-    def getinstance():
-        if cls not in instances:
-            instances[cls] = cls()
-        return instances[cls]
-    return getinstance

octavia/amphorae/backends/health_daemon/status_message.py

@@ -1,4 +1,4 @@
-#    Copyright 2014 Hewlett-Packard Development Company, L.P.
+# Copyright 2014 Hewlett-Packard Development Company, L.P.
 #
 #    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
@@ -12,22 +12,61 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import binascii
 import hashlib
 import hmac
 import json
+import zlib
+
+from oslo_log import log as logging
+
+from octavia.common import exceptions
+from octavia.i18n import _LW
+
+LOG = logging.getLogger(__name__)
+
+hash_algo = hashlib.sha256
+hash_len = 32
 
 
-def encode(msg, key):
-    result = {}
-    src = json.dumps(msg)
-    hmc = hmac.new(key.encode('ascii'), src.encode('ascii'), hashlib.sha1)
-    result['msg'] = msg
-    result['hmac'] = hmc.hexdigest()
-    return json.dumps(result)
+def to_hex(byte_array):
+    return binascii.hexlify(byte_array).decode()
 
 
-def checkhmac(envelope_str, key):
-    envelope = json.loads(envelope_str)
-    src = json.dumps(envelope['msg'])
-    hmc = hmac.new(key.encode('ascii'), src.encode('ascii'), hashlib.sha1)
-    return hmc.hexdigest() == envelope['hmac']
+def encode_obj(obj):
+    json_bytes = json.dumps(obj).encode('utf-8')
+    binary_array = zlib.compress(json_bytes, 9)
+    return binary_array
+
+
+def decode_obj(binary_array):
+    json_str = zlib.decompress(binary_array).decode('utf-8')
+    obj = json.loads(json_str)
+    return obj
+
+
+def wrap_envelope(obj, key):
+    payload = encode_obj(obj)
+    hmc = get_hmac(payload, key)
+    envelope = payload + hmc
+    return envelope
+
+
+def unwrap_envelope(envelope, key):
+    payload = envelope[:-hash_len]
+    expected_hmc = envelope[-hash_len:]
+    calculated_hmc = get_hmac(payload, key)
+    if expected_hmc != calculated_hmc:
+        LOG.warn(_LW('calculated hmac: %(s1)s not equal to msg hmac: '
+                     '%(s2)s dropping packet'), {'s1': to_hex(calculated_hmc),
+                                                 's2': to_hex(expected_hmc)})
+        fmt = 'calculated hmac: {0} not equal to msg hmac: {1} dropping packet'
+        raise exceptions.InvalidHMACException(fmt.format(
+            to_hex(calculated_hmc), to_hex(expected_hmc)))
+    obj = decode_obj(payload)
+    return obj
+
+
+def get_hmac(payload, key):
+    hmc = hmac.new(key.encode("utf-8"), payload, hashlib.sha256)
+    return hmc.digest()

octavia/amphorae/backends/utils/haproxy_query.py

@@ -90,8 +90,7 @@ class HAProxyQuery(object):
         """
 
         results = self._query(
-            'show stat {proxy_iid} {object_type}'
-            + '{server_id}'.format(
+            'show stat {proxy_iid} {object_type} {server_id}'.format(
                 proxy_iid=proxy_iid,
                 object_type=object_type,
                 server_id=server_id))
@@ -118,16 +117,16 @@ class HAProxyQuery(object):
             # pxname: pool, svname: server_name, status: status
 
             # All the way up is UP, otherwise call it DOWN
-            if line['status'] != consts.AMPHORA_UP:
-                line['status'] = consts.AMPHORA_DOWN
+            if line['status'] != consts.UP:
+                line['status'] = consts.DOWN
 
             if line['pxname'] not in final_results:
-                final_results[line['pxname']] = dict(members=[])
+                final_results[line['pxname']] = dict(members={})
 
             if line['svname'] == 'BACKEND':
                 final_results[line['pxname']]['uuid'] = line['pxname']
                 final_results[line['pxname']]['status'] = line['status']
             else:
-                final_results[line['pxname']]['members'].append(
-                    {line['svname']: line['status']})
+                final_results[line['pxname']]['members'][line['svname']] = (
+                    line['status'])
         return final_results

octavia/amphorae/drivers/haproxy/rest_api_driver.py

@@ -22,6 +22,7 @@ import requests
 import six
 from stevedore import driver as stevedore_driver
 
+from octavia.amphorae.driver_exceptions import exceptions as driver_except
 from octavia.amphorae.drivers import driver_base as driver_base
 from octavia.amphorae.drivers.haproxy import exceptions as exc
 from octavia.amphorae.drivers.haproxy.jinja import jinja_cfg
@@ -244,10 +245,10 @@ class AmphoraAPIClient(object):
                 LOG.warn(_LW("Could not talk  to instance"))
                 time.sleep(CONF.haproxy_amphora.connection_retry_interval)
                 if a >= CONF.haproxy_amphora.connection_max_retries:
-                    raise exc.TimeOutException()
+                    raise driver_except.TimeOutException()
             else:
                 return r
-        raise exc.UnavailableException()
+        raise driver_except.UnavailableException()
 
     def upload_config(self, amp, listener_id, config):
         r = self.put(

octavia/amphorae/drivers/health/heartbeat_udp.py

@@ -0,0 +1,181 @@
+# Copyright 2014 Rackspace
+#
+#    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 socket
+
+from concurrent import futures
+from oslo_config import cfg
+from oslo_log import log as logging
+
+from octavia.amphorae.backends.health_daemon import status_message
+from octavia.common import exceptions
+from octavia.db import repositories
+from octavia.i18n import _LI
+
+
+UDP_MAX_SIZE = 64 * 1024
+LOG = logging.getLogger(__name__)
+
+
+class UDPStatusGetter(object):
+    """This class defines methods that will gather heatbeats
+
+    The heartbeats are transmitted via UDP and this class will bind to a port
+    and absorb them
+    """
+    def __init__(self, health_update, stats_update):
+        self.stats_update = stats_update
+        self.health_update = health_update
+        self.key = cfg.CONF.health_manager.heartbeat_key
+        self.ip = cfg.CONF.health_manager.bind_ip
+        self.port = cfg.CONF.health_manager.bind_port
+        self.sockaddr = None
+        LOG.info(_LI(
+            'attempting to listen on {0} port {1}').format(
+                self.ip, self.port))
+        self.sock = None
+        self.update(self.key, self.ip, self.port)
+
+        self.executor = futures.ThreadPoolExecutor(
+            max_workers=cfg.CONF.health_manager.status_update_threads)
+        self.repo = repositories.Repositories().amphorahealth
+
+    def update(self, key, ip, port):
+        """Update the running config for the udp socket server
+
+        :param key: The hmac key used to verify the UDP packets. String
+        :param ip: The ip address the UDP server will read from
+        :param port: The port the UDP server will read from
+        :return: None
+        """
+        self.key = key
+        for addrinfo in socket.getaddrinfo(ip, port, 0, socket.SOCK_DGRAM):
+            ai_family = addrinfo[0]
+            self.sockaddr = addrinfo[4]
+            if self.sock is not None:
+                self.sock.close()
+            self.sock = socket.socket(ai_family, socket.SOCK_DGRAM)
+            self.sock.bind((ip, port))
+            if cfg.CONF.health_manager.sock_rlimit > 0:
+                rlimit = cfg.CONF.health_manager.sock_rlimit
+                LOG.info(_LI("setting sock rlimit to {0}").format(rlimit))
+                self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF,
+                                     rlimit)
+            break  # just used the first addr getaddrinfo finds
+        if self.sock is None:
+            raise exceptions.NetworkConfig("unable to find suitable socket")
+
+    def dorecv(self, *args, **kw):
+        """Waits for a UDP heart beat to be sent.
+
+        :return: Returns the unwrapped payload and addr that sent the \
+                 heart beat. The format of the obj from the UDP sender is of
+                 the form. Not that listener_1 has not pools and listener_4
+                 has no nodes.
+
+        {"listeners": {
+        "listener_uuid_1": {
+            "pools": {},
+            "status": "OPEN",
+            "stats": {
+                "conns": 0,
+                "rx": 0,
+                "tx": 0
+            }
+        },
+        "listener_uuid_2": {
+            "pools": {
+                "pool_uuid_1": {
+                    "members": [
+                        {
+                            "member_uuid_1": "DOWN"
+                        },
+                        {
+                            "member_uuid_2": "DOWN"
+                        },
+                        {
+                            "member_uuid_3": "DOWN"
+                        },
+                        {
+                            "member_uuid_4": "DOWN"
+                        }
+                    ]
+                }
+            },
+            "status": "OPEN",
+            "stats": {
+                "conns": 0,
+                "rx": 0,
+                "tx": 0
+            }
+        },
+        "listener_uuid_3": {
+            "pools": {
+                "pool_uuid_2": {
+                    "members": [
+                        {
+                            "member_uuid_5": "DOWN"
+                        },
+                        {
+                            "member_uuid_6": "DOWN"
+                        },
+                        {
+                            "member_uuid_7": "DOWN"
+                        },
+                        {
+                            "member_uuid_8": "DOWN"
+                        }
+                    ]
+                }
+            },
+            "status": "OPEN",
+            "stats": {
+                "conns": 0,
+                "rx": 0,
+                "tx": 0
+            }
+        },
+        "listener_uuid_4": {
+            "pools": {
+                "pool_uuid_3": {
+                    "members": []
+                }
+            },
+            "status": "OPEN",
+            "stats": {
+                "conns": 0,
+                "rx": 0,
+                "tx": 0
+            }
+        }
+    },
+    "id": "amphora_uuid",
+    "seq": 1033
+}
+
+        """
+        (data, srcaddr) = self.sock.recvfrom(UDP_MAX_SIZE)
+        obj = status_message.unwrap_envelope(data, self.key)
+        return obj, srcaddr
+
+    def check(self):
+        try:
+            (obj, srcaddr) = self.dorecv()
+            if self.health_update:
+                self.executor.submit(self.health_update.update_health, obj)
+            if self.stats_update:
+                self.executor.submit(self.stats_update.update_stats, obj)
+        except exceptions.InvalidHMACException:
+            # Pass here as the packet was dropped and logged already
+            pass

octavia/cmd/agent.py

@@ -17,6 +17,8 @@
 # make sure PYTHONPATH includes the home directory if you didn't install
 
 import logging
+import multiprocessing as multiproc
+import os
 import ssl
 import sys
 
@@ -24,11 +26,14 @@ from oslo_config import cfg
 from werkzeug import serving
 
 from octavia.amphorae.backends.agent.api_server import server
+from octavia.amphorae.backends.health_daemon import health_daemon
 from octavia.common import service
 
 LOG = logging.getLogger(__name__)
 CONF = cfg.CONF
+CONF.import_group('amphora_agent', 'octavia.common.config')
 CONF.import_group('haproxy_amphora', 'octavia.common.config')
+HM_SENDER_CMD_QUEUE = multiproc.Queue()
 
 
 # Hack: Use werkzeugs context
@@ -59,12 +64,21 @@ def main():
     # comment out to improve logging
     service.prepare_service(sys.argv)
 
+    # Workaround for an issue with the auto-reload used below in werkzeug
+    # Without it multiple health senders get started when werkzeug reloads
+    if not os.environ.get('WERKZEUG_RUN_MAIN'):
+        health_sender_proc = multiproc.Process(name='HM_sender',
+                                               target=health_daemon.run_sender,
+                                               args=(HM_SENDER_CMD_QUEUE,))
+        health_sender_proc.daemon = True
+        health_sender_proc.start()
+
     # We will only enforce that the client cert is from the good authority
     # todo(german): Watch this space for security improvements
     ctx = OctaviaSSLContext(ssl.PROTOCOL_SSLv23)
 
-    ctx.load_cert_chain(CONF.haproxy_amphora.agent_server_cert,
-                        ca=CONF.haproxy_amphora.agent_server_ca)
+    ctx.load_cert_chain(CONF.amphora_agent.agent_server_cert,
+                        ca=CONF.amphora_agent.agent_server_ca)
 
     # This will trigger a reload if any files change and
     # in particular the certificate file
@@ -74,4 +88,4 @@ def main():
                        use_debugger=CONF.debug,
                        ssl_context=ctx,
                        use_reloader=True,
-                       extra_files=[CONF.haproxy_amphora.agent_server_cert])
+                       extra_files=[CONF.amphora_agent.agent_server_cert])

octavia/cmd/health_manager.py

@@ -14,13 +14,15 @@
 #
 import multiprocessing
 import sys
-import time
 
 from oslo_config import cfg
 from oslo_log import log as logging
 
+from octavia.amphorae.drivers.health import heartbeat_udp
 from octavia.common import service
 from octavia.controller.healthmanager import health_manager
+from octavia.controller.healthmanager import update_health_mixin
+from octavia.controller.healthmanager import update_stats_mixin
 from octavia.i18n import _LI
 
 CONF = cfg.CONF
@@ -28,16 +30,18 @@ LOG = logging.getLogger(__name__)
 CONF.import_group('health_manager', 'octavia.common.config')
 
 
-def HM_listener():
+def hm_listener():
+    # TODO(german): steved'or load those drivers
+    udp_getter = heartbeat_udp.UDPStatusGetter(
+        update_health_mixin.UpdateHealthMixin(),
+        update_stats_mixin.UpdateStatsMixin())
     while True:
-        time.sleep(5)
-        # to do by Carlos
+        udp_getter.check()
 
 
-def HM_health_check():
+def hm_health_check():
+    hm = health_manager.HealthManager()
     while True:
-        time.sleep(CONF.health_manager.interval)
-        hm = health_manager.HealthManager()
         hm.health_check()
 
 
@@ -45,21 +49,21 @@ def main():
     service.prepare_service(sys.argv)
     processes = []
 
-    HM_listener_proc = multiprocessing.Process(name='HM_listener',
-                                               target=HM_listener)
-    processes.append(HM_listener_proc)
-    HM_health_check_proc = multiprocessing.Process(name='HM_health_check',
-                                                   target=HM_health_check)
-    processes.append(HM_health_check_proc)
+    hm_listener_proc = multiprocessing.Process(name='HM_listener',
+                                               target=hm_listener)
+    processes.append(hm_listener_proc)
+    hm_health_check_proc = multiprocessing.Process(name='HM_health_check',
+                                                   target=hm_health_check)
+    processes.append(hm_health_check_proc)
     LOG.info(_LI("Health Manager listener process starts:"))
-    HM_listener_proc.start()
+    hm_listener_proc.start()
     LOG.info(_LI("Health manager check process starts:"))
-    HM_health_check_proc.start()
+    hm_health_check_proc.start()
 
     try:
         for process in processes:
             process.join()
     except KeyboardInterrupt:
         LOG.info(_LI("Health Manager existing due to signal"))
-        HM_listener_proc.terminate()
-        HM_health_check_proc.terminate()
+        hm_listener_proc.terminate()
+        hm_health_check_proc.terminate()

octavia/common/config.py

@@ -80,21 +80,59 @@ core_opts = [
                help=_('Name of the controller plugin to use'))
 ]
 
+# Options only used by the amphora agent
+amphora_agent_opts = [
+    cfg.StrOpt('agent_server_ca', default='/etc/octavia/certs/client_ca.pem',
+               help=_("The ca which signed the client certificates")),
+    cfg.StrOpt('agent_server_cert', default='/etc/octavia/certs/server.pem',
+               help=_("The server certificate for the agent.py server "
+                      "to use")),
+    cfg.StrOpt('agent_server_network_dir',
+               default='/etc/network/interfaces.d/',
+               help=_("The directory where new network interfaces "
+                      "are located")),
+    # Do not specify in octavia.conf, loaded at runtime
+    cfg.StrOpt('amphora_id', help=_("The amphora ID.")),
+]
+
 networking_opts = [
     cfg.StrOpt('lb_network_name', help=_('Name of amphora internal network')),
 ]
 
 healthmanager_opts = [
+    cfg.StrOpt('bind_ip', default='0.0.0.0',
+               help=_('IP address the controller will listen on for '
+                      'heart beats')),
+    cfg.IntOpt('bind_port', default=5555,
+               help=_('Port number the controller will listen on'
+                      'for heart beats')),
     cfg.IntOpt('failover_threads',
                default=10,
                help=_('Number of threads performing amphora failovers.')),
-    cfg.IntOpt('interval',
-               default=3,
-               help=_('Sleep time between health checks in seconds.')),
+    cfg.IntOpt('status_update_threads',
+               default=50,
+               help=_('Number of threads performing amphora failovers.')),
+    cfg.StrOpt('heartbeat_key',
+               help=_('key used to validate amphora sending'
+                      'the message'), secret=True),
     cfg.IntOpt('heartbeat_timeout',
-               default=10,
+               default=60,
                help=_('Interval, in seconds, to wait before failing over an '
                       'amphora.')),
+    cfg.IntOpt('health_check_interval',
+               default=3,
+               help=_('Sleep time between health checks in seconds.')),
+    cfg.IntOpt('sock_rlimit', default=0,
+               help=_(' sets the value of the heartbeat recv buffer')),
+
+    # Used by the health manager on the amphora
+    cfg.ListOpt('controller_ip_port_list',
+                help=_('List of controller ip and port pairs for the '
+                       'heartbeat receivers. Example [\'127.0.0.1:5555\', '
+                       '\'127.0.0.1:5555\']')),
+    cfg.IntOpt('heartbeat_interval',
+               default=10,
+               help=_('Sleep time between sending hearthbeats.'))
 ]
 
 oslo_messaging_opts = [
@@ -133,7 +171,7 @@ haproxy_amphora_opts = [
     # REST server
     cfg.StrOpt('bind_host', default='0.0.0.0',
                help=_("The host IP to bind to")),
-    cfg.IntOpt('bind_port', default=8443,
+    cfg.IntOpt('bind_port', default=9443,
                help=_("The port to bind to")),
     cfg.StrOpt('haproxy_cmd', default='/usr/sbin/haproxy',
                help=_("The full path to haproxy")),
@@ -143,15 +181,6 @@ haproxy_amphora_opts = [
                help=_("The respawn interval for haproxy's upstart script")),
     cfg.StrOpt('haproxy_cert_dir', default='/tmp/',
                help=_("The directory to store haproxy cert files in")),
-    cfg.StrOpt('agent_server_cert', default='/etc/octavia/certs/server.pem',
-               help=_("The server certificate for the agent.py server "
-                      "to use")),
-    cfg.StrOpt('agent_server_ca', default='/etc/octavia/certs/client_ca.pem',
-               help=_("The ca which signed the client certificates")),
-    cfg.StrOpt('agent_server_network_dir',
-               default='/etc/network/interfaces.d/',
-               help=_("The directory where new network interfaces "
-                      "are located")),
     # REST client
     cfg.StrOpt('client_cert', default='/etc/octavia/certs/client.pem',
                help=_("The client certificate to talk to the agent")),
@@ -236,6 +265,7 @@ house_keeping_opts = [
 
 # Register the configuration options
 cfg.CONF.register_opts(core_opts)
+cfg.CONF.register_opts(amphora_agent_opts, group='amphora_agent')
 cfg.CONF.register_opts(networking_opts, group='networking')
 cfg.CONF.register_opts(oslo_messaging_opts, group='oslo_messaging')
 cfg.CONF.register_opts(haproxy_amphora_opts, group='haproxy_amphora')

octavia/common/constants.py

@@ -42,12 +42,6 @@ SUPPORTED_PROTOCOLS = (PROTOCOL_TCP, PROTOCOL_HTTPS, PROTOCOL_HTTP,
 #       the provisioning_status table
 # Amphora has been allocated to a load balancer
 AMPHORA_ALLOCATED = 'ALLOCATED'
-# Amphora healthy with listener(s) deployed
-# TODO(johnsom) This doesn't exist
-AMPHORA_UP = 'UP'
-# Amphora unhealthy with listener(s) deployed
-# TODO(johnsom) This doesn't exist
-AMPHORA_DOWN = 'DOWN'
 # Amphora is being built
 AMPHORA_BOOTING = 'BOOTING'
 # Amphora is ready to be allocated to a load balancer
@@ -65,9 +59,8 @@ SUPPORTED_PROVISIONING_STATUSES = (ACTIVE, AMPHORA_ALLOCATED,
                                    PENDING_UPDATE, DELETED, ERROR)
 MUTABLE_STATUSES = (ACTIVE,)
 
-SUPPORTED_AMPHORA_STATUSES = (AMPHORA_ALLOCATED, AMPHORA_UP, AMPHORA_DOWN,
-                              AMPHORA_BOOTING, AMPHORA_READY, DELETED,
-                              PENDING_DELETE)
+SUPPORTED_AMPHORA_STATUSES = (AMPHORA_ALLOCATED, AMPHORA_BOOTING,
+                              AMPHORA_READY, DELETED, PENDING_DELETE)
 
 ONLINE = 'ONLINE'
 OFFLINE = 'OFFLINE'
@@ -150,3 +143,18 @@ SUPPORTED_LB_TOPOLOGIES = (TOPOLOGY_ACTIVE_STANDBY, TOPOLOGY_SINGLE)
 SUPPORTED_AMPHORA_ROLES = (ROLE_BACKUP, ROLE_MASTER, ROLE_STANDALONE)
 
 AGENT_API_TEMPLATES = '/templates'
+AGENT_CONF_TEMPLATE = 'amphora_agent_conf.template'
+
+OPEN = 'OPEN'
+FULL = 'FULL'
+
+# OPEN = HAProxy listener status nbconn < maxconn
+# FULL = HAProxy listener status not nbconn < maxconn
+HAPROXY_LISTENER_STATUSES = (OPEN, FULL)
+
+UP = 'UP'
+DOWN = 'DOWN'
+
+# UP = HAProxy backend has working or no servers
+# DOWN = HAProxy backend has no working servers
+HAPROXY_BACKEND_STATUSES = (UP, DOWN)

octavia/common/exceptions.py

@@ -80,10 +80,18 @@ class InvalidOption(APIException):
     code = 400
 
 
+class InvalidHMACException(OctaviaException):
+    message = _("HMAC hashes didn't match")
+
+
 class MissingArguments(OctaviaException):
     message = _("Missing arguments.")
 
 
+class NetworkConfig(OctaviaException):
+    message = _("Unable to allocate network resource from config")
+
+
 class NeedsPassphrase(OctaviaException):
     message = _("Passphrase needed to decrypt key but client "
                 "did not provide one.")

octavia/compute/drivers/nova_driver.py

@@ -106,11 +106,11 @@ class VirtualMachineManager(compute_base.ComputeBase):
         try:
             amphora = self.get_amphora(amphora_id=amphora_id)
             if amphora and amphora.status == 'ACTIVE':
-                return constants.AMPHORA_UP
+                return constants.UP
         except Exception:
             LOG.exception(_LE("Error retrieving nova virtual machine status."))
             raise exceptions.ComputeStatusException()
-        return constants.AMPHORA_DOWN
+        return constants.DOWN
 
     def get_amphora(self, amphora_id):
         '''Retrieve the information in nova of a virtual machine.

octavia/controller/healthmanager/health_manager.py

@@ -39,7 +39,7 @@ class HealthManager(object):
         with futures.ThreadPoolExecutor(max_workers=self.threads) as executor:
             try:
                 while True:
-                    time.sleep(CONF.health_manager.interval)
+                    time.sleep(CONF.health_manager.health_check_interval)
                     session = db_api.get_session()
                     LOG.debug("Starting amphora health check")
                     failover_count = 0

octavia/controller/healthmanager/update_health_mixin.py

@@ -12,14 +12,16 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import datetime
+
 from oslo_log import log as logging
 import sqlalchemy
-from sqlalchemy.sql import func
 
 from octavia.amphorae.drivers import driver_base as driver_base
 from octavia.common import constants
 from octavia.db import api as db_api
 from octavia.db import repositories as repo
+from octavia.i18n import _LE, _LW
 
 import six
 
@@ -31,8 +33,10 @@ class UpdateHealthMixin(driver_base.HealthMixin):
     def __init__(self):
         super(UpdateHealthMixin, self).__init__()
         # first setup repo for amphora, listener,member(nodes),pool repo
+        self.amphora_repo = repo.AmphoraRepository()
         self.amphora_health_repo = repo.AmphoraHealthRepository()
         self.listener_repo = repo.ListenerRepository()
+        self.loadbalancer_repo = repo.LoadBalancerRepository()
         self.member_repo = repo.MemberRepository()
         self.pool_repo = repo.PoolRepository()
 
@@ -43,76 +47,131 @@ class UpdateHealthMixin(driver_base.HealthMixin):
         :type map: string
         :returns: null
 
-        This function has the following 3 goals:
-        1)Update the health_manager table based on amphora status is up/down
-        2)Update related DB status to be ERROR/DOWN when amphora is down
-        3)Update related DB status to be ACTIVATE/UP when amphora is up
-        4)Track the status of the members
-
         The input health data structure is shown as below:
 
         health = {
-            "amphora-status": "AMPHORA_UP",
-            "amphora-id": FAKE_UUID_1,
+            "id": self.FAKE_UUID_1,
             "listeners": {
-                "listener-id-1": {"listener-status": "ONLINE",
-                                  "members": {
-                                      "member-id-1": "ONLINE",
-                                      "member-id-2": "ONLINE"
-                                  }
-                },
-                "listener-id-2": {"listener-status": "ONLINE",
-                                  "members": {
-                                      "member-id-3": "ERROR",
-                                      "member-id-4": "ERROR",
-                                      "member-id-5": "ONLINE"
+                "listener-id-1": {"status": constants.OPEN, "pools": {
+                    "pool-id-1": {"status": constants.UP,
+                                  "members": {"member-id-1": constants.ONLINE}
                                   }
                 }
+                }
             }
         }
+
         """
         session = db_api.get_session()
 
-        # if the input amphora is healthy, we update its db info
-        # before update db, we need to check if the db has been created,
-        # if not, we need to create first
-        if health["amphora-status"] == constants.AMPHORA_UP:
-            amphora_id = health["amphora-id"]
-            amphora = self.amphora_health_repo.get(