Merge 38136d7 into 94eb6ce

tendrl/node_agent/alerts/alert_socket.py

@@ -0,0 +1,60 @@
+from _socket import error as _socket_error
+import gevent.event
+import gevent.greenlet
+from gevent.server import StreamServer
+from io import BlockingIOError
+import json
+import logging
+from tendrl.commons.alert import AlertUtils
+from tendrl.node_agent.alerts.base_alert_handler import AlertHandlerManager
+from tendrl.node_agent.alerts.base_alert_handler import NoHandlerException
+import tendrl.node_agent.manager.utils as utils
+import uuid
+
+LOG = logging.getLogger(__name__)
+RECEIVE_DATA_SIZE = 4096
+db_client = None
+
+
+class AlertsManager(gevent.greenlet.Greenlet):
+
+    def read_socket(self, sock, address):
+        try:
+            data = sock.recv(RECEIVE_DATA_SIZE)
+            alert_utils = AlertUtils(db_client)
+            alert_json = json.loads(data)
+            alert_json['alert_id'] = str(uuid.uuid4())
+            alert_json['significance'] = 'HIGH'
+            alert_json['node_id'] = utils.get_local_node_context()
+            alert_json['ackedby'] = ''
+            alert_json['acked'] = False
+            alert = alert_utils.to_obj(alert_json)
+            AlertHandlerManager(db_client).handle(alert)
+        except (
+            KeyError,
+            TypeError,
+            ValueError,
+            NoHandlerException
+        ) as ex:
+            LOG.error('Failed to handle data on alert socket.Error %s' % ex)
+
+    def __init__(self, alerts_socket_addr, alerts_socket_port, etcd_client):
+        super(AlertsManager, self).__init__()
+        self.hostname = alerts_socket_addr
+        self.port = alerts_socket_port
+        global db_client
+        db_client = etcd_client
+        self.server = StreamServer(
+            (self.hostname, int(self.port)),
+            self.read_socket
+        )
+
+    def _run(self):
+        try:
+            self.server.serve_forever()
+        except (TypeError, BlockingIOError, _socket_error, ValueError) as ex:
+            LOG.error('Error trying to serve the alerting socket forever.\
+                Error %s' % ex)
+
+    def stop(self):
+        self.server.close()

tendrl/node_agent/alerts/base_alert_handler.py

@@ -0,0 +1,123 @@
+import datetime
+import etcd
+import importlib
+import inspect
+import logging
+import os
+import six
+from tendrl.commons.alert import AlertUtils
+from tendrl.commons.singleton import to_singleton
+
+LOG = logging.getLogger(__name__)
+
+
+class NoHandlerException(Exception):
+    pass
+
+
+class HandlerMount(type):
+
+    def __init__(cls, name, bases, attrs):
+        if not hasattr(cls, 'handlers'):
+            cls.handlers = []
+        else:
+            cls.register_handler(cls)
+
+    def register_handler(cls, handler):
+        instance = handler()
+        cls.handlers.append(instance)
+
+
+@six.add_metaclass(HandlerMount)
+class AlertHandler(object):
+    def __init__(self):
+        self.time_stamp = datetime.datetime.now().isoformat()
+        self.alert = None
+        self.handles = ''
+
+    def update_alert(self, etcd_client):
+        # Fetch alerts in etcd
+        try:
+            alerts = AlertUtils(etcd_client).get_alerts()
+            # Check if similar alert already exists
+            for curr_alert in alerts:
+                # If similar alert exists, update the similar alert to etcd
+                if AlertUtils(etcd_client).is_same(self.alert, curr_alert):
+                    self.alert = AlertUtils(
+                        etcd_client
+                    ).update(
+                        self.alert,
+                        curr_alert
+                    )
+                    if not AlertUtils(
+                        etcd_client
+                    ).equals(
+                        self.alert,
+                        curr_alert
+                    ):
+                        AlertUtils(etcd_client).store_alert(self.alert)
+                    return
+                # else add this new alert to etcd
+            AlertUtils(etcd_client).store_alert(self.alert)
+        except etcd.EtcdKeyNotFound:
+            AlertUtils(etcd_client).store_alert(self.alert)
+        except etcd.EtcdConnectionFailed as ex:
+            LOG.error(
+                'Failed to fetch existing alerts.Error %s' % ex,
+                exc_info=True
+            )
+
+    def handle(self, alert_obj, etcd_client):
+        try:
+            self.alert = alert_obj
+            self.alert.significance = 'HIGH'
+            self.update_alert(etcd_client)
+        except Exception as ex:
+            LOG.error(
+                'Failed to handle the alert %s.Error %s'
+                % (str(alert_obj.to_json_string()), str(ex)),
+                exc_info=True
+            )
+
+
+@to_singleton
+class AlertHandlerManager(object):
+    def load_handlers(self):
+        try:
+            path = os.path.dirname(os.path.abspath(__file__)) + '/handlers'
+            pkg = 'tendrl.node_agent.alerts.handlers'
+            for py in [f[:-3] for f in os.listdir(path)
+                       if f.endswith('.py') and f != '__init__.py']:
+                handler_name = '.'.join([pkg, py])
+                mod = importlib.import_module(handler_name)
+                clsmembers = inspect.getmembers(mod, inspect.isclass)
+                for name, cls in clsmembers:
+                    exec("from %s import %s" % (handler_name, name))
+        except (SyntaxError, ValueError, ImportError) as ex:
+            LOG.error('Failed to load the alert handlers. Error %s' %
+                      ex, exc_info=True)
+            raise ex
+
+    def __init__(self, etcd_client):
+        try:
+            self.load_handlers()
+            self.etcd_client = etcd_client
+            alert_handlers = []
+            for handler in AlertHandler.handlers:
+                alert_handlers.append(handler.handles)
+            self.etcd_client.write(
+                '/alerting/alert_types/node_agent',
+                alert_handlers
+            )
+        except (SyntaxError, ValueError, ImportError) as ex:
+            raise ex
+
+    def handle(self, alert):
+        for handler in AlertHandler.handlers:
+            if handler.handles == alert.resource:
+                handler.handle(alert, self.etcd_client)
+                return
+        raise NoHandlerException(
+            'No alert handler defined for %s and hence cannot handle alert %s'
+            % (alert['resource'], str(alert))
+        )

