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
|
#!/usr/bin/env python3
# Copyright 2021 The ChromiumOS Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Utilities to file bugs."""
import base64
import datetime
import enum
import json
import os
from typing import Any, Dict, List, Optional
X20_PATH = "/google/data/rw/teams/c-compiler-chrome/prod_bugs"
class WellKnownComponents(enum.IntEnum):
"""A listing of "well-known" components recognized by our infra."""
CrOSToolchainPublic = -1
CrOSToolchainPrivate = -2
def _WriteBugJSONFile(object_type: str, json_object: Dict[str, Any]):
"""Writes a JSON file to X20_PATH with the given bug-ish object."""
final_object = {
"type": object_type,
"value": json_object,
}
# The name of this has two parts:
# - An easily sortable time, to provide uniqueness and let our service send
# things in the order they were put into the outbox.
# - 64 bits of entropy, so two racing bug writes don't clobber the same file.
now = datetime.datetime.utcnow().isoformat("T", "seconds") + "Z"
entropy = base64.urlsafe_b64encode(os.getrandom(8))
entropy_str = entropy.rstrip(b"=").decode("utf-8")
file_path = os.path.join(X20_PATH, f"{now}_{entropy_str}.json")
temp_path = file_path + ".in_progress"
try:
with open(temp_path, "w") as f:
json.dump(final_object, f)
os.rename(temp_path, file_path)
except:
os.remove(temp_path)
raise
return file_path
def AppendToExistingBug(bug_id: int, body: str):
"""Sends a reply to an existing bug."""
_WriteBugJSONFile(
"AppendToExistingBugRequest",
{
"body": body,
"bug_id": bug_id,
},
)
def CreateNewBug(
component_id: int,
title: str,
body: str,
assignee: Optional[str] = None,
cc: Optional[List[str]] = None,
):
"""Sends a request to create a new bug.
Args:
component_id: The component ID to add. Anything from WellKnownComponents
also works.
title: Title of the bug. Must be nonempty.
body: Body of the bug. Must be nonempty.
assignee: Assignee of the bug. Must be either an email address, or a
"well-known" assignee (detective, mage).
cc: A list of emails to add to the CC list. Must either be an email
address, or a "well-known" individual (detective, mage).
"""
obj = {
"component_id": component_id,
"subject": title,
"body": body,
}
if assignee:
obj["assignee"] = assignee
if cc:
obj["cc"] = cc
_WriteBugJSONFile("FileNewBugRequest", obj)
def SendCronjobLog(cronjob_name: str, failed: bool, message: str):
"""Sends the record of a cronjob to our bug infra.
cronjob_name: The name of the cronjob. Expected to remain consistent over
time.
failed: Whether the job failed or not.
message: Any seemingly relevant context. This is pasted verbatim in a bug, if
the cronjob infra deems it worthy.
"""
_WriteBugJSONFile(
"ChrotomationCronjobUpdate",
{
"name": cronjob_name,
"message": message,
"failed": failed,
},
)
|
