1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
|
##############################################################################
# Copyright (c) 2016 Max Breitenfeldt and others.
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Apache License, Version 2.0
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
import logging
import re
from django.conf import settings
import requests
from django.core.cache import cache
logger = logging.getLogger(__name__)
# TODO: implement caching decorator, cache get_* functions
def get_json(url):
if cache.get(url) is None:
try:
response = requests.get(url)
json = response.json()
cache.set(url, json, 180) # cache result for 180 seconds
return json
except requests.exceptions.RequestException as e:
logger.exception(e)
except ValueError as e:
logger.exception(e)
else:
return cache.get(url)
def get_all_slaves():
url = settings.ALL_SLAVES_URL
json = get_json(url)
if json is not None:
return json['computer'] # return list of dictionaries
return []
def get_slave(slavename):
slaves = get_all_slaves()
for slave in slaves:
if slave['displayName'] == slavename:
return slave
return {}
def get_ci_slaves():
url = settings.CI_SLAVES_URL
json = get_json(url)
if json is not None:
return json['nodes']
return []
def get_all_jobs():
url = settings.ALL_JOBS_URL
json = get_json(url)
if json is not None:
return json['jobs'] # return list of dictionaries
return []
def get_jenkins_job(slavename):
jobs = get_all_jobs()
max_time = 0
last_job = None
for job in jobs:
if job['lastBuild'] is not None:
if job['lastBuild']['builtOn'] == slavename:
if job['lastBuild']['building'] is True:
return job # return active build
if job['lastBuild']['timestamp'] > max_time:
last_job = job
max_time = job['lastBuild']['timestamp']
return last_job
def is_ci_slave(slavename):
ci_slaves = get_ci_slaves()
for ci_slave in ci_slaves:
if ci_slave['nodeName'] == slavename:
return True
return False
def is_dev_pod(slavename):
if is_ci_slave(slavename):
return False
if slavename.find('pod') != -1:
return True
return False
def parse_job(job):
result = parse_job_string(job['lastBuild']['fullDisplayName'])
result['building'] = job['lastBuild']['building']
result['result'] = ''
if not job['lastBuild']['building']:
result['result'] = job['lastBuild']['result']
result['url'] = job['url']
return result
def parse_job_string(full_displayname):
job = {}
job['scenario'] = ''
job['installer'] = ''
job['branch'] = ''
tokens = re.split(r'[ -]', full_displayname)
for i in range(len(tokens)):
if tokens[i] == 'os':
job['scenario'] = '-'.join(tokens[i: i + 4])
elif tokens[i] in ['fuel', 'joid', 'apex', 'compass']:
job['installer'] = tokens[i]
elif tokens[i] in ['master', 'arno', 'brahmaputra', 'colorado']:
job['branch'] = tokens[i]
tokens = full_displayname.split(' ')
job['name'] = tokens[0]
return job
def get_slave_url(slave):
return settings.GET_SLAVE_URL + slave['displayName']
def get_slave_status(slave):
if not slave['offline'] and slave['idle']:
return 'online / idle'
if not slave['offline']:
return 'online'
return 'offline'
|
