1.獲取數(shù)據(jù)
python version 2.7
假設(shè)我們要獲取的Jenkins job名字為test_flow,該job觸發(fā)了另外兩個(gè)Jenkins job
test1 和test2 job.
我們要獲取test_flow job的運(yùn)行時(shí)間，狀態(tài)，number，jobname.
我們把功能性函數(shù)寫在一個(gè)文件中 Build class類中，方便主函數(shù)調(diào)用
python-jenkins模塊
build.py
import
jenkins
import
ssl
import
re
import
datetime
ssl
.
_create_default_https_context
=
ssl
.
_create_unverified_context
#build 類主要獲取Jenkins job參數(shù)
class
Build
:
def
__init__
(
self
,
jobname
,
url
,
number
)
:
self
.
jobname
=
jobname
self
.
number
=
number
self
.
jenkins_url
=
url
self
.
jenkins_server
=
jenkins
.
Jenkins
(
self
.
jenkins_url
)
self
.
build_console
=
''
def
_getConsoleFromJenkins
(
self
)
:
self
.
build_console
=
self
.
jenkins_server
.
get_build_console_output
(
self
.
jobname
,
self
.
number
)
return
self
.
build_console
def
get_next_job_build_number
(
self
,
name
)
:
number
=
0
status
=
"Running"
if
self
.
build_console
==
''
:
self
.
_getConsoleFromJenkins
(
)
pattern
=
re
.
compile
(
r
'Starting building: '
+
str
(
name
)
+
' #(.*)'
)
match
=
re
.
search
(
pattern
,
self
.
build_console
,
0
)
if
match
:
numberstr
=
match
.
group
(
1
)
number
=
int
(
numberstr
)
status
=
"Completed"
return
{
"job"
:
name
,
"number"
:
number
,
"status"
:
status
}
def
_setAttrFromJenkins
(
self
)
:
try
:
build_info
=
self
.
jenkins_server
.
get_build_info
(
self
.
jobname
,
self
.
number
)
except
:
return
False
self
.
href
=
build_info
[
'url'
]
self
.
duration
=
build_info
[
'duration'
]
self
.
build_time
=
build_info
[
'timestamp'
]
if
build_info
[
'building'
]
:
self
.
status
=
"Running"
else
:
self
.
status
=
build_info
[
'result'
]
return
True
#獲取jenkins job的各種信息然后組裝到字典attr中
def
attr_to_dict
(
self
)
:
attr
=
dict
(
)
if
self
.
_setAttrFromJenkins
(
)
==
False
:
return
None
attr
[
'job'
]
=
self
.
jobname
attr
[
'number'
]
=
self
.
number
attr
[
'href'
]
=
self
.
href
attr
[
'duration'
]
=
int
(
self
.
duration
/
1000
)
attr
[
'status'
]
=
self
.
status
attr
[
'timestamp'
]
=
self
.
build_time
# cet time zone
attr
[
'build_time'
]
=
datetime
.
datetime
.
utcfromtimestamp
(
float
(
self
.
build_time
)
/
1000
+
3600
)
.
strftime
(
'%Y-%m-%d %H:%M:%S'
)
if
self
.
status
!=
"Running"
:
attr
[
'finish_time'
]
=
datetime
.
datetime
.
utcfromtimestamp
(
float
(
self
.
build_time
)
/
1000
+
3600
+
float
(
self
.
duration
)
/
1000
)
.
strftime
(
'%Y-%m-%d %H:%M:%S'
)
return
attr
class
Job
:
def
__init__
(
self
,
jobname
,
url
)
:
self
.
jenkins_job
=
jobname
self
.
jenkins_url
=
url
self
.
jenkins_server
=
jenkins
.
Jenkins
(
self
.
jenkins_url
)
#獲取最后完成的number
def
get_last_completed
(
self
)
:
job_info
=
self
.
jenkins_server
.
get_job_info
(
self
.
jenkins_job
)
if
job_info
is
None
:
print
(
"yes it is none \n"
)
return
0
return
int
(
job_info
[
'lastCompletedBuild'
]
[
'number'
]
)
#獲取jenkins job的已經(jīng)運(yùn)行的number隊(duì)列
def
get_build_list
(
self
)
:
build
=
self
.
jenkins_server
.
get_job_info
(
self
.
jenkins_job
)
if
build
is
None
:
print
(
"yes it is none \n"
)
return
build
[
'builds'
]
2.es數(shù)據(jù)庫(kù)的創(chuàng)建
獲取的數(shù)據(jù)需要存儲(chǔ)到es數(shù)據(jù)庫(kù)中，這里用到的是Elasticsearch6.01版本
es.py
from
elasticsearch
import
Elasticsearch
class
Elastic
:
def
__init__
(
self
,
host
,
index
)
:
self
.
index
=
index
self
.
es
=
Elasticsearch
(
host
)
def
exists
(
self
)
:
return
self
.
es
.
indices
.
exists
(
self
.
index
)
def
create_index
(
self
)
:
build_mapping
=
{
'properties'
:
{
'timestamp'
:
{
'type'
:
'date'
}
,
}
}
create_index_body
=
{
'setting'
:
{
'number_of_shards'
:
1
,
'number_of_replicas'
:
0
}
,
'mappings'
:
{
'_doc'
:
{
'properties'
:
{
'job'
:
{
'type'
:
'keyword'
}
,
'timestamp'
:
{
'type'
:
'date'
}
,
'status'
:
{
'type'
:
'keyword'
}
,
'item'
:
{
'type'
:
'object'
,
'properties'
:
{
'test1'
:
build_mapping
,
}
}
}
}
}
}
try
:
self
.
es
.
indices
.
create
(
index
=
self
.
index
,
body
=
create_index_body
)
except
TransportError
as
e
:
if
e
.
error
==
'index_already_exists_exception'
:
pass
else
:
raise
def
save
(
self
,
body
)
:
try
:
self
.
es
.
index
(
self
.
index
,
doc_type
=
'_doc'
,
body
=
body
)
except
:
print
(
"save error !!!"
)
def
update
(
self
,
_id
,
body
)
:
try
:
self
.
es
.
index
(
self
.
index
,
doc_type
=
'_doc'
,
id
=
_id
,
body
=
body
)
except
:
print
(
"update error !!!"
)
3.主函數(shù)
getdata.py
#!/usr/local/bin/python
import
re
import
sys
import
argparse
from
elasticsearch
import
Elasticsearch
from
module
.
es
import
*
from
module
.
build
import
*
job_name
=
test_flow
jenkins_url
=
http
:
//
jenkins
.
test
.
com
/
jenkins
job
=
Job
(
job_name
,
jenkins_url
)
doc
=
dict
(
)
testjob
=
Build
(
job_name
,
jenkins_url
,
number
)
#建立對(duì)象
doc
=
testjob
.
attr_to_dict
(
)
#獲取job參數(shù)存到字典中
es
.
save
(
doc
)
#把數(shù)據(jù)存儲(chǔ)到es數(shù)據(jù)庫(kù)中
更多文章、技術(shù)交流、商務(wù)合作、聯(lián)系博主
微信掃碼或搜索：z360901061
微信掃一掃加我為好友
QQ號(hào)聯(lián)系： 360901061
您的支持是博主寫作最大的動(dòng)力，如果您喜歡我的文章，感覺(jué)我的文章對(duì)您有幫助，請(qǐng)用微信掃描下面二維碼支持博主2元、5元、10元、20元等您想捐的金額吧，狠狠點(diǎn)擊下面給點(diǎn)支持吧，站長(zhǎng)非常感激您！手機(jī)微信長(zhǎng)按不能支付解決辦法：請(qǐng)將微信支付二維碼保存到相冊(cè)，切換到微信，然后點(diǎn)擊微信右上角掃一掃功能，選擇支付二維碼完成支付。
【本文對(duì)您有幫助就好】元
