<commit_before>import os
import sys
import json
import time
import shutil
import subprocess
import logging
import argparse
from contextlib import contextmanager

from cloudify import broker_config, dispatch
from cloudify.logs import setup_agent_logger
from cloudify.utils import get_admin_api_token, get_tenant
from cloudify.constants import MGMTWORKER_QUEUE
from cloudify.models_states import ExecutionState
from cloudify.state import current_workflow_ctx
from cloudify.manager import get_rest_client, update_execution_status
from cloudify.amqp_client import AMQPConnection, get_client, SendHandler

from cloudify_rest_client.executions import Execution
from cloudify_rest_client.exceptions import InvalidExecutionUpdateStatus

from cloudify_agent.worker import (
    ProcessRegistry,
    ServiceTaskConsumer,
    CloudifyOperationConsumer
)
from cloudify_agent import worker as agent_worker

from .hooks import HookConsumer
from .monitoring import (
    get_broker_hosts,
    get_db_hosts,
    get_manager_hosts,
    update_broker_targets,
    update_broker_alerts,
    update_db_targets,
    update_db_alerts,
    update_manager_targets,
    update_manager_alerts,
)
try:
    from cloudify_premium import syncthing_utils
except ImportError:
    syncthing_utils = None

DEFAULT_MAX_WORKERS = 10
logger = logging.getLogger('mgmtworker')


class MgmtworkerOperationConsumer(CloudifyOperationConsumer):
    """CloudifyOperationConsumer but with late_ack enabled.

    late_ack means that operations are acked after they're finished, so
    if one mgmtworker dies while handling an operation, the operation
    will be re-sent to another mgmtworker.
    """
    late_ack = True


class CloudifyWorkflowConsumer(CloudifyOperationConsumer):
    routing_key = 'workflow'
    handler = dispatch.WorkflowHandler
    late_ack = True

    @contextmanager
    def _update_operation_state(self, *args, **kwargs):
        # noop - override superclass method, which tries to update the
        # operation state: we're not working with operations, but workflows
        yield

    def handle_task(self, full_task):
        if self.is_scheduled_execution(full_task):
            task = full_task['cloudify_task']
            ctx = task['kwargs']['__cloudify_context']<commit_msg>RD-3407 Remove unused attribute (#3367)<commit_after>import os
import sys
import json
import time
import shutil
import subprocess
import logging
import argparse
from contextlib import contextmanager

from cloudify import broker_config
from cloudify.logs import setup_agent_logger
from cloudify.utils import get_admin_api_token, get_tenant
from cloudify.constants import MGMTWORKER_QUEUE
from cloudify.models_states import ExecutionState
from cloudify.state import current_workflow_ctx
from cloudify.manager import get_rest_client, update_execution_status
from cloudify.amqp_client import AMQPConnection, get_client, SendHandler

from cloudify_rest_client.executions import Execution
from cloudify_rest_client.exceptions import InvalidExecutionUpdateStatus

from cloudify_agent.worker import (
    ProcessRegistry,
    ServiceTaskConsumer,
    CloudifyOperationConsumer
)
from cloudify_agent import worker as agent_worker

from .hooks import HookConsumer
from .monitoring import (
    get_broker_hosts,
    get_db_hosts,
    get_manager_hosts,
    update_broker_targets,
    update_broker_alerts,
    update_db_targets,
    update_db_alerts,
    update_manager_targets,
    update_manager_alerts,
)
try:
    from cloudify_premium import syncthing_utils
except ImportError:
    syncthing_utils = None

DEFAULT_MAX_WORKERS = 10
logger = logging.getLogger('mgmtworker')


class MgmtworkerOperationConsumer(CloudifyOperationConsumer):
    """CloudifyOperationConsumer but with late_ack enabled.

    late_ack means that operations are acked after they're finished, so
    if one mgmtworker dies while handling an operation, the operation
    will be re-sent to another mgmtworker.
    """
    late_ack = True


class CloudifyWorkflowConsumer(CloudifyOperationConsumer):
    routing_key = 'workflow'
    late_ack = True

    @contextmanager
    def _update_operation_state(self, *args, **kwargs):
        # noop - override superclass method, which tries to update the
        # operation state: we're not working with operations, but workflows
        yield

    def handle_task(self, full_task):
        if self.is_scheduled_execution(full_task):
            task = full_task['cloudify_task']
            ctx = task['kwargs']['__cloudify_context']
{'fraction_non_alphanumeric': 0.048147326381184824, 'fraction_numerical': 0.002662524961171511, 'mean_word_length': 4.816774193548387, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '948329', 'n_tokens_mistral': 1287, 'n_tokens_neox': 1242, 'n_words': 364}