reanahub / reana-job-controller / 282

pending completion

Build # 282

Build Type

Pull #103

travis-ci

Committed by

web-flow

Commit Message

k8s: return V1Job when creating job

* Reverts change done in d660142af
  since it breaks `k8s_delete_job` when called from `watch_jobs`
  (Closes #101).

Pull Request Pull Request #103: k8s: return V1Job when creating job

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

156 of 324 relevant lines covered (48.15%)

0.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

27.08

/reana_job_controller/k8s.py

# -*- coding: utf-8 -*-
#
# This file is part of REANA.
# Copyright (C) 2017, 2018 CERN.
#
# REANA is free software; you can redistribute it and/or modify it
# under the terms of the MIT License; see LICENSE file for more details.

"""Kubernetes wrapper."""

import logging
import os
import threading
import traceback

from flask import current_app as app
from kubernetes import client, watch
from kubernetes.client.models.v1_delete_options import V1DeleteOptions
from kubernetes.client.rest import ApiException
from reana_db.database import Session
from reana_db.models import Job

from reana_job_controller import config, volume_templates
from reana_job_controller.api_client import (current_k8s_batchv1_api_client,
                                             current_k8s_corev1_api_client)
from reana_job_controller.errors import ComputingBackendSubmissionError


def add_shared_volume(job):
    """Add shared CephFS volume to a given job spec.

    :param job: Kubernetes job spec.
    """
    storage_backend = os.getenv('REANA_STORAGE_BACKEND', 'LOCAL')
    if storage_backend == 'CEPHFS':
        volume = volume_templates.get_k8s_cephfs_volume(
            job['metadata']['namespace'])
    else:
        volume = volume_templates.get_k8s_hostpath_volume(
            job['metadata']['namespace'])
    mount_path = config.SHARED_FS_MAPPING['MOUNT_DEST_PATH']

    job['spec']['template']['spec']['containers'][0]['volumeMounts'].append(
        {'name': volume['name'], 'mountPath': mount_path}
    )
    job['spec']['template']['spec']['volumes'].append(volume)


def k8s_instantiate_job(job_id, docker_img, cmd, cvmfs_repos, env_vars,
                        namespace, shared_file_system, job_type):
    """Create Kubernetes job.

    :param job_id: Job uuid.
    :param docker_img: Docker image to run the job.
    :param cmd: Command provided to the docker container.
    :param cvmfs_repos: List of CVMFS repository names.
    :param env_vars: Dictionary representing environment variables
        as {'var_name': 'var_value'}.
    :param namespace: Job's namespace.
    :shared_file_system: Boolean which represents whether the job
        should have a shared file system mounted.
    :returns: A :class:`kubernetes.client.models.v1_job.V1Job` corresponding
        to the created job, None if the creation could not take place.
    """
    job = {
        'kind': 'Job',
        'apiVersion': 'batch/v1',
        'metadata': {
            'name': job_id,
            'namespace': namespace
        },
        'spec': {
            'backoffLimit': app.config['MAX_JOB_RESTARTS'],
            'autoSelector': True,
            'template': {
                'metadata': {
                    'name': job_id
                },
                'spec': {
                    'containers': [
                        {
                            'name': job_id,
                            'image': docker_img,
                            'env': [],
                            'volumeMounts': []
                        },
                    ],
                    'volumes': [],
                    'restartPolicy': 'Never'
                }
            }
        }
    }

    if cmd:
        import shlex
        (job['spec']['template']['spec']['containers']
         [0]['command']) = shlex.split(cmd)

    if env_vars:
        for var, value in env_vars.items():
            job['spec']['template']['spec']['containers'][0]['env'].append(
                {'name': var, 'value': value}
            )

    if shared_file_system:
        add_shared_volume(job)

    if cvmfs_repos:
        for num, repo in enumerate(cvmfs_repos):
            volume = volume_templates.get_k8s_cvmfs_volume(namespace, repo)
            mount_path = volume_templates.get_cvmfs_mount_point(repo)

            volume['name'] += '-{}'.format(num)
            (job['spec']['template']['spec']['containers'][0]
                ['volumeMounts'].append(
                    {'name': volume['name'], 'mountPath': mount_path}
            ))
            job['spec']['template']['spec']['volumes'].append(volume)

    # add better handling
    try:
        api_response = \
            current_k8s_batchv1_api_client.create_namespaced_job(
                namespace=namespace, body=job)
        return api_response
    except client.rest.ApiException as e:
        logging.debug("Error while connecting to Kubernetes API: {}".format(e))
    except Exception as e:
        logging.error(traceback.format_exc())
        logging.debug("Unexpected error: {}".format(e))


def k8s_watch_jobs(job_db):
    """Open stream connection to k8s apiserver to watch all jobs status.

    :param job_db: Dictionary which contains all current jobs.
    :param config: configuration to connect to k8s apiserver.
    """
    while True:
        logging.debug('Starting a new stream request to watch Jobs')
        try:
            w = watch.Watch()
            for event in w.stream(
                    current_k8s_batchv1_api_client.list_job_for_all_namespaces
            ):
                logging.info(
                    'New Job event received: {0}'.format(event['type']))
                job = event['object']

                # Taking note of the remaining jobs since deletion might not
                # happend straight away.
                remaining_jobs = [j for j in job_db.keys()
                                  if not job_db[j]['deleted']]
                if (not job_db.get(job.metadata.name) or
                        job.metadata.name not in remaining_jobs):
                    # Ignore jobs not created by this specific instance
                    # or already deleted jobs.
                    continue
                elif job.status.succeeded:
                    logging.info(
                        'Job {} succeeded.'.format(
                            job.metadata.name)
                    )
                    job_db[job.metadata.name]['status'] = 'succeeded'
                elif (job.status.failed and
                      job.status.failed >= config.MAX_JOB_RESTARTS):
                    logging.info('Job {} failed.'.format(
                        job.metadata.name))
                    job_db[job.metadata.name]['status'] = 'failed'
                else:
                    continue
                # Grab logs when job either succeeds or fails.
                logging.info('Getting last spawned pod for job {}'.format(
                    job.metadata.name))
                last_spawned_pod = \
                    current_k8s_corev1_api_client.list_namespaced_pod(
                        job.metadata.namespace,
                        label_selector='job-name={job_name}'.format(
                            job_name=job.metadata.name)).items[-1]
                logging.info('Grabbing pod {} logs...'.format(
                    last_spawned_pod.metadata.name))
                job_db[job.metadata.name]['log'] = \
                    current_k8s_corev1_api_client.read_namespaced_pod_log(
                        namespace=last_spawned_pod.metadata.namespace,
                        name=last_spawned_pod.metadata.name)
                # Store job logs
                try:
                    logging.info('Storing job logs: {}'.
                                 format(job_db[job.metadata.name]['log']))
                    Session.query(Job).filter_by(id_=job.metadata.name). \
                        update(dict(logs=job_db[job.metadata.name]['log']))
                    Session.commit()

                except Exception as e:
                    logging.debug('Could not retrieve'
                                  ' logs for object: {}'.
                                  format(last_spawned_pod))
                    logging.debug('Exception: {}'.format(str(e)))

                logging.info('Cleaning job {} ...'.format(
                    job.metadata.name))
                k8s_delete_job(job)
                job_db[job.metadata.name]['deleted'] = True
        except client.rest.ApiException as e:
            logging.debug(
                "Error while connecting to Kubernetes API: {}".format(e))
        except Exception as e:
            logging.error(traceback.format_exc())
            logging.debug("Unexpected error: {}".format(e))


def k8s_delete_job(job, asynchronous=True):
    """Delete Kubernetes job.

    :param job: The :class:`kubernetes.client.models.v1_job.V1Job` to be
        deleted.
    :param asynchronous: Whether the function waits for the action to be
        performed or does it asynchronously.
    """
    try:
        propagation_policy = 'Background' if asynchronous else 'Foreground'
        delete_options = V1DeleteOptions(
            propagation_policy=propagation_policy)
        current_k8s_batchv1_api_client.delete_namespaced_job(
            job.metadata.name, job.metadata.namespace, delete_options)
    except ApiException as e:
        logging.error(
            'An error has occurred while connecting to Kubernetes API Server'
            ' \n {}'.format(e))
        raise ComputingBackendSubmissionError(e.reason)


def start_watch_jobs_thread(JOB_DB):
    """Watch changes on job objects on kubernetes."""
    job_event_reader_thread = threading.Thread(target=k8s_watch_jobs,
                                               args=(JOB_DB,))
    job_event_reader_thread.daemon = True
    job_event_reader_thread.start()

1	# -- coding: utf-8 --
2	#
3	# This file is part of REANA.
4	# Copyright (C) 2017, 2018 CERN.
5	#
6	# REANA is free software; you can redistribute it and/or modify it
7	# under the terms of the MIT License; see LICENSE file for more details.
8
9	"""Kubernetes wrapper."""	1✔
10
11	import logging	1✔
12	import os	1✔
13	import threading	1✔
14	import traceback	1✔
15
16	from flask import current_app as app	1✔
17	from kubernetes import client, watch	1✔
18	from kubernetes.client.models.v1_delete_options import V1DeleteOptions	1✔
19	from kubernetes.client.rest import ApiException	1✔
20	from reana_db.database import Session	1✔
21	from reana_db.models import Job	1✔
22
23	from reana_job_controller import config, volume_templates	1✔
24	from reana_job_controller.api_client import (current_k8s_batchv1_api_client,	1✔
25	current_k8s_corev1_api_client)
26	from reana_job_controller.errors import ComputingBackendSubmissionError	1✔
27
28
29	def add_shared_volume(job):	1✔
30	"""Add shared CephFS volume to a given job spec.
31
32	:param job: Kubernetes job spec.
33	"""
34	storage_backend = os.getenv('REANA_STORAGE_BACKEND', 'LOCAL')	×
35	if storage_backend == 'CEPHFS':	×
36	volume = volume_templates.get_k8s_cephfs_volume(	×
37	job['metadata']['namespace'])
38	else:
39	volume = volume_templates.get_k8s_hostpath_volume(	×
40	job['metadata']['namespace'])
41	mount_path = config.SHARED_FS_MAPPING['MOUNT_DEST_PATH']	×
42
43	job['spec']['template']['spec']['containers'][0]['volumeMounts'].append(	×
44	{'name': volume['name'], 'mountPath': mount_path}
45	)
46	job['spec']['template']['spec']['volumes'].append(volume)	×
47
48
49	def k8s_instantiate_job(job_id, docker_img, cmd, cvmfs_repos, env_vars,	1✔
50	namespace, shared_file_system, job_type):
51	"""Create Kubernetes job.
52
53	:param job_id: Job uuid.
54	:param docker_img: Docker image to run the job.
55	:param cmd: Command provided to the docker container.
56	:param cvmfs_repos: List of CVMFS repository names.
57	:param env_vars: Dictionary representing environment variables
58	as {'var_name': 'var_value'}.
59	:param namespace: Job's namespace.
60	:shared_file_system: Boolean which represents whether the job
61	should have a shared file system mounted.
62	:returns: A :class:`kubernetes.client.models.v1_job.V1Job` corresponding
63	to the created job, None if the creation could not take place.
64	"""
65	job = {	×
66	'kind': 'Job',
67	'apiVersion': 'batch/v1',
68	'metadata': {
69	'name': job_id,
70	'namespace': namespace
71	},
72	'spec': {
73	'backoffLimit': app.config['MAX_JOB_RESTARTS'],
74	'autoSelector': True,
75	'template': {
76	'metadata': {
77	'name': job_id
78	},
79	'spec': {
80	'containers': [
81	{
82	'name': job_id,
83	'image': docker_img,
84	'env': [],
85	'volumeMounts': []
86	},
87	],
88	'volumes': [],
89	'restartPolicy': 'Never'
90	}
91	}
92	}
93	}
94
95	if cmd:	×
96	import shlex	×
97	(job['spec']['template']['spec']['containers']	×
98	[0]['command']) = shlex.split(cmd)
99
100	if env_vars:	×
101	for var, value in env_vars.items():	×
102	job['spec']['template']['spec']['containers'][0]['env'].append(	×
103	{'name': var, 'value': value}
104	)
105
106	if shared_file_system:	×
107	add_shared_volume(job)	×
108
109	if cvmfs_repos:	×
110	for num, repo in enumerate(cvmfs_repos):	×
111	volume = volume_templates.get_k8s_cvmfs_volume(namespace, repo)	×
112	mount_path = volume_templates.get_cvmfs_mount_point(repo)	×
113
114	volume['name'] += '-{}'.format(num)	×
115	(job['spec']['template']['spec']['containers'][0]	×
116	['volumeMounts'].append(
117	{'name': volume['name'], 'mountPath': mount_path}
118	))
119	job['spec']['template']['spec']['volumes'].append(volume)	×
120
121	# add better handling
122	try:	×
123	api_response = \	×
124	current_k8s_batchv1_api_client.create_namespaced_job(
125	namespace=namespace, body=job)
126	return api_response	×
127	except client.rest.ApiException as e:	×
128	logging.debug("Error while connecting to Kubernetes API: {}".format(e))	×
129	except Exception as e:	×
130	logging.error(traceback.format_exc())	×
131	logging.debug("Unexpected error: {}".format(e))	×
132
133
134	def k8s_watch_jobs(job_db):	1✔
135	"""Open stream connection to k8s apiserver to watch all jobs status.
136
137	:param job_db: Dictionary which contains all current jobs.
138	:param config: configuration to connect to k8s apiserver.
139	"""
140	while True:	×
141	logging.debug('Starting a new stream request to watch Jobs')	×
142	try:	×
143	w = watch.Watch()	×
144	for event in w.stream(	×
145	current_k8s_batchv1_api_client.list_job_for_all_namespaces
146	):
147	logging.info(	×
148	'New Job event received: {0}'.format(event['type']))
149	job = event['object']	×
150
151	# Taking note of the remaining jobs since deletion might not
152	# happend straight away.
153	remaining_jobs = [j for j in job_db.keys()	×
154	if not job_db[j]['deleted']]
155	if (not job_db.get(job.metadata.name) or	×
156	job.metadata.name not in remaining_jobs):
157	# Ignore jobs not created by this specific instance
158	# or already deleted jobs.
159	continue	×
160	elif job.status.succeeded:	×
161	logging.info(	×
162	'Job {} succeeded.'.format(
163	job.metadata.name)
164	)
165	job_db[job.metadata.name]['status'] = 'succeeded'	×
166	elif (job.status.failed and	×
167	job.status.failed >= config.MAX_JOB_RESTARTS):
168	logging.info('Job {} failed.'.format(	×
169	job.metadata.name))
170	job_db[job.metadata.name]['status'] = 'failed'	×
171	else:
172	continue	×
173	# Grab logs when job either succeeds or fails.
174	logging.info('Getting last spawned pod for job {}'.format(	×
175	job.metadata.name))
176	last_spawned_pod = \	×
177	current_k8s_corev1_api_client.list_namespaced_pod(
178	job.metadata.namespace,
179	label_selector='job-name={job_name}'.format(
180	job_name=job.metadata.name)).items[-1]
181	logging.info('Grabbing pod {} logs...'.format(	×
182	last_spawned_pod.metadata.name))
183	job_db[job.metadata.name]['log'] = \	×
184	current_k8s_corev1_api_client.read_namespaced_pod_log(
185	namespace=last_spawned_pod.metadata.namespace,
186	name=last_spawned_pod.metadata.name)
187	# Store job logs
188	try:	×
189	logging.info('Storing job logs: {}'.	×
190	format(job_db[job.metadata.name]['log']))
191	Session.query(Job).filter_by(id_=job.metadata.name). \	×
192	update(dict(logs=job_db[job.metadata.name]['log']))
193	Session.commit()	×
194
195	except Exception as e:	×
196	logging.debug('Could not retrieve'	×
197	' logs for object: {}'.
198	format(last_spawned_pod))
199	logging.debug('Exception: {}'.format(str(e)))	×
200
201	logging.info('Cleaning job {} ...'.format(	×
202	job.metadata.name))
203	k8s_delete_job(job)	×
204	job_db[job.metadata.name]['deleted'] = True	×
205	except client.rest.ApiException as e:	×
206	logging.debug(	×
207	"Error while connecting to Kubernetes API: {}".format(e))
208	except Exception as e:	×
209	logging.error(traceback.format_exc())	×
210	logging.debug("Unexpected error: {}".format(e))	×
211
212
213	def k8s_delete_job(job, asynchronous=True):	1✔
214	"""Delete Kubernetes job.
215
216	:param job: The :class:`kubernetes.client.models.v1_job.V1Job` to be
217	deleted.
218	:param asynchronous: Whether the function waits for the action to be
219	performed or does it asynchronously.
220	"""
221	try:	1✔
222	propagation_policy = 'Background' if asynchronous else 'Foreground'	1✔
223	delete_options = V1DeleteOptions(	1✔
224	propagation_policy=propagation_policy)
225	current_k8s_batchv1_api_client.delete_namespaced_job(	1✔
226	job.metadata.name, job.metadata.namespace, delete_options)
227	except ApiException as e:	1✔
228	logging.error(	1✔
229	'An error has occurred while connecting to Kubernetes API Server'
230	' \n {}'.format(e))
231	raise ComputingBackendSubmissionError(e.reason)	1✔
232
233
234	def start_watch_jobs_thread(JOB_DB):	1✔
235	"""Watch changes on job objects on kubernetes."""
236	job_event_reader_thread = threading.Thread(target=k8s_watch_jobs,	×
237	args=(JOB_DB,))
238	job_event_reader_thread.daemon = True	×
239	job_event_reader_thread.start()	×

reanahub / reana-job-controller / 282

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous