crossbench/probes/internal.py - platform/external/chromium-crossbench - Git at Google

 # Copyright 2023 The Chromium Authors
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.

 from __future__ import annotations

 import json
 import logging
 from typing import TYPE_CHECKING, Iterable, Optional

 from crossbench.probes import probe
 from crossbench.probes.json import JsonResultProbe, JsonResultProbeContext
 from crossbench.probes.metric import MetricsMerger
 from crossbench.probes.results import (EmptyProbeResult, ProbeResult,
                                        ProbeResultDict)

 if TYPE_CHECKING:
   from crossbench.runner.actions import Actions
   from crossbench.runner.groups.browsers import BrowsersRunGroup
   from crossbench.runner.groups.repetitions import RepetitionsRunGroup
   from crossbench.runner.groups.stories import StoriesRunGroup
   from crossbench.runner.run import Run
   from crossbench.types import Json, JsonDict, JsonList


 class InternalProbe(probe.Probe):
   IS_GENERAL_PURPOSE = False

   @property
   def is_internal(self) -> bool:
     return True


 class InternalJsonResultProbe(JsonResultProbe, InternalProbe):
   IS_GENERAL_PURPOSE = False
   FLATTEN = False

   def get_context(self, run: Run) -> InternalJsonResultProbeContext:
     return InternalJsonResultProbeContext(self, run)


 class InternalJsonResultProbeContext(
     JsonResultProbeContext[InternalJsonResultProbe]):

   def stop(self) -> None:
     # Only extract data in the late teardown phase.
     pass

   def teardown(self) -> ProbeResult:
     self._json_data = self.extract_json(self.run)  # pylint: disable=no-member
     return super().teardown()


 class LogProbe(InternalProbe):
   """
   Runner-internal meta-probe: Collects the python logging data from the runner
   itself.
   """
   NAME = "cb.log"

   def get_context(self, run: Run) -> LogProbeContext:
     return LogProbeContext(self, run)


 class LogProbeContext(probe.ProbeContext[LogProbe]):

   def __init__(self, probe_instance: LogProbe, run: Run) -> None:
     super().__init__(probe_instance, run)
     self._log_handler: Optional[logging.Handler] = None

   def setup(self) -> None:
     log_formatter = logging.Formatter(
         "%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s] "
         "[%(name)s]  %(message)s")
     self._log_handler = logging.FileHandler(self.result_path)
     self._log_handler.setFormatter(log_formatter)
     self._log_handler.setLevel(logging.DEBUG)
     logging.getLogger().addHandler(self._log_handler)

   def start(self) -> None:
     pass

   def stop(self) -> None:
     pass

   def teardown(self) -> ProbeResult:
     assert self._log_handler
     logging.getLogger().removeHandler(self._log_handler)
     self._log_handler = None
     return ProbeResult(file=(self.local_result_path,))


 class SystemDetailsProbe(InternalJsonResultProbe):
   """
   Runner-internal meta-probe: Collects the browser's system/platform details.
   """
   NAME = "cb.system.details"

   def to_json(self, actions: Actions) -> Json:
     return actions.run.browser_platform.system_details()

   def merge_repetitions(self, group: RepetitionsRunGroup) -> ProbeResult:
     return EmptyProbeResult()


 class ErrorsProbe(InternalJsonResultProbe):
   """
   Runner-internal meta-probe: Collects all errors from running stories and/or
   from merging probe data.
   """
   NAME = "cb.errors"

   def to_json(self, actions: Actions) -> Json:
     return actions.run.exceptions.to_json()

   def merge_repetitions(self, group: RepetitionsRunGroup) -> ProbeResult:
     return self._merge_group(group, (run.results for run in group.runs))

   def merge_stories(self, group: StoriesRunGroup) -> ProbeResult:
     return self._merge_group(
         group, (rep_group.results for rep_group in group.repetitions_groups))

   def merge_browsers(self, group: BrowsersRunGroup) -> ProbeResult:
     return self._merge_group(
         group, (story_group.results for story_group in group.story_groups))

   def _merge_group(self, group,
                    results_iter: Iterable[ProbeResultDict]) -> ProbeResult:
     merged_errors = []

     for results in results_iter:
       result = results[self]
       if not result:
         continue
       source_file = result.json
       assert source_file.is_file()
       with source_file.open(encoding="utf-8") as f:
         repetition_errors = json.load(f)
         assert isinstance(repetition_errors, list)
         merged_errors.extend(repetition_errors)

     group_errors = group.exceptions.to_json()
     assert isinstance(group_errors, list)
     merged_errors.extend(group_errors)

     if not merged_errors:
       return EmptyProbeResult()

     return self.write_group_result(group, merged_errors)


 class DurationsProbe(InternalJsonResultProbe):
   """
   Runner-internal meta-probe: Collects timing information for various components
   of the runner (and the times spent in individual stories as well).
   """
   NAME = "cb.durations"

   def to_json(self, actions: Actions) -> Json:
     return actions.run.durations.to_json()

   def merge_stories(self, group: StoriesRunGroup) -> ProbeResult:
     merged = MetricsMerger.merge_json_list(
         (repetitions_group.results[self].json
          for repetitions_group in group.repetitions_groups),
         merge_duplicate_paths=True)
     return self.write_group_result(group, merged, csv_formatter=None)

   def merge_browsers(self, group: BrowsersRunGroup) -> ProbeResult:
     merged = MetricsMerger.merge_json_list(
         (story_group.results[self].json for story_group in group.story_groups),
         merge_duplicate_paths=True)
     return self.write_group_result(group, merged, csv_formatter=None)


 class ResultsSummaryProbe(InternalJsonResultProbe):
   """
   Runner-internal meta-probe: Collects a summary results.json with all the Run
   information, including all paths to the results of all attached Probes.
   """
   NAME = "cb.results"
   # Given that this is  a meta-Probe that summarizes the data from other
   # probes we exclude it from the default results lists.
   PRODUCES_DATA = False

   @property
   def is_attached(self) -> bool:
     return True

   def to_json(self, actions: Actions) -> JsonDict:
     return actions.run.details_json()

   def merge_repetitions(self, group: RepetitionsRunGroup) -> ProbeResult:
     repetitions: JsonList = []
     browser: Optional[JsonDict] = None

     for run in group.runs:
       source_file = run.results[self].json
       assert source_file.is_file()
       with source_file.open(encoding="utf-8") as f:
         repetition_data = json.load(f)
       if browser is None:
         browser = repetition_data["browser"]
         del browser["log"]
       repetitions.append({
           "cwd": repetition_data["cwd"],
           "probes": repetition_data["probes"],
           "success": repetition_data["success"],
           "errors": repetition_data["errors"],
       })

     merged_data: JsonDict = {
         "cwd": str(group.path),
         "story": group.story.details_json(),
         "browser": browser,
         "group": group.info,
         "repetitions": repetitions,
         "probes": group.results.to_json(),
         "success": group.is_success,
         "errors": group.exceptions.error_messages(),
     }
     return self.write_group_result(group, merged_data, csv_formatter=None)

   def merge_stories(self, group: StoriesRunGroup) -> ProbeResult:
     stories: JsonDict = {}
     browser = None

     for repetitions_group in group.repetitions_groups:
       source_file = repetitions_group.results[self].json
       assert source_file.is_file()
       with source_file.open(encoding="utf-8") as f:
         merged_story_data = json.load(f)
       if browser is None:
         browser = merged_story_data["browser"]
       story_info = merged_story_data["story"]
       stories[story_info["name"]] = {
           "cwd": merged_story_data["cwd"],
           "duration": story_info["duration"],
           "probes": merged_story_data["probes"],
           "errors": merged_story_data["errors"],
       }

     merged_data: JsonDict = {
         "cwd": str(group.path),
         "browser": browser,
         "stories": stories,
         "group": group.info,
         "probes": group.results.to_json(),
         "success": group.is_success,
         "errors": group.exceptions.error_messages(),
     }
     return self.write_group_result(group, merged_data, csv_formatter=None)

   def merge_browsers(self, group: BrowsersRunGroup) -> ProbeResult:
     browsers: JsonDict = {}
     for story_group in group.story_groups:
       source_file = story_group.results[self].json
       assert source_file.is_file()
       with source_file.open(encoding="utf-8") as f:
         merged_browser_data = json.load(f)
       browser_info = merged_browser_data["browser"]
       browsers[browser_info["unique_name"]] = {
           "cwd": merged_browser_data["cwd"],
           "probes": merged_browser_data["probes"],
           "errors": merged_browser_data["errors"],
       }

     merged_data: JsonDict = {
         "cwd": str(group.path),
         "browsers": browsers,
         "probes": group.results.to_json(),
         "success": group.is_success,
         "errors": group.exceptions.error_messages(),
     }
     return self.write_group_result(group, merged_data, csv_formatter=None)
	# Copyright 2023 The Chromium Authors
	# Use of this source code is governed by a BSD-style license that can be
	# found in the LICENSE file.

	from __future__ import annotations

	import json
	import logging
	from typing import TYPE_CHECKING, Iterable, Optional

	from crossbench.probes import probe
	from crossbench.probes.json import JsonResultProbe, JsonResultProbeContext
	from crossbench.probes.metric import MetricsMerger
	from crossbench.probes.results import (EmptyProbeResult, ProbeResult,
	ProbeResultDict)

	if TYPE_CHECKING:
	from crossbench.runner.actions import Actions
	from crossbench.runner.groups.browsers import BrowsersRunGroup
	from crossbench.runner.groups.repetitions import RepetitionsRunGroup
	from crossbench.runner.groups.stories import StoriesRunGroup
	from crossbench.runner.run import Run
	from crossbench.types import Json, JsonDict, JsonList


	class InternalProbe(probe.Probe):
	IS_GENERAL_PURPOSE = False

	@property
	def is_internal(self) -> bool:
	return True


	class InternalJsonResultProbe(JsonResultProbe, InternalProbe):
	IS_GENERAL_PURPOSE = False
	FLATTEN = False

	def get_context(self, run: Run) -> InternalJsonResultProbeContext:
	return InternalJsonResultProbeContext(self, run)


	class InternalJsonResultProbeContext(
	JsonResultProbeContext[InternalJsonResultProbe]):

	def stop(self) -> None:
	# Only extract data in the late teardown phase.
	pass

	def teardown(self) -> ProbeResult:
	self._json_data = self.extract_json(self.run) # pylint: disable=no-member
	return super().teardown()


	class LogProbe(InternalProbe):
	"""
	Runner-internal meta-probe: Collects the python logging data from the runner
	itself.
	"""
	NAME = "cb.log"

	def get_context(self, run: Run) -> LogProbeContext:
	return LogProbeContext(self, run)


	class LogProbeContext(probe.ProbeContext[LogProbe]):

	def __init__(self, probe_instance: LogProbe, run: Run) -> None:
	super().__init__(probe_instance, run)
	self._log_handler: Optional[logging.Handler] = None

	def setup(self) -> None:
	log_formatter = logging.Formatter(
	"%(asctime)s [%(threadName)-12.12s] [%(levelname)-5.5s] "
	"[%(name)s] %(message)s")
	self._log_handler = logging.FileHandler(self.result_path)
	self._log_handler.setFormatter(log_formatter)
	self._log_handler.setLevel(logging.DEBUG)
	logging.getLogger().addHandler(self._log_handler)

	def start(self) -> None:
	pass

	def stop(self) -> None:
	pass

	def teardown(self) -> ProbeResult:
	assert self._log_handler
	logging.getLogger().removeHandler(self._log_handler)
	self._log_handler = None
	return ProbeResult(file=(self.local_result_path,))


	class SystemDetailsProbe(InternalJsonResultProbe):
	"""
	Runner-internal meta-probe: Collects the browser's system/platform details.
	"""
	NAME = "cb.system.details"

	def to_json(self, actions: Actions) -> Json:
	return actions.run.browser_platform.system_details()

	def merge_repetitions(self, group: RepetitionsRunGroup) -> ProbeResult:
	return EmptyProbeResult()


	class ErrorsProbe(InternalJsonResultProbe):
	"""
	Runner-internal meta-probe: Collects all errors from running stories and/or
	from merging probe data.
	"""
	NAME = "cb.errors"

	def to_json(self, actions: Actions) -> Json:
	return actions.run.exceptions.to_json()

	def merge_repetitions(self, group: RepetitionsRunGroup) -> ProbeResult:
	return self._merge_group(group, (run.results for run in group.runs))

	def merge_stories(self, group: StoriesRunGroup) -> ProbeResult:
	return self._merge_group(
	group, (rep_group.results for rep_group in group.repetitions_groups))

	def merge_browsers(self, group: BrowsersRunGroup) -> ProbeResult:
	return self._merge_group(
	group, (story_group.results for story_group in group.story_groups))

	def _merge_group(self, group,
	results_iter: Iterable[ProbeResultDict]) -> ProbeResult:
	merged_errors = []

	for results in results_iter:
	result = results[self]
	if not result:
	continue
	source_file = result.json
	assert source_file.is_file()
	with source_file.open(encoding="utf-8") as f:
	repetition_errors = json.load(f)
	assert isinstance(repetition_errors, list)
	merged_errors.extend(repetition_errors)

	group_errors = group.exceptions.to_json()
	assert isinstance(group_errors, list)
	merged_errors.extend(group_errors)

	if not merged_errors:
	return EmptyProbeResult()

	return self.write_group_result(group, merged_errors)


	class DurationsProbe(InternalJsonResultProbe):
	"""
	Runner-internal meta-probe: Collects timing information for various components
	of the runner (and the times spent in individual stories as well).
	"""
	NAME = "cb.durations"

	def to_json(self, actions: Actions) -> Json:
	return actions.run.durations.to_json()

	def merge_stories(self, group: StoriesRunGroup) -> ProbeResult:
	merged = MetricsMerger.merge_json_list(
	(repetitions_group.results[self].json
	for repetitions_group in group.repetitions_groups),
	merge_duplicate_paths=True)
	return self.write_group_result(group, merged, csv_formatter=None)

	def merge_browsers(self, group: BrowsersRunGroup) -> ProbeResult:
	merged = MetricsMerger.merge_json_list(
	(story_group.results[self].json for story_group in group.story_groups),
	merge_duplicate_paths=True)
	return self.write_group_result(group, merged, csv_formatter=None)


	class ResultsSummaryProbe(InternalJsonResultProbe):
	"""
	Runner-internal meta-probe: Collects a summary results.json with all the Run
	information, including all paths to the results of all attached Probes.
	"""
	NAME = "cb.results"
	# Given that this is a meta-Probe that summarizes the data from other
	# probes we exclude it from the default results lists.
	PRODUCES_DATA = False

	@property
	def is_attached(self) -> bool:
	return True

	def to_json(self, actions: Actions) -> JsonDict:
	return actions.run.details_json()

	def merge_repetitions(self, group: RepetitionsRunGroup) -> ProbeResult:
	repetitions: JsonList = []
	browser: Optional[JsonDict] = None

	for run in group.runs:
	source_file = run.results[self].json
	assert source_file.is_file()
	with source_file.open(encoding="utf-8") as f:
	repetition_data = json.load(f)
	if browser is None:
	browser = repetition_data["browser"]
	del browser["log"]
	repetitions.append({
	"cwd": repetition_data["cwd"],
	"probes": repetition_data["probes"],
	"success": repetition_data["success"],
	"errors": repetition_data["errors"],
	})

	merged_data: JsonDict = {
	"cwd": str(group.path),
	"story": group.story.details_json(),
	"browser": browser,
	"group": group.info,
	"repetitions": repetitions,
	"probes": group.results.to_json(),
	"success": group.is_success,
	"errors": group.exceptions.error_messages(),
	}
	return self.write_group_result(group, merged_data, csv_formatter=None)

	def merge_stories(self, group: StoriesRunGroup) -> ProbeResult:
	stories: JsonDict = {}
	browser = None

	for repetitions_group in group.repetitions_groups:
	source_file = repetitions_group.results[self].json
	assert source_file.is_file()
	with source_file.open(encoding="utf-8") as f:
	merged_story_data = json.load(f)
	if browser is None:
	browser = merged_story_data["browser"]
	story_info = merged_story_data["story"]
	stories[story_info["name"]] = {
	"cwd": merged_story_data["cwd"],
	"duration": story_info["duration"],
	"probes": merged_story_data["probes"],
	"errors": merged_story_data["errors"],
	}

	merged_data: JsonDict = {
	"cwd": str(group.path),
	"browser": browser,
	"stories": stories,
	"group": group.info,
	"probes": group.results.to_json(),
	"success": group.is_success,
	"errors": group.exceptions.error_messages(),
	}
	return self.write_group_result(group, merged_data, csv_formatter=None)

	def merge_browsers(self, group: BrowsersRunGroup) -> ProbeResult:
	browsers: JsonDict = {}
	for story_group in group.story_groups:
	source_file = story_group.results[self].json
	assert source_file.is_file()
	with source_file.open(encoding="utf-8") as f:
	merged_browser_data = json.load(f)
	browser_info = merged_browser_data["browser"]
	browsers[browser_info["unique_name"]] = {
	"cwd": merged_browser_data["cwd"],
	"probes": merged_browser_data["probes"],
	"errors": merged_browser_data["errors"],
	}

	merged_data: JsonDict = {
	"cwd": str(group.path),
	"browsers": browsers,
	"probes": group.results.to_json(),
	"success": group.is_success,
	"errors": group.exceptions.error_messages(),
	}
	return self.write_group_result(group, merged_data, csv_formatter=None)