voltha/coordinator_etcd.py

#
# Copyright 2017 the original author or authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

""" Etcd-based coordinator services """

from consul import ConsulException
from consul.twisted import Consul
from requests import ConnectionError
from structlog import get_logger
from twisted.internet import reactor
from twisted.internet.defer import inlineCallbacks, returnValue, Deferred
from txaioetcd import Client, KeySet, Transaction, CompVersion, OpGet, OpSet, Failed
from zope.interface import implementer

from leader import Leader
from common.utils.asleep import asleep
from common.utils.message_queue import MessageQueue
from voltha.registry import IComponent
from worker import Worker
from simplejson import dumps, loads
from common.utils.deferred_utils import DeferredWithTimeout, TimeOutError

log = get_logger()


class StaleMembershipEntryException(Exception):
    pass


@implementer(IComponent)
class CoordinatorEtcd(object):
    """
    An app shall instantiate only one Coordinator (singleton).
    A single instance of this object shall take care of all external
    with consul, and via consul, all coordination activities with its
    clustered peers. Roles include:
    - registering an ephemeral membership entry (k/v record) in consul
    - participating in a symmetric leader election, and potentially assuming
      the leader's role. What leadership entails is not a concern for the
      coordination, it simply instantiates (and shuts down) a leader class
      when it gains (or looses) leadership.
    """

    CONNECT_RETRY_INTERVAL_SEC = 1
    RETRY_BACKOFF = [0.05, 0.1, 0.2, 0.5, 1, 2, 5]

    # Public methods:

    def __init__(self,
                 internal_host_address,
                 external_host_address,
                 instance_id,
                 rest_port,
                 config,
                 consul='localhost:8500',
                 etcd='localhost:2379'):

        log.info('initializing-coordinator')
        self.config = config['coordinator']
        self.worker_config = config['worker']
        self.leader_config = config['leader']
        self.membership_watch_relatch_delay = config.get(
            'membership_watch_relatch_delay', 0.1)
        self.tracking_loop_delay = self.config.get(
            'tracking_loop_delay', 1)
        self.session_renewal_timeout = self.config.get(
            'session_renewal_timeout', 5)
        self.session_renewal_loop_delay = self.config.get(
            'session_renewal_loop_delay', 3)
        self.membership_maintenance_loop_delay = self.config.get(
            'membership_maintenance_loop_delay', 5)
        self.session_time_to_live = self.config.get(
            'session_time_to_live', 10)
        self.prefix = self.config.get('voltha_kv_prefix', 'service/voltha')
        self.leader_prefix = '/'.join((self.prefix, self.config.get(
            self.config['leader_key'], 'leader')))
        self.membership_prefix = '/'.join((self.prefix, self.config.get(
            self.config['membership_key'], 'members'), ''))
        self.assignment_prefix = '/'.join((self.prefix, self.config.get(
            self.config['assignment_key'], 'assignments'), ''))
        self.workload_prefix = '/'.join((self.prefix, self.config.get(
            self.config['workload_key'], 'work'), ''))
        self.core_store_prefix = '/'.join((self.prefix, self.config.get(
            self.config['core_store_key'], 'data/core')))
        self.core_store_assignment_key = self.core_store_prefix + \
                                         '/assignment'
        self.core_storage_suffix = 'core_store'

        self.retries = 0
        self.instance_id = instance_id
        self.internal_host_address = internal_host_address
        self.external_host_address = external_host_address
        self.rest_port = rest_port
        self.membership_record_key = self.membership_prefix + self.instance_id

        self.lease = None
        # session_id refers to either a Consul session ID or an Etcd lease object
        self.session_id = None
        self.i_am_leader = False
        self.leader_id = None  # will be the instance id of the current leader
        self.shutting_down = False
        self.leader = None
        self.membership_callback = None

        self.worker = Worker(self.instance_id, self)

        self.host = consul.split(':')[0].strip()
        self.port = int(consul.split(':')[1].strip())

        # TODO need to handle reconnect events properly
        self.consul = Consul(host=self.host, port=self.port)

        # Create etcd client
        kv_host = etcd.split(':')[0].strip()
        kv_port = etcd.split(':')[1].strip()
        self.etcd_url = u'http://' + kv_host + u':' + kv_port
        self.etcd = Client(reactor, self.etcd_url)

        self.wait_for_leader_deferreds = []

        self.peers_mapping_queue = MessageQueue()

    def start(self):
        log.debug('starting')
        reactor.callLater(0, self._async_init)
        log.info('started')
        return self

    @inlineCallbacks
    def stop(self):
        log.debug('stopping')
        self.shutting_down = True
        yield self._delete_session()  # this will delete the leader lock too
        yield self.worker.stop()
        if self.leader is not None:
            yield self.leader.stop()
            self.leader = None
        log.info('stopped')

    def wait_for_a_leader(self):
        """
        Async wait till a leader is detected/elected. The deferred will be
        called with the leader's instance_id
        :return: Deferred.
        """
        d = Deferred()
        if self.leader_id is not None:
            d.callback(self.leader_id)
            return d
        else:
            self.wait_for_leader_deferreds.append(d)
            return d

    # Wait for a core data id to be assigned to this voltha instance
    @inlineCallbacks
    def get_core_store_id_and_prefix(self):
        core_store_id = yield self.worker.get_core_store_id()
        returnValue((core_store_id, self.core_store_prefix))

    def recv_peers_map(self):
        return self.peers_mapping_queue.get()

    def publish_peers_map_change(self, msg):
        self.peers_mapping_queue.put(msg)

    # Proxy methods for etcd with retry support

    def kv_get(self, *args, **kw):
        # Intercept 'index' argument
        for name, value in kw.items():
            if name == 'index':
                kw.pop('index')
                break
        return self._retry('GET', *args, **kw)

    def kv_put(self, *args, **kw):
        return self._retry('PUT', *args, **kw)

    def kv_delete(self, *args, **kw):
        return self._retry('DELETE', *args, **kw)

    # Methods exposing key membership information

    @inlineCallbacks
    def get_members(self):
        """Return list of all members"""
        _, members = yield self.kv_get(self.membership_prefix, recurse=True)
        returnValue([member['Key'][len(self.membership_prefix):]
                     for member in members])

    # Private (internal) methods:

    @inlineCallbacks
    def _async_init(self):
        yield self._create_session()
        yield self._create_membership_record()
        yield self._start_leader_tracking()
        yield self.worker.start()

    def _backoff(self, msg):
        wait_time = self.RETRY_BACKOFF[min(self.retries,
                                           len(self.RETRY_BACKOFF) - 1)]
        self.retries += 1
        log.error(msg, retry_in=wait_time)
        return asleep(wait_time)

    def _clear_backoff(self):
        if self.retries:
            log.info('reconnected-to-consul', after_retries=self.retries)
            self.retries = 0

    @inlineCallbacks
    def _create_session(self):

        @inlineCallbacks
        def _create_session():
            etcd = yield self.get_kv_client()
            # Create etcd lease
            self.lease = yield etcd.lease(self.session_time_to_live)
            self.session_id = self.lease
            log.info('created-etcd-lease', lease=self.session_id)
            self._start_session_tracking()

        yield self._retry(_create_session)

    @inlineCallbacks
    def _delete_session(self):
        try:
            yield self.lease.revoke()
        except Exception as e:
            log.exception('failed-to-delete-session',
                          session_id=self.session_id)

    @inlineCallbacks
    def _create_membership_record(self):
        yield self._do_create_membership_record_with_retries()
        reactor.callLater(0, self._maintain_membership_record)

    @inlineCallbacks
    def _maintain_membership_record(self):
        try:
            while 1:
                valid_membership = yield self._assert_membership_record_valid()
                if not valid_membership:
                    log.info('recreating-membership-before',
                             session=self.session_id)
                    yield self._do_create_membership_record_with_retries()
                    log.info('recreating-membership-after',
                             session=self.session_id)
                else:
                    log.debug('valid-membership', session=self.session_id)
                # Async sleep before checking the membership record again
                yield asleep(self.membership_maintenance_loop_delay)

        except Exception, e:
            log.exception('unexpected-error-leader-trackin', e=e)
        finally:
            # except in shutdown, the loop must continue (after a short delay)
            if not self.shutting_down:
                reactor.callLater(self.membership_watch_relatch_delay,
                                  self._maintain_membership_record)

    def _create_membership_record_data(self):
        member_record = dict()
        member_record['status'] = 'alive'
        member_record['host_address'] = self.external_host_address
        return member_record

    @inlineCallbacks
    def _assert_membership_record_valid(self):
        try:
            log.info('membership-record-before')
            is_timeout, (_, record) = yield \
                                        self.consul_get_with_timeout(
                                                key=self.membership_record_key,
                                                index=0,
                                                timeout=5)
            if is_timeout:
                returnValue(False)

            log.info('membership-record-after', record=record)
            if record is None or \
                            'Session' not in record:
                log.info('membership-record-change-detected',
                         old_session=self.session_id,
                         record=record)
                returnValue(False)
            else:
                returnValue(True)
        except Exception as e:
            log.exception('membership-validation-exception', e=e)
            returnValue(False)

    @inlineCallbacks
    def _do_create_membership_record_with_retries(self):
        while 1:
            log.info('recreating-membership', session=self.session_id)
            result = yield self._retry(
                'PUT',
                self.membership_record_key,
                dumps(self._create_membership_record_data()),
                acquire=self.session_id)
            if result:
                log.info('new-membership-record-created',
                         session=self.session_id)
                break
            else:
                log.warn('cannot-create-membership-record')
                yield self._backoff('stale-membership-record')

    def _start_session_tracking(self):
        reactor.callLater(0, self._session_tracking_loop)

    @inlineCallbacks
    def _session_tracking_loop(self):

        @inlineCallbacks
        def _redo_session():
            log.info('_redo_session-before')
            yield self._delete_session()

            # Create a new etcd connection/session with a new lease
            try:
                self.etcd = Client(reactor, self.etcd_url)
                self.lease = yield self.etcd.lease(self.session_time_to_live)
                self.session_id = self.lease
                log.info('new-etcd-session', session=self.session_id)

            except Exception as e:
                log.exception('could-not-create-an-etcd-lease', e=e)

        @inlineCallbacks
        def _renew_session(m_callback):
            try:
                time_left = yield self.lease.remaining()
                log.info('_renew_session', time_left=time_left)
                result = yield self.lease.refresh()
                log.info('just-renewed-session', result=result)
                if not m_callback.called:
                    # Triggering callback will cancel the timeout timer
                    log.info('trigger-callback-to-cancel-timeout-timer')
                    m_callback.callback(result)
                else:
                    # Timeout event has already been called.  Just ignore
                    # this event
                    log.info('renew-called-after-timeout, etcd ref changed?')
            except Exception, e:
                # Let the invoking method receive a timeout
                log.exception('could-not-renew-session', e=e)

        try:
            while 1:
                log.debug('session-tracking-start')
                rcvd = DeferredWithTimeout(
                    timeout=self.session_renewal_timeout)
                _renew_session(rcvd)
                try:
                    _ = yield rcvd
                except TimeOutError as e:
                    log.info('session-renew-timeout', e=e)
                    # Redo the session
                    yield _redo_session()
                except Exception as e:
                    log.exception('session-renew-exception', e=e)
                else:
                    log.debug('successfully-renewed-session')

                # Async sleep before the next session tracking
                yield asleep(self.session_renewal_loop_delay)

        except Exception as e:
            log.exception('renew-exception', e=e)
        finally:
            reactor.callLater(self.session_renewal_loop_delay,
                              self._session_tracking_loop)

    def _start_leader_tracking(self):
        reactor.callLater(0, self._leadership_tracking_loop)

    @inlineCallbacks
    def _leadership_tracking_loop(self):
        log.info('leadership-attempt-before')
            
        # Try to acquire leadership lease via test-and-set operation.
        # Success means the leader key was previously absent and was
        # just re-created by this instance.
        leader_prefix = bytes(self.leader_prefix)
        txn = Transaction(
            compare=[
                CompVersion(leader_prefix, '==', 0)
            ],
            success=[
                OpSet(leader_prefix, bytes(self.instance_id), lease=self.lease),
                OpGet(leader_prefix)
            ],
            failure=[]
        )
        newly_asserted = False
        try:
            result = yield self.etcd.submit(txn)
        except Failed as failed:
            log.info('Leader key PRESENT')
            for response in failed.responses:
                log.info('Leader key already present', response=response)
        else:
            newly_asserted = True
            log.info('Leader key ABSENT')
            for response in result.responses:
                log.info('Leader key was absent', response=response)

        log.info('leadership-attempt-after')

        # Confirm that the assertion succeeded by reading back the value
        # of the leader key.
        leader = None
        result = yield self.etcd.get(b'service/voltha/leader')
        if result.kvs:
            kv = result.kvs[0]
            leader = kv.value
            log.info('Leader readback', leader=leader, instance=self.instance_id)    

        if leader is None:
            log.info('Failed to read leader key')
        elif leader == self.instance_id:
            if newly_asserted:
                log.info("I JUST BECAME LEADER!")
                yield self._assert_leadership()
            else:
                log.info("I'm an aging LEADER")
        else:
            log.info('The LEADER is another', leader=leader)
            yield self._assert_nonleadership(leader)

        # May have to add code here to handle case where, for some reason, the lease
        # had been blown away and the txn failed for that reason

        # except in shutdown, the loop must continue (after a short delay)
        if not self.shutting_down:
            reactor.callLater(self.tracking_loop_delay,
                              self._leadership_tracking_loop)

    @inlineCallbacks
    def _assert_leadership(self):
        """(Re-)assert leadership"""
        if not self.i_am_leader:
            self.i_am_leader = True
            self._set_leader_id(self.instance_id)
            yield self._just_gained_leadership()

    @inlineCallbacks
    def _assert_nonleadership(self, leader_id):
        """(Re-)assert non-leader role"""

        # update leader_id anyway
        self._set_leader_id(leader_id)

        if self.i_am_leader:
            self.i_am_leader = False
            yield self._just_lost_leadership()

    def _set_leader_id(self, leader_id):
        self.leader_id = leader_id
        deferreds, self.wait_for_leader_deferreds = \
            self.wait_for_leader_deferreds, []
        for d in deferreds:
            d.callback(leader_id)

    def _just_gained_leadership(self):
        log.info('became-leader')
        self.leader = Leader(self)
        return self.leader.start()

    def _just_lost_leadership(self):
        log.info('lost-leadership')
        return self._halt_leader()

    def _halt_leader(self):
        if self.leader:
            d = self.leader.stop()
            self.leader = None
        return d

    def get_kv_client(self):
        return self.etcd

    @inlineCallbacks
    def _retry(self, operation, *args, **kw):
        prefix = False
        for name, value in kw.items():
            if name == 'acquire':
                lease = value
                kw['lease'] = lease
                kw.pop('acquire')
            elif name == 'keys':
                kw['keys_only'] = True
                kw.pop('keys')
            elif name=='recurse':