tendrl/node_agent/alerts/handlers/cpu_handler.py

@@ -0,0 +1,7 @@
+from tendrl.node_agent.alerts.base_alert_handler import AlertHandler
+
+
+class CpuHandler(AlertHandler):
+    def __init__(self):
+        AlertHandler.__init__(self)
+        self.handles = 'cpu'

tendrl/node_agent/alerts/handlers/memory_handler.py

@@ -0,0 +1,7 @@
+from tendrl.node_agent.alerts.base_alert_handler import AlertHandler
+
+
+class MemoryHandler(AlertHandler):
+    def __init__(self):
+        AlertHandler.__init__(self)
+        self.handles = 'memory'

tendrl/node_agent/alerts/handlers/mount_point_handler.py

@@ -0,0 +1,7 @@
+from tendrl.node_agent.alerts.base_alert_handler import AlertHandler
+
+
+class MountPointHandler(AlertHandler):
+    def __init__(self):
+        AlertHandler.__init__(self)
+        self.handles = 'df'

tendrl/node_agent/alerts/handlers/swap_handler.py

@@ -0,0 +1,7 @@
+from tendrl.node_agent.alerts.base_alert_handler import AlertHandler
+
+
+class SwapHandler(AlertHandler):
+    def __init__(self):
+        AlertHandler.__init__(self)
+        self.handles = 'swap'

tendrl/node_agent/manager/manager.py

@@ -13,10 +13,9 @@
 from tendrl.commons.etcdobj import etcdobj
 from tendrl.commons.log import setup_logging
 from tendrl.commons.manager import manager as common_manager
+from tendrl.node_agent.alerts.alert_socket import AlertsManager
 from tendrl.node_agent.discovery.platform.manager import PlatformManager
 from tendrl.node_agent.discovery.sds.manager import SDSDiscoveryManager
-from tendrl.node_agent.manager.tendrl_definitions_node_agent import data as \
-    def_data
 from tendrl.node_agent.manager import utils
 from tendrl.node_agent.persistence.cpu import Cpu
 from tendrl.node_agent.persistence.disk import Disk
@@ -28,7 +27,6 @@
 from tendrl.node_agent.persistence.platform import Platform
 from tendrl.node_agent.persistence.service import Service
 from tendrl.node_agent.persistence.tendrl_context import TendrlContext
-from tendrl.node_agent.persistence.tendrl_definitions import TendrlDefinitions
 
 config = load_config("node-agent",
                      "/etc/tendrl/node-agent/node-agent.conf.yaml")
@@ -123,6 +121,19 @@ def __init__(self, machine_id):
         self.register_node(machine_id)
         self.load_and_execute_platform_discovery_plugins()
         self.load_and_execute_sds_discovery_plugins()
