Finished feature POL1-463-multicast-subscription-polling.

brian_polling_manager/configuration.py

@@ -36,12 +36,19 @@ _DEFAULT_CONFIG = {
         'gws-direct-interface-check': {
             'script': '/var/lib/sensu/bin/poll-gws-direct.sh',
             'measurement': 'gwsd_counters',
-            'command': '{script} {measurement} {nren} {isp} {hostname} {tag}'
+            'command': '{script} --inventory http://localhost:18080'
+                       ' {measurement} {nren} {isp} {hostname} {tag}'
         },
         'dscp32-service-check': {
             'script': '/var/lib/sensu/bin/poll-gws-indirect.sh',
             'measurement': 'dscp32_counters',
             'command': '{script} {measurement} {service}'
+        },
+        'eumetsat-multicast-check': {
+          'script': '/home/brian_checks/venv/eumetsat-multicast',
+          'measurement': 'multicast',
+          'command': '{script} --inventory http://localhost:18080'
+                     ' --measurement {measurement} --hostname {hostname}'
         }
     },
     'statedir': '/tmp/',
@@ -75,16 +82,21 @@ CONFIG_SCHEMA = {
                     'minItems': 1
                 },
                 'api-key': {'type': 'string'},
-                'interface-check': {'$ref': '#/definitions/influx-check'},
+                'interface-check':
+                    {'$ref': '#/definitions/influx-check'},
                 'gws-direct-interface-check':
                     {'$ref': '#/definitions/influx-check'},
-                'dscp32-service-check': {'$ref': '#/definitions/influx-check'},
+                'dscp32-service-check':
+                    {'$ref': '#/definitions/influx-check'},
+                'eumetsat-multicast-check':
+                    {'$ref': '#/definitions/influx-check'},
             },
             'required': [
                 'api-base', 'api-key',
                 'interface-check',
                 'gws-direct-interface-check',
-                'dscp32-service-check'],
+                'dscp32-service-check',
+                'eumetsat-multicast-check'],
             'additionalProperties': False
         },
         'statsd': {
@@ -121,6 +133,7 @@ class State(object):
     GWS_INDIRECT = 'gws-indirect.json'
     INTERFACES = 'interfaces.json'
     STATE = 'state.json'
+    EUMET_MC = 'eumetsat-multicast.json'
 
     STATE_SCHEMA = {
         '$schema': 'http://json-schema.org/draft-07/schema#',
@@ -138,7 +151,8 @@ class State(object):
             'state': os.path.join(state_dir, State.STATE),
             'interfaces': os.path.join(state_dir, State.INTERFACES),
             'gws-direct': os.path.join(state_dir, State.GWS_DIRECT),
-            'gws-indirect': os.path.join(state_dir, State.GWS_INDIRECT)
+            'gws-indirect': os.path.join(state_dir, State.GWS_INDIRECT),
+            'eumetsat-multicast': os.path.join(state_dir, State.EUMET_MC)
         }
 
     @staticmethod
@@ -219,6 +233,19 @@ class State(object):
             new_services,
             inventory.GWS_INDIRECT_SCHEMA)
 