#               if value == 'True':
                prefix = True
                keyset = KeySet(bytes(args[0]), prefix=True)
                kw.pop('recurse')
        log.info('start-op', operation=operation, args=args, kw=kw)

        while 1:
            try:
                etcd = yield self.get_kv_client()
                if operation == 'GET':
                    key = bytes(args[0])
                    # If multiple keys requested, return a list
                    # else return a single record
                    if not prefix:
                        index = 0
                        record = dict()
                        res = yield etcd.get(key, **kw)
                        if res.kvs:
                            if len(res.kvs) == 1:
                                kv = res.kvs[0]
                                index = kv.mod_revision
                                record['Key'] = kv.key
                                record['Value'] = kv.value
                                record['ModifyIndex'] = index
                                record['Session'] = self.lease.lease_id if self.lease else ''
                                result = (index, record)
                    else:
                        # Get values for all keys that match the prefix
                        index = 0
                        records = []
                        res = yield etcd.get(keyset, **kw)
                        if args[0] == 'service/voltha/assignments/':
                            log.info('assignments', result=res)
                        if res.kvs and len(res.kvs) > 0:
                            for kv in res.kvs:
                                # Which index should be returned? The max over all keys?
                                if kv.mod_revision > index:
                                    index = kv.mod_revision
                                rec = dict()
                                rec['Key'] = kv.key
                                rec['Value'] = kv.value
                                rec['ModifyIndex'] = kv.mod_revision
                                rec['Session'] = self.lease.lease_id if self.lease else ''
                                records.append(rec)
                        result = (index, records)
                elif operation == 'PUT':
                    key = bytes(args[0])
                    result = yield etcd.set(key, args[1], **kw)
                elif operation == 'DELETE':
                    key = bytes(args[0])
                    result = yield etcd.delete(key, **kw)
                else:
                    # Default case - consider operation as a function call
                    result = yield operation(*args, **kw)
                self._clear_backoff()
                break
            except Exception, e:
                if not self.shutting_down:
                    log.exception(e)
                yield self._backoff('unknown-error')

        log.info('end-op', operation=operation, args=args, kw=kw, result=result)
        returnValue(result)

    @inlineCallbacks
    def consul_get_with_timeout(self, key, timeout, **kw):
        """
        Query etcd with a timeout
        :param key: Key to query
        :param timeout: timeout value
        :param kw: additional key-value params
        :return: (is_timeout, (index, result)).

        The Consul version of this method performed a 'wait-type' get operation
        that returned a result when the key's value had a ModifyIndex greater
        than the 'index' argument. Not sure etcd supports this functionality.
        """

        # Intercept 'index' argument
        for name, value in kw.items():
            if name == 'index':
                mod_revision = value
                log.info('consul_get_with_timeout', index=mod_revision)
                kw.pop('index')
                break

        @inlineCallbacks
        def _get(key, m_callback):
            try:
                (index, result) = yield self._retry('GET', key, **kw)
                if index > mod_revision and not m_callback.called:
                    log.debug('got-result-cancelling-timer')
                    m_callback.callback((index, result))
            except Exception as e:
                log.exception('got-exception', e=e)

        try:
            rcvd = DeferredWithTimeout(timeout=timeout)
            _get(key, rcvd)
            try:
                result = yield rcvd
                log.debug('result-received', result=result)
                returnValue((False, result))
            except TimeOutError as e:
                log.debug('timeout-or-no-data-change', key=key)
            except Exception as e:
                log.exception('exception', e=e)
        except Exception as e:
            log.exception('exception', e=e)

        returnValue((True, (None, None)))