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
|
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import logging
import os
import taskcluster_urls as liburls
from taskcluster import Hooks
from taskgraph.util import taskcluster as tc_util
from taskgraph.util.taskcluster import (
_do_request,
get_index_url,
get_root_url,
get_task_definition,
get_task_url,
)
logger = logging.getLogger(__name__)
def insert_index(index_path, task_id, data=None, use_proxy=False):
index_url = get_index_url(index_path, use_proxy=use_proxy)
# Find task expiry.
expires = get_task_definition(task_id, use_proxy)["expires"]
response = _do_request(
index_url,
method="put",
json={
"taskId": task_id,
"rank": 0,
"data": data or {},
"expires": expires,
},
)
return response
def status_task(task_id, use_proxy=False):
"""Gets the status of a task given a task_id.
In testing mode, just logs that it would have retrieved status.
Args:
task_id (str): A task id.
use_proxy (bool): Whether to use taskcluster-proxy (default: False)
Returns:
dict: A dictionary object as defined here:
https://docs.taskcluster.net/docs/reference/platform/queue/api#status
"""
if tc_util.testing:
logger.info(f"Would have gotten status for {task_id}.")
else:
resp = _do_request(get_task_url(task_id, use_proxy) + "/status")
status = resp.json().get("status", {})
return status
def state_task(task_id, use_proxy=False):
"""Gets the state of a task given a task_id.
In testing mode, just logs that it would have retrieved state. This is a subset of the
data returned by :func:`status_task`.
Args:
task_id (str): A task id.
use_proxy (bool): Whether to use taskcluster-proxy (default: False)
Returns:
str: The state of the task, one of
``pending, running, completed, failed, exception, unknown``.
"""
if tc_util.testing:
logger.info(f"Would have gotten state for {task_id}.")
else:
status = status_task(task_id, use_proxy=use_proxy).get("state") or "unknown"
return status
def trigger_hook(hook_group_id, hook_id, hook_payload):
hooks = Hooks({"rootUrl": get_root_url(True)})
response = hooks.triggerHook(hook_group_id, hook_id, hook_payload)
logger.info(
"Task seen here: {}/tasks/{}".format(
get_root_url(os.environ.get("TASKCLUSTER_PROXY_URL")),
response["status"]["taskId"],
)
)
def list_task_group_tasks(task_group_id):
"""Generate the tasks in a task group"""
params = {}
while True:
url = liburls.api(
get_root_url(False),
"queue",
"v1",
f"task-group/{task_group_id}/list",
)
resp = _do_request(url, method="get", params=params).json()
yield from resp["tasks"]
if resp.get("continuationToken"):
params = {"continuationToken": resp.get("continuationToken")}
else:
break
def list_task_group_incomplete_task_ids(task_group_id):
states = ("running", "pending", "unscheduled")
for task in [t["status"] for t in list_task_group_tasks(task_group_id)]:
if task["state"] in states:
yield task["taskId"]
def list_task_group_complete_tasks(task_group_id):
tasks = {}
for task in list_task_group_tasks(task_group_id):
if task.get("status", {}).get("state", "") == "completed":
tasks[task.get("task", {}).get("metadata", {}).get("name", "")] = task.get(
"status", {}
).get("taskId", "")
return tasks
|
