pantsbuild / pants / 21378504907

import csv

import logging

from typing import Any, Dict, Mapping, Tuple

from pants.engine.internals.scheduler import Workunit

from pants.engine.rules import collect_rules, rule

from pants.engine.streaming_workunit_handler import (

from pants.engine.unions import UnionRule

from pants.option.option_types import BoolOption, StrOption

from pants.option.subsystem import Subsystem

from pants.util.dirutil import safe_open

logger = logging.getLogger(__name__)

class RemoteCacheLoggerCallback(WorkunitsCallback):

    def __init__(self, wulogger: "RemoteCacheLogger"):

        self.wulogger = wulogger

        self._completed_workunits: Dict[str, Mapping[str, Any]] = {}

    @property

    def can_finish_async(self) -> bool:

        return False

    def __call__(

        for wu in completed_workunits:

            if wu["name"] == "remote_cache_read_speculation":

                metadata = wu.get("metadata", {})

                if not metadata or "request_description" not in metadata:

                    continue

                self._completed_workunits[wu["span_id"]] = {

        if finished:

            logger.debug(f"RemoteCacheLogger finished: captured {len(self._completed_workunits)} workunits")

            filepath = f"{self.wulogger.logdir}/{context.run_tracker.run_id}.csv"

            with safe_open(filepath, "w", newline="") as f:

                fieldnames = ["description", "action_digest", "outcome", "request"]

                writer = csv.DictWriter(f, fieldnames=fieldnames)

                writer.writeheader()

                writer.writerows(self._completed_workunits.values())

                logger.debug(f"Wrote {len(self._completed_workunits)} entries to {filepath}")

class RemoteCacheLoggerCallbackFactoryRequest:

class RemoteCacheLogger(Subsystem):

    options_scope = "remote-cache-logger"

    help = "Remote Cache Logger subsystem. Useful for debugging remote cache."

    enabled = BoolOption("--enabled", default=False, help="Whether to enable remote cache logging.")

    logdir = StrOption("--logdir", default=".pants.d/workdir", help="Where to write the log to.")

@rule

def construct_callback(

    return WorkunitsCallbackFactory(

def rules():

    return [