+        self._alerts_manager = AlertsManager(
+            config['configuration']['tendrl_alerts_socket_addr'],
+            config['configuration']['tendrl_alerts_socket_port'],
+            self.etcd_orm.client
+        )
+
+    def start(self):
+        super(NodeAgentManager, self).start()
+        self._alerts_manager.start()
+
+    def stop(self):
+        super(NodeAgentManager, self).stop()
+        self._alerts_manager.stop()
 
     def register_node(self, machine_id):
         update_node_context(self, machine_id)
@@ -143,8 +154,7 @@ def register_node(self, machine_id):
                 status="UP"
             )
         )
-        self.persister_thread.update_tendrl_definitions(TendrlDefinitions(
-            updated=str(time.time()), data=def_data))
+        self.persister_thread.append_definitions()
 
     def on_pull(self, raw_data):
         LOG.info("on_pull, Updating Node_context data")

tendrl/node_agent/manager/tendrl_definitions_node_agent.py

@@ -318,6 +318,16 @@
           run: tendrl.node_agent.atoms.service.configure.Configure
           type: Update
           uuid: b90a0d97-8c9f-4ab1-8f64-dbb5638159a3
+        check_service_status:
+          enabled: true
+          inputs:
+            mandatory:
+              - Node.fqdn
+              - Service.name
+          name: "check whether the service is running"
+          run: tendrl.node_agent.objects.service.atoms.check_service_status.CheckServiceStatus
+          type: Create
+          uuid: eda0b13a-7362-48d5-b5ca-4b6d6533a5ab
       attrs:
         config_data:
           help: "Configuration data for the service"

tendrl/node_agent/objects/node/atoms/cmd.py

@@ -1,3 +1,4 @@
+import shlex
 import subprocess
 
 from tendrl.commons.atoms import base_atom
@@ -6,6 +7,6 @@
 class Cmd(base_atom.BaseAtom):
     def run(self, parameters):
         cmd = parameters.get("Node.cmd_str")
-        cmd = ["nohup"] + cmd.split(" ")
+        cmd = ["nohup"] + shlex.split(cmd)
         subprocess.Popen(cmd)
         return True

tendrl/node_agent/objects/service/atoms/check_service_status.py

@@ -0,0 +1,12 @@
+import os
+
+
+class CheckServiceStatus(object):
+    def run(self, parameters):
+        service_name = parameters.get("Service.name")
+        response = os.system("systemctl status %s" % service_name)
+        # and then check the response...
+        if response == 0:
+            return True
+        else:
+            return False

tendrl/node_agent/persistence/persister.py

@@ -1,10 +1,21 @@
+import etcd
+from tendrl.commons.etcdobj.etcdobj import Server as etcd_server
 from tendrl.commons.persistence.etcd_persister import EtcdPersister
+from tendrl.node_agent.manager.tendrl_definitions_node_agent import data as \
+    def_data
+from tendrl.node_agent.persistence.tendrl_definitions import TendrlDefinitions
+import time
+import yaml
 
 
 class NodeAgentEtcdPersister(EtcdPersister):
-    def __init__(self, config):
-        super(NodeAgentEtcdPersister, self).__init__(config)
-        self._store = self.get_store()
+    def __init__(self, config, etcd_client):
+        super(NodeAgentEtcdPersister, self).__init__(etcd_client)
+        etcd_kwargs = {
+            'port': int(config["configuration"]["etcd_port"]),
+            'host': config["configuration"]["etcd_connection"]
+        }
+        self._store = etcd_server(etcd_kwargs=etcd_kwargs)
 
     def update_cpu(self, cpu):
         self._store.save(cpu)
@@ -32,3 +43,25 @@ def update_tendrl_definitions(self, definition):
 
     def update_platform(self, platform):
         self._store.save(platform)
+
+    def append_definitions(self):
+        try:
+            defs_path = 'tendrl_definitions_node_agent/data'
+            defs = yaml.load(self._store.client.read(
+                defs_path).value.decode("utf-8"))
+            perf_defs = yaml.load(def_data)
+            for key in perf_defs:
+                if key.startswith('namespace.'):
+                    defs[key] = perf_defs[key]
+            self.update_tendrl_definitions(
+                TendrlDefinitions(updated=str(
+                    time.time()),
+                    data=yaml.safe_dump(defs))
+            )
+        except etcd.EtcdKeyNotFound:
+            self.update_tendrl_definitions(
+                TendrlDefinitions(
+                    updated=str(time.time()),
+                    data=def_data
+                )
+            )

tendrl/node_agent/persistence/tendrl_definitions.py

@@ -7,6 +7,6 @@ class TendrlDefinitions(EtcdObj):
 
     """
     # TODO(rohan) add the definitions in etcd at startup
-    __name__ = '/tendrl_definitions_node-agent'
+    __name__ = '/tendrl_definitions_node_agent'
 
     data = fields.StrField("data")