common/service.py

"""
Copyright 2020 The Magma Authors.

This source code is licensed under the BSD-style license found in the
LICENSE file in the root directory of this source tree.

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 asyncio
import faulthandler
import functools
import logging
import os
import signal
import time
from concurrent import futures
from typing import Any, Dict, List, Optional

import grpc
import pkg_resources
from common.log_count_handler import MsgCounterHandler
from common.log_counter import ServiceLogErrorReporter
from common.metrics_export import get_metrics
from common.service_registry import ServiceRegistry
from configuration.exceptions import LoadConfigError
from configuration.mconfig_managers import get_mconfig_manager
from configuration.service_configs import load_service_config
from orc8r.protos.common_pb2 import LogLevel, Void
from orc8r.protos.metricsd_pb2 import MetricsContainer
from orc8r.protos.service303_pb2 import (
    GetOperationalStatesResponse,
    ReloadConfigResponse,
    ServiceInfo,
    State,
)
from orc8r.protos.service303_pb2_grpc import (
    Service303Servicer,
    Service303Stub,
    add_Service303Servicer_to_server,
)

MAX_DEFAULT_WORKER = 10


async def loop_exit():
    """
    Stop the loop in an async context
    """
    loop = asyncio.get_event_loop()
    loop.stop()


class MagmaService(Service303Servicer):
    """
    MagmaService provides the framework for all Magma services.
    This class also implements the Service303 interface for external
    entities to interact with the service.
    """

    def __init__(self, name, empty_mconfig, loop=None):
        self._name = name
        self._port = 0
        self._get_status_callback = None
        self._get_operational_states_cb = None
        self._log_count_handler = MsgCounterHandler()

        # Init logging before doing anything
        logging.basicConfig(
            level=logging.INFO,
            format='[%(asctime)s %(levelname)s %(name)s] %(message)s',
        )
        # Add a handler to count errors
        logging.root.addHandler(self._log_count_handler)

        # Set gRPC polling strategy
        self._set_grpc_poll_strategy()

        # Load the managed config if present
        self._mconfig = empty_mconfig
        self._mconfig_metadata = None
        self._mconfig_manager = get_mconfig_manager()
        self.reload_mconfig()

        self._state = ServiceInfo.STARTING
        self._health = ServiceInfo.APP_UNHEALTHY
        if loop is None:
            loop = asyncio.get_event_loop()
        self._loop = loop
        self._start_time = int(time.time())
        self._register_signal_handlers()

        # Load the service config if present
        self._config = None
        self.reload_config()

        # Count errors
        self.log_counter = ServiceLogErrorReporter(
            loop=self._loop,
            service_config=self._config,
            handler=self._log_count_handler,
        )
        self.log_counter.start()

        # Operational States
        self._operational_states = []

        self._version = '0.0.0'
        # Load the service version if available
        try:
            # Check if service on docker
            if self._config and 'init_system' in self._config \
                    and self._config['init_system'] == 'docker':
                # image comes in form of "feg_gateway_python:<IMAGE_TAG>\n"
                # Skip the "feg_gateway_python:" part
                image = os.popen(
                    'docker ps --filter name=magmad --format "{{.Image}}" | '
                    'cut -d ":" -f 2',
                )
                image_tag = image.read().strip('\n')
                self._version = image_tag
            else:
                self._version = pkg_resources.get_distribution('orc8r').version
        except pkg_resources.ResolutionError as e:
            logging.info(e)

        if self._config and 'grpc_workers' in self._config:
            self._server = grpc.server(
                futures.ThreadPoolExecutor(
                    max_workers=self._config['grpc_workers'],
                ),
            )
        else:
            self._server = grpc.server(
                futures.ThreadPoolExecutor(max_workers=MAX_DEFAULT_WORKER),
            )
        add_Service303Servicer_to_server(self, self._server)

    @property
    def version(self):
        """Return the current running version of the Magma service"""
        return self._version

    @property
    def name(self):
        """Return the name of service

        Returns:
            tr: name of service
        """
        return self._name

    @property
    def rpc_server(self):
        """Return the RPC server used by the service"""
        return self._server

    @property
    def port(self):
        """Return the listening port of the service"""
        return self._port

    @property
    def loop(self):
        """Return the asyncio event loop used by the service"""
        return self._loop

    @property
    def state(self):
        """Return the state of the service"""
        return self._state

    @property
    def config(self) -> Dict[str, Any]:
        """Return the service config"""
        return self._config

    @property
    def mconfig(self):
        """Return the managed config"""
        return self._mconfig

    @property
    def mconfig_metadata(self):
        """Return the metadata of the managed config"""
        return self._mconfig_metadata

    @property
    def mconfig_manager(self):
        """Return the mconfig manager for this service"""
        return self._mconfig_manager

    def reload_config(self):
        """Reload the local config for the service"""
        try:
            self._config = load_service_config(self._name)
            self._setup_logging()
        except LoadConfigError as e:
            logging.warning(e)

    def reload_mconfig(self):
        """Reload the managed config for the service"""
        try:
            # reload mconfig manager in case feature flag for streaming changed
            self._mconfig_manager = get_mconfig_manager()
            self._mconfig = self._mconfig_manager.load_service_mconfig(
                self._name,
                self._mconfig,
            )
            self._mconfig_metadata = \
                self._mconfig_manager.load_mconfig_metadata()
        except LoadConfigError as e:
            logging.warning(e)

    def add_operational_states(self, states: List[State]):
        """Add a list of states into the service

        Args:
            states (List[State]): [description]
        """
        self._operational_states.extend(states)

    def run(self):
        """
        Start the service and runs the event loop until a term signal
        is received or a StopService rpc call is made on the Service303
        interface.
        """
        logging.info("Starting %s...", self._name)
        (host, port) = ServiceRegistry.get_service_address(self._name)
        self._port = self._server.add_insecure_port('{}:{}'.format(host, port))
        logging.info("Listening on address %s:%d", host, self._port)
        self._state = ServiceInfo.ALIVE
        # Python services are healthy immediately when run
        self._health = ServiceInfo.APP_HEALTHY
        self._server.start()
        self._loop.run_forever()
        # Waiting for the term signal or StopService rpc call

    def close(self):
        """
        Clean up the service before termination. This needs to be
        called atleast once after the service has been inited.
        """
        self._loop.close()
        self._server.stop(0).wait()

    def register_get_status_callback(self, get_status_callback):
        """Register function for getting status

        Must return a map(string, string)
        """
        self._get_status_callback = get_status_callback

    def register_operational_states_callback(self, get_operational_states_cb):
        """Register the callback function that gets called on GetOperationalStates rpc

        Args:
            get_operational_states_cb ([type]): callback function
        """
        self._get_operational_states_cb = get_operational_states_cb

    def _stop(self, reason):
        """Stop the service gracefully"""
        logging.info("Stopping %s with reason %s...", self._name, reason)
        self._state = ServiceInfo.STOPPING
        self._server.stop(0)

        for pending_task in asyncio.Task.all_tasks(self._loop):
            pending_task.cancel()

        self._state = ServiceInfo.STOPPED
        self._health = ServiceInfo.APP_UNHEALTHY
        asyncio.ensure_future(loop_exit())

    def _set_grpc_poll_strategy(self):
        """
        The new default 'epollex' poll strategy is causing fd leaks, leading to
        service crashes after 1024 open fds.
        See https://github.com/grpc/grpc/issues/15759
        """
        os.environ['GRPC_POLL_STRATEGY'] = 'epoll1,poll'

    def _get_log_level_from_config(self) -> Optional[int]:
        if self._config is None:
            return None
        log_level = self._config.get('log_level', None)
        if log_level is None:
            return None
        # convert from log level string to LogLevel enum value
        try:
            proto_level = LogLevel.Value(log_level)
        except ValueError:
            logging.error(
                'Unknown logging level in config: %s, ignoring',
                log_level,
            )
            return None
        return proto_level

    def _get_log_level_from_mconfig(self) -> Optional[int]:
        if self._mconfig is None:
            return None
        return self._mconfig.log_level

    def _setup_logging(self):
        """Set up log level from config values

        The config file on the AGW takes precedence over the mconfig
        If neither config file nor mconfig has the log level config, default to INFO
        """
        log_level_from_config = self._get_log_level_from_config()
        log_level_from_mconfig = self._get_log_level_from_mconfig()

        if log_level_from_config is not None:
            log_level = log_level_from_config
        elif log_level_from_mconfig is not None:
            log_level = log_level_from_mconfig
        else:
            logging.warning(
                'logging level is not specified in either yml config '
                'or mconfig, defaulting to INFO',
            )
            log_level = LogLevel.Value('INFO')
        self._set_log_level(log_level)

    @staticmethod
    def _set_log_level(proto_level: int):
        """Set log level based on proto-enum level

        Args:
            proto_level (int): proto enum defined in common.proto
        """
        if proto_level == LogLevel.Value('DEBUG'):
            level = logging.DEBUG
        elif proto_level == LogLevel.Value('INFO'):
            level = logging.INFO
        elif proto_level == LogLevel.Value('WARNING'):
            level = logging.WARNING
        elif proto_level == LogLevel.Value('ERROR'):
            level = logging.ERROR
        elif proto_level == LogLevel.Value('FATAL'):
            level = logging.FATAL
        else:
            logging.error(
                'Unknown logging level: %d, defaulting to INFO',
                proto_level,
            )
            level = logging.INFO

        logging.info(
            "Setting logging level to %s",
            logging.getLevelName(level),
        )
        logger = logging.getLogger('')
        logger.setLevel(level)

    def _register_signal_handlers(self):
        """Register signal handlers

        Right now we just exit on SIGINT/SIGTERM/SIGQUIT
        """
        for signame in ['SIGINT', 'SIGTERM', 'SIGQUIT']:
            self._loop.add_signal_handler(
                getattr(signal, signame),
                functools.partial(self._stop, signame),
            )

        def _signal_handler():
            logging.info('Handling SIGHUP...')
            faulthandler.dump_traceback()
        self._loop.add_signal_handler(
            signal.SIGHUP, functools.partial(_signal_handler),
        )

    def GetServiceInfo(self, request, context):
        """Return the service info (name, version, state, meta, etc.)"""
        service_info = ServiceInfo(
            name=self._name,
            version=self._version,
            state=self._state,
            health=self._health,
            start_time_secs=self._start_time,
        )
        if self._get_status_callback is not None:
            status = self._get_status_callback()
            try:
                service_info.status.meta.update(status)
            except (TypeError, ValueError) as exp:
                logging.error("Error getting service status: %s", exp)
        return service_info

    def StopService(self, request, context):
        """Handle request to stop the service"""
        logging.info("Request to stop service.")
        self._loop.call_soon_threadsafe(self._stop, 'RPC')
        return Void()

    def GetMetrics(self, request, context):
        """
        Collects timeseries samples from prometheus python client on this
        process
        """
        metrics = MetricsContainer()
        metrics.family.extend(get_metrics())
        return metrics

    def SetLogLevel(self, request, context):
        """Handle request to set the log level"""
        self._set_log_level(request.level)
        return Void()

    def SetLogVerbosity(self, request, context):
        pass  # Not Implemented

    def ReloadServiceConfig(self, request, context):
        """Handle request to reload the service config file"""
        self.reload_config()
        return ReloadConfigResponse(result=ReloadConfigResponse.RELOAD_SUCCESS)

    def GetOperationalStates(self, request, context):
        """Return the  operational states of devices managed by this service."""
        res = GetOperationalStatesResponse()
        if self._get_operational_states_cb is not None:
            states = self._get_operational_states_cb()
            res.states.extend(states)
        return res


def get_service303_client(service_name: str, location: str) \
        -> Optional[Service303Stub]:
    """
    Return a grpc client attached to the given service
    name and location.
    Example Use: client = get_service303_client("state", ServiceRegistry.LOCAL)
    """
    try:
        chan = ServiceRegistry.get_rpc_channel(
            service_name,
            location,
        )
        return Service303Stub(chan)
    except ValueError:
        # Service can't be contacted
        logging.error('Failed to get RPC channel to %s', service_name)
        return None