+    @property
+    def eumetsat_multicast(self) -> list:
+        return State._load_json(
+            self.cache_filenames['eumetsat-multicast'],
+            inventory.MULTICAST_SUBSCRIPTION_LIST_SCHEMA)
+
+    @eumetsat_multicast.setter
+    def eumetsat_multicast(self, new_subscriptions):
+        State._save_json(
+            self.cache_filenames['eumetsat-multicast'],
+            new_subscriptions,
+            inventory.MULTICAST_SUBSCRIPTION_LIST_SCHEMA)
+
 
 def _setup_logging(filename=None):
     """

brian_polling_manager/eumetsat_multicast.py

+from brian_polling_manager import sensu
+
+_CHECK_PREFIX = 'eumetmc'
+
+
+def load_eumetsat_multicast_checks(sensu_params):
+    def _is_eumetsat_multicast_check(check):
+        name = check['metadata']['name']
+        return name.startswith(_CHECK_PREFIX)
+    ifc_checks = filter(
+        _is_eumetsat_multicast_check, sensu.load_all_checks(sensu_params))
+    return {c['metadata']['name']: c for c in ifc_checks}
+
+
+class EUMETSATMulticastHostCheck(sensu.AbstractCheck):
+
+    def __init__(self, check_config, hostname):
+        super().__init__()
+        self.check_config = check_config
+        self.hostname = hostname
+
+    @sensu.AbstractCheck.name.getter
+    def name(self):
+        return f'{_CHECK_PREFIX}-{self.hostname}'
+
+    @sensu.AbstractCheck.command.getter
+    def command(self):
+        return self.check_config['command'].format(
+            script=self.check_config['script'],
+            measurement=self.check_config['measurement'],
+            hostname=self.hostname)
+
+    @sensu.AbstractCheck.proxy_entity_name.getter
+    def proxy_entity_name(self):
+        return self.hostname
+
+
+def refresh(sensu_params, subscriptions):
+
+    # one check per unique host
+    all_routers = {x['router'] for x in subscriptions}
+    required_checks = [
+        EUMETSATMulticastHostCheck(
+            sensu_params['eumetsat-multicast-check'], hostname)
+        for hostname in all_routers]
+
+    return sensu.refresh(
+        sensu_params,
+        required_checks,
+        load_eumetsat_multicast_checks(sensu_params))

brian_polling_manager/gws_direct.py

 from brian_polling_manager import sensu
 
+_CHECK_PREFIX = 'gwsd'
+
 
 def load_gws_direct_checks(sensu_params):
     def _is_gws_direct_check(check):
         name = check['metadata']['name']
-        return name.startswith('gwsd')
+        return name.startswith(_CHECK_PREFIX)
     ifc_checks = filter(
         _is_gws_direct_check, sensu.load_all_checks(sensu_params))
     return {c['metadata']['name']: c for c in ifc_checks}
@@ -23,7 +25,7 @@ class GwSDirectInterfaceCheck(sensu.AbstractCheck):
         isp = isp.replace(' ', '_')
         tag = self.interface['tag']
         tag = tag.replace(' ', '_')
-        return f'gwsd-{self.interface["nren"]}-{isp}-{tag}'
+        return f'{_CHECK_PREFIX}-{self.interface["nren"]}-{isp}-{tag}'
 
     @sensu.AbstractCheck.command.getter
     def command(self):

brian_polling_manager/gws_indirect.py

 import re
 from brian_polling_manager import sensu
 
+_CHECK_PREFIX = 'dscp32'
+
 
 def load_dscp32_checks(sensu_params):
     def _is_dscp32_check(check):
         name = check['metadata']['name']
-        return name.startswith('dscp32')
+        return name.startswith(_CHECK_PREFIX)
     ifc_checks = filter(
         _is_dscp32_check, sensu.load_all_checks(sensu_params))
     return {c['metadata']['name']: c for c in ifc_checks}
@@ -21,7 +23,7 @@ class DSCP32CountersCheck(sensu.AbstractCheck):
     @sensu.AbstractCheck.name.getter
     def name(self):
         name = re.sub(r'[\s_-]+', '_', self.service['name'])
-        return f'dscp32-{name}'
+        return f'{_CHECK_PREFIX}-{name}'
 
     @sensu.AbstractCheck.command.getter
     def command(self):

brian_polling_manager/inventory.py

@@ -97,6 +97,27 @@ GWS_INDIRECT_SCHEMA = {
 }
 
 
+# much less strict version of the actual schema
+MULTICAST_SUBSCRIPTION_LIST_SCHEMA = {
+    '$schema': 'http://json-schema.org/draft-07/schema#',
+
+    'definitions': {
+        'subscription': {
+            'type': 'object',
+            'properties': {
+                # we really only use this field
+                # don't depend strictly on unused data
+                'router': {'type': 'string'}
+            },
+            'required': ['router']
+        }
+    },
+
+    'type': 'array',
+    'items': {'$ref': '#/definitions/subscription'}
+}
+
+
 def _pick_one(haystack):
     if not isinstance(haystack, (list, tuple, set)):
         haystack = [haystack]
@@ -159,6 +180,19 @@ def load_gws_indirect_services(base_urls):
         'poller/gws/indirect', base_urls, GWS_INDIRECT_SCHEMA)
 
 
+def load_eumetsat_multicast_subscriptions(base_urls):
+    """
+    Load /poller/eumetsat-multicast from inventory provider
+
+    :param base_urls: inventory provider base api url, or a list of them
+    :return: a list of dicts, each with a 'router' key
+    """
+    return _load_inventory_json(
+        'poller/eumetsat-multicast',
+        base_urls,
+        MULTICAST_SUBSCRIPTION_LIST_SCHEMA)
+
+
 def last_update_timestamp(base_urls) -> float:
     try:
         r = requests.get(

brian_polling_manager/main.py

@@ -30,7 +30,7 @@ import jsonschema
 from statsd import StatsClient
 
 from brian_polling_manager import inventory, configuration, \
-    interfaces, gws_direct, gws_indirect, sensu
+    interfaces, gws_direct, gws_indirect, eumetsat_multicast, sensu
 
 logger = logging.getLogger(__name__)
 
@@ -54,7 +54,8 @@ REFRESH_RESULT_SCHEMA = {
     'properties': {
         'interfaces': {'$ref': '#/definitions/refresh-result'},
         'gws_direct': {'$ref': '#/definitions/refresh-result'},
-        'gws_indirect': {'$ref': '#/definitions/refresh-result'}
+        'gws_indirect': {'$ref': '#/definitions/refresh-result'},
+        'eumetsat_multicast': {'$ref': '#/definitions/refresh-result'},
     },
     'required': ['interfaces'],
     'additionalProperties': False
@@ -85,11 +86,17 @@ def refresh(config, force=False):
             config['inventory'])
         state.gws_indirect = inventory.load_gws_indirect_services(
             config['inventory'])
+        state.eumetsat_multicast \
+            = inventory.load_eumetsat_multicast_subscriptions(
+                config['inventory'])
     result = {
         'interfaces': interfaces.refresh(config['sensu'], state.interfaces),
         'gws_direct': gws_direct.refresh(config['sensu'], state.gws_direct),
         'gws_indirect': gws_indirect.refresh(
             config['sensu'], state.gws_indirect),
+        'eumetsat_multicast': eumetsat_multicast.refresh(
+            config['sensu'], state.eumetsat_multicast),
+
     }
     jsonschema.validate(result, REFRESH_RESULT_SCHEMA)  # sanity
 

test/conftest.py

@@ -36,20 +36,28 @@ def config():
                 'interface-check': {
                     'script': '/var/lib/sensu/bin/counter2influx.sh',
                     'measurement': 'counters',
-                    'command': ('{script} {measurement} '
-                                '{community} {hostname} '
-                                '{interface} {ifIndex}'),
+                    'command': '{script} {measurement}'
+                               ' {community} {hostname}'
+                               ' {interface} {ifIndex}',
                 },
                 'gws-direct-interface-check': {
                     'script': '/var/lib/sensu/bin/poll-gws-direct.sh',
                     'measurement': 'gwsd_counters',
-                    'command': ('{script} {measurement} '
-                                '{nren} {isp} {hostname} {tag}')
+                    'command': '{script} {measurement}'
+                               ' {nren} {isp} {hostname} {tag}'
                 },
                 'dscp32-service-check': {
                     'script': '/var/lib/sensu/bin/poll-gws-indirect.sh',
                     'measurement': 'dscp32_counters',
                     'command': '{script} {measurement} {service}'
+                },
+                'eumetsat-multicast-check': {
+                    'script': '/home/brian_checks/venv/eumetsat-multicast',
+                    'measurement': 'multicast',
+                    'command': '{script}'
+                               ' --inventory http://localhost:18080'
+                               ' --measurement {measurement}'
+                               ' --hostname {hostname}'
                 }
             },
             'statedir': state_dir_name,
@@ -184,6 +192,11 @@ def mocked_inventory():
         url=re.compile(r'.*inventory.+/poller/gws/indirect.*'),
         body=_load_test_data('gws-indirect.json'))
 
+    responses.add(
+        method=responses.GET,
+        url=re.compile(r'.*inventory.+/poller/eumetsat-multicast'),
+        body=_load_test_data('eumetsat-multicast.json'))
+
     bogus_version = {'latch': {'timestamp': 10000 * random.random()}}
     # mocked api for returning all checks
     responses.add(