<commit_before># Copyright 2014-2020 Scalyr Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


from __future__ import unicode_literals
from __future__ import absolute_import

import copy
import datetime
import os
import sys
import threading
import time
import operator
from six.moves import range

if False:
    from typing import Dict
    from typing import List
    from typing import Optional
    from typing import Tuple

from scalyr_agent import (
    scalyr_logging as scalyr_logging,
    StoppableThread,
    util as scalyr_util,
)
from scalyr_agent.agent_status import (
    CopyingManagerStatus,
    ApiKeyWorkerPoolStatus,
)
from scalyr_agent.copying_manager.worker import (
    CopyingManagerThreadedWorker,
    SharedObjectManager,
)
from scalyr_agent.log_processing import LogMatcher, LogFileProcessor
from scalyr_agent.log_watcher import LogWatcher
from scalyr_agent.configuration import Configuration
from scalyr_agent.scalyr_client import ScalyrClientSessionStatus
from scalyr_agent.copying_manager.common import write_checkpoint_state_to_file

import six

log = scalyr_logging.getLogger(__name__)

SCHEDULED_DELETION = "scheduled-deletion"
CONSOLIDATED_CHECKPOINTS_FILE_NAME = "checkpoints.json"


class ApiKeyWorkerPool(object):
    """
    This abstraction is responsible for maintaining the workers
    for a particular entry in the 'api_keys' list in the configuration.
    It also creates worker instances according to the number in the "workers" field in each entry.
    Depending on workers type (multiprocess or not), it will operate with
    instances of the CopyingManagerWorker or their proxy objects.

    The instances if this class itself live only in the original process,
    controlled by the copying manager and do not have any proxy versions.
    """

    def __init__(self, config, api_key_config):
        # type: (Configuration, Dict) -> None

        self.__api_key_id = api_key_config["id"]
        self.__config = config

        self.__workers = []  # type: List[CopyingManagerThreadedWorker]

        # collection of shared object managers. (Subclasses of the multiprocessing.BaseManager)
        # Those managers allow to communicate with workers if they are in the multiprocessing mode.
        self.__shared_object_managers = (
            dict()
        )  # type: Dict[six.text_type, SharedObjectManager]

        for i in range(api_key_config["workers"]):
            # combine the id of the api key and worker's position in the list to get a workers id.
            worker_id = "%s_%s" % (self.__api_key_id, i)

            if not self.__config.use_multiprocess_copying_workers:
                worker = CopyingManagerThreadedWorker(
                    self.__config, api_key_config, worker_id
                )
            else:
                shared_object_manager = SharedObjectManager()

                # Important for the understanding. When the shared_object_manager is started, it creates a new process.
                # We use this process as a process for the worker.

                worker_agent_log_path = os.path.join(
                    self.__config.agent_log_path, "agent-%s.log" % worker_id
                )

                # change agent log path for the new worker.
                # this initializer function will be invoked in the worker's process.
                def initializer():
                    self._change_worker_process_agent_log_path(worker_agent_log_path)

                shared_object_manager.start(initializer=initializer)

                # create proxy object of the worker. The real worker instance is created in the new process
                # which was created when shared_object_manager started.
                # pylint: disable=E1101
                worker = shared_object_manager.CopyingManagerWorkerProxy(
                    self.__config, api_key_config, worker_id
                )
                # pylint: enable=E1101

                # also save new shared object manager.
                self.__shared_object_managers[worker.get_id()] = shared_object_manager

            self.__workers.append(worker)
            log.log(
                scalyr_logging.DEBUG_LEVEL_1,
                "CopyingManagerWorker #%s is created." % worker.get_id(),
            )

        # The index of the next worker which will be used to handle the log processor.
        self.__current_worker = 0

    def _change_worker_process_agent_log_path(self, path):
        """
        Reconfigure the agent logger, mainly, to change to path of the agent.log file
        for the worker when it is running in the separate thread.
        Multiple concurrent processes that write to the same 'agent.log' file may cause an incorrect results, so,
        to be on a safe side, we just create separate agent-<worker_id>.log files for each worker.
        """
        scalyr_logging.set_log_destination(agent_log_file_path=path, use_disk=True)

    @property
    def api_key_id(self):<commit_msg>Add __str__ and __repr__ to ApiKeyWorkerPool to make troubleshooting easier.<commit_after># Copyright 2014-2020 Scalyr Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


from __future__ import unicode_literals
from __future__ import absolute_import

import copy
import datetime
import os
import sys
import threading
import time
import operator
from six.moves import range

if False:
    from typing import Dict
    from typing import List
    from typing import Optional
    from typing import Tuple

