<commit_before>"""Contains the NighthawkTestServer class, which wraps the nighthawk_test_servern binary."""

import collections
import http.client
import json
import logging
import os
import random
import re
import requests
import socket
import subprocess
import sys
import tempfile
import threading
import time
import yaml
from string import Template
from pathlib import Path
from rules_python.python.runfiles import runfiles

from test.integration.common import IpVersion, NighthawkException
from test.integration.subprocess_mixin import SubprocessMixin


def _substitute_yaml_values(runfiles_instance, obj, params):
  if isinstance(obj, dict):
    for k, v in obj.items():
      obj[k] = _substitute_yaml_values(runfiles_instance, v, params)
  elif isinstance(obj, list):
    for i in range(len(obj)):
      obj[i] = _substitute_yaml_values(runfiles_instance, obj[i], params)
  else:
    if isinstance(obj, str):
      # Inspect string values and substitute where applicable.
      INJECT_RUNFILE_MARKER = '@inject-runfile:'
      if obj[0] == '$':
        return Template(obj).substitute(params)
      elif obj.startswith(INJECT_RUNFILE_MARKER):
        with open(runfiles_instance.Rlocation(obj[len(INJECT_RUNFILE_MARKER):].strip()),
                  'r') as file:
          return file.read()
  return obj


class _TestCaseWarnErrorIgnoreList(
    collections.namedtuple("_TestCaseWarnErrorIgnoreList", "test_case_regexp ignore_list")):
  """Maps test case names to messages that should be ignored in the test server logs.

  If the name of the currently executing test case matches the test_case_regexp,
  any messages logged by the test server as either a WARNING or an ERROR
  will be checked against the ignore_list. If the logged messages contain any of
  the messages in the ignore list as a substring, they will be ignored.
  Any unmatched messages of either a WARNING or an ERROR severity will fail the
  test case.

  Attributes:
    test_case_regexp: A compiled regular expression as returned by re.compile(),
      the regexp that will be used to match test case names.
    ignore_list: A tuple of strings, messages to ignore for matching test cases.
  """


# A list of _TestCaseWarnErrorIgnoreList instances, message pieces that should
# be ignored even if logged by the test server at a WARNING or an ERROR
# severity.
#
# If multiple test_case_regexp entries match the current test case name, all the
# corresponding ignore lists will be used.
_TEST_SERVER_WARN_ERROR_IGNORE_LIST = frozenset([
    # This test case purposefully uses the deprecated Envoy v2 API which emits
    # the following warnings.
    _TestCaseWarnErrorIgnoreList(
        re.compile('test_nighthawk_test_server_envoy_deprecated_v2_api'),
        (
            "Configuration does not parse cleanly as v3. v2 configuration is deprecated",
            "Deprecated field: type envoy.api.v2.listener.Filter",
            "Deprecated field: type envoy.config.filter.network.http_connection_manager.v2.HttpFilter",
            "Using deprecated extension name 'envoy.http_connection_manager'",
            "Using deprecated extension name 'envoy.router'",
        ),
    ),

    # A catch-all that applies to all remaining test cases.
    _TestCaseWarnErrorIgnoreList(
        re.compile('.*'),
        (
            # TODO(#582): Identify these and file issues or add explanation as necessary.
            "Unable to use runtime singleton for feature envoy.http.headermap.lazy_map_min_size",
            "Unable to use runtime singleton for feature envoy.reloadable_features.header_map_correctly_coalesce_cookies",
            "Using deprecated extension name 'envoy.listener.tls_inspector' for 'envoy.filters.listener.tls_inspector'.",
            "there is no configured limit to the number of allowed active connections. Set a limit via the runtime key overload.global_downstream_max_connections",

            # A few of our filters use the same typed configuration, specifically
            # 'test-server', 'time-tracking' and 'dynamic-delay'.
            # For now this is by design.
            "Double registration for type: 'nighthawk.server.ResponseOptions'",

            # Logged for normal termination, not really a warning.
            "caught ENVOY_SIGTERM",
        ),
    ),
])


