<commit_before>import abc
import functools
import os
import textwrap
import threading
import time
import traceback
from typing import Any
from typing import Callable
from typing import Dict
from typing import Generator
from typing import List
from typing import Optional
from typing import Tuple

import pytimeparse
try:
    from slo_utils.yelpsoa_configs import get_slo_files_from_soaconfigs
    from slo_transcoder.composite_sinks.signalform_detectors import make_signalform_detector_composite_sink
    from slo_transcoder.sources.yelpsoaconfigs import YelpSoaConfigsSource
    SLO_TRANSCODER_LOADED = True
except ImportError:
    SLO_TRANSCODER_LOADED = False

import tempfile

from paasta_tools.utils import DEFAULT_SOA_DIR
from paasta_tools.automatic_rollbacks.signalfx import tail_signalfx
from paasta_tools.automatic_rollbacks.slack import SlackDeploymentProcess


def get_slos_for_service(service, soa_dir=DEFAULT_SOA_DIR) -> Generator:
    if not SLO_TRANSCODER_LOADED:
        return

    slo_files = [f for f in get_slo_files_from_soaconfigs(soa_dir) if f.startswith(os.path.join(soa_dir, service))]
    with tempfile.TemporaryDirectory() as temp_dir:
        SFDCS = make_signalform_detector_composite_sink(output_directory=temp_dir)
        sources = [YelpSoaConfigsSource(file) for file in slo_files]
        if not sources:
            return   # SFDCS instantiation crashes with IndexError if sources is an empty list.
        composite_sink = SFDCS(service=service, sources=sources)

        for sink in composite_sink.sinks():
            signalflow, rules = sink.generate_signalflow_signals_and_rules()
            query = textwrap.dedent('\n'.join(signalflow))
            yield sink, query


class SLODemultiplexer:
    def __init__(
        self,
        sink: Any,
        individual_slo_callback: Callable[['SLOWatcher'], None],
        start_timestamp: Optional[float] = None,
        max_duration: float = 300,
    ) -> None:
        self.sink = sink
        self.individual_slo_callback = individual_slo_callback
        self.max_duration = max_duration
        if start_timestamp is None:
            self.start_timestamp = time.time()
        else:
            self.start_timestamp = start_timestamp

        self.slo_watchers_by_label: Dict[str, 'SLOWatcher'] = {}
        for slo in sink.slos:
            label = sink._get_detector_label(slo)
            watcher = SLOWatcher(slo, individual_slo_callback, self.start_timestamp, label, max_duration=max_duration)
            self.slo_watchers_by_label[label] = watcher

    def process_datapoint(self, props, datapoint, timestamp) -> None:
        slo_label = props['dimensions']['sf_metric'].rsplit('.', 1)[0]
        watcher = self.slo_watchers_by_label[slo_label]
        watcher.process_datapoint(props, datapoint, timestamp)


class SLOWatcher:
    def __init__(
        self,
        slo: Any,
        callback: Callable[['SLOWatcher'], Any],
        start_timestamp: float,
        label: str,
        max_duration: float = 300,
    ) -> None:
        self.slo = slo
        self.window: List[Tuple[float, float]] = []
        self.max_duration = max_duration
        self.failing: Optional[bool] = None
        self.bad_after_mark: Optional[bool] = None
        self.bad_before_mark: Optional[bool] = None
        self.callback = callback
        self.start_timestamp = start_timestamp
        self.label = label

    def process_datapoint(self, props, datapoint, timestamp) -> None:<commit_msg>Expand max_duration on SLOWatcher to 3600 seconds.<commit_after>import abc
import functools
import os
import textwrap
import threading
import time
import traceback
from typing import Any
from typing import Callable
from typing import Dict
from typing import Generator
from typing import List
from typing import Optional
from typing import Tuple

import pytimeparse
try:
    from slo_utils.yelpsoa_configs import get_slo_files_from_soaconfigs
    from slo_transcoder.composite_sinks.signalform_detectors import make_signalform_detector_composite_sink
    from slo_transcoder.sources.yelpsoaconfigs import YelpSoaConfigsSource
    SLO_TRANSCODER_LOADED = True
except ImportError:
    SLO_TRANSCODER_LOADED = False

import tempfile

from paasta_tools.utils import DEFAULT_SOA_DIR
from paasta_tools.automatic_rollbacks.signalfx import tail_signalfx
from paasta_tools.automatic_rollbacks.slack import SlackDeploymentProcess


def get_slos_for_service(service, soa_dir=DEFAULT_SOA_DIR) -> Generator:
    if not SLO_TRANSCODER_LOADED:
        return

    slo_files = [f for f in get_slo_files_from_soaconfigs(soa_dir) if f.startswith(os.path.join(soa_dir, service))]
    with tempfile.TemporaryDirectory() as temp_dir:
        SFDCS = make_signalform_detector_composite_sink(output_directory=temp_dir)
        sources = [YelpSoaConfigsSource(file) for file in slo_files]
        if not sources:
            return   # SFDCS instantiation crashes with IndexError if sources is an empty list.
        composite_sink = SFDCS(service=service, sources=sources)

        for sink in composite_sink.sinks():
            signalflow, rules = sink.generate_signalflow_signals_and_rules()
            query = textwrap.dedent('\n'.join(signalflow))
            yield sink, query


class SLODemultiplexer:
    def __init__(
        self,
        sink: Any,
        individual_slo_callback: Callable[['SLOWatcher'], None],
        start_timestamp: Optional[float] = None,
        max_duration: float = 3600,
    ) -> None:
        self.sink = sink
        self.individual_slo_callback = individual_slo_callback
        self.max_duration = max_duration
        if start_timestamp is None:
            self.start_timestamp = time.time()
        else:
            self.start_timestamp = start_timestamp

        self.slo_watchers_by_label: Dict[str, 'SLOWatcher'] = {}
        for slo in sink.slos:
            label = sink._get_detector_label(slo)
            watcher = SLOWatcher(slo, individual_slo_callback, self.start_timestamp, label, max_duration=max_duration)
            self.slo_watchers_by_label[label] = watcher

    def process_datapoint(self, props, datapoint, timestamp) -> None:
        slo_label = props['dimensions']['sf_metric'].rsplit('.', 1)[0]
        watcher = self.slo_watchers_by_label[slo_label]
        watcher.process_datapoint(props, datapoint, timestamp)


class SLOWatcher:
    def __init__(
        self,
        slo: Any,
        callback: Callable[['SLOWatcher'], Any],
        start_timestamp: float,
        label: str,
        max_duration: float,
    ) -> None:
        self.slo = slo
        self.window: List[Tuple[float, float]] = []
        self.max_duration = max_duration
        self.failing: Optional[bool] = None
        self.bad_after_mark: Optional[bool] = None
        self.bad_before_mark: Optional[bool] = None
        self.callback = callback
        self.start_timestamp = start_timestamp
        self.label = label

    def process_datapoint(self, props, datapoint, timestamp) -> None:
{'fraction_non_alphanumeric': 0.07228571428571429, 'fraction_numerical': 0.0025714285714285713, 'mean_word_length': 3.348447204968944, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5987386', 'n_tokens_mistral': 2215, 'n_tokens_neox': 2147, 'n_words': 494}