from scalyr_agent import (
    scalyr_logging as scalyr_logging,
    StoppableThread,
    util as scalyr_util,
)
from scalyr_agent.agent_status import (
    CopyingManagerStatus,
    ApiKeyWorkerPoolStatus,
)
from scalyr_agent.copying_manager.worker import (
    CopyingManagerThreadedWorker,
    SharedObjectManager,
)
from scalyr_agent.log_processing import LogMatcher, LogFileProcessor
from scalyr_agent.log_watcher import LogWatcher
from scalyr_agent.configuration import Configuration
from scalyr_agent.scalyr_client import ScalyrClientSessionStatus
from scalyr_agent.copying_manager.common import write_checkpoint_state_to_file

import six

log = scalyr_logging.getLogger(__name__)

SCHEDULED_DELETION = "scheduled-deletion"
CONSOLIDATED_CHECKPOINTS_FILE_NAME = "checkpoints.json"


class ApiKeyWorkerPool(object):
    """
    This abstraction is responsible for maintaining the workers
    for a particular entry in the 'api_keys' list in the configuration.
    It also creates worker instances according to the number in the "workers" field in each entry.
    Depending on workers type (multiprocess or not), it will operate with
    instances of the CopyingManagerWorker or their proxy objects.

    The instances if this class itself live only in the original process,
    controlled by the copying manager and do not have any proxy versions.
    """

    def __init__(self, config, api_key_config):
        # type: (Configuration, Dict) -> None

        self.__api_key_config = api_key_config
        self.__api_key_id = api_key_config["id"]
        self.__config = config

        self.__workers = []  # type: List[CopyingManagerThreadedWorker]

        # collection of shared object managers. (Subclasses of the multiprocessing.BaseManager)
        # Those managers allow to communicate with workers if they are in the multiprocessing mode.
        self.__shared_object_managers = (
            dict()
        )  # type: Dict[six.text_type, SharedObjectManager]

        for i in range(api_key_config["workers"]):
            # combine the id of the api key and worker's position in the list to get a workers id.
            worker_id = "%s_%s" % (self.__api_key_id, i)

            if not self.__config.use_multiprocess_copying_workers:
                worker = CopyingManagerThreadedWorker(
                    self.__config, api_key_config, worker_id
                )
            else:
                shared_object_manager = SharedObjectManager()

                # Important for the understanding. When the shared_object_manager is started, it creates a new process.
                # We use this process as a process for the worker.

                worker_agent_log_path = os.path.join(
                    self.__config.agent_log_path, "agent-%s.log" % worker_id
                )

                # change agent log path for the new worker.
                # this initializer function will be invoked in the worker's process.
                def initializer():
                    self._change_worker_process_agent_log_path(worker_agent_log_path)

                shared_object_manager.start(initializer=initializer)

                # create proxy object of the worker. The real worker instance is created in the new process
                # which was created when shared_object_manager started.
                # pylint: disable=E1101
                worker = shared_object_manager.CopyingManagerWorkerProxy(
                    self.__config, api_key_config, worker_id
                )
                # pylint: enable=E1101

                # also save new shared object manager.
                self.__shared_object_managers[worker.get_id()] = shared_object_manager

            self.__workers.append(worker)
            log.log(
                scalyr_logging.DEBUG_LEVEL_1,
                "CopyingManagerWorker #%s is created." % worker.get_id(),
            )

        # The index of the next worker which will be used to handle the log processor.
        self.__current_worker = 0

    def _change_worker_process_agent_log_path(self, path):
        """
        Reconfigure the agent logger, mainly, to change to path of the agent.log file
        for the worker when it is running in the separate thread.
        Multiple concurrent processes that write to the same 'agent.log' file may cause an incorrect results, so,
        to be on a safe side, we just create separate agent-<worker_id>.log files for each worker.
        """
        scalyr_logging.set_log_destination(agent_log_file_path=path, use_disk=True)

    def __repr__(self):
        api_key_config = copy.deepcopy(self.__api_key_config)
        api_key = api_key_config.get("api_key", "")
        api_key_config["api_key"] = "..." + api_key[-4:]
        return "<ApiKeyWorkerPool api_key_config=%s>" % (api_key_config)

    def __str__(self):
        return str(self.__repr__())

    @property
    def api_key_id(self):
{'fraction_non_alphanumeric': 0.05702254583735415, 'fraction_numerical': 0.003947714711816826, 'mean_word_length': 2.8229376257545273, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '5387965', 'n_tokens_mistral': 3073, 'n_tokens_neox': 2957, 'n_words': 1054}