Skip to content

S
slapos.core
Commit 63a9e08a authored by Alain Takoudjou's avatar Alain Takoudjou
Browse files
Options

fixup to make it work

parent 24fb62a4
Hide whitespace changes
Inline Side-by-side
Showing with 139 additions and 98 deletions
slapos/grid/promise/__init__.py
View file @ 63a9e08a
......@@ -35,10 +35,13 @@ import time
import importlib
import ConfigParser
import re
import traceback
import psutil
import subprocess
import slapos.slap
from abc import ABCMeta, abstractmethod
from multiprocessing import Process, Queue
from multiprocessing import Process, Queue as MQueue
import Queue
from datetime import datetime, timedelta
from threading import Thread
from slapos.grid.promise import interface
......@@ -49,13 +52,15 @@ class PromiseError(Exception):
pass
class BaseResult(object):
def __init__(self, problem=False, message=None, date):
def __init__(self, problem=False, message=None, date=None):
self.__problem = problem
self.__message = message
self.__date = date
if self.__date is None:
self.__date = datetime.now()
def hasFailed(self):
return not self.__problem
return self.__problem
@property
def type(self):
......@@ -78,8 +83,8 @@ class TestResult(BaseResult):
class AnomalyResult(BaseResult):
@property
def type(self):
return "Anomaly Result"
def type(self):
return "Anomaly Result"
class PromiseQueueResult(object):
......@@ -99,22 +104,22 @@ class GenericPromise(object):
self.__log_folder = self.__config.pop('log-folder', None)
self.__partition_folder = self.__config.pop('partition-folder', None)
sef.__check_anomaly = self.___config.pop('check-anomaly', False)
self.__title = self.___config.pop('title', None)
self.__periodicity = self.___config.pop('periodicity', None)
self.__debug = self.___config.pop('debug', True)
self.__name = self.___config.pop('name', None)
self.__promise_path = self.___config.pop('path', None)
self.queue = self.___config.pop('queue', None)
self.__check_anomaly = self.__config.pop('check-anomaly', False)
self.__title = self.__config.pop('title', None)
self.__periodicity = self.__config.pop('periodicity', None)
self.__debug = self.__config.pop('debug', True)
self.__name = self.__config.pop('name', None)
self.__promise_path = self.__config.pop('path', None)
self.queue = self.__config.pop('queue', None)
self.__logger_buffer = None
self._validateConf()
self._configureLogger()
for key, value in config.items():
setattr(self, key.replace('-', '_', value)
setattr(self, key.replace('-', '_'), value)
def _configureLogger(self):
self.logger = logging.getLogger(__name__)
self.logger = logging.getLogger(self.__name)
if self.__log_folder is None:
# configure logger with StringIO
import cStringIO
......@@ -130,8 +135,8 @@ class GenericPromise(object):
self.logger.setLevel(logging.DEBUG if self.__debug else logging.INFO)
logger_handler.setFormatter(
format=logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"),
datefmt="%Y-%m-%d %H:%M:%S"
fmt=logging.Formatter("%(asctime)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S")
)
self.logger.addHandler(logger_handler)
......@@ -156,8 +161,9 @@ class GenericPromise(object):
"(?P<day_of_week>\*|[0-6](\-[0-6])?)"\
)
)
if validate_crontab_time_format_regex.match(period) is None:
raise ValueError("Periodicity %r is not a valid Cron time format." % period)
#if validate_crontab_time_format_regex.match(self.__periodicity) is None:
# raise ValueError("Promise periodicity %r is not a valid Cron time " \
# "format." % self.__periodicity)
def getConfig(self):
return self.__config
......@@ -171,7 +177,7 @@ class GenericPromise(object):
def getPartitionFolder(self):
return self.__partition_folder
de getPromiseFile(self):
def getPromiseFile(self):
return self.__promise_path
def __bang(self, message):
......@@ -194,18 +200,24 @@ class GenericPromise(object):
)
computer_partition.bang(message)
def __getResultFromString(self, regex, result_string, only_failure=False):
def __getLogRegex(self):
return re.compile(r"(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+\-?\s*(\w{4,7})\s+\-?\s*(.*)")
def __getResultFromString(self, result_string, only_failure=False):
line_list = result_string.split('\n')
result_list = []
line_part = ""
regex = self.__getLogRegex()
for line in line_list:
if not line:
continue
match = regex.match(line)
if match is not None:
if not only_failure or (only_failure and match.groups()[1] == 'ERROR'):
result_list.append([
match.groups()[0],
match.groups()[1],
match.groups()[2] + line_part,
(match.groups()[2] + line_part).strip(),
])
line_part = ""
else:
......@@ -224,12 +236,9 @@ class GenericPromise(object):
ex: [['2018-02-02 17:49:01', 'ERROR', "Promise has failed"], ...]
"""
regex = re.compile(r"(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})\s+(\w{4,7})\s+(.*)")
if self.__log_file is None:
if self.__logger_buffer is not None:
return self.__getResultFromString(regex,
self.__logger_buffer.getvalue(),
return self.__getResultFromString(self.__logger_buffer.getvalue(),
only_failure)
else:
return []
......@@ -237,6 +246,7 @@ class GenericPromise(object):
if not os.path.exists(self.__log_file):
return []
regex = self.__getLogRegex()
date = datetime.now() - timedelta(minutes=latest_minute)
date_string = date.strftime('%Y-%m-%d %H:%M:%S')
......@@ -266,7 +276,7 @@ class GenericPromise(object):
line_list.append([
result.groups()[0],
result.groups()[1],
result.groups()[2] + line_part,
(result.groups()[2] + line_part).strip(),
])
else:
line_part += '\n' + line
......@@ -276,15 +286,38 @@ class GenericPromise(object):
line_list.reverse()
return line_list
def defaultTest(self, latest_minute=3):
"""
Fail if the latest 2 messages has failed.
If only there is only one result, fail if that result has error
"""
problem = False
message = ""
latest_result_list = self.getLastPromiseResultList(
latest_minute=latest_minute,
only_failure=False
)
result_size = len(latest_result_list)
if result_size == 0:
return TestResult(problem=False, message="No result!")
i = 0
latest_result_list.reverse()
# we test at most the 2 latest results
while i < result_size and i < 2:
if latest_result_list[i][1] == 'ERROR':
problem = True
i += 1
return TestResult(problem=problem, message=latest_result_list[0][2])
@abstractmethod
def sense(self):
"""Run the promise code and store the result"""
@abstractmethod
def anomaly(self):
"""Called to detect if there is an anomaly which require to bang."""
@abstractmethod
def test(self):
"""Test promise and say if problem is detected or not"""
......@@ -299,7 +332,7 @@ class GenericPromise(object):
except Exception, e:
# log the result
self.logger.error(str(e))
if sef.__check_anomaly:
if self.__check_anomaly:
# run sense, anomaly
try:
result = self.anomaly()
......@@ -336,7 +369,6 @@ class PromiseWrapper(GenericPromise):
A wrapper promise used to run old promises style and bash promises
"""
import subprocess
zope_interface.implements(interface.IPromise)
def __init__(self, config):
......@@ -357,32 +389,14 @@ class PromiseWrapper(GenericPromise):
message = output or ""
if error:
message += "\n" + error
if promise_process.return_code != 0:
self.logger.error(message)
if promise_process.returncode != 0:
self.logger.error(message.strip())
else:
self.logger.info(message)
self.logger.info(message.strip())
def test(self):
"""
Fail if the latest 2 messages has failed.
"""
failed_count = 0
message = ""
latest_result_list = self.getLastPromiseResultList(latest_minute=4,
only_failure=False)
return self.defaultTest(latest_minute=3)
if len(latest_result_list) == 0:
return TestResult(problem=False, message="No result")
latest_result_list.reverse()
for in in range(0, 2):
if latest_result_list[i][1] == 'ERROR':
failed_count += 1
return TestResult(
problem=failed_count == 2,
message=latest_result_list[0][2]
)
class PromiseRunner(Process):
......@@ -399,9 +413,9 @@ class PromiseRunner(Process):
@param allow_bang: Bolean saying if bang should be called in case of
anomaly failure.
"""
Process.__init__()
self.promise_instance = promise_instance
self.config = config
Process.__init__(self)
self.promise = promise_instance
self.logger = logger
self.allow_bang = allow_bang
self.uid = uid
self.gid = gid
......@@ -409,11 +423,11 @@ class PromiseRunner(Process):
# self.daemon = False
def run(self):
if self.uid is not None and self.gid is not None:
dropPrivileges(uid, gid, logger=logger)
if self.uid and self.gid:
dropPrivileges(self.uid, self.gid, logger=self.logger)
if self.cwd is not None:
os.chdir(self.cwd)
promise_instance.run(self.allow_bang)
self.promise.run(self.allow_bang)
class PromiseLauncher(object):
......@@ -462,6 +476,7 @@ class PromiseLauncher(object):
If True, show Promise consumption and execution time information
"""
self.save_method = save_method
self.__config = {
'promise-timeout': 20,
'promise-dir': None,
......@@ -469,21 +484,26 @@ class PromiseLauncher(object):
'log-folder': None,
'profile': False,
'uid': None,
'gid': None
'gid': None,
'master-url': None,
'partition-cert': None,
'partition-key': None,
'partition-id': None,
'computer-id': None,
'debug': True,
'check-anomaly': False
}
if self.config_file is not None:
if config_file is not None:
self.loadConfigFromFile(config_file)
if config is not None:
self.__config.update(config)
if save_method is not None:
self.save_method = save_method
for key, value in self.__config.items():
setattr(self, key.replace('-', '_'), value or None)
if self.promise_dir is None:
raise ValueError("Promise folder is missing in configuration!")
if logger is None:
self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging.DEBUG)
......@@ -492,16 +512,10 @@ class PromiseLauncher(object):
logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
)
self.logger.addHandler(handler)
else:
self.logger = logger
init_file = os.path.join(self.promise_dir, '__init__.py')
if not os.path.exists(init_file):
with open(ini_file, 'w') as f:
f.write("")
os.chmod(init_file, 0644)
if sys.path[0] != self.promise_dir:
sys.path[0:0] = [self.promise_dir]
self.queue_result = Queue()
self.queue_result = MQueue()
self.bang_called = False
def _loadConfigFromFile(self, config_file):
......@@ -549,14 +563,22 @@ class PromiseLauncher(object):
Launch the promise and save the result if `self.save_method` is not None
If no save method is set, raise PromiseError in case of failure
"""
self.logger.info("Checking promise %s..." % promise_name)
if promise_module is None:
promise_instance = PromiseWrapper(argument_dict)
else:
promise_instance = promise_module.RunPromise(argument_dict)
if self.profile:
self.logger.info("Checking promise %s..." % promise_name)
try:
if promise_module is None:
promise_instance = PromiseWrapper(argument_dict)
else:
promise_instance = promise_module.RunPromise(argument_dict)
except Exception:
# to not prevent run other promises
self.logger.error(traceback.format_exc())
self.logger.warning("Promise %s skipped." % promise_name)
return
promise_process = PromiseRunner(
promise_module,
logger=self.logger
promise_instance,
logger=self.logger,
allow_bang=not self.bang_called and self.check_anomaly,
uid=self.uid,
gid=self.gid,
......@@ -577,7 +599,7 @@ class PromiseLauncher(object):
pass
for current_increment in range(0, increment_limit):
execution_time = current_increment * sleep_time
if not process.is_alive():
if not promise_process.is_alive():
try:
queue_item = self.queue_result.get(False, 2)
except Queue.Empty:
......@@ -643,23 +665,36 @@ class PromiseLauncher(object):
Run all promises
"""
promise_list = []
base_config = {
'log-folder': self.log_folder,
'partition-folder': self.partition_folder,
'debug': True,
'master-url': self.master_url,
'partition-cert': self.partition_cert,
'partition-key': self.partition_key,
'partition-id': self.partition_id,
'computer-id': self.computer_id,
'queue': self.queue_result,
}
if os.path.exists(self.promise_dir) and os.path.isdir(self.promise_dir):
# if there is no __init file, add it
init_file = os.path.join(self.promise_dir, '__init__.py')
if not os.path.exists(init_file):
with open(init_file, 'w') as f:
f.write("")
os.chmod(init_file, 0644)
if sys.path[0] != self.promise_dir:
sys.path[0:0] = [self.promise_dir]
promise_list = []
# load all promises so we can catch import errors before launch them
promise_list = [(promise_name, self._loadPromiseModule(promise_name))
for promise_name in os.listdir(self.promise_dir)]
base_config = {
'log-folder': self.log_folder,
'partition-folder': self.partition_folder,
'debug': True,
'master-url': self.master_url,
'partition-cert': self.partition_cert,
'partition-key': self.partition_key,
'partition-id': self.partition_id,
'computer-id': self.computer_id,
'queue': self.queue_result,
}
for promise_name in os.listdir(self.promise_dir):
if promise_name.startswith('__init__'):
continue
promise_list.append((promise_name,
self._loadPromiseModule(promise_name)))
for promise in promise_list:
config = {
'path': os.path.join(self.promise_dir, promise[0]),
......@@ -670,11 +705,17 @@ class PromiseLauncher(object):
if os.path.exists(self.old_promise_dir) and os.path.isdir(self.old_promise_dir):
# run old promise styles
for promise_name in self.old_promise_dir:
for promise_name in os.listdir(self.old_promise_dir):
promise_path = os.path.join(self.old_promise_dir, promise_name)
if not os.path.isfile(promise_path) or \
not os.access(promise_path, os.X_OK):
self.logger.warning("Bad promise file at %r." % promise_path)
continue
config = {
'path': os.path.join(self.old_promise_dir, promise_name),
'path': promise_path,
'name': promise_name
}
config.update(base_config)
# We will use PromiseWrapper class to run this
# We will use promise wrapper to run this
self._launchPromise(promise_name, config)
slapos/grid/slapgrid.py
View file @ 63a9e08a
......@@ -61,7 +61,7 @@ from slapos.grid.svcbackend import (launchSupervisord,
_getSupervisordConfigurationDirectory,
_getSupervisordSocketPath)
from slapos.grid.utils import (md5digest, dropPrivileges, SlapPopen, updateFile)
from slapos.grid.promise import PromiseLauncher
from slapos.grid.promise import PromiseLauncher, PromiseError
from slapos.human import human2bytes
import slapos.slap
from netaddr import valid_ipv4, valid_ipv6
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7