-
Notifications
You must be signed in to change notification settings - Fork 237
/
debug_logs.py
135 lines (122 loc) · 4.75 KB
/
debug_logs.py
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
import datetime
import os
import csv
from cumulusci.tasks.salesforce import BaseSalesforceApiTask
class TurnOnDebugLogs(BaseSalesforceApiTask):
task_docs = """
Turn on debugging information in Salesforce
"""
debug_level_id = None # Class variable
def _run_task(self):
"""Create a TraceFlag for a given user."""
self._delete_debug_levels()
self._delete_trace_flags()
self.logger.info("Creating DebugLevel object")
DebugLevel = self._get_tooling_object("DebugLevel")
result = DebugLevel.create(
{
"ApexCode": "Info",
"ApexProfiling": "Debug",
"Callout": "Info",
"Database": "Info",
"DeveloperName": "CumulusCI",
"MasterLabel": "CumulusCI",
"System": "Info",
"Validation": "Info",
"Visualforce": "Info",
"Workflow": "Info",
}
)
if TurnOnDebugLogs.debug_level_id:
self.logger.info(
"Debug logs are already turned on ({TurnOnDebugLogs.debug_level_id})"
)
return
TurnOnDebugLogs.debug_level_id = result["id"]
self.logger.info("Setting up trace flag to capture debug logs")
# New TraceFlag expires 12 hours from now
expiration_date = datetime.datetime.utcnow() + datetime.timedelta(
seconds=60 * 60 * 12
)
TraceFlag = self._get_tooling_object("TraceFlag")
result = TraceFlag.create(
{
"DebugLevelId": result["id"],
"ExpirationDate": expiration_date.isoformat(),
"LogType": "USER_DEBUG",
"TracedEntityId": self.org_config.user_id,
}
)
self.logger.info("Created TraceFlag for user")
def _delete_trace_flags(self):
"""
Delete existing DebugLevel objects.
This will automatically delete associated TraceFlags as well.
"""
self.logger.info("Deleting existing TraceFlag objects")
result = self.tooling.query(
"Select Id from TraceFlag Where TracedEntityId = '{}'".format(
self.org_config.user_id
)
)
if result["totalSize"]:
TraceFlag = self._get_tooling_object("TraceFlag")
for record in result["records"]:
TraceFlag.delete(str(record["Id"]))
def _delete_debug_levels(self):
"""
Delete existing DebugLevel objects.
This will automatically delete associated TraceFlags as well.
"""
self.logger.info("Deleting existing DebugLevel objects")
result = self.tooling.query("Select Id from DebugLevel")
if result["totalSize"]:
DebugLevel = self._get_tooling_object("DebugLevel")
for record in result["records"]:
DebugLevel.delete(str(record["Id"]))
# Based on https://github.com/SFDO-Tooling/CumulusCI/blob/225c6fcc6d625330abebe9efb087bba84f2daaa5/cumulusci/tasks/salesforce.py#L1511
class DownloadDebugLogs(BaseSalesforceApiTask):
task_docs = """
Download debug logs from Salesforce.
"""
task_options = {
"debug_log_dir": {"description": "Where to put the logs", "required": True}
}
def _run_task(self):
result = self.tooling.query_all(
"SELECT Id, Application, "
+ "DurationMilliseconds, Location, LogLength, LogUserId, "
+ "Operation, Request, StartTime, Status "
+ "from ApexLog"
)
debug_log_dir = self.options.get("debug_log_dir")
if not os.path.exists(debug_log_dir):
os.makedirs(debug_log_dir)
filename = "%s/%s.log" % (
debug_log_dir,
str(datetime.datetime.utcnow()).replace(" ", "-"),
)
with open(filename, "w", newline="") as f:
# the original code had a one-class-per-file layout
# which could be restored if helpful. See URL above.
fieldnames = [
"Id",
"Application",
"DurationMilliseconds",
"Location",
"LogLength",
"LogUserId",
"Operation",
"Request",
"StartTime",
"Status",
]
writer = csv.DictWriter(f, fieldnames, extrasaction="ignore")
writer.writeheader()
for log in result["records"]:
writer.writerow(log)
self.logger.info("Created: " + filename)
if TurnOnDebugLogs.debug_level_id:
DebugLevel = self._get_tooling_object("DebugLevel")
DebugLevel.delete(str(TurnOnDebugLogs.debug_level_id))
TurnOnDebugLogs.debug_level_id = None