class TestServerBase(SubprocessMixin):
  """Base class for running a server in a separate process.

  Attributes:
    ip_version: IP version that the proxy should use when listening.
    server_ip: string containing the server ip that will be used to listen
    server_port: Integer, get the port used by the server to listen for traffic.
    docker_image: String, supplies a docker image for execution of the test server binary. Sourced from environment variable NH_DOCKER_IMAGE.
    tmpdir: String, indicates the location used to store outputs like logs.
  """

  def __init__(self, server_binary_path, config_template_path, server_ip, ip_version, request,
               server_binary_config_path_arg, parameters, tag):
    """Initialize a TestServerBase instance.

    Args:
        server_binary_path (str): specify the path to the server binary.
        config_template_path (str): specify the path to the test server configuration template.
        server_ip (str): Specify the ip address the test server should use to listen for traffic.
        ip_version (IPAddress): Specify the ip version the server should use to listen for traffic.
        request: The pytest `request` fixture used to determin information about the currently executed test.
        server_binary_config_path_arg (str): Specify the name of the CLI argument the test server binary uses to accept a configuration path.
        parameters (dict): Supply to provide configuration template parameter replacement values.
        tag (str): Supply to get recognizeable output locations.
    """
    SubprocessMixin.__init__(self)
    assert ip_version != IpVersion.UNKNOWN
    self.ip_version = ip_version
    self.server_ip = server_ip
    self.server_port = -1
    self.docker_image = os.getenv("NH_DOCKER_IMAGE", "")
    self.tmpdir = os.path.join(os.getenv("TMPDIR", "/tmp/nighthawk_benchmark/"), tag + "/")
    self._server_binary_path = server_binary_path
    self._config_template_path = config_template_path
    self._parameters = dict(parameters)
    self._parameters["server_ip"] = self.server_ip
    self._parameters["tmpdir"] = self.tmpdir
    self._parameters["tag"] = tag
    self._server_process = None
    self._server_thread = threading.Thread(target=self._serverThreadRunner)
    self._admin_address_path = ""
    self._parameterized_config_path = ""
    self._instance_id = str(random.randint(1, 1024 * 1024 * 1024))
    self._server_binary_config_path_arg = server_binary_config_path_arg
    self._prepareForExecution()
    self._request = request

  def _prepareForExecution(self):
    runfiles_instance = runfiles.Create()
    with open(runfiles_instance.Rlocation(self._config_template_path)) as f:
      data = yaml.load(f, Loader=yaml.FullLoader)
      data = _substitute_yaml_values(runfiles_instance, data, self._parameters)

    Path(self.tmpdir).mkdir(parents=True, exist_ok=True)

    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".config.yaml",
                                     dir=self.tmpdir) as tmp:
      self._parameterized_config_path = tmp.name
      yaml.safe_dump(data,
                     tmp,
                     default_flow_style=False,
                     explicit_start=True,
                     allow_unicode=True,
                     encoding='utf-8')

    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".adminport",
                                     dir=self.tmpdir) as tmp:
      self._admin_address_path = tmp.name

  def _argsForSubprocess(self):
    args = []
    if self.docker_image != "":
      # TODO(#383): As of https://github.com/envoyproxy/envoy/commit/e8a2d1e24dc9a0da5273442204ec3cdfad1e7ca8
      # we need to have ENVOY_UID=0 in the environment, or this will break on docker runs, as Envoy
      # will not be able to read the configuration files we stub here in docker runs.
      args = [
          "docker", "run", "--network=host", "--rm", "-v", "{t}:{t}".format(t=self.tmpdir), "-e",
          "ENVOY_UID=0", self.docker_image
      ]
    args = args + [
        self._server_binary_path, self._server_binary_config_path_arg,
        self._parameterized_config_path, "-l", "debug", "--base-id", self._instance_id,
        "--admin-address-path", self._admin_address_path, "--concurrency", "1"
    ]
    return args

  def _serverThreadRunner(self):
    stdout, stderr = super()._serverThreadRunner()
    warnings, errors = _extractWarningsAndErrors(stdout.decode() + stderr.decode(),
                                                 self._request.node.name,
                                                 _TEST_SERVER_WARN_ERROR_IGNORE_LIST)
    if warnings:
      [logging.warn("Process logged a warning: %s", w) for w in warnings]
    if errors:<commit_msg>Use logging.warning as logging.warn is deprecated. (#897)<commit_after>"""Contains the NighthawkTestServer class, which wraps the nighthawk_test_servern binary."""

import collections
import http.client
import json
import logging
import os
import random
import re
import requests
import socket
import subprocess
import sys
import tempfile
import threading
import time
import yaml
from string import Template
from pathlib import Path
from rules_python.python.runfiles import runfiles

from test.integration.common import IpVersion, NighthawkException
from test.integration.subprocess_mixin import SubprocessMixin


def _substitute_yaml_values(runfiles_instance, obj, params):
  if isinstance(obj, dict):
    for k, v in obj.items():
      obj[k] = _substitute_yaml_values(runfiles_instance, v, params)
  elif isinstance(obj, list):
    for i in range(len(obj)):
      obj[i] = _substitute_yaml_values(runfiles_instance, obj[i], params)
  else:
    if isinstance(obj, str):
      # Inspect string values and substitute where applicable.
      INJECT_RUNFILE_MARKER = '@inject-runfile:'
      if obj[0] == '$':
        return Template(obj).substitute(params)
      elif obj.startswith(INJECT_RUNFILE_MARKER):
        with open(runfiles_instance.Rlocation(obj[len(INJECT_RUNFILE_MARKER):].strip()),
                  'r') as file:
          return file.read()
  return obj


class _TestCaseWarnErrorIgnoreList(
    collections.namedtuple("_TestCaseWarnErrorIgnoreList", "test_case_regexp ignore_list")):
  """Maps test case names to messages that should be ignored in the test server logs.

  If the name of the currently executing test case matches the test_case_regexp,
  any messages logged by the test server as either a WARNING or an ERROR
  will be checked against the ignore_list. If the logged messages contain any of
  the messages in the ignore list as a substring, they will be ignored.
  Any unmatched messages of either a WARNING or an ERROR severity will fail the
  test case.

  Attributes:
    test_case_regexp: A compiled regular expression as returned by re.compile(),
      the regexp that will be used to match test case names.
    ignore_list: A tuple of strings, messages to ignore for matching test cases.
  """


# A list of _TestCaseWarnErrorIgnoreList instances, message pieces that should
# be ignored even if logged by the test server at a WARNING or an ERROR
# severity.
#
# If multiple test_case_regexp entries match the current test case name, all the
# corresponding ignore lists will be used.
_TEST_SERVER_WARN_ERROR_IGNORE_LIST = frozenset([
    # This test case purposefully uses the deprecated Envoy v2 API which emits
    # the following warnings.
    _TestCaseWarnErrorIgnoreList(
        re.compile('test_nighthawk_test_server_envoy_deprecated_v2_api'),
        (
            "Configuration does not parse cleanly as v3. v2 configuration is deprecated",
            "Deprecated field: type envoy.api.v2.listener.Filter",
            "Deprecated field: type envoy.config.filter.network.http_connection_manager.v2.HttpFilter",
            "Using deprecated extension name 'envoy.http_connection_manager'",
            "Using deprecated extension name 'envoy.router'",
        ),
    ),

    # A catch-all that applies to all remaining test cases.
    _TestCaseWarnErrorIgnoreList(
        re.compile('.*'),
        (
            # TODO(#582): Identify these and file issues or add explanation as necessary.
            "Unable to use runtime singleton for feature envoy.http.headermap.lazy_map_min_size",
            "Unable to use runtime singleton for feature envoy.reloadable_features.header_map_correctly_coalesce_cookies",
            "Using deprecated extension name 'envoy.listener.tls_inspector' for 'envoy.filters.listener.tls_inspector'.",
            "there is no configured limit to the number of allowed active connections. Set a limit via the runtime key overload.global_downstream_max_connections",

            # A few of our filters use the same typed configuration, specifically
            # 'test-server', 'time-tracking' and 'dynamic-delay'.
            # For now this is by design.
            "Double registration for type: 'nighthawk.server.ResponseOptions'",

            # Logged for normal termination, not really a warning.
            "caught ENVOY_SIGTERM",
        ),
    ),
])


class TestServerBase(SubprocessMixin):
  """Base class for running a server in a separate process.

  Attributes:
    ip_version: IP version that the proxy should use when listening.
    server_ip: string containing the server ip that will be used to listen
    server_port: Integer, get the port used by the server to listen for traffic.
    docker_image: String, supplies a docker image for execution of the test server binary. Sourced from environment variable NH_DOCKER_IMAGE.
    tmpdir: String, indicates the location used to store outputs like logs.
  """

  def __init__(self, server_binary_path, config_template_path, server_ip, ip_version, request,
               server_binary_config_path_arg, parameters, tag):
    """Initialize a TestServerBase instance.

    Args:
        server_binary_path (str): specify the path to the server binary.
        config_template_path (str): specify the path to the test server configuration template.
        server_ip (str): Specify the ip address the test server should use to listen for traffic.
        ip_version (IPAddress): Specify the ip version the server should use to listen for traffic.
        request: The pytest `request` fixture used to determin information about the currently executed test.
        server_binary_config_path_arg (str): Specify the name of the CLI argument the test server binary uses to accept a configuration path.
        parameters (dict): Supply to provide configuration template parameter replacement values.
        tag (str): Supply to get recognizeable output locations.
    """
    SubprocessMixin.__init__(self)
    assert ip_version != IpVersion.UNKNOWN
    self.ip_version = ip_version
    self.server_ip = server_ip
    self.server_port = -1
    self.docker_image = os.getenv("NH_DOCKER_IMAGE", "")
    self.tmpdir = os.path.join(os.getenv("TMPDIR", "/tmp/nighthawk_benchmark/"), tag + "/")
    self._server_binary_path = server_binary_path
    self._config_template_path = config_template_path
    self._parameters = dict(parameters)
    self._parameters["server_ip"] = self.server_ip
    self._parameters["tmpdir"] = self.tmpdir
    self._parameters["tag"] = tag
    self._server_process = None
    self._server_thread = threading.Thread(target=self._serverThreadRunner)
    self._admin_address_path = ""
    self._parameterized_config_path = ""
    self._instance_id = str(random.randint(1, 1024 * 1024 * 1024))
    self._server_binary_config_path_arg = server_binary_config_path_arg
    self._prepareForExecution()
    self._request = request

  def _prepareForExecution(self):
    runfiles_instance = runfiles.Create()
    with open(runfiles_instance.Rlocation(self._config_template_path)) as f:
      data = yaml.load(f, Loader=yaml.FullLoader)
      data = _substitute_yaml_values(runfiles_instance, data, self._parameters)

    Path(self.tmpdir).mkdir(parents=True, exist_ok=True)

    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".config.yaml",
                                     dir=self.tmpdir) as tmp:
      self._parameterized_config_path = tmp.name
      yaml.safe_dump(data,
                     tmp,
                     default_flow_style=False,
                     explicit_start=True,
                     allow_unicode=True,
                     encoding='utf-8')

    with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".adminport",
                                     dir=self.tmpdir) as tmp:
      self._admin_address_path = tmp.name

  def _argsForSubprocess(self):
    args = []
    if self.docker_image != "":
      # TODO(#383): As of https://github.com/envoyproxy/envoy/commit/e8a2d1e24dc9a0da5273442204ec3cdfad1e7ca8
      # we need to have ENVOY_UID=0 in the environment, or this will break on docker runs, as Envoy
      # will not be able to read the configuration files we stub here in docker runs.
      args = [
          "docker", "run", "--network=host", "--rm", "-v", "{t}:{t}".format(t=self.tmpdir), "-e",
          "ENVOY_UID=0", self.docker_image
      ]
    args = args + [
        self._server_binary_path, self._server_binary_config_path_arg,
        self._parameterized_config_path, "-l", "debug", "--base-id", self._instance_id,
        "--admin-address-path", self._admin_address_path, "--concurrency", "1"
    ]
    return args

  def _serverThreadRunner(self):
    stdout, stderr = super()._serverThreadRunner()
    warnings, errors = _extractWarningsAndErrors(stdout.decode() + stderr.decode(),
                                                 self._request.node.name,
                                                 _TEST_SERVER_WARN_ERROR_IGNORE_LIST)
    if warnings:
      [logging.warning("Process logged a warning: %s", w) for w in warnings]
    if errors:
{'fraction_non_alphanumeric': 0.07942198444652805, 'fraction_numerical': 0.005901494677623959, 'mean_word_length': 3.242395882077679, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1874764', 'n_tokens_mistral': 5094, 'n_tokens_neox': 4918, 'n_words': 1597}