<commit_before>import argparse
import logging
import os
import time
import traceback
import json

import ray
from ray.autoscaler.autoscaler import StandardAutoscaler
from ray.autoscaler.load_metrics import LoadMetrics
import ray.gcs_utils
import ray.utils
import ray.ray_constants as ray_constants
from ray.utils import binary_to_hex, setup_logger
from ray.autoscaler.commands import teardown_cluster
import redis

logger = logging.getLogger(__name__)


class Monitor:
    """A monitor for Ray processes.

    The monitor is in charge of cleaning up the tables in the global state
    after processes have died. The monitor is currently not responsible for
    detecting component failures.

    Attributes:
        redis: A connection to the Redis server.
        primary_subscribe_client: A pubsub client for the Redis server.
            This is used to receive notifications about failed components.
    """

    def __init__(self, redis_address, autoscaling_config, redis_password=None):
        # Initialize the Redis clients.
        ray.state.state._initialize_global_state(
            redis_address, redis_password=<PASSWORD>)
        self.redis = ray.services.create_redis_client(
            redis_address, password=<PASSWORD>)
        # Set the redis client and mode so _internal_kv works for autoscaler.
        worker = ray.worker.global_worker
        worker.redis_client = self.redis
        worker.mode = 0
        # Setup subscriptions to the primary Redis server and the Redis shards.
        self.primary_subscribe_client = self.redis.pubsub(
            ignore_subscribe_messages=True)
        # Keep a mapping from raylet client ID to IP address to use
        # for updating the load metrics.
        self.raylet_id_to_ip_map = {}
        self.load_metrics = LoadMetrics()
        if autoscaling_config:
            self.autoscaler = StandardAutoscaler(autoscaling_config,
                                                 self.load_metrics)
            self.autoscaling_config = autoscaling_config
        else:
            self.autoscaler = None
            self.autoscaling_config = None

    def __del__(self):
        """Destruct the monitor object."""
        # We close the pubsub client to avoid leaking file descriptors.
        try:
            primary_subscribe_client = self.primary_subscribe_client
        except AttributeError:
            primary_subscribe_client = None
        if primary_subscribe_client is not None:
            primary_subscribe_client.close()

    def subscribe(self, channel):
        """Subscribe to the given channel on the primary Redis shard.

        Args:
            channel (str): The channel to subscribe to.

        Raises:
            Exception: An exception is raised if the subscription fails.
        """
        self.primary_subscribe_client.subscribe(channel)

    def psubscribe(self, pattern):
        """Subscribe to the given pattern on the primary Redis shard.

        Args:
            pattern (str): The pattern to subscribe to.

        Raises:
            Exception: An exception is raised if the subscription fails.
        """
        self.primary_subscribe_client.psubscribe(pattern)

    def parse_resource_demands(self, resource_load_by_shape):
        """Handle the message.resource_load_by_shape protobuf for the demand
        based autoscaling. Catch and log all exceptions so this doesn't
        interfere with the utilization based autoscaler until we're confident
        this is stable.

        Args:
            resource_load_by_shape (pb2.gcs.ResourceLoad): The resource demands
                in protobuf form or None.
        """
        waiting_bundles, infeasible_bundles = [], []
        try:
            if self.autoscaler:
                for resource_demand_pb in list(
                        resource_load_by_shape.resource_demands):
                    request_shape = dict(resource_demand_pb.shape)
                    for _ in range(resource_demand_pb.num_ready_requests_queued):
                        waiting_bundles.append(request_shape)
                    for _ in range(
                            resource_demand_pb.num_infeasible_requests_queued):
                        infeasible_bundles.append(request_shape)
        except Exception as e:
            logger.exception(e)
        return waiting_bundles, infeasible_bundles

    def xray_heartbeat_batch_handler(self, unused_channel, data):
        """Handle an xray heartbeat batch message from Redis."""

        pub_message = ray.gcs_utils.PubSubMessage.FromString(data)
        heartbeat_data = pub_message.data

        message = ray.gcs_utils.HeartbeatBatchTableData.FromString(<commit_msg>FIX: Lint (#10384)<commit_after>import argparse
import logging
import os
import time
import traceback
import json

import ray
from ray.autoscaler.autoscaler import StandardAutoscaler
from ray.autoscaler.load_metrics import LoadMetrics
import ray.gcs_utils
import ray.utils
import ray.ray_constants as ray_constants
from ray.utils import binary_to_hex, setup_logger
from ray.autoscaler.commands import teardown_cluster
import redis

logger = logging.getLogger(__name__)


class Monitor:
    """A monitor for Ray processes.

    The monitor is in charge of cleaning up the tables in the global state
    after processes have died. The monitor is currently not responsible for
    detecting component failures.

    Attributes:
        redis: A connection to the Redis server.
        primary_subscribe_client: A pubsub client for the Redis server.
            This is used to receive notifications about failed components.
    """

    def __init__(self, redis_address, autoscaling_config, redis_password=None):
        # Initialize the Redis clients.
        ray.state.state._initialize_global_state(
            redis_address, redis_password=<PASSWORD>_password)
        self.redis = ray.services.create_redis_client(
            redis_address, password=<PASSWORD>)
        # Set the redis client and mode so _internal_kv works for autoscaler.
        worker = ray.worker.global_worker
        worker.redis_client = self.redis
        worker.mode = 0
        # Setup subscriptions to the primary Redis server and the Redis shards.
        self.primary_subscribe_client = self.redis.pubsub(
            ignore_subscribe_messages=True)
        # Keep a mapping from raylet client ID to IP address to use
        # for updating the load metrics.
        self.raylet_id_to_ip_map = {}
        self.load_metrics = LoadMetrics()
        if autoscaling_config:
            self.autoscaler = StandardAutoscaler(autoscaling_config,
                                                 self.load_metrics)
            self.autoscaling_config = autoscaling_config
        else:
            self.autoscaler = None
            self.autoscaling_config = None

    def __del__(self):
        """Destruct the monitor object."""
        # We close the pubsub client to avoid leaking file descriptors.
        try:
            primary_subscribe_client = self.primary_subscribe_client
        except AttributeError:
            primary_subscribe_client = None
        if primary_subscribe_client is not None:
            primary_subscribe_client.close()

    def subscribe(self, channel):
        """Subscribe to the given channel on the primary Redis shard.

        Args:
            channel (str): The channel to subscribe to.

        Raises:
            Exception: An exception is raised if the subscription fails.
        """
        self.primary_subscribe_client.subscribe(channel)

    def psubscribe(self, pattern):
        """Subscribe to the given pattern on the primary Redis shard.

        Args:
            pattern (str): The pattern to subscribe to.

        Raises:
            Exception: An exception is raised if the subscription fails.
        """
        self.primary_subscribe_client.psubscribe(pattern)

    def parse_resource_demands(self, resource_load_by_shape):
        """Handle the message.resource_load_by_shape protobuf for the demand
        based autoscaling. Catch and log all exceptions so this doesn't
        interfere with the utilization based autoscaler until we're confident
        this is stable.

        Args:
            resource_load_by_shape (pb2.gcs.ResourceLoad): The resource demands
                in protobuf form or None.
        """
        waiting_bundles, infeasible_bundles = [], []
        try:
            if self.autoscaler:
                for resource_demand_pb in list(
                        resource_load_by_shape.resource_demands):
                    request_shape = dict(resource_demand_pb.shape)
                    for _ in range(
                            resource_demand_pb.num_ready_requests_queued):
                        waiting_bundles.append(request_shape)
                    for _ in range(
                            resource_demand_pb.num_infeasible_requests_queued):
                        infeasible_bundles.append(request_shape)
        except Exception as e:
            logger.exception(e)
        return waiting_bundles, infeasible_bundles

    def xray_heartbeat_batch_handler(self, unused_channel, data):
        """Handle an xray heartbeat batch message from Redis."""

        pub_message = ray.gcs_utils.PubSubMessage.FromString(data)
        heartbeat_data = pub_message.data

        message = ray.gcs_utils.HeartbeatBatchTableData.FromString(
{'fraction_non_alphanumeric': 0.055100521221146684, 'fraction_numerical': 0.0009573449633017764, 'mean_word_length': 2.3966763005780347, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6162017', 'n_tokens_mistral': 2502, 'n_tokens_neox': 2367, 'n_words': 732}