doctor_tests/monitor/collectd_plugin.py

   1 ##############################################################################
   2 # Copyright (c) 2017 NEC Corporation and others.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9
  10 import collectd
  11 import sys
  12 from datetime import datetime
  13 import json
  14 import requests
  15 import time
  16 from requests.exceptions import ConnectionError
  17
  18 from keystoneauth1 import loading
  19 from keystoneauth1 import session
  20 from congressclient.v1 import client
  21
  22
  23 def write_debug(str_write, write_type, compute_user):
  24     file_name = ('/home/%s/monitor.log' % compute_user)
  25     file_tmp = open(file_name, write_type)
  26     file_tmp.write("%s" % str_write)
  27     file_tmp.close()
  28
  29
  30 class DoctorMonitorCollectd(object):
  31     def __init__(self):
  32         self.control_ip = ''
  33         self.compute_user = ''
  34         self.compute_ip = ''
  35         self.host_name = ''
  36         self.inspector_type = ''
  37         self.inspector_url = ''
  38         self.os_auth_url = ''
  39         self.os_username = ''
  40         self.os_password = ''
  41         self.os_project_name = ''
  42         self.os_user_domain_name = ''
  43         self.os_user_domain_id = ''
  44         self.os_project_domain_name = ''
  45         self.os_project_domain_id = ''
  46         self.sess = ''
  47         self.auth = ''
  48         self.inspector_notified = 0
  49         self.start_notifications = 0
  50         self.monitor_type = 'sample'
  51
  52     def config_func(self, config):
  53         for node in config.children:
  54             key = node.key.lower()
  55             val = node.values[0]
  56
  57             if key == 'compute_host':
  58                 self.host_name = val
  59             elif key == 'control_ip':
  60                 self.control_ip = val
  61             elif key == 'compute_ip':
  62                 self.compute_ip = val
  63             elif key == 'compute_user':
  64                 self.compute_user = val
  65             elif key == 'inspector_type':
  66                 self.inspector_type = val
  67             elif key == 'os_auth_url':
  68                 self.os_auth_url = val
  69             elif key == 'os_username':
  70                 self.os_username = val
  71             elif key == 'os_password':
  72                 self.os_password = val
  73             elif key == 'os_project_name':
  74                 self.os_project_name = val
  75             elif key == 'os_user_domain_name':
  76                 self.os_user_domain_name = val
  77             elif key == 'os_user_domain_id':
  78                 self.os_user_domain_id = val
  79             elif key == 'os_project_domain_name':
  80                 self.os_project_domain_name = val
  81             elif key == 'os_project_domain_id':
  82                 self.os_project_domain_id = val
  83             else:
  84                 collectd.info('Unknown config key "%s"' % key)
  85
  86     def init_collectd(self):
  87         write_debug("Compute node collectd monitor start at %s\n\n"
  88                     % datetime.now().isoformat(), "w", self.compute_user)
  89
  90         if self.inspector_type == 'sample':
  91             self.inspector_url = ('http://%s:12345/events' % self.control_ip)
  92         elif self.inspector_type == 'congress':
  93             loader = loading.get_plugin_loader('password')
  94             self.auth = loader.load_from_options(
  95                 auth_url=self.os_auth_url,
  96                 username=self.os_username,
  97                 password=self.os_password,
  98                 project_name=self.os_project_name,
  99                 user_domain_name=self.os_user_domain_name,
 100                 user_domain_id=self.os_user_domain_id,
 101                 project_domain_name=self.os_project_domain_name,
 102                 project_domain_id=self.os_project_domain_id)
 103             self.sess = session.Session(auth=self.auth)
 104             congress = client.Client(session=self.sess, service_type='policy')
 105             ds = congress.list_datasources()['results']
 106             doctor_ds = next(
 107                 (item for item in ds if item['driver'] == 'doctor'), None)
 108
 109             congress_endpoint = \
 110                 congress.httpclient.get_endpoint(auth=self.auth)
 111             self.inspector_url = ('%s/v1/data-sources/%s/tables/events/rows'
 112                                   % (congress_endpoint, doctor_ds['id']))
 113         else:
 114             sys.exit()
 115         self.start_notifications = 1
 116
 117     def notify_inspector(self):
 118         event_type = "compute.host.down"
 119         payload = [
 120             {
 121                 'id': ("monitor_%s_id1" % self.monitor_type),
 122                 'time': datetime.now().isoformat(),
 123                 'type': event_type,
 124                 'details': {
 125                     'hostname': self.host_name,
 126                     'status': 'down',
 127                     'monitor': ("monitor_%s" % self.monitor_type),
 128                     'monitor_event_id':
 129                         ("monitor_%s_event1" % self.monitor_type)
 130                 },
 131             },
 132         ]
 133         data = json.dumps(payload)
 134         self.inspector_notified = 1
 135
 136         if self.inspector_type == 'sample':
 137             headers = {'content-type': 'application/json'}
 138             try:
 139                 requests.post(self.inspector_url, data=data, headers=headers)
 140             except ConnectionError as err:
 141                 print(err)
 142         elif self.inspector_type == 'congress':
 143             # TODO(umar) enhance for token expiry case
 144             headers = {
 145                 'Content-Type': 'application/json',
 146                 'Accept': 'application/json',
 147                 'X-Auth-Token': self.sess.get_token()
 148             }
 149             requests.put(self.inspector_url, data=data, headers=headers)
 150
 151     def handle_notify(self, notification, data=None):
 152         if (notification.seerity == collectd.NOTIF_FAILURE or
 153                 notification.severity == collectd.NOTIF_WARNING):
 154             if (self.start_notifications == 1 and
 155                     self.inspector_notified == 0):
 156                 write_debug("Received down notification:"
 157                             "doctor monitor detected at %s\n"
 158                             % time.time(), "a", self.compute_user)
 159                 self.notify_inspector()
 160
 161         elif notification.severity == collectd.NOTIF_OKAY:
 162             collectd.info("Interface status: UP again %s\n" % time.time())
 163         else:
 164             collectd.info("Unknown notification severity %s\n"
 165                           % notification.severity)
 166
 167
 168 monitor = DoctorMonitorCollectd()
 169
 170 collectd.register_config(monitor.config_func)
 171 collectd.register_init(monitor.init_collectd)
 172 collectd.register_notification(monitor.handle_notify)