promise: import cpu load check to new promise format

slapos/promise/plugin/check_server_cpu_load.py

+from zope import interface as zope_interface
+from slapos.grid.promise import interface
+from slapos.grid.promise.generic import GenericPromise
+
+import subprocess
+import os
+
+class RunPromise(GenericPromise):
+
+  zope_interface.implements(interface.IPromise)
+
+  def __init__(self, config):
+    GenericPromise.__init__(self, config)
+    self.setPeriodicity(minute=3)
+
+  def checkCPULoad(self, tolerance=2.2):
+
+    # tolerance=1.5 => accept CPU load up to 1.5 =150%
+    uptime_result = subprocess.check_output(['uptime'])
+    line = uptime_result.strip().split(' ')
+    load, load5, long_load = line[-3:]
+    long_load = float(long_load.replace(',', '.'))
+    core_count = int(subprocess.check_output(['nproc']).strip())
+    max_load = core_count * tolerance
+    if long_load > max_load:
+      # display top statistics
+      top_result = subprocess.check_output(['top', '-n', '1', '-b'])
+      message = "CPU load is high: %s %s %s\n\n" % (load, load5, long_load)
+      i = 0
+      result_list = top_result.split('\n')
+      # display first 5 lines
+      while i < len(result_list) and i < 5:
+        message += "\n%s" % result_list[i]
+        i += 1
+      self.logger.error(message)
+    else:
+      self.logger.info("CPU load is OK")
+
+  def sense(self):
+    load_threshold = self.getConfig('cpu-load-threshold')
+    threshold = 0
+
+    if load_threshold is not None:
+      try:
+        threshold = float(load_threshold)
+      except ValueError, e:
+        self.logger.error("CPU load threshold %r is not valid: %s" % (load_threshold, e))
+        return
+
+    self.checkCPULoad(threshold or 2.2)
+
+  def test(self):
+    return self._test(result_count=5, failure_amount=5)
+
+  def anomaly(self):
+    return self._test(result_count=5, failure_amount=5)

slapos/test/promise/plugin/test_check_server_cpu_load.py

+##############################################################################
+#
+# Copyright (c) 2018 Vifib SARL and Contributors. All Rights Reserved.
+#
+# WARNING: This program as such is intended to be used by professional
+# programmers who take the whole responsibility of assessing all potential
+# consequences resulting from its eventual inadequacies and bugs
+# End users who are looking for a ready-to-use solution with commercial
+# guarantees and support are strongly adviced to contract a Free Software
+# Service Company
+#
+# This program is Free Software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; either version 3
+# of the License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
+#
+##############################################################################
+
+from slapos.test.promise.plugin import TestPromisePluginMixin
+from slapos.grid.promise import PromiseError
+import os
+
+class TestCheckServerCPULoad(TestPromisePluginMixin):
+
+  def setUp(self):
+    TestPromisePluginMixin.setUp(self)
+    self.promise_name = "server-cpu-load-promise.py"
+
+    content = """from slapos.promise.plugin.check_server_cpu_load import RunPromise
+
+extra_config_dict = {
+  'cpu-load-threshold': '2.0',
+}
+"""
+    self.writePromise(self.promise_name, content)
+
+  def test_check_cpu_load_run(self):
+    self.configureLauncher()
+    self.launcher.run()
+    result = self.getPromiseResult(self.promise_name)
+    if result['result']['failed']:
+      self.assertTrue("CPU load is high" in result['result']['message'])
+    else:
+      self.assertEquals("CPU load is OK", result['result']['message'].strip())
+
+if __name__ == '__main__':
+  unittest.main()