forked from demisto/content
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathgitlab_slack_notifier.py
More file actions
176 lines (153 loc) · 6.97 KB
/
gitlab_slack_notifier.py
File metadata and controls
176 lines (153 loc) · 6.97 KB
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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
import argparse
import logging
import os
from typing import Tuple
import gitlab
from slack import WebClient as SlackClient
from Tests.Marketplace.marketplace_services import get_upload_data
from Tests.Marketplace.marketplace_constants import BucketUploadFlow
from Tests.scripts.utils.log_util import install_logging
from Tests.scripts.slack_notifier import get_fields, get_failing_unit_tests_file_data
DEMISTO_GREY_ICON = 'https://3xqz5p387rui1hjtdv1up7lw-wpengine.netdna-ssl.com/wp-content/' \
'uploads/2018/07/Demisto-Icon-Dark.png'
ARTIFACTS_FOLDER = os.getenv('ARTIFACTS_FOLDER', './artifacts')
ENV_RESULTS_PATH = os.getenv('ENV_RESULTS_PATH', os.path.join(ARTIFACTS_FOLDER, 'env_results.json'))
PACK_RESULTS_PATH = os.path.join(ARTIFACTS_FOLDER, BucketUploadFlow.PACKS_RESULTS_FILE)
CONTENT_CHANNEL = 'dmst-content-team'
GITLAB_PROJECT_ID = os.getenv('CI_PROJECT_ID', 2596) # the default is the id of the content repo in code.pan.run
GITLAB_SERVER_URL = os.getenv('CI_SERVER_URL', 'https://code.pan.run') # disable-secrets-detection
CONTENT_NIGHTLY = 'Content Nightly'
BUCKET_UPLOAD = 'Upload Packs to Marketplace Storage'
SDK_NIGHTLY = 'Demisto SDK Nightly'
PRIVATE_NIGHTLY = 'Private Nightly'
WORKFLOW_TYPES = {CONTENT_NIGHTLY, SDK_NIGHTLY, BUCKET_UPLOAD, PRIVATE_NIGHTLY}
def options_handler():
parser = argparse.ArgumentParser(description='Parser for slack_notifier args')
parser.add_argument('-u', '--url', help='The gitlab server url', default=GITLAB_SERVER_URL)
parser.add_argument('-p', '--pipeline_id', help='The pipeline id to check the status of', required=True)
parser.add_argument('-s', '--slack_token', help='The token for slack', required=True)
parser.add_argument('-c', '--ci_token', help='The token for circleci/gitlab', required=True)
parser.add_argument(
'-f', '--env_results_path', help='The env results file containing the dns address', default=ENV_RESULTS_PATH
)
parser.add_argument('-ca', '--ci_artifacts', help="The path to the ci artifacts directory")
parser.add_argument(
'-ch', '--slack_channel', help='The slack channel in which to send the notification', default=CONTENT_CHANNEL
)
parser.add_argument('-gp', '--gitlab_project_id', help='The gitlab project_id.', default=GITLAB_PROJECT_ID)
parser.add_argument(
'-tw', '--triggering-workflow', help='The type of ci pipeline workflow the notifier is reporting on',
choices=WORKFLOW_TYPES)
options = parser.parse_args()
return options
def unit_tests_results():
failing_unit_tests = get_failing_unit_tests_file_data()
slack_results = []
if failing_unit_tests:
slack_results.append({
"title": f'{"Failed Unit Tests"} - ({len(failing_unit_tests)})',
"value": '\n'.join(failing_unit_tests),
"short": False
})
return slack_results
def test_playbooks_results():
playbooks_data, _, _ = get_fields()
return playbooks_data
def bucket_upload_results():
steps_fields = []
logging.info(f'retrieving upload data from "{PACK_RESULTS_PATH}"')
successful_packs, failed_packs, successful_private_packs, _ = get_upload_data(
PACK_RESULTS_PATH, BucketUploadFlow.UPLOAD_PACKS_TO_MARKETPLACE_STORAGE
)
if successful_packs:
steps_fields += [{
"title": "Successful Packs:",
"value": "\n".join(sorted([pack_name for pack_name in {*successful_packs}], key=lambda s: s.lower())),
"short": False
}]
if failed_packs:
steps_fields += [{
"title": "Failed Packs:",
"value": "\n".join(sorted([pack_name for pack_name in {*failed_packs}], key=lambda s: s.lower())),
"short": False
}]
if successful_private_packs:
steps_fields += [{
"title": "Successful Private Packs:",
"value": "\n".join(sorted([pack_name for pack_name in {*successful_private_packs}],
key=lambda s: s.lower())),
"short": False
}]
return steps_fields
def construct_slack_msg(triggering_workflow, pipeline_url, pipeline_failed_jobs) -> list:
title = triggering_workflow
if pipeline_failed_jobs:
title += ' - Failure'
color = 'danger'
else:
title += ' - Success'
color = 'good'
content_fields = []
failed_jobs_names = {job.name for job in pipeline_failed_jobs}
if failed_jobs_names:
content_fields.append({
"title": f'{"Failed Jobs"} - ({len(failed_jobs_names)})',
"value": '\n'.join(failed_jobs_names),
"short": False
})
triggering_workflow_lower = triggering_workflow.lower()
check_unittests_substrings = {'lint', 'unit', 'demisto sdk nightly'}
if any({substr in triggering_workflow_lower for substr in check_unittests_substrings}):
content_fields += unit_tests_results()
if 'upload' in triggering_workflow_lower:
content_fields += bucket_upload_results()
if 'content nightly' in triggering_workflow_lower:
content_fields += test_playbooks_results()
slack_msg = [{
'fallback': title,
'color': color,
'title': title,
'title_link': pipeline_url,
'fields': content_fields
}]
return slack_msg
def collect_pipeline_data(gitlab_client, project_id, pipeline_id) -> Tuple[str, list]:
failed_jobs = []
project = gitlab_client.projects.get(int(project_id))
pipeline = project.pipelines.get(int(pipeline_id))
jobs = pipeline.jobs.list()
for job in jobs:
logging.info(f'status of gitlab job with id {job.id} and name {job.name} is {job.status}')
if job.status == 'failed':
logging.info(f'collecting failed job {job.name}')
logging.info(f'pipeline associated with failed job is {job.pipeline.get("web_url")}')
failed_jobs.append(job)
return pipeline.web_url, failed_jobs
def main():
install_logging('Slack_Notifier.log')
options = options_handler()
server_url = options.url
slack_token = options.slack_token
ci_token = options.ci_token
# env_results_file_name = options.env_results_file_name
# ci_artifacts_path = options.ci_artifacts
project_id = options.gitlab_project_id
pipeline_id = options.pipeline_id
triggering_workflow = options.triggering_workflow # ci workflow type that is triggering the slack notifier
slack_channel = options.slack_channel
gitlab_client = gitlab.Gitlab(server_url, private_token=ci_token)
pipeline_url, pipeline_failed_jobs = collect_pipeline_data(gitlab_client, project_id, pipeline_id)
slack_msg_data = construct_slack_msg(triggering_workflow, pipeline_url, pipeline_failed_jobs)
slack_client = SlackClient(slack_token)
username = 'Content GitlabCI'
slack_client.api_call(
"chat.postMessage",
json={
'channel': slack_channel,
'username': username,
'as_user': 'False',
'attachments': slack_msg_data
}
)
if __name__ == '__main__':
main()