charm-interface-ovsdb/src/lib/ovsdb.py

# Copyright 2019 Canonical Ltd
#
# 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.

# NOTE: Do not use the ``charms.reactive`` decorators that take flags
#       as arguments to wrap functions or methods in this shared library.
#
#       Consume the shared code from the interface specific files and declare
#       which flags to react to there.

import inspect
import ipaddress

import charmhelpers.core as ch_core

import charms.reactive as reactive


class OVSDB(reactive.Endpoint):
    DB_NB_PORT = 6641
    DB_SB_PORT = 6642

    def _format_addr(self, addr):
        """Validate and format IP address

        :param addr: IPv6 or IPv4 address
        :type addr: str
        :returns: Address string, optionally encapsulated in brackets ([])
        :rtype: str
        :raises: ValueError
        """
        ipaddr = ipaddress.ip_address(addr)
        if isinstance(ipaddr, ipaddress.IPv6Address):
            fmt = '[{}]'
        else:
            fmt = '{}'
        return fmt.format(ipaddr)

    def _endpoint_local_bound_addr(self):
        """Retrieve local address bound to endpoint.

        :returns: IPv4 or IPv6 address bound to endpoint
        :rtype: str
        """
        for relation in self.relations:
            ng_data = ch_core.hookenv.network_get(
                self.expand_name('{endpoint_name}'),
                relation_id=relation.relation_id)
            for interface in ng_data.get('bind-addresses', []):
                for addr in interface.get('addresses', []):
                    return self._format_addr(addr['address'])

    @property
    def cluster_local_addr(self):
        """Retrieve local address bound to endpoint.

        :returns: IPv4 or IPv6 address bound to endpoint
        :rtype: str
        """
        return self._endpoint_local_bound_addr()

    def _remote_addrs(self, key):
        """Retrieve addresses published by remote units.

        :param key: Relation data key to retrieve value from.
        :type key: str
        :returns: IPv4 or IPv6 addresses published by remote units.
        :rtype: Iterator[str]
        """
        for relation in self.relations:
            for unit in relation.units:
                try:
                    addr = self._format_addr(
                        unit.received.get(key, ''))
                    yield addr
                except ValueError:
                    continue

    @property
    def cluster_remote_addrs(self):
        """Retrieve remote addresses bound to remote endpoint.

        :returns: IPv4 or IPv6 addresses bound to remote endpoints.
        :rtype: Iterator[str]
        """
        return self._remote_addrs('bound-address')

    @property
    def db_nb_port(self):
        """Provide port number for OVN Northbound OVSDB.

        :returns: port number for OVN Northbound OVSDB.
        :rtype: int
        """
        return self.DB_NB_PORT

    @property
    def db_sb_port(self):
        """Provide port number for OVN Southbound OVSDB.

        :returns: port number for OVN Southbound OVSDB.
        :rtype: int
        """
        return self.DB_SB_PORT

    def db_connection_strs(self, addrs, port, proto='ssl'):
        """Provide connection strings.

        :param port: Port number
        :type port: int
        :param proto: Protocol
        :type proto: str
        :returns: connection strings
        :rtype: Iterator[str]
        """
        for addr in addrs:
            if all([proto, addr, str(port)]):
                yield ':'.join((proto, addr, str(port)))

    @property
    def db_nb_connection_strs(self):
        """Provide OVN Northbound OVSDB connection strings.

        :returns: OVN Northbound OVSDB connection strings.
        :rtpye: Iterator[str]
        """
        return self.db_connection_strs(self.cluster_remote_addrs,
                                       self.db_nb_port)

    @property
    def db_sb_connection_strs(self):
        """Provide OVN Southbound OVSDB connection strings.

        :returns: OVN Southbound OVSDB connection strings.
        :rtpye: Iterator[str]
        """
        return self.db_connection_strs(self.cluster_remote_addrs,
                                       self.db_sb_port)

    def expected_units_available(self):
        """Whether expected units have joined and published data on a relation.

        NOTE: This does not work for the peer relation, see separate method
              for that in the peer relation implementation.
        :returns: True if expected units have joined and published data,
                  False otherwise.
        :rtype: bool
        """
        expected_related_units = len(
            list(ch_core.hookenv.expected_related_units(
                self.expand_name('{endpoint_name}'))))
        # A minimum of 3 ovn-central units are required for operation, if
        # this number is 1 chances are we are consuming the realation over CMR
        # which would not provide us with an accurate number.
        if expected_related_units == 1:
            expected_related_units = 3
        if len(self.all_joined_units) >= expected_related_units:
            bound_addrs = [unit.received.get('bound-address', None) is not None
                           for relation in self.relations
                           for unit in relation.units]
            return all(bound_addrs)
        return False

    def publish_cluster_local_addr(self, addr=None):
        """Announce address on relation.

        This will be used by our peers and clients to build a connection
        string to the remote cluster.

        :param addr: Override address to announce.
        :type addr: Optional[str]
        """
        for relation in self.relations:
            relation.to_publish['bound-address'] = (
                addr or self.cluster_local_addr)

    def joined(self):
        ch_core.hookenv.log('{}: {} -> {}'
                            .format(self._endpoint_name,
                                    type(self).__name__,
                                    inspect.currentframe().f_code.co_name),
                            level=ch_core.hookenv.INFO)
        reactive.set_flag(self.expand_name('{endpoint_name}.connected'))

    def broken(self):
        reactive.clear_flag(self.expand_name('{endpoint_name}.available'))
        reactive.clear_flag(self.expand_name('{endpoint_name}.connected'))