Skip to content

Add utask waiting time metric #5545

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
brun0-matheus merged 4 commits into from
Oct 28, 2026
Merged

Add utask waiting time metric #5545

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
29de530
Add preprocess_end_time to uworker input proto
brun0-matheus Oct 14, 2026
809430f
Add UTASK_SUBTASK_WAIT_DURATION_SECS metric
brun0-matheus Oct 14, 2026
b511985
Add and change unittests
brun0-matheus Oct 14, 2026
75aecb9
Add extra unittests
brun0-matheus Oct 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 34 additions & 7 deletions src/clusterfuzz/_internal/bot/tasks/utasks/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,8 @@ def __init__(self, subtask: _Subtask):
else:
self._preprocess_start_time_ns = None

self._previous_subtask_end_time_ns = None

# Events-related metadata.
self._event_data = None
# Error type from utask main handled in postprocess.
Expand All @@ -123,7 +125,8 @@ def set_task_details(
execution_mode: Mode,
platform: str,
preprocess_start_time: Optional[Timestamp] = None,
task_argument: Optional[str | uworker_msg_pb2.Input] = None): # pylint: disable=no-member
task_argument: Optional[str | uworker_msg_pb2.Input] = None, # pylint: disable=no-member
previous_subtask_end_time: Optional[Timestamp] = None):
"""Sets task details that might not be known at instantation time.

Must be called once for metrics to be recorded when exiting the context.
Expand All @@ -141,6 +144,8 @@ def set_task_details(
task_argument: Task argument, which for preprocess is usually the
`testcase_id` or `fuzzer_name`, and for main/postprocess is the uworker
input proto.
previous_subtask_end_time: Timestamp at which the previous subtask for
this task finished executing. None if there is no previous subtask.
"""
task_command = task_utils.get_command_from_module(utask_module.__name__)
self._labels = {
Expand All @@ -161,6 +166,11 @@ def set_task_details(
# Ensure we always have a value after this method returns.
assert self._preprocess_start_time_ns is not None

if previous_subtask_end_time is not None:
end_time = previous_subtask_end_time.ToNanoseconds()
if end_time > 0:
Comment thread
brun0-matheus marked this conversation as resolved.
Resolved
self._previous_subtask_end_time_ns = end_time

def emit_utask_events(self, task_status: str,
task_outcome: str | None = None) -> None:
"""Helper to emit task execution events during the recorder context."""
Expand Down Expand Up @@ -237,6 +247,12 @@ def __exit__(self, _exc_type, _exc_value, _traceback):
monitoring_metrics.UTASK_SUBTASK_E2E_DURATION_SECS.add(
e2e_duration_secs, self._labels)

if self._previous_subtask_end_time_ns is not None:
wait_duration_secs = max(
0, (self.start_time_ns - self._previous_subtask_end_time_ns) / 10**9)
Comment thread
decoNR marked this conversation as resolved.
Resolved
monitoring_metrics.UTASK_SUBTASK_WAIT_DURATION_SECS.add(
Comment thread
decoNR marked this conversation as resolved.
Resolved
wait_duration_secs, self._labels)

# The only case where a task might fail without throwing, is in
# utask_main, by returning an ErrorType proto which indicates
# failure.
Expand Down Expand Up @@ -340,6 +356,7 @@ def tworker_preprocess_no_io(utask_module, task_argument, job_type,
if not uworker_input:
return None

uworker_input.preprocess_end_time.FromNanoseconds(time.time_ns())
return uworker_io.serialize_uworker_input(uworker_input)


Expand All @@ -356,9 +373,13 @@ def uworker_main_no_io(utask_module, serialized_uworker_input):
uworker_input.uworker_env.clear()
task_utils.reset_task_stage_env()

recorder.set_task_details(utask_module, uworker_input.job_type, Mode.QUEUE,
environment.platform(),
uworker_input.preprocess_start_time)
recorder.set_task_details(
utask_module,
uworker_input.job_type,
Mode.QUEUE,
environment.platform(),
uworker_input.preprocess_start_time,
previous_subtask_end_time=uworker_input.preprocess_end_time)
Comment thread
decoNR marked this conversation as resolved.
Resolved

uworker_output = utask_module.utask_main(uworker_input)
if uworker_output is None:
Expand Down Expand Up @@ -428,7 +449,9 @@ def tworker_preprocess(utask_module, task_argument, job_type, uworker_env):
# to pass to the batch job and for the local executor to download locally,
# and the uworker output download URL for the local executor to download
# the output after local execution of `utask_main`.
return uworker_io.serialize_and_upload_uworker_input(uworker_input)
urls = uworker_io.prepare_uworker_io_urls()
uworker_input.preprocess_end_time.FromNanoseconds(time.time_ns())
return uworker_io.serialize_and_upload_uworker_input(uworker_input, urls)


def set_uworker_env(uworker_env: dict) -> None:
Expand Down Expand Up @@ -463,8 +486,12 @@ def uworker_main(input_download_url) -> None:
execution_mode = Mode.SWARMING if environment.is_running_on_swarming(
) else Mode.BATCH
recorder.set_task_details(
utask_module, uworker_input.job_type, execution_mode,
environment.platform(), uworker_input.preprocess_start_time)
utask_module,
uworker_input.job_type,
execution_mode,
environment.platform(),
uworker_input.preprocess_start_time,
previous_subtask_end_time=uworker_input.preprocess_end_time)

logs.info('Starting utask_main: %s.' % utask_module)
uworker_output = utask_module.utask_main(uworker_input)
Expand Down
18 changes: 13 additions & 5 deletions src/clusterfuzz/_internal/bot/tasks/utasks/uworker_io.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@
# limitations under the License.
"""Module for dealing with input and output (I/O) to a uworker."""

from typing import Tuple
from typing import Type
from typing import TypeVar
import uuid
Expand Down Expand Up @@ -99,9 +98,8 @@ def serialize_uworker_input(uworker_input: uworker_msg_pb2.Input) -> bytes: # p
return uworker_input.SerializeToString()


def serialize_and_upload_uworker_input(
uworker_input: uworker_msg_pb2.Input) -> Tuple[str, str]: # pylint: disable=no-member
"""Serializes input for the untrusted portion of a task."""
def prepare_uworker_io_urls() -> tuple[str, str, str, str]:
"""Generates URLs for GCS upload/download."""
signed_input_download_url, input_gcs_url = get_uworker_input_urls()
logs.info(f'input_gcs_url: {input_gcs_url}')
# Get URLs for the uworker'ps output. We need a signed upload URL so it can
Expand All @@ -111,12 +109,22 @@ def serialize_and_upload_uworker_input(
input_gcs_url)
logs.info(f'output_gcs_url: {output_gcs_url}')

return (signed_input_download_url, signed_output_upload_url, input_gcs_url,
Comment thread
brun0-matheus marked this conversation as resolved.
Resolved
output_gcs_url)


def serialize_and_upload_uworker_input(
Comment thread
brun0-matheus marked this conversation as resolved.
Resolved
uworker_input: uworker_msg_pb2.Input, # pylint: disable=no-member
urls: tuple[str, str, str, str]) -> tuple[str, str]:
"""Serializes input for the untrusted portion of a task."""
(signed_input_download_url, signed_output_upload_url, input_gcs_url,
output_gcs_url) = urls

assert not uworker_input.HasField('uworker_output_upload_url')
uworker_input.uworker_output_upload_url = signed_output_upload_url

serialized_uworker_input = zlib.compress(uworker_input.SerializeToString())
upload_uworker_input(serialized_uworker_input, input_gcs_url)

return signed_input_download_url, output_gcs_url


Expand Down
16 changes: 16 additions & 0 deletions src/clusterfuzz/_internal/metrics/monitoring_metrics.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -374,6 +374,22 @@
],
)

UTASK_SUBTASK_WAIT_DURATION_SECS = monitor.CumulativeDistributionMetric(
'utask/subtask_wait_duration_secs',
description=(
'Time elapsed between the previous subtask finishing and the current '
'subtask starting, in seconds, per subtask. '
'Mode is "batch", "swarming", or "queue".'),
bucketer=monitor.GeometricBucketer(),
field_spec=[
monitor.StringField('task'),
monitor.StringField('job'),
monitor.StringField('subtask'),
monitor.StringField('mode'),
monitor.StringField('platform'),
],
)

ANDROID_UPTIME = monitor.CounterMetric(
'android_device_uptime',
description='Android device uptime',
Expand Down
1 change: 1 addition & 0 deletions src/clusterfuzz/_internal/protos/uworker_msg.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,7 @@ message Input {
optional google.protobuf.Timestamp preprocess_start_time = 19;
repeated string global_blacklisted_functions = 20;
optional google.protobuf.Any fuzz_target = 21;
optional google.protobuf.Timestamp preprocess_end_time = 22;
}

message VariantTaskInput {
Expand Down
100 changes: 50 additions & 50 deletions src/clusterfuzz/_internal/protos/uworker_msg_pb2.py
View file
Open in desktop

Large diffs are not rendered by default.

10 changes: 8 additions & 2 deletions src/clusterfuzz/_internal/protos/uworker_msg_pb2.pyi
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -636,6 +636,7 @@ class Input(google.protobuf.message.Message):
PREPROCESS_START_TIME_FIELD_NUMBER: builtins.int
GLOBAL_BLACKLISTED_FUNCTIONS_FIELD_NUMBER: builtins.int
FUZZ_TARGET_FIELD_NUMBER: builtins.int
PREPROCESS_END_TIME_FIELD_NUMBER: builtins.int
@property
def testcase(self) -> google.protobuf.any_pb2.Any: ...
@property
Expand Down Expand Up @@ -672,6 +673,8 @@ class Input(google.protobuf.message.Message):
def global_blacklisted_functions(self) -> google.protobuf.internal.containers.RepeatedScalarFieldContainer[builtins.str]: ...
@property
def fuzz_target(self) -> google.protobuf.any_pb2.Any: ...
@property
def preprocess_end_time(self) -> google.protobuf.timestamp_pb2.Timestamp: ...
def __init__(
self,
*,
Expand All @@ -695,9 +698,10 @@ class Input(google.protobuf.message.Message):
preprocess_start_time: google.protobuf.timestamp_pb2.Timestamp | None = ...,
global_blacklisted_functions: collections.abc.Iterable[builtins.str] | None = ...,
fuzz_target: google.protobuf.any_pb2.Any | None = ...,
preprocess_end_time: google.protobuf.timestamp_pb2.Timestamp | None = ...,
) -> None: ...
def HasField(self, field_name: typing_extensions.Literal["_analyze_task_input", b"_analyze_task_input", "_corpus_pruning_task_input", b"_corpus_pruning_task_input", "_fuzz_target", b"_fuzz_target", "_fuzz_task_input", b"_fuzz_task_input", "_fuzzer_name", b"_fuzzer_name", "_job_type", b"_job_type", "_minimize_task_input", b"_minimize_task_input", "_module_name", b"_module_name", "_preprocess_start_time", b"_preprocess_start_time", "_progression_task_input", b"_progression_task_input", "_regression_task_input", b"_regression_task_input", "_setup_input", b"_setup_input", "_symbolize_task_input", b"_symbolize_task_input", "_testcase", b"_testcase", "_testcase_id", b"_testcase_id", "_testcase_upload_metadata", b"_testcase_upload_metadata", "_uworker_output_upload_url", b"_uworker_output_upload_url", "_variant_task_input", b"_variant_task_input", "analyze_task_input", b"analyze_task_input", "corpus_pruning_task_input", b"corpus_pruning_task_input", "fuzz_target", b"fuzz_target", "fuzz_task_input", b"fuzz_task_input", "fuzzer_name", b"fuzzer_name", "job_type", b"job_type", "minimize_task_input", b"minimize_task_input", "module_name", b"module_name", "preprocess_start_time", b"preprocess_start_time", "progression_task_input", b"progression_task_input", "regression_task_input", b"regression_task_input", "setup_input", b"setup_input", "symbolize_task_input", b"symbolize_task_input", "testcase", b"testcase", "testcase_id", b"testcase_id", "testcase_upload_metadata", b"testcase_upload_metadata", "uworker_output_upload_url", b"uworker_output_upload_url", "variant_task_input", b"variant_task_input"]) -> builtins.bool: ...
def ClearField(self, field_name: typing_extensions.Literal["_analyze_task_input", b"_analyze_task_input", "_corpus_pruning_task_input", b"_corpus_pruning_task_input", "_fuzz_target", b"_fuzz_target", "_fuzz_task_input", b"_fuzz_task_input", "_fuzzer_name", b"_fuzzer_name", "_job_type", b"_job_type", "_minimize_task_input", b"_minimize_task_input", "_module_name", b"_module_name", "_preprocess_start_time", b"_preprocess_start_time", "_progression_task_input", b"_progression_task_input", "_regression_task_input", b"_regression_task_input", "_setup_input", b"_setup_input", "_symbolize_task_input", b"_symbolize_task_input", "_testcase", b"_testcase", "_testcase_id", b"_testcase_id", "_testcase_upload_metadata", b"_testcase_upload_metadata", "_uworker_output_upload_url", b"_uworker_output_upload_url", "_variant_task_input", b"_variant_task_input", "analyze_task_input", b"analyze_task_input", "corpus_pruning_task_input", b"corpus_pruning_task_input", "fuzz_target", b"fuzz_target", "fuzz_task_input", b"fuzz_task_input", "fuzzer_name", b"fuzzer_name", "global_blacklisted_functions", b"global_blacklisted_functions", "job_type", b"job_type", "minimize_task_input", b"minimize_task_input", "module_name", b"module_name", "preprocess_start_time", b"preprocess_start_time", "progression_task_input", b"progression_task_input", "regression_task_input", b"regression_task_input", "setup_input", b"setup_input", "symbolize_task_input", b"symbolize_task_input", "testcase", b"testcase", "testcase_id", b"testcase_id", "testcase_upload_metadata", b"testcase_upload_metadata", "uworker_env", b"uworker_env", "uworker_output_upload_url", b"uworker_output_upload_url", "variant_task_input", b"variant_task_input"]) -> None: ...
def HasField(self, field_name: typing_extensions.Literal["_analyze_task_input", b"_analyze_task_input", "_corpus_pruning_task_input", b"_corpus_pruning_task_input", "_fuzz_target", b"_fuzz_target", "_fuzz_task_input", b"_fuzz_task_input", "_fuzzer_name", b"_fuzzer_name", "_job_type", b"_job_type", "_minimize_task_input", b"_minimize_task_input", "_module_name", b"_module_name", "_preprocess_end_time", b"_preprocess_end_time", "_preprocess_start_time", b"_preprocess_start_time", "_progression_task_input", b"_progression_task_input", "_regression_task_input", b"_regression_task_input", "_setup_input", b"_setup_input", "_symbolize_task_input", b"_symbolize_task_input", "_testcase", b"_testcase", "_testcase_id", b"_testcase_id", "_testcase_upload_metadata", b"_testcase_upload_metadata", "_uworker_output_upload_url", b"_uworker_output_upload_url", "_variant_task_input", b"_variant_task_input", "analyze_task_input", b"analyze_task_input", "corpus_pruning_task_input", b"corpus_pruning_task_input", "fuzz_target", b"fuzz_target", "fuzz_task_input", b"fuzz_task_input", "fuzzer_name", b"fuzzer_name", "job_type", b"job_type", "minimize_task_input", b"minimize_task_input", "module_name", b"module_name", "preprocess_end_time", b"preprocess_end_time", "preprocess_start_time", b"preprocess_start_time", "progression_task_input", b"progression_task_input", "regression_task_input", b"regression_task_input", "setup_input", b"setup_input", "symbolize_task_input", b"symbolize_task_input", "testcase", b"testcase", "testcase_id", b"testcase_id", "testcase_upload_metadata", b"testcase_upload_metadata", "uworker_output_upload_url", b"uworker_output_upload_url", "variant_task_input", b"variant_task_input"]) -> builtins.bool: ...
def ClearField(self, field_name: typing_extensions.Literal["_analyze_task_input", b"_analyze_task_input", "_corpus_pruning_task_input", b"_corpus_pruning_task_input", "_fuzz_target", b"_fuzz_target", "_fuzz_task_input", b"_fuzz_task_input", "_fuzzer_name", b"_fuzzer_name", "_job_type", b"_job_type", "_minimize_task_input", b"_minimize_task_input", "_module_name", b"_module_name", "_preprocess_end_time", b"_preprocess_end_time", "_preprocess_start_time", b"_preprocess_start_time", "_progression_task_input", b"_progression_task_input", "_regression_task_input", b"_regression_task_input", "_setup_input", b"_setup_input", "_symbolize_task_input", b"_symbolize_task_input", "_testcase", b"_testcase", "_testcase_id", b"_testcase_id", "_testcase_upload_metadata", b"_testcase_upload_metadata", "_uworker_output_upload_url", b"_uworker_output_upload_url", "_variant_task_input", b"_variant_task_input", "analyze_task_input", b"analyze_task_input", "corpus_pruning_task_input", b"corpus_pruning_task_input", "fuzz_target", b"fuzz_target", "fuzz_task_input", b"fuzz_task_input", "fuzzer_name", b"fuzzer_name", "global_blacklisted_functions", b"global_blacklisted_functions", "job_type", b"job_type", "minimize_task_input", b"minimize_task_input", "module_name", b"module_name", "preprocess_end_time", b"preprocess_end_time", "preprocess_start_time", b"preprocess_start_time", "progression_task_input", b"progression_task_input", "regression_task_input", b"regression_task_input", "setup_input", b"setup_input", "symbolize_task_input", b"symbolize_task_input", "testcase", b"testcase", "testcase_id", b"testcase_id", "testcase_upload_metadata", b"testcase_upload_metadata", "uworker_env", b"uworker_env", "uworker_output_upload_url", b"uworker_output_upload_url", "variant_task_input", b"variant_task_input"]) -> None: ...
@typing.overload
def WhichOneof(self, oneof_group: typing_extensions.Literal["_analyze_task_input", b"_analyze_task_input"]) -> typing_extensions.Literal["analyze_task_input"] | None: ...
@typing.overload
Expand All @@ -715,6 +719,8 @@ class Input(google.protobuf.message.Message):
@typing.overload
def WhichOneof(self, oneof_group: typing_extensions.Literal["_module_name", b"_module_name"]) -> typing_extensions.Literal["module_name"] | None: ...
@typing.overload
def WhichOneof(self, oneof_group: typing_extensions.Literal["_preprocess_end_time", b"_preprocess_end_time"]) -> typing_extensions.Literal["preprocess_end_time"] | None: ...
@typing.overload
def WhichOneof(self, oneof_group: typing_extensions.Literal["_preprocess_start_time", b"_preprocess_start_time"]) -> typing_extensions.Literal["preprocess_start_time"] | None: ...
@typing.overload
def WhichOneof(self, oneof_group: typing_extensions.Literal["_progression_task_input", b"_progression_task_input"]) -> typing_extensions.Literal["progression_task_input"] | None: ...
Expand Down
Loading
Loading