# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

"""
This module contains contains classes implementing X-Ray Agent behaviour
"""

import collections
import io
import json
import logging
import os
import queue
import re
import signal
import socket
import subprocess
import time
import typing
from threading import Thread, current_thread, Lock, Event, Timer
from typing import Any, Optional, Dict, Tuple
from dataclasses import dataclass

import psutil

from .executor import BoundedThreadExecutor
from xray import gettext as _
from xray.apiclient import get_client
from xray.internal.constants import local_tasks_storage, tasks_base_storage, safe_id_pattern
from xray.internal.exceptions import XRayError, XRayAPIError
from .fault_detector import FaultDetector
from xray.internal.local_counters import open_local_storage, flush_memory_storage, get_task_ids
from xray.internal.types import Task
from xray.internal.user_plugin_utils import extract_creds
from xray.internal.utils import dbm_storage, get_current_cpu_throttling_time

if typing.TYPE_CHECKING:
    from xray.apiclient.api_client import SendClient, SmartAdviceAPIClient, APIClient


@dataclass
class APIDataContainer:
    client: 'APIClient'
    task: Task


logger = logging.getLogger(__name__)


class Agent:
    """
    X-Ray Agent class
    """

    COUNTERS_FLUSH_INTERVAL = 15
    MONGO_FLUSH_INTERVAL = 60
    CLEANUP_INTERVAL = 43200  # once in 12 hours
    _MAX_CACHE_ENTRIES = 10000

    def __init__(
        self,
        system_id,
        # keep max_connections quite big to handle spikes
        max_connections=psutil.cpu_count() * 8,
        # max_workers can also be quite big because they are not cpu-bound
        max_workers=psutil.cpu_count() * 4,
        maxqueuesize=psutil.cpu_count() * 16,
        max_connections_per_uid=None,
        min_jobs_per_uid=None,
        reserved_pool_slots=None,
    ):
        self.sys_id = system_id
        self.maxqueuesize = maxqueuesize
        self.max_connections = max_connections
        self.max_workers = max_workers
        self.max_connections_per_uid = (
            max_connections_per_uid if max_connections_per_uid is not None else max_connections // 2
        )
        # Real capacity of the shared workers_pool: the BoundedThreadExecutor
        # admits up to (maxqueuesize + max_workers) outstanding jobs before
        # raising queue.Full (executor.py:37 — BoundedSemaphore(maxqueuesize +
        # max_workers)). Defaults: cpu_count*16 + cpu_count*4 = cpu_count*20
        # (e.g. 80 on a 4-CPU box). The per-UID fairness logic below derives
        # from THIS real capacity, not maxqueuesize alone.
        self._pool_capacity = maxqueuesize + max_workers
        # The shared workers_pool is a single bounded queue used by ALL uids;
        # over a world-writable socket any local user can flood it and starve
        # other tenants' traces. We bound each uid to a DYNAMIC max-min fair
        # share of the real capacity, computed at submit time from the number
        # of currently-active uids (see _effective_uid_cap / _submit_uid_job).
        # The cap is enforced UNCONDITIONALLY — there is no self-driven
        # "pressure gate" a flooder can defeat by supplying its own load:
        #   * a sole active uid gets the whole capacity (minus the reserve
        #     below), so a legitimate single tenant is NOT regressed; it is
        #     bounded only by the real queue.Full;
        #   * with N active uids each gets capacity // N, so a flooder converges
        #     to its fair share as its queued jobs drain and a victim arriving
        #     after the burst reaches capacity // N within ~one drain cycle.
        #
        # _min_floor keeps the per-uid cap from collapsing to 0 under very high
        # contention (many active uids): with a tiny floor the cap stays usable
        # for normal bursty tracing and extreme contention simply falls through
        # to the real queue.Full. max_workers // 4 (floored at 4) is small
        # relative to capacity yet always admits a few jobs per uid.
        self._min_floor = min_jobs_per_uid if min_jobs_per_uid is not None else max(4, max_workers // 4)
        # Reserve headroom so a NEWLY-arriving tenant is never totally locked
        # out by a sole flooder that already filled the pool: a single uid can
        # never hold more than (capacity - reserve) outstanding, guaranteeing
        # at least `reserve` slots remain reachable by a new uid / for draining.
        # reserve = max_workers because that is exactly how many jobs the pool
        # actively executes (and thus frees) per drain step — reserving one
        # full worker-set keeps a fresh tenant admittable within one step. A
        # legitimate sole user capped at capacity - max_workers (e.g. 64 of 80)
        # still has far more concurrent outstanding traces than any real single
        # tenant needs, so this is not a meaningful regression.
        self._reserve = reserved_pool_slots if reserved_pool_slots is not None else max_workers

        # don't process SIGUSR2 with default handler
        signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGUSR2})

        # initialize ClickHouse API client to send requests data
        clickhouse_client_object: typing.Type[SendClient] = get_client('api_req')
        self.send_client: SendClient = clickhouse_client_object(system_id=self.sys_id)
        # initialize Adviser API client to send requests data
        adviser_client_object: typing.Type[SmartAdviceAPIClient] = get_client('adviser')
        self.adviser_client: SmartAdviceAPIClient = adviser_client_object()
        # initial state of MongoDB API client to interact with tasks
        self.task_client_object: typing.Type[APIClient] = get_client()
        # initialize storage for cache of remote API data
        self.api_data_cache_lock = Lock()
        self.api_data_cache: Dict[str, APIDataContainer] = dict()
        # initialize Fault Detector
        self.fault_detector = FaultDetector()
        # per-UID concurrent connection counter (N6 hardening)
        self._uid_connections: Dict[int, int] = collections.defaultdict(int)
        self._uid_connections_lock = Lock()
        # per-UID outstanding worker-job counter: maps uid -> number of jobs it
        # currently has queued/in-flight in the shared workers_pool. Drives the
        # dynamic max-min fair-share cap (the dict size is the count of active
        # uids; see _effective_uid_cap). Zero-count entries are deleted on
        # release so the dict stays a faithful active-uid set and cannot grow
        # unbounded. _uid_jobs_total is the sum of all per-UID counts, kept for
        # observability; both are mutated together under _uid_jobs_lock.
        self._uid_jobs: Dict[int, int] = collections.defaultdict(int)
        self._uid_jobs_total: int = 0
        self._uid_jobs_lock = Lock()

        self.signal_handler_thread: Optional[Thread] = None
        self.flusher_thread: Optional[Thread] = None

    def _wait_for_sigusr2(self):
        siginfo = signal.sigwaitinfo({signal.SIGUSR2})
        logging.info('Received SIGUSR2 from pid=%s, flushing database storage on disk', siginfo.si_pid)

        self._flush_mongodb_counters()
        flush_memory_storage()

        logging.info('Sending signal back to process that requested storage flush')
        try:
            os.kill(siginfo.si_pid, signal.SIGUSR2)
        except OSError:
            logging.warning('Process that requested storage flush no longer exists')

    def _setup_signal_handler(self):
        """
        Setup SIGUSR2 handler that starts in-memory
        storage flush when received.

        When flushed, send SIGUSR2 back to the process that send signal.
        """
        signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGUSR2})
        while True:
            try:
                self._wait_for_sigusr2()
            except Exception:
                logging.exception('Unable to process signal, see traceback for details.')

    def _flush_mongodb_counters(self, task_id=None):
        tasks_to_flush = [task_id] if task_id is not None else get_task_ids()
        for task_id in tasks_to_flush:
            logger.info('Updating task requests counters in mongo for task_id=%s', task_id)

            try:
                apiclient, task = self.get_cached_or_load(task_id)
            except XRayError:
                logging.warning('Unable to get client and task %s', task_id)
                continue

            # read stored request_id
            with open_local_storage(task_id) as storage:
                task.update_with_local_data(next_request_id=storage.next_request_id)

            if task.tracing_by == 'time':
                # tracing_count for task by time represents number of minutes
                # left to active tracing and is updated by stop|continue
                # task routines only
                self.update_counts(apiclient, task.request_count)
            else:
                # tracing_count for task by request_qty depends on number of
                # collected requests, thus should be updated alongside
                self.update_counts(apiclient, task.request_count, task.tracing_count)

    def _flusher(self):
        """
        This method flushes data from memory to
        local storage periodically.
        """
        last_mongo_flush_time = 0
        last_api_data_cache_cleanup = 0
        while True:
            time.sleep(self.COUNTERS_FLUSH_INTERVAL)
            if time.time() - last_mongo_flush_time > self.MONGO_FLUSH_INTERVAL:
                self._flush_mongodb_counters()
                flush_memory_storage()
                last_mongo_flush_time = time.time()
                # we should cleanup API data cache only after flushing counters
                # in order not to lose counters for already inactive tasks
                if time.time() - last_api_data_cache_cleanup > self.CLEANUP_INTERVAL:
                    self.cleanup_api_data_cache()
                    last_api_data_cache_cleanup = time.time()
            else:
                flush_memory_storage(remove=False)

    def start(self, sock: socket.socket, background_routine: bool = False, loops: Optional[int] = None) -> None:
        """
        Start listening socket
        """
        logger.info('Starting daemon')
        if background_routine:
            # setup signal handlers
            self.signal_handler_thread = Thread(target=self._setup_signal_handler)
            self.signal_handler_thread.start()

            # start periodical database flushing
            self.flusher_thread = Thread(target=self._flusher)
            self.flusher_thread.start()

        with (
            BoundedThreadExecutor(max_workers=self.max_workers, maxqueuesize=self.maxqueuesize) as workers_pool,
            BoundedThreadExecutor(
                max_workers=self.max_connections,
                # turn off the queue for connections because we
                # don't want the php processes to wait in queue
                # and slow down the php processes
                maxqueuesize=0,
            ) as connections_pool,
        ):
            while loops if loops is not None else True:
                connection, address = sock.accept()

                try:
                    connections_pool.submit(self.handle_incoming_connection, connection, workers_pool)
                except queue.Full:
                    logger.error(
                        'Request %s was rejected because our connections thread pool '
                        'is full of tasks. Increase max_connections in configuration.'
                    )
                    connection.close()

                if loops is not None:
                    loops -= 1

    def add_limit_faults(self, data, t_key, cpu_value):
        """
        Calculate throttling values and update given data
        with resulting throttling stat
        """
        data['hitting_limits'], data['throttled_time'] = self.fault_detector(t_key, cpu_value)

    def _handle_request_init(self, php_pid: int, cpu_usage: int, caller_uid: Optional[int] = None):
        """
        Called when php request starts and sends us welcome
        request meaning that request started on the php side.
        """
        logger.info('Received request init trigger from php=%s uid=%s cpu_usage=%s', php_pid, caller_uid, cpu_usage)
        # save current CPU throttling time and timestamp; scope the shared cache
        # per SO_PEERCRED uid so one uid's init flood cannot starve other tenants
        self.fault_detector.save(php_pid, cpu_usage, uid=caller_uid)
        # attempt to flush expired entries
        self.fault_detector.flush()

    def _handle_request_end(self, php_pid: int, cpu_usage: int, request_data: dict, caller_uid: int):
        # otherwise calculate throttling fact, add it to data
        # and send gathered stat to CH
        if request_data.get('hitting_limits') is None:
            # only calculate faults if extension failed to get them itself
            self.add_limit_faults(request_data, php_pid, cpu_usage)

        logger.info(
            '[%s] Processing trace for task %s (%s)',
            current_thread().name,
            request_data.get('tracing_task_id'),
            request_data.get('url'),
        )
        self.process_request_data(request_data, caller_uid)

    # Maximum payload size for a single request.
    # The PHP C extension can emit up to ~3 MB in worst-case configuration:
    #   331 spans (100 mysql + 100 external + 30 plugin + 100 slow + 1 shortcode)
    #   × ~9 KB each (4 KB query + 20-frame backtrace with ~200 B/frame + overhead).
    # 4 MB provides 2× headroom over the calculated worst case while still
    # bounding memory against a malicious or misbehaving client.
    _MAX_READ_SIZE = 4 * 1024 * 1024
    # Socket read timeout in seconds
    _READ_TIMEOUT = 5.0
    # Absolute wall-clock cap on the whole read, in seconds. settimeout above is
    # only a per-recv idle timeout: a slow-loris client trickling one byte every
    # < _READ_TIMEOUT seconds resets it on each chunk and holds the read (and the
    # connection-pool thread + per-uid slot) open indefinitely. This bounds the
    # total duration of the read regardless of how the client paces its bytes.
    _MAX_READ_DURATION = 30.0

    def handle_incoming_connection(self, connection: socket.socket, workers_pool: BoundedThreadExecutor) -> None:
        """
        Handle incoming connection
        :param connection: socket object usable to
                           send and receive data on the connection
        :param workers_pool: pool where we can place
                             tasks for the futher processing
        """
        try:
            _pid, _uid, _gid = extract_creds(connection)
        except OSError:
            logger.warning('Failed to extract credentials, closing connection')
            connection.close()
            return

        with self._uid_connections_lock:
            if self._uid_connections.get(_uid, 0) >= self.max_connections_per_uid:
                logger.warning('Too many concurrent connections from uid=%s, rejecting', _uid)
                connection.close()
                return
            self._uid_connections[_uid] += 1

        try:
            self._process_connection(connection, workers_pool, _pid, _uid)
        finally:
            with self._uid_connections_lock:
                self._uid_connections[_uid] -= 1
                if self._uid_connections[_uid] <= 0:
                    del self._uid_connections[_uid]

    def _process_connection(
        self, connection: socket.socket, workers_pool: BoundedThreadExecutor, _pid: int, _uid: int
    ) -> None:
        current_cpu = get_current_cpu_throttling_time(_uid)
        connection.settimeout(self._READ_TIMEOUT)
        fileobj: io.TextIOBase = connection.makefile(errors='ignore')
        # Wall-clock deadline: settimeout only bounds idle time per recv, so a
        # slow-loris client pacing its bytes can hold the read open forever. The
        # watchdog shuts the socket down at the deadline (which unblocks the read)
        # and flips deadline_exceeded so the dispatch below is skipped: a buffer
        # collected past the deadline must NOT be processed, whether it is empty,
        # partial, or a complete-but-held request.
        deadline_exceeded = Event()

        def _abort_on_deadline():
            deadline_exceeded.set()
            try:
                connection.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass

        watchdog = Timer(self._MAX_READ_DURATION, _abort_on_deadline)
        watchdog.start()
        try:
            input_data = self.read_input(fileobj)
        except json.JSONDecodeError as e:
            logger.error('JSON decode failed: %s', str(e), extra={'t_name': current_thread().name})
            return
        except (OSError, socket.timeout):
            logger.warning('Connection read timed out or failed for pid=%s uid=%s', _pid, _uid)
            return
        finally:
            watchdog.cancel()
            # close connection as soon as possible to
            # allow new clients to be connected
            connection.close()

        if deadline_exceeded.is_set():
            # The watchdog fired: the read returned only because the socket was
            # shut down at the deadline (SHUT_RDWR surfaces EOF, so read RETURNS
            # the partial buffer instead of raising). Drop the connection without
            # dispatching any job.
            logger.warning(
                'Connection read exceeded %ss deadline for pid=%s uid=%s, dropping', self._MAX_READ_DURATION, _pid, _uid
            )
            return

        # continue in different pool and threads because we don't want
        # our php processes to wait until processing is complete
        if input_data is None:
            self._submit_uid_job(
                workers_pool, _uid, self._handle_request_init, php_pid=_pid, cpu_usage=current_cpu, caller_uid=_uid
            )
        else:
            self._submit_uid_job(
                workers_pool,
                _uid,
                self._handle_request_end,
                php_pid=_pid,
                cpu_usage=current_cpu,
                request_data=input_data,
                caller_uid=_uid,
            )

    def _effective_uid_cap(self, _uid: int) -> int:
        """Dynamic max-min fair-share cap for one uid, in outstanding jobs.

        MUST be called while holding _uid_jobs_lock (it reads _uid_jobs).

        active = number of currently-active uids (those with > 0 outstanding
        jobs). _uid_jobs is kept free of zero-count entries, so its size is the
        active set; _uid itself is counted even on its first job (it is about
        to become active), so a flood never under-counts itself out of a share.

        The cap is:
            max(min_floor, min(capacity - reserve, max(min_floor, capacity // active)))

        * active == 1  -> capacity // 1 == capacity, clamped to capacity-reserve
          (e.g. 80-16=64): a lone user is bounded only by the real queue.Full,
          no regression, yet still leaves `reserve` slots for a new tenant.
        * active == N  -> capacity // N: a fair split; a flooder is held to its
          share going forward and converges to it as its jobs drain.
        * min_floor stops the cap collapsing to 0 with very many active uids.
        * the OUTER max(min_floor, ...) keeps the cap >= min_floor even when the
          capacity - reserve clamp is <= 0 (e.g. maxqueuesize == 0, so
          capacity == max_workers == reserve): without it the cap would be 0 and
          _submit_uid_job would drop EVERY request from EVERY uid (total
          self-DoS). On such a tiny pool the per-uid cap is min_floor and the
          executor's real queue.Full (BoundedSemaphore = maxqueuesize +
          max_workers) is the backstop.
        """
        active = len(self._uid_jobs) + (0 if _uid in self._uid_jobs else 1)
        fair_share = max(self._min_floor, self._pool_capacity // active)
        reserve_clamp = self._pool_capacity - self._reserve
        return max(self._min_floor, min(reserve_clamp, fair_share))

    def _submit_uid_job(self, workers_pool: BoundedThreadExecutor, _uid: int, fn, **kwargs) -> None:
        """
        Submit a worker job into the shared workers_pool, enforcing a DYNAMIC
        max-min per-UID fair share of the real pool capacity, UNCONDITIONALLY.

        At submit time the per-UID cap is recomputed from the number of
        currently-active uids (see _effective_uid_cap): a sole active uid gets
        the whole capacity minus a reserved headroom (no regression — bounded
        only by the real queue.Full), while with N active uids each gets
        capacity // N (fair split). There is NO pressure gate: the cap alone
        provides both no-regression for a lone user and anti-monopolization
        under contention, and a flooder cannot defeat it by supplying its own
        load. The reserved headroom guarantees a newly-arriving tenant always
        finds free slots even when one uid has filled its share.
        """
        with self._uid_jobs_lock:
            effective_cap = self._effective_uid_cap(_uid)
            if self._uid_jobs.get(_uid, 0) >= effective_cap:
                logger.warning(
                    'Per-UID fair-share drop: uid=%s holds %s '
                    'outstanding jobs (dynamic fair-share cap %s for '
                    '%s active uid(s), pool capacity %s, reserve %s); '
                    'dropping this request to protect other tenants',
                    _uid,
                    self._uid_jobs.get(_uid, 0),
                    effective_cap,
                    len(self._uid_jobs) + (0 if _uid in self._uid_jobs else 1),
                    self._pool_capacity,
                    self._reserve,
                )
                return
            self._uid_jobs[_uid] += 1
            self._uid_jobs_total += 1

        # Decrement when the job actually COMPLETES (not when submit returns):
        # wrap the callable so the per-UID slot is released in a finally on
        # every path (success or exception inside the job).
        def _job():
            try:
                return fn(**kwargs)
            finally:
                self._release_uid_job(_uid)

        try:
            workers_pool.submit(_job)
        except queue.Full:
            # global pool is full; release the per-UID + total slots we just
            # took so we don't leak the count, then keep the existing global
            # handling (distinct from the per-UID fairness drop above)
            self._release_uid_job(_uid)
            logger.error(
                'Request %s was rejected because our workers thread pool '
                'is full of tasks. Increase queuemaxsize or max_threads in configuration.'
            )
            return
        except BaseException:
            # any other submit failure must not leak the per-UID/total slot either
            self._release_uid_job(_uid)
            raise

    def _release_uid_job(self, _uid: int) -> None:
        """Decrement the per-UID and total outstanding-job counters under the
        lock. The total counter is floored at 0 so it can never underflow if a
        release is somehow called more often than expected, and zero-count
        per-UID entries are deleted to keep _uid_jobs from growing unbounded."""
        with self._uid_jobs_lock:
            self._uid_jobs[_uid] -= 1
            if self._uid_jobs[_uid] <= 0:
                del self._uid_jobs[_uid]
            if self._uid_jobs_total > 0:
                self._uid_jobs_total -= 1

    def read_input(self, fileio: io.TextIOBase) -> Any:
        """
        Read input data and return decoded json
        :param fileio: a file-like object providing read method
        """
        data = fileio.read(self._MAX_READ_SIZE)
        logger.debug('Received %d bytes', len(data))
        if len(data.strip()) == 0:
            return
        return json.loads(data.strip(), strict=False)

    def instantiate_mongo_client(self, fake_task_id: str) -> 'APIClient':
        """
        Initialize MongoDB client for current task
        """
        try:
            with dbm_storage(local_tasks_storage) as task_storage:
                try:
                    real_id = task_storage[fake_task_id].decode()
                except KeyError:
                    raise XRayError(
                        _("Cannot resolve tracing_task_id: no match found in storage"),
                        extra={'id': fake_task_id, 'all_ids': task_storage.keys()},
                    )
        except RuntimeError as e:
            raise XRayError(_("Cannot resolve tracing_task_id: %s") % str(e))

        return self.task_client_object(system_id=self.sys_id, tracing_task_id=real_id)

    def get_cached_or_load(self, fake_task_id: str) -> Tuple['APIClient', Task]:
        """
        Returns a client and task from cache of API data or
        initialize client and GET task from MongoDB and add to cache
        """
        logger.debug('Cached API data: %s', self.api_data_cache)

        # Fast-path: lookup outside the lock for the common cache-hit case.
        cached_data = self.api_data_cache.get(fake_task_id)
        if cached_data is not None:
            return cached_data.client, cached_data.task

        apiclient = self.instantiate_mongo_client(fake_task_id)
        _t = apiclient.get_task()
        logger.debug('Adding new container in cache: %s --> %s, %s', fake_task_id, _t, apiclient)
        with self.api_data_cache_lock:
            # Re-check inside the lock: another thread may have inserted the
            # same fake_task_id while we were fetching from MongoDB. If so,
            # discard our duplicate (it will be garbage-collected) and reuse
            # the cached entry to avoid spurious eviction of an unrelated
            # valid entry and a redundant MongoDB client.
            cached_data = self.api_data_cache.get(fake_task_id)
            if cached_data is not None:
                return cached_data.client, cached_data.task
            if len(self.api_data_cache) >= self._MAX_CACHE_ENTRIES:
                oldest_key = next(iter(self.api_data_cache))
                logger.info(
                    'api_data_cache at capacity (%d), evicting oldest entry %s', len(self.api_data_cache), oldest_key
                )
                self.api_data_cache.pop(oldest_key)
            self.api_data_cache[fake_task_id] = APIDataContainer(client=apiclient, task=_t)
        return apiclient, _t

    def cleanup_api_data_cache(self) -> None:
        """
        Cleanup an API data im-memory cache dict in order not store
        inactive (stopped, already completed) tasks there
        """
        try:
            with dbm_storage(local_tasks_storage) as task_storage:
                active_tasks = [k.decode() for k in task_storage.keys()]
        except RuntimeError:
            logger.warning('Unable to cleanup cache, storage unavailable')
            return

        for _task in list(self.api_data_cache.keys()):
            with self.api_data_cache_lock:
                if _task in self.api_data_cache and _task not in active_tasks:
                    logger.info('Cleaning up inactive container %s', _task)
                    self.api_data_cache.pop(_task)

    def process_request_data(self, request_data: dict, caller_uid: int = None) -> None:
        """
        Increment request ID in /usr/share/alt-php-xray/requests/{tracing_task_id} file
        Substitute request_id and tracing_task_id in request_data.
        Send request_data to ClickHouse
        :param request_data: original request data
        :param caller_uid: UID of the connecting process (from SO_PEERCRED)
        """
        fake_task_id = request_data['tracing_task_id']

        # Verify the caller owns this task to prevent cross-user data injection.
        # Task files live at /usr/share/alt-php-xray-tasks/{uid}/{fake_id},
        # created by the manager at task start. Only root can write there,
        # so the presence of the file is a reliable ownership proof.
        if caller_uid is not None:
            if not safe_id_pattern.match(fake_task_id):
                logger.warning('Rejecting request with invalid task id: %s', fake_task_id)
                return
            owner_task_file = os.path.join(tasks_base_storage, str(caller_uid), fake_task_id)
            if not os.path.isfile(owner_task_file):
                logger.warning(
                    'Rejecting request from uid=%s for task %s: no matching task file in per-UID storage',
                    caller_uid,
                    fake_task_id,
                )
                return

        _, task = self.get_cached_or_load(fake_task_id)
        logger.info('Processing task: %s', task.task_id)
        with open_local_storage(request_data['tracing_task_id'], flush=task.is_manual) as storage:
            # read stored request_id
            task.update_with_local_data(next_request_id=storage.next_request_id)

            if task.tracing_count <= 0:
                logger.info('Tracing count is 0, nothing should be done')
                return
            # update input data with stored request_id
            updated_request_data = self.update_request_data(request_data, task)
            # send data with updated ids
            logger.info('Sending to ClickHouse')
            self.send_client(updated_request_data)
            try:
                logger.info('Sending to SmartAdvice')
                self.adviser_client(updated_request_data)
            except XRayAPIError:
                # ignore all errors occurring within smart advice
                # microservice intercommunication
                pass
            # then increment request_id counter
            storage.next_request_id += 1
            # locally recalculate how much requests left to process
            task.update_with_local_data(next_request_id=storage.next_request_id)

        if task.is_manual:
            self._flush_mongodb_counters(task.fake_id)
        if task.tracing_by != 'time' and task.tracing_count <= 0:
            self.complete_task(task)

    def update_request_data(self, data: dict, task: Task) -> dict:
        """
        Substitute request_id and tracing_task_id
        :param data: original input
        :param task: a Task instance
        :return: updated input
        """
        data['request_id'] = task.request_count + 1
        data['tracing_task_id'] = task.task_id
        for item in data['data']:
            item['request_id'] = task.request_count + 1
            item['tracing_task_id'] = task.task_id
            if item['type'] == 'mysql_query':
                item['query'] = self.hide_symbols(item['query'])

        logger.info(
            'Input updated: tracing_task_id = %s & request_id = %s', data.get('tracing_task_id'), data.get('request_id')
        )
        logger.debug(
            'Updated input: %d item(s) for tracing_task_id = %s', len(data.get('data', [])), data.get('tracing_task_id')
        )
        return data

    def update_counts(self, client: 'APIClient', request_count: int, tracing_count: Optional[int] = None) -> None:
        """
        Update task counters in mongodb instance
        """
        client.update_counts_only(tracing_count=tracing_count, request_count=request_count)

    def complete_task(self, _task: Task) -> None:
        """
        Stop and complete request_qty task
        :param _task: tracing task to stop
        """
        logger.info('Task %s should be completed', _task.task_id)
        # delay for MongoDB to process counts, received lately (see XRAY-87)
        time.sleep(1)
        self._run_complete_task_cmd(_task.task_id)

    def _run_complete_task_cmd(self, task_id):
        try:
            subprocess.check_output(
                ['cloudlinux-xray-manager', 'stop', '--system_id', self.sys_id, '--tracing_task_id', task_id],
                timeout=120,
                stderr=subprocess.PIPE,
            )
        except (subprocess.TimeoutExpired, subprocess.CalledProcessError) as e:
            logger.error('Failed to complete task %s: %s', task_id, e)

    @staticmethod
    def hide_symbols(mysql_query: str) -> str:
        """
        Sanitize data in single quotes from MySQL query
        """

        def replacer(m):
            """
            Works with whole string in single or double quotes
            """
            q = m.group('quote')
            t = m.group('trunc')

            def inner_repl(inner_m):
                """
                Works with characters inside quotes
                """
                if inner_m.group('digit'):
                    return '0'
                elif inner_m.group('symbol'):
                    return 'x'

            sanitized = re.sub(r"((?P<digit>\d)|(?P<symbol>[^0-9_:;\-/',. \\]))", inner_repl, m.group('in_quote'))
            # wrap sanitized string back with originally detected characters
            # (quotes/truncation marker)
            return f'{q}{sanitized}{t or q}'

        # string either wrapped in quotes (single or double) or
        # starting from quote and finishing with ... (truncation marker)
        # including escaped with either / or \ quote
        pattern = re.compile(r"""(?P<quote>['"])(?P<in_quote>.*?)((?<![\\|/])(?P=quote)|(?P<trunc>\.{3}))""")
        return re.sub(pattern, replacer, mysql_query)