sporada/nopaque
1
0
Fork 0
mirror of https://gitlab.ub.uni-bielefeld.de/sfb1288inf/nopaque.git synced 2025-06-20 21:10:36 +00:00
Code Issues Projects Releases Wiki Activity

Huge config update and smtp fix for daemon

Browse Source
This commit is contained in:
Patrick Jentsch
2020-10-08 12:34:02 +02:00
parent 5e221d90ad
commit dff92cbf4d
43 changed files with 613 additions and 1204 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
daemon/.dockerignore
View File

@ -1,3 +1,6 @@
# Docker related files
Dockerfile
.dockerignore
*.bak
# Packages
__pycache__

6
daemon/Dockerfile
View File

@ -1,7 +1,7 @@
FROM python:3.6-slim-stretch
LABEL maintainer="inf_sfb1288@lists.uni-bielefeld.de"
LABEL authors="Patrick Jentsch <p.jentsch@uni-bielefeld.de>, Stephan Porada <sporada@uni-bielefeld.de>"
ARG DOCKER_GID
@ -15,7 +15,7 @@ RUN apt-get update \
build-essential \
libpq-dev \
wait-for-it \
&& rm -rf /var/lib/apt/lists/*
&& rm -r /var/lib/apt/lists/*
RUN groupadd --gid ${DOCKER_GID} --system docker \
@ -31,4 +31,4 @@ RUN python -m venv venv \
&& mkdir logs
ENTRYPOINT ["./docker-entrypoint.sh"]
ENTRYPOINT ["./boot.sh"]

9
daemon/boot.sh Executable file
View File

@ -0,0 +1,9 @@
#!/bin/bash
echo "Waiting for db..."
wait-for-it "${NOPAQUE_DB_HOST}:${NOPAQUE_DB_PORT:-5432}" --strict --timeout=0
echo "Waiting for nopaque..."
wait-for-it nopaque:5000 --strict --timeout=0
source venv/bin/activate
python nopaqued.py

61
daemon/config.py Normal file
View File

@ -0,0 +1,61 @@
import logging
import os
root_dir = os.path.abspath(os.path.dirname(__file__))
DEFAULT_DATA_DIR = os.path.join('/mnt/data')
DEFAULT_DB_PORT = '5432'
DEFAULT_DOMAIN = 'localhost'
DEFAULT_LOG_DATE_FORMAT = '%Y-%m-%d %H:%M:%S'
DEFAULT_LOG_FILE = os.path.join(root_dir, 'nopaqued.log')
DEFAULT_LOG_FORMAT = ('[%(asctime)s] %(levelname)s in %(pathname)s '
'(function: %(funcName)s, line: %(lineno)d): '
'%(message)s')
DEFAULT_LOG_LEVEL = 'ERROR'
DEFAULT_MAIL_USE_SSL = 'False'
DEFAULT_MAIL_USE_TLS = 'False'
DEFAULT_PROTOCOL = 'http'
class Config:
''' ### Database ### '''
DB_HOST = os.environ.get('NOPAQUE_DB_HOST')
DB_NAME = os.environ.get('NOPAQUE_DB_NAME')
DB_PASSWORD = os.environ.get('NOPAQUE_DB_PASSWORD')
DB_PORT = os.environ.get('NOPAQUE_DB_PORT', DEFAULT_DB_PORT)
DB_USERNAME = os.environ.get('NOPAQUE_DB_USERNAME')
SQLALCHEMY_DATABASE_URI = 'postgresql://{}:{}@{}:{}/{}'.format(
DB_USERNAME, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME)
''' ### SMTP ### '''
SMTP_DEFAULT_SENDER = os.environ.get('NOPAQUE_SMTP_DEFAULT_SENDER')
SMTP_PASSWORD = os.environ.get('NOPAQUE_SMTP_PASSWORD')
SMTP_PORT = os.environ.get('NOPAQUE_SMTP_PORT')
SMTP_SERVER = os.environ.get('NOPAQUE_SMTP_SERVER')
SMTP_USERNAME = os.environ.get('NOPAQUE_SMTP_USERNAME')
SMTP_USE_SSL = os.environ.get('NOPAQUE_SMTP_USE_SSL',
DEFAULT_MAIL_USE_SSL).lower() == 'true'
SMTP_USE_TLS = os.environ.get('NOPAQUE_SMTP_USE_TLS',
DEFAULT_MAIL_USE_TLS).lower() == 'true'
''' ### General ### '''
DATA_DIR = os.environ.get('NOPAQUE_DATA_DIR', DEFAULT_DATA_DIR)
DOMAIN = os.environ.get('NOPAQUE_DOMAIN', DEFAULT_DOMAIN)
PROTOCOL = os.environ.get('NOPAQUE_PROTOCOL', DEFAULT_PROTOCOL)
''' ### Logging ### '''
LOG_DATE_FORMAT = os.environ.get('NOPAQUE_LOG_DATE_FORMAT',
DEFAULT_LOG_DATE_FORMAT)
LOG_FILE = os.environ.get('NOPAQUE_DAEMON_LOG_FILE', DEFAULT_LOG_FILE)
LOG_FORMAT = os.environ.get('NOPAQUE_LOG_FORMAT', DEFAULT_LOG_FORMAT)
LOG_LEVEL = os.environ.get('NOPAQUE_LOG_LEVEL', DEFAULT_LOG_LEVEL)
def init_app(self):
# Configure logging according to the corresponding (LOG_*) config
# entries
logging.basicConfig(datefmt=self.LOG_DATE_FORMAT,
filename=self.LOG_FILE,
format=self.LOG_FORMAT,
level=self.LOG_LEVEL)

9
daemon/docker-entrypoint.sh
View File

@ -1,9 +0,0 @@
#!/bin/bash
echo "Waiting for db..."
wait-for-it db:5432 --strict --timeout=0
echo "Waiting for web..."
wait-for-it web:5000 --strict --timeout=0
source venv/bin/activate
python nopaqued.py

0
daemon/logger/__init__.py
View File

30
daemon/logger/logger.py
View File

@ -1,30 +0,0 @@
import os
import logging
def init_logger():
'''
Functions initiates a logger instance.
'''
os.makedirs('logs', exist_ok=True)
logging.basicConfig(filename='logs/nopaqued.log',
format='[%(asctime)s] %(levelname)s in '
'%(pathname)s:%(lineno)d - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S', filemode='w')
NOPAQUE_LOG_LEVEL = os.environ.get('NOPAQUE_LOG_LEVEL')
if NOPAQUE_LOG_LEVEL is None:
FLASK_CONFIG = os.environ.get('FLASK_CONFIG')
if FLASK_CONFIG == 'development':
logging.basicConfig(level='DEBUG')
elif FLASK_CONFIG == 'testing':
# TODO: Set an appropriate log level
pass
elif FLASK_CONFIG == 'production':
logging.basicConfig(level='ERROR')
else:
logging.basicConfig(level=NOPAQUE_LOG_LEVEL)
return logging.getLogger(__name__)
if __name__ == '__main__':
init_logger()

24
daemon/nopaqued.py
View File

@ -2,26 +2,20 @@ from tasks.check_corpora import check_corpora
from tasks.check_jobs import check_jobs
from tasks.notify import notify
from time import sleep
import os
def nopaqued():
NOPAQUE_EXECUTE_NOTIFICATIONS = os.environ.get('NOPAQUE_EXECUTE_NOTIFICATIONS', 'True').lower() == 'true' # noqa
threads = {'check_corpora': None, 'check_jobs': None, 'notify': None}
check_corpora_thread = check_corpora()
check_jobs_thread = check_jobs()
notify_thread = notify()
threads['check_corpora'] = check_corpora()
threads['check_jobs'] = check_jobs()
threads['notify'] = notify(NOPAQUE_EXECUTE_NOTIFICATIONS)
while True:
if not threads['check_corpora'].is_alive():
threads['check_corpora'] = check_corpora()
if not threads['check_jobs'].is_alive():
threads['check_jobs'] = check_jobs()
if not threads['notify'].is_alive():
threads['notify'] = notify(NOPAQUE_EXECUTE_NOTIFICATIONS)
# If execute_notifications True mails are sent.
# If execute_notifications False no mails are sent.
# But notification status will be set nonetheless.
if not check_corpora_thread.is_alive():
check_corpora_thread = check_corpora()
if not check_jobs_thread.is_alive():
check_jobs_thread = check_jobs()
if not notify_thread.is_alive():
notify_thread = notify()
sleep(3)

455
daemon/nopaqued.py.bak
View File

@ -1,455 +0,0 @@
from notify.notification import Notification
from notify.service import NotificationService
from sqlalchemy import create_engine, asc
from sqlalchemy.orm import Session, relationship
from sqlalchemy.ext.automap import automap_base
from datetime import datetime
from time import sleep
import docker
import json
import logging
import os
import shutil
''' Global constants '''
NOPAQUE_STORAGE = os.environ.get('NOPAQUE_STORAGE')
''' Global variables '''
docker_client = None
session = None
# Classes for database models
Base = automap_base()
class Corpus(Base):
__tablename__ = 'corpora'
files = relationship('CorpusFile', collection_class=set)
class CorpusFile(Base):
__tablename__ = 'corpus_files'
class Job(Base):
__tablename__ = 'jobs'
inputs = relationship('JobInput', collection_class=set)
results = relationship('JobResult', collection_class=set)
notification_data = relationship('NotificationData', collection_class=list)
notification_email_data = relationship('NotificationEmailData', collection_class=list)
class NotificationData(Base):
__tablename__ = 'notification_data'
job = relationship('Job', collection_class=set)
class NotificationEmailData(Base):
__tablename__ = 'notification_email_data'
job = relationship('Job', collection_class=set)
class JobInput(Base):
__tablename__ = 'job_results'
class JobResult(Base):
__tablename__ = 'job_results'
class User(Base):
__tablename__ = 'users'
jobs = relationship('Job', collection_class=set)
corpora = relationship('Corpus', collection_class=set)
def check_corpora():
corpora = session.query(Corpus).all()
for corpus in filter(lambda corpus: corpus.status == 'submitted', corpora):
__create_build_corpus_service(corpus)
for corpus in filter(lambda corpus: (corpus.status == 'queued'
or corpus.status == 'running'),
corpora):
__checkout_build_corpus_service(corpus)
for corpus in filter(lambda corpus: corpus.status == 'start analysis',
corpora):
__create_cqpserver_container(corpus)
for corpus in filter(lambda corpus: corpus.status == 'stop analysis',
corpora):
__remove_cqpserver_container(corpus)
def __create_build_corpus_service(corpus):
corpus_dir = os.path.join(NOPAQUE_STORAGE, str(corpus.user_id),
'corpora', str(corpus.id))
corpus_data_dir = os.path.join(corpus_dir, 'data')
corpus_file = os.path.join(corpus_dir, 'merged', 'corpus.vrt')
corpus_registry_dir = os.path.join(corpus_dir, 'registry')
if os.path.exists(corpus_data_dir):
shutil.rmtree(corpus_data_dir)
if os.path.exists(corpus_registry_dir):
shutil.rmtree(corpus_registry_dir)
os.mkdir(corpus_data_dir)
os.mkdir(corpus_registry_dir)
service_args = {'command': 'docker-entrypoint.sh build-corpus',
'constraints': ['node.role==worker'],
'labels': {'origin': 'nopaque',
'type': 'corpus.prepare',
'corpus_id': str(corpus.id)},
'mounts': [corpus_file + ':/root/files/corpus.vrt:ro',
corpus_data_dir + ':/corpora/data:rw',
corpus_registry_dir + ':/usr/local/share/cwb/registry:rw'],
'name': 'build-corpus_{}'.format(corpus.id),
'restart_policy': docker.types.RestartPolicy()}
service_image = ('gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest')
try:
service = docker_client.services.get(service_args['name'])
except docker.errors.NotFound:
pass
except docker.errors.DockerException:
return
else:
service.remove()
try:
docker_client.services.create(service_image, **service_args)
except docker.errors.DockerException:
corpus.status = 'failed'
else:
corpus.status = 'queued'
def __checkout_build_corpus_service(corpus):
service_name = 'build-corpus_{}'.format(corpus.id)
try:
service = docker_client.services.get(service_name)
except docker.errors.NotFound:
logger.error('__checkout_build_corpus_service({}):'.format(corpus.id)
+ ' The service does not exist.'
+ ' (stauts: {} -> failed)'.format(corpus.status))
corpus.status = 'failed'
return
except docker.errors.DockerException:
return
service_tasks = service.tasks()
if not service_tasks:
return
task_state = service_tasks[0].get('Status').get('State')
if corpus.status == 'queued' and task_state != 'pending':
corpus.status = 'running'
elif corpus.status == 'running' and task_state == 'complete':
service.remove()
corpus.status = 'prepared'
elif corpus.status == 'running' and task_state == 'failed':
service.remove()
corpus.status = task_state
def __create_cqpserver_container(corpus):
corpus_dir = os.path.join(NOPAQUE_STORAGE, str(corpus.user_id),
'corpora', str(corpus.id))
corpus_data_dir = os.path.join(corpus_dir, 'data')
corpus_registry_dir = os.path.join(corpus_dir, 'registry')
container_args = {'command': 'cqpserver',
'detach': True,
'volumes': [corpus_data_dir + ':/corpora/data:rw',
corpus_registry_dir + ':/usr/local/share/cwb/registry:rw'],
'name': 'cqpserver_{}'.format(corpus.id),
'network': 'opaque_default'}
container_image = ('gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest')
try:
container = docker_client.containers.get(container_args['name'])
except docker.errors.NotFound:
pass
except docker.errors.DockerException:
return
else:
container.remove(force=True)
try:
docker_client.containers.run(container_image, **container_args)
except docker.errors.DockerException:
return
else:
corpus.status = 'analysing'
def __remove_cqpserver_container(corpus):
container_name = 'cqpserver_{}'.format(corpus.id)
try:
container = docker_client.containers.get(container_name)
except docker.errors.NotFound:
pass
except docker.errors.DockerException:
return
else:
container.remove(force=True)
corpus.status = 'prepared'
def check_jobs():
jobs = session.query(Job).all()
for job in filter(lambda job: job.status == 'submitted', jobs):
__create_job_service(job)
for job in filter(lambda job: (job.status == 'queued'), jobs):
__checkout_job_service(job)
# __add_notification_data(job, 'queued')
for job in filter(lambda job: (job.status == 'running'), jobs):
__checkout_job_service(job)
# __add_notification_data(job, 'running')
# for job in filter(lambda job: job.status == 'complete', jobs):
# __add_notification_data(job, 'complete')
# for job in filter(lambda job: job.status == 'failed', jobs):
#__add_notification_data(job, 'failed')
for job in filter(lambda job: job.status == 'canceling', jobs):
__remove_job_service(job)
def __add_notification_data(job, notified_on_status):
# checks if user wants any notifications at all
if (job.user.setting_job_status_mail_notifications == 'none'):
# logger.warning('User does not want any notifications!')
return
# checks if user wants only notification on completed jobs
elif (job.user.setting_job_status_mail_notifications == 'end'
and notified_on_status != 'complete'):
# logger.warning('User only wants notifications on job completed!')
return
else:
# check if a job already has associated NotificationData
notification_exists = len(job.notification_data)
# create notification_data for current job if there is none
if (notification_exists == 0):
notification_data = NotificationData(job_id=job.id)
session.add(notification_data)
session.commit() # If no commit job will have no NotificationData
# logger.warning('Created NotificationData for current Job.'))
else:
pass
# logger.warning('Job already had notification: {}'.format(notification_exists))
if (job.notification_data[0].notified_on != notified_on_status):
notification_email_data = NotificationEmailData(job_id=job.id)
notification_email_data.notify_status = notified_on_status
notification_email_data.creation_date = datetime.utcnow()
job.notification_data[0].notified_on = notified_on_status
session.add(notification_email_data)
# logger.warning('Created NotificationEmailData for current Job.')
else:
# logger.warning('NotificationEmailData has already been created for current Job!')
pass
def __create_job_service(job):
job_dir = os.path.join(NOPAQUE_STORAGE, str(job.user_id), 'jobs',
str(job.id))
service_args = {'command': ('{} /files /files/output'.format(job.service)
+ ' {}'.format(job.secure_filename if job.service == 'file-setup' else '')
+ ' --log-dir /files'
+ ' --zip [{}]_{}'.format(job.service, job.secure_filename)
+ ' ' + ' '.join(json.loads(job.service_args))),
'constraints': ['node.role==worker'],
'labels': {'origin': 'nopaque',
'type': 'service.{}'.format(job.service),
'job_id': str(job.id)},
'mounts': [job_dir + ':/files:rw'],
'name': 'job_{}'.format(job.id),
'resources': docker.types.Resources(
cpu_reservation=job.n_cores * (10 ** 9),
mem_reservation=job.mem_mb * (10 ** 6)),
'restart_policy': docker.types.RestartPolicy()}
service_image = ('gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/'
+ job.service + ':' + job.service_version)
try:
service = docker_client.services.get(service_args['name'])
except docker.errors.NotFound:
pass
except docker.errors.DockerException:
return
else:
service.remove()
try:
docker_client.services.create(service_image, **service_args)
except docker.errors.DockerException:
job.status = 'failed'
else:
job.status = 'queued'
def __checkout_job_service(job):
service_name = 'job_{}'.format(job.id)
try:
service = docker_client.services.get(service_name)
except docker.errors.NotFound:
logger.error('__checkout_job_service({}):'.format(job.id)
+ ' The service does not exist.'
+ ' (stauts: {} -> failed)'.format(job.status))
job.status = 'failed'
return
except docker.errors.DockerException:
return
service_tasks = service.tasks()
if not service_tasks:
return
task_state = service_tasks[0].get('Status').get('State')
if job.status == 'queued' and task_state != 'pending':
job.status = 'running'
elif (job.status == 'running'
and (task_state == 'complete' or task_state == 'failed')):
service.remove()
job.end_date = datetime.utcnow()
job.status = task_state
if task_state == 'complete':
results_dir = os.path.join(NOPAQUE_STORAGE, str(job.user_id),
'jobs', str(job.id), 'output')
results = filter(lambda x: x.endswith('.zip'),
os.listdir(results_dir))
for result in results:
job_result = JobResult(dir=results_dir, filename=result,
job_id=job.id)
session.add(job_result)
def __remove_job_service(job):
service_name = 'job_{}'.format(job.id)
try:
service = docker_client.services.get(service_name)
except docker.errors.NotFound:
job.status = 'canceled'
except docker.errors.DockerException:
return
else:
service.update(mounts=None)
service.remove()
def handle_jobs():
check_jobs()
def handle_corpora():
check_corpora()
# Email notification functions
def create_mail_notifications(notification_service):
notification_email_data = session.query(NotificationEmailData).order_by(asc(NotificationEmailData.creation_date)).all()
notifications = {}
for data in notification_email_data:
notification = Notification()
notification.set_addresses(notification_service.email_address,
data.job.user.email)
subject_template = '[nopaque] Status update for your Job/Corpora: {title}!'
subject_template_values_dict = {'title': data.job.title}
protocol = os.environ.get('NOPAQUE_PROTOCOL')
domain = os.environ.get('NOPAQUE_DOMAIN')
url = '{protocol}://{domain}/{jobs}/{id}'.format(
protocol=protocol, domain=domain, jobs='jobs', id=data.job.id)
body_template_values_dict = {'username': data.job.user.username,
'id': data.job.id,
'title': data.job.title,
'status': data.notify_status,
'time': data.creation_date,
'url': url}
notification.set_notification_content(subject_template,
subject_template_values_dict,
'templates/notification_messages/notification.txt',
'templates/notification_messages/notification.html',
body_template_values_dict)
notifications[data.job.id] = notification
# Using a dictionary for notifications avoids sending multiple mails
# if the status of a job changes in a few seconds. The user will not get
# swamped with mails for queued, running and complete if those happen in
# in a few seconds. Only the last update will be sent.
session.delete(data)
return notifications
def send_mail_notifications(notifications, notification_service):
for key, notification in notifications.items():
try:
notification_service.send(notification)
notification_service.mail_limit_exceeded = False
except Exception as e:
# Adds notifications to unsent if mail server exceded limit for
# consecutive mail sending
notification_service.not_sent[key] = notification
notification_service.mail_limit_exceeded = True
def notify():
# Initialize notification service
notification_service = NotificationService()
notification_service.get_smtp_configs()
notification_service.set_server()
# create notifications (content, recipient etc.)
notifications = create_mail_notifications(notification_service)
# only login and send mails if there are any notifications
if (len(notifications) > 0):
try:
notification_service.login()
# combine new and unsent notifications
notifications.update(notification_service.not_sent)
# send all notifications
send_mail_notifications(notifications, notification_service)
# remove unsent notifications because they have been sent now
# but only if mail limit has not been exceeded
if (notification_service.mail_limit_exceeded is not True):
notification_service.not_sent = {}
notification_service.quit()
except Exception as e:
notification_service.not_sent.update(notifications)
# Logger functions #
def init_logger():
'''
Functions initiates a logger instance.
'''
global logger
if not os.path.isfile('logs/nopaqued.log'):
file_path = os.path.join(os.getcwd(), 'logs/nopaqued.log')
log = open(file_path, 'w+')
log.close()
logging.basicConfig(datefmt='%Y-%m-%d %H:%M:%S',
filemode='w', filename='logs/nopaqued.log',
format='%(asctime)s - %(levelname)s - %(name)s - '
'%(filename)s - %(lineno)d - %(message)s')
logger = logging.getLogger(__name__)
if os.environ.get('FLASK_CONFIG') == 'development':
logger.setLevel(logging.DEBUG)
if os.environ.get('FLASK_CONFIG') == 'production':
logger.setLevel(logging.WARNING)
def nopaqued():
global Base
global docker_client
global session
engine = create_engine(
'postgresql://{}:{}@db/{}'.format(
os.environ.get('POSTGRES_USER'),
os.environ.get('POSTGRES_PASSWORD'),
os.environ.get('POSTGRES_DB_NAME')))
Base.prepare(engine, reflect=True)
session = Session(engine)
session.commit()
docker_client = docker.from_env()
docker_client.login(password=os.environ.get('GITLAB_PASSWORD'),
registry="gitlab.ub.uni-bielefeld.de:4567",
username=os.environ.get('GITLAB_USERNAME'))
# executing background functions
while True:
handle_jobs()
handle_corpora()
# notify()
session.commit()
sleep(3)
if __name__ == '__main__':
init_logger()
nopaqued()

11
daemon/notify/notification.py
View File

@ -11,16 +11,17 @@ class Notification(EmailMessage):
body_html_template_path,
body_template_values_dict):
# Create subject with subject_template_values_dict
self['subject'] = subject_template.format(**subject_template_values_dict)
self['subject'] = subject_template.format(
**subject_template_values_dict)
# Open template files and insert values from body_template_values_dict
with open(body_txt_template_path) as nfile:
self.body_txt = nfile.read().format(**body_template_values_dict)
self.body = nfile.read().format(**body_template_values_dict)
with open(body_html_template_path) as nfile:
self.body_html = nfile.read().format(**body_template_values_dict)
self.html = nfile.read().format(**body_template_values_dict)
# Set txt of email
self.set_content(self.body_txt)
self.set_content(self.body)
# Set html alternative
self.add_alternative(self.body_html, subtype='html')
self.add_alternative(self.html, subtype='html')
def set_addresses(self, sender, recipient):
self['From'] = sender

45
daemon/notify/service.py
View File

@ -1,41 +1,16 @@
import os
import smtplib
class NotificationService(object):
class NotificationService:
"""This is a nopaque notifcation service object."""
def __init__(self, execute_flag):
super(NotificationService, self).__init__()
self.execute_flag = execute_flag # If True mails are sent normaly
# If False mails are not sent. Used to avoid sending mails for jobs
# that have been completed a long time ago. Use this if you implement
# notify into an already existing nopaque instance. Change it to True
# after the daemon has run one time with the flag set to False
self.not_sent = {} # Holds due to an error unsent email notifications
self.mail_limit_exceeded = False # Bool to show if the mail server
# stoped sending mails due to exceeding its sending limit
def get_smtp_configs(self):
self.password = os.environ.get('MAIL_PASSWORD')
self.port = os.environ.get('MAIL_PORT')
self.server_str = os.environ.get('MAIL_SERVER')
self.tls = os.environ.get('MAIL_USE_TLS')
self.username = os.environ.get('MAIL_USERNAME').split("@")[0]
self.email_address = os.environ.get('MAIL_USERNAME')
def set_server(self):
self.smtp_server = smtplib.SMTP(host=self.server_str, port=self.port)
def login(self):
self.smtp_server.starttls()
self.smtp_server.login(self.username, self.password)
def __init__(self, smtp):
# Bool to show if the mail server stoped sending mails due to exceeding
# its sending limit
self.mail_limit_exceeded = False
# Holds due to an error unsent email notifications
self.not_sent = {}
self.smtp = smtp
def send(self, email):
if self.execute_flag:
self.smtp_server.send_message(email)
else:
return
self.smtp.send_message(email)
def quit(self):
self.smtp_server.quit()
self.smtp.quit()

2
daemon/tasks/Models.py
View File

@ -1,6 +1,6 @@
from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import relationship
from tasks import engine
from . import engine
Base = automap_base()

21
daemon/tasks/__init__.py
View File

@ -1,22 +1,11 @@
from config import Config
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
import os
import docker
''' Global constants '''
NOPAQUE_STORAGE = os.environ.get('NOPAQUE_STORAGE')
''' Docker client '''
config = Config()
config.init_app()
docker_client = docker.from_env()
docker_client.login(password=os.environ.get('GITLAB_PASSWORD'),
registry="gitlab.ub.uni-bielefeld.de:4567",
username=os.environ.get('GITLAB_USERNAME'))
''' Scoped session '''
engine = create_engine(
'postgresql://{}:{}@db/{}'.format(
os.environ.get('POSTGRES_USER'),
os.environ.get('POSTGRES_PASSWORD'),
os.environ.get('POSTGRES_DB_NAME')))
session_factory = sessionmaker(bind=engine)
Session = scoped_session(session_factory)
engine = create_engine(config.SQLALCHEMY_DATABASE_URI)
Session = scoped_session(sessionmaker(bind=engine))

39
daemon/tasks/check_corpora.py
View File

@ -1,16 +1,16 @@
from logger.logger import init_logger
from tasks import Session, docker_client, NOPAQUE_STORAGE
from tasks.decorators import background
from tasks.Models import Corpus
from . import config, docker_client, Session
from .decorators import background
from .models import Corpus
import docker
import logging
import os
import shutil
@background
def check_corpora():
c_session = Session()
corpora = c_session.query(Corpus).all()
session = Session()
corpora = session.query(Corpus).all()
for corpus in filter(lambda corpus: corpus.status == 'submitted', corpora):
__create_build_corpus_service(corpus)
for corpus in filter(lambda corpus: (corpus.status == 'queued'
@ -23,13 +23,15 @@ def check_corpora():
for corpus in filter(lambda corpus: corpus.status == 'stop analysis',
corpora):
__remove_cqpserver_container(corpus)
c_session.commit()
session.commit()
Session.remove()
def __create_build_corpus_service(corpus):
corpus_dir = os.path.join(NOPAQUE_STORAGE, str(corpus.user_id),
'corpora', str(corpus.id))
corpus_dir = os.path.join(config.DATA_DIR,
str(corpus.user_id),
'corpora',
str(corpus.id))
corpus_data_dir = os.path.join(corpus_dir, 'data')
corpus_file = os.path.join(corpus_dir, 'merged', 'corpus.vrt')
corpus_registry_dir = os.path.join(corpus_dir, 'registry')
@ -49,7 +51,8 @@ def __create_build_corpus_service(corpus):
corpus_registry_dir + ':/usr/local/share/cwb/registry:rw'],
'name': 'build-corpus_{}'.format(corpus.id),
'restart_policy': docker.types.RestartPolicy()}
service_image = ('gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest')
service_image = \
'gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest'
try:
service = docker_client.services.get(service_args['name'])
except docker.errors.NotFound:
@ -67,14 +70,13 @@ def __create_build_corpus_service(corpus):
def __checkout_build_corpus_service(corpus):
logger = init_logger()
service_name = 'build-corpus_{}'.format(corpus.id)
try:
service = docker_client.services.get(service_name)
except docker.errors.NotFound:
logger.error('__checkout_build_corpus_service({}):'.format(corpus.id)
+ ' The service does not exist.'
+ ' (stauts: {} -> failed)'.format(corpus.status))
logging.error('__checkout_build_corpus_service({}):'.format(corpus.id)
+ ' The service does not exist.'
+ ' (stauts: {} -> failed)'.format(corpus.status))
corpus.status = 'failed'
return
except docker.errors.DockerException:
@ -94,8 +96,10 @@ def __checkout_build_corpus_service(corpus):
def __create_cqpserver_container(corpus):
corpus_dir = os.path.join(NOPAQUE_STORAGE, str(corpus.user_id),
'corpora', str(corpus.id))
corpus_dir = os.path.join(config.DATA_DIR,
str(corpus.user_id),
'corpora',
str(corpus.id))
corpus_data_dir = os.path.join(corpus_dir, 'data')
corpus_registry_dir = os.path.join(corpus_dir, 'registry')
container_args = {'command': 'cqpserver',
@ -104,7 +108,8 @@ def __create_cqpserver_container(corpus):
corpus_registry_dir + ':/usr/local/share/cwb/registry:rw'],
'name': 'cqpserver_{}'.format(corpus.id),
'network': 'nopaque_default'}
container_image = ('gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest')
container_image = \
'gitlab.ub.uni-bielefeld.de:4567/sfb1288inf/cqpserver:latest'
try:
container = docker_client.containers.get(container_args['name'])
except docker.errors.NotFound:

75
daemon/tasks/check_jobs.py
View File

@ -1,46 +1,42 @@
from datetime import datetime
from logger.logger import init_logger
from tasks import Session, docker_client, NOPAQUE_STORAGE
from tasks.decorators import background
from tasks.Models import Job, NotificationData, NotificationEmailData, JobResult
from . import config, docker_client, Session
from .decorators import background
from .models import Job, JobResult, NotificationData, NotificationEmailData
import docker
import logging
import json
import os
@background
def check_jobs():
# logger = init_logger()
cj_session = Session()
jobs = cj_session.query(Job).all()
session = Session()
jobs = session.query(Job).all()
for job in filter(lambda job: job.status == 'submitted', jobs):
__create_job_service(job)
for job in filter(lambda job: (job.status == 'queued'), jobs):
__checkout_job_service(job, cj_session)
__add_notification_data(job, 'queued', cj_session)
for job in filter(lambda job: (job.status == 'running'), jobs):
__checkout_job_service(job, cj_session)
__add_notification_data(job, 'running', cj_session)
for job in filter(lambda job: job.status == 'queued', jobs):
__checkout_job_service(job, session)
__add_notification_data(job, 'queued', session)
for job in filter(lambda job: job.status == 'running', jobs):
__checkout_job_service(job, session)
__add_notification_data(job, 'running', session)
for job in filter(lambda job: job.status == 'complete', jobs):
__add_notification_data(job, 'complete', cj_session)
__add_notification_data(job, 'complete', session)
for job in filter(lambda job: job.status == 'failed', jobs):
__add_notification_data(job, 'failed', cj_session)
__add_notification_data(job, 'failed', session)
for job in filter(lambda job: job.status == 'canceling', jobs):
__remove_job_service(job)
cj_session.commit()
session.commit()
Session.remove()
def __add_notification_data(job, notified_on_status, scoped_session):
logger = init_logger()
def __add_notification_data(job, notified_on_status, session):
# checks if user wants any notifications at all
if (job.user.setting_job_status_mail_notifications == 'none'):
# logger.warning('User does not want any notifications!')
return
# checks if user wants only notification on completed jobs
elif (job.user.setting_job_status_mail_notifications == 'end'
and notified_on_status != 'complete'):
# logger.warning('User only wants notifications on job completed!')
return
else:
# check if a job already has associated NotificationData
@ -48,27 +44,21 @@ def __add_notification_data(job, notified_on_status, scoped_session):
# create notification_data for current job if there is none
if (notification_exists == 0):
notification_data = NotificationData(job_id=job.id)
scoped_session.add(notification_data)
scoped_session.commit()
session.add(notification_data)
# If no commit job will have no NotificationData
# logger.warning('Created NotificationData for current Job.'))
else:
pass
# logger.warning('Job already had notification: {}'.format(notification_exists))
session.commit()
if (job.notification_data[0].notified_on != notified_on_status):
notification_email_data = NotificationEmailData(job_id=job.id)
notification_email_data.notify_status = notified_on_status
notification_email_data.creation_date = datetime.utcnow()
job.notification_data[0].notified_on = notified_on_status
scoped_session.add(notification_email_data)
logger.warning('Created NotificationEmailData for current Job.')
else:
# logger.warning('NotificationEmailData has already been created for current Job!')
pass
session.add(notification_email_data)
def __create_job_service(job):
job_dir = os.path.join(NOPAQUE_STORAGE, str(job.user_id), 'jobs',
job_dir = os.path.join(config.DATA_DIR,
str(job.user_id),
'jobs',
str(job.id))
cmd = '{} -i /files -o /files/output'.format(job.service)
if job.service == 'file-setup':
@ -105,15 +95,14 @@ def __create_job_service(job):
job.status = 'queued'
def __checkout_job_service(job, scoped_session):
logger = init_logger()
def __checkout_job_service(job, session):
service_name = 'job_{}'.format(job.id)
try:
service = docker_client.services.get(service_name)
except docker.errors.NotFound:
logger.error('__checkout_job_service({}):'.format(job.id)
+ ' The service does not exist.'
+ ' (stauts: {} -> failed)'.format(job.status))
logging.error('__checkout_job_service({}): '.format(job.id)
+ 'The service does not exist. '
+ '(status: {} -> failed)'.format(job.status))
job.status = 'failed'
return
except docker.errors.DockerException:
@ -130,14 +119,18 @@ def __checkout_job_service(job, scoped_session):
job.end_date = datetime.utcnow()
job.status = task_state
if task_state == 'complete':
results_dir = os.path.join(NOPAQUE_STORAGE, str(job.user_id),
'jobs', str(job.id), 'output')
results_dir = os.path.join(config.DATA_DIR,
str(job.user_id),
'jobs',
str(job.id),
'output')
results = filter(lambda x: x.endswith('.zip'),
os.listdir(results_dir))
for result in results:
job_result = JobResult(dir=results_dir, filename=result,
job_result = JobResult(dir=results_dir,
filename=result,
job_id=job.id)
scoped_session.add(job_result)
session.add(job_result)
def __remove_job_service(job):

98
daemon/tasks/notify.py
View File

@ -1,58 +1,71 @@
from notify.notification import Notification
from notify.service import NotificationService
from sqlalchemy import asc
from tasks import Session
from tasks.decorators import background
from tasks.Models import NotificationEmailData
import os
from . import config, Session
from .decorators import background
from .models import NotificationEmailData
import logging
import smtplib
@background
def notify(execute_flag):
# If True mails are sent normaly
# If False mails are not sent. Used to avoid sending mails for jobs that
# have been completed a long time ago. Use this if you implement notify
# into an already existing nopaque instance. Change it to True after the
# daemon has run one time with the flag set to False.
# Initialize notification service
notification_service = NotificationService(execute_flag)
notification_service.get_smtp_configs()
notification_service.set_server()
def notify():
session = Session()
if config.SMTP_USE_SSL:
smtp = smtplib.SMTP_SSL(host=config.SMTP_SERVER, port=config.SMTP_PORT)
else:
smtp = smtplib.SMTP(host=config.SMTP_SERVER, port=config.SMTP_PORT)
if config.SMTP_USE_TLS:
smtp.starttls()
try:
smtp.login(config.SMTP_USERNAME, config.SMTP_PASSWORD)
except smtplib.SMTPHeloError:
logging.warning('The server didn’t reply properly to the HELO '
'greeting.')
return
except smtplib.SMTPAuthenticationError as e:
logging.warning('The server didn’t accept the username/password '
'combination.')
logging.warning(e)
return
except smtplib.SMTPNotSupportedError:
logging.warning('The AUTH command is not supported by the server.')
return
except smtplib.SMTPException:
logging.warning('No suitable authentication method was found.')
return
notification_service = NotificationService(smtp)
# create notifications (content, recipient etc.)
notifications = __create_mail_notifications(notification_service)
notifications = __create_mail_notifications(notification_service, session)
# only login and send mails if there are any notifications
if (len(notifications) > 0):
try:
notification_service.login()
# combine new and unsent notifications
notifications.update(notification_service.not_sent)
# send all notifications
__send_mail_notifications(notifications, notification_service)
# remove unsent notifications because they have been sent now
# but only if mail limit has not been exceeded
if (notification_service.mail_limit_exceeded is not True):
notification_service.not_sent = {}
notification_service.quit()
except Exception as e:
notification_service.not_sent.update(notifications)
notification_service.quit()
# combine new and unsent notifications
notifications.update(notification_service.not_sent)
# send all notifications
__send_mail_notifications(notifications, notification_service)
# remove unsent notifications because they have been sent now
# but only if mail limit has not been exceeded
if (notification_service.mail_limit_exceeded is not True):
notification_service.not_sent = {}
smtp.quit()
Session.remove()
# Email notification functions
def __create_mail_notifications(notification_service):
mn_session = Session()
notification_email_data = mn_session.query(NotificationEmailData).order_by(asc(NotificationEmailData.creation_date)).all()
def __create_mail_notifications(notification_service, session):
notification_email_data = session.query(NotificationEmailData).order_by(asc(NotificationEmailData.creation_date)).all() # noqa
notifications = {}
for data in notification_email_data:
notification = Notification()
notification.set_addresses(notification_service.email_address,
notification.set_addresses(config.SMTP_DEFAULT_SENDER,
data.job.user.email)
subject_template = '[nopaque] Status update for your Job/Corpora: {title}!'
subject_template = ('[nopaque] Status update for your Job/Corpora: '
'{title}!')
subject_template_values_dict = {'title': data.job.title}
protocol = os.environ.get('NOPAQUE_PROTOCOL')
domain = os.environ.get('NOPAQUE_DOMAIN')
url = '{protocol}://{domain}/{jobs}/{id}'.format(
protocol=protocol, domain=domain, jobs='jobs', id=data.job.id)
url = '{}://{}/{}/{}'.format(config.PROTOCOL,
config.DOMAIN,
'jobs',
data.job.id)
body_template_values_dict = {'username': data.job.user.username,
'id': data.job.id,
'title': data.job.title,
@ -72,9 +85,8 @@ def __create_mail_notifications(notification_service):
# get swamped with mails for queued, running and complete if those
# happen in in a few seconds. Only the last update will be sent.
# This depends on the sleep time interval though.
mn_session.delete(data)
mn_session.commit()
Session.remove()
session.delete(data)
session.commit()
return notifications
@ -83,8 +95,10 @@ def __send_mail_notifications(notifications, notification_service):
try:
notification_service.send(notification)
notification_service.mail_limit_exceeded = False
except Exception as e:
except Exception:
# Adds notifications to unsent if mail server exceded limit for
# consecutive mail sending
logging.warning('limit')
notification_service.not_sent[key] = notification
notification_service.mail_limit_exceeded = True
notification_service.not_sent.update(notifications)