split requirements and venv into oss and yelpy for clog use

Author: sidsatyelp

MANIFEST.in

@@ -0,0 +1,2 @@
+include requirements-oss.txt
+include requirements-yelp.txt

Makefile

@@ -15,13 +15,33 @@ UID:=`id -u`
 GID:=`id -g`
 ITERATION=yelp1
 
-.PHONY: test tests coverage clean
+# Determine environment (YELP or OSS)
+# Set SCL_ENV to 'YELP' if FQDN ends in '.yelpcorp.com'
+# Otherwise, set SCL_ENV to 'OSS'
+ifneq ($(findstring .yelpcorp.com,$(shell hostname -f)),)
+	SCL_ENV ?= YELP
+else
+	SCL_ENV ?= OSS
+endif
 
+.PHONY: test tests-yelp tests-oss coverage clean venv venv-yelp venv-oss
 
+
+# Main test target dispatches to environment-specific test target
 test:
-	tox
+ifeq ($(SCL_ENV),YELP)
+	$(MAKE) tests-yelp
+else
+	$(MAKE) tests-oss
+endif
+
+tests-yelp: venv-yelp
+	tox -e py3-yelp # Assumes py3-yelp will be the tox env for yelp
+
+tests-oss: venv-oss
+	tox -e py3 # Assumes py3 will be the default/OSS tox env
 
-tests: test
+tests: test # Alias for backward compatibility or general use
 coverage: test
 
 itest_%: package_%
@@ -51,8 +71,19 @@ package_%:
 		'
 	docker run -v $(CURDIR):/work:rw docker-dev.yelpcorp.com/$*_yelp chown -R $(UID):$(GID) /work
 
-venv: requirements.txt setup.py tox.ini
-	tox -e venv
+# Main venv target dispatches to environment-specific target
+venv:
+ifeq ($(SCL_ENV),YELP)
+	$(MAKE) venv-yelp
+else
+	$(MAKE) venv-oss
+endif
+
+venv-yelp: requirements-yelp.txt requirements-oss.txt setup.py tox.ini
+	tox -e venv-yelp # This tox env should install .[yelp]
+
+venv-oss: requirements-oss.txt setup.py tox.ini
+	tox -e venv-oss # This tox env should install normally
 
 clean:
 	rm -rf .cache

requirements.txt renamed to requirements-oss.txt

@@ -1,7 +1,9 @@
+boto3
+ephemeral-port-reserve >= 1.1.0
 importlib-resources==5.4.0
 pyinotify==0.9.6
-pyyaml >= 3.0
+PyYAML >= 5.1
+requests>=2.18.4
 typing-extensions==4.13.2
 # To resolve the error: botocore 1.29.125 has requirement urllib3<1.27,>=1.25.4, but you'll have urllib3 2.0.1 which is incompatible.
 urllib3==1.26.15
-yelp-clog==7.2.0

requirements-yelp.txt

@@ -0,0 +1 @@
+yelp-clog==7.2.0

service_configuration_lib/spark_config.py

@@ -24,7 +24,12 @@
 from service_configuration_lib.text_colors import TextColors
 from service_configuration_lib.utils import EPHEMERAL_PORT_END
 from service_configuration_lib.utils import EPHEMERAL_PORT_START
-from service_configuration_lib.utils import get_clog_handler
+
+# Only works inside yelpy environments
+try:
+    import clog
+except ImportError:
+    clog = None
 
 AWS_CREDENTIALS_DIR = '/etc/boto_cfg/'
 AWS_ENV_CREDENTIALS_PROVIDER = 'com.amazonaws.auth.EnvironmentVariableCredentialsProvider'
@@ -441,16 +446,13 @@ def __init__(self, is_driver_on_k8s_tron: bool = False):
         try:
             (
                 self.spark_srv_conf, self.spark_constants, self.default_spark_srv_conf,
-                self.mandatory_default_spark_srv_conf, self.spark_costs, _,
+                self.mandatory_default_spark_srv_conf, self.spark_costs,
             ) = utils.load_spark_srv_conf()
         except Exception as e:
             log.error(f'Failed to load Spark srv configs: {e}')
             # should fail because Spark config calculation depends on values in srv-configs
             raise e
 
-        # Get a MonkHandler instance for JIRA validation warnings.
-        self.jira_monk_handler_ref = get_clog_handler(client_id='service_configuration_lib.spark_config')
-
     def _append_spark_prometheus_conf(self, spark_opts: Dict[str, str]) -> Dict[str, str]:
         spark_opts['spark.ui.prometheus.enabled'] = 'true'
         spark_opts[
@@ -1020,7 +1022,6 @@ def _handle_jira_ticket_validation(
             user: The user running the job
             jira_ticket: The Jira ticket provided by the user
         """
-        # Get the jira ticket validation setting
         flag_enabled = self.mandatory_default_spark_srv_conf.get('spark.yelp.jira_ticket.enabled', 'false')
         valid_ticket = self._get_valid_jira_ticket(jira_ticket)
 
@@ -1051,21 +1052,26 @@ def _handle_jira_ticket_validation(
                     'For more information: http://y/spark-jira-ticket-param '
                     'If you have questions, please reach out to #spark on Slack. '
                 )
-                if self.jira_monk_handler_ref:
-                    log_payload = {
-                        'timestamp': int(time.time()),
-                        'scribe_log': self.jira_monk_handler_ref.stream,
-                        'event': 'jira_ticket_validation_warning',
-                        'level': 'WARNING',
-                        'reason': 'Ticket missing or invalid. See http://y/spark-jira-ticket-param',
-                        'user': user,
-                        'jira_ticket_provided': jira_ticket,
-                    }
-                    self.jira_monk_handler_ref.logger.log_line(
-                        self.jira_monk_handler_ref.stream, json.dumps(log_payload),
-                    )
+                if clog:
+                    try:
+                        clog.config.configure(
+                            scribe_host='169.254.255.254',  # Standard Scribe host
+                            scribe_port=1463,             # Standard Scribe port
+                            monk_disable=False,           # Ensure Monk (for clog) is enabled
+                            scribe_disable=False,         # Ensure Scribe is enabled
+                        )
+                        log_payload = {
+                            'timestamp': int(time.time()),
+                            'event': 'jira_ticket_validation_warning',
+                            'level': 'WARNING',
+                            'reason': 'Ticket missing or invalid. See http://y/spark-jira-ticket-param',
+                            'user': user,
+                            'jira_ticket_provided': jira_ticket,
+                        }
+                        clog.log_line('spark_jira_ticket', json.dumps(log_payload))
+                    except Exception as e:
+                        log.warning(f'{warning_message} Clog operation failed with: {e}')
                 else:
-                    # Fallback to default logger if clog handler setup failed or ref is missing/invalid
                     log.warning(warning_message)
         return valid_ticket
 

service_configuration_lib/utils.py

@@ -3,7 +3,6 @@
 import errno
 import hashlib
 import logging
-import os
 import random
 import string
 import uuid
@@ -14,16 +13,9 @@
 from socket import SOL_SOCKET
 from typing import Any
 from typing import Dict
-from typing import List
-from typing import Optional
 from typing import Tuple
 
-import clog.config
-import srv_configs
 import yaml
-from clog.config import monk_host
-from clog.config import monk_port
-from clog.handlers import MonkHandler
 from typing_extensions import Literal
 
 DEFAULT_SPARK_RUN_CONFIG = '/nail/srv/configs/spark.yaml'
@@ -40,7 +32,7 @@
 SPARK_DRIVER_MEM_OVERHEAD_FACTOR_DEFAULT = 0.1
 
 
-log = logging.Logger(__name__)
+log = logging.getLogger(__name__)
 log.setLevel(logging.INFO)
 
 
@@ -50,7 +42,6 @@ def load_spark_srv_conf(preset_values=None) -> Tuple[
     Dict[str, Any],  # default_spark_srv_conf
     Dict[str, Any],  # mandatory_default_spark_srv_conf
     Dict[str, Dict[str, float]],  # spark_costs
-    List[Dict[str, Any]],  # module_configs
 ]:
     if preset_values is None:
         preset_values = dict()
@@ -62,10 +53,9 @@ def load_spark_srv_conf(preset_values=None) -> Tuple[
             default_spark_srv_conf = spark_constants['defaults']
             mandatory_default_spark_srv_conf = spark_constants['mandatory_defaults']
             spark_costs = spark_constants['cost_factor']
-            module_configs = loaded_values['module_config']
             return (
                 spark_srv_conf, spark_constants, default_spark_srv_conf,
-                mandatory_default_spark_srv_conf, spark_costs, module_configs,
+                mandatory_default_spark_srv_conf, spark_costs,
             )
     except Exception as e:
         log.warning(f'Failed to load {DEFAULT_SPARK_RUN_CONFIG}: {e}')
@@ -227,161 +217,3 @@ def get_spark_driver_memory_overhead_mb(spark_conf: Dict[str, str]) -> float:
         )
         driver_mem_overhead_mb = driver_mem_mb * driver_mem_overhead_factor
     return round(driver_mem_overhead_mb, 5)
-
-
-def _load_default_service_configurations_for_clog() -> Optional[Dict[str, Any]]:
-    """
-    Loads the external configuration file for the 'clog' namespace if specified in
-    DEFAULT_SPARK_RUN_CONFIG's 'module_config' section.
-    Returns the inline 'config' dictionary for the 'clog' namespace if found,
-    otherwise None.
-    """
-    clog_config_file_path = None
-    clog_inline_config = None
-    found_clog_module_config = False
-
-    try:
-        _, _, _, _, _, module_configs = load_spark_srv_conf()
-
-        for mc_item in module_configs:
-            if isinstance(mc_item, dict) and mc_item.get('namespace') == 'clog':
-                found_clog_module_config = True
-                clog_config_file_path = mc_item.get('file')
-                clog_inline_config = mc_item.get('config')
-                break
-
-        if not found_clog_module_config:
-            log.warning(
-                f"Could not find 'clog' namespace entry in 'module_config' "
-                f'section within {DEFAULT_SPARK_RUN_CONFIG}.',
-            )
-            return None
-
-        if clog_config_file_path:
-            if os.path.exists(clog_config_file_path):
-                try:
-                    srv_configs.use_file(clog_config_file_path, namespace='clog')
-                    log.info(
-                        f'Successfully loaded clog configuration file {clog_config_file_path} '
-                        f"into namespace 'clog'.",
-                    )
-                except Exception as e_use_file:
-                    log.error(
-                        f'Error loading clog configuration file {clog_config_file_path} '
-                        f'using srv_configs.use_file: {e_use_file}',
-                    )
-            else:
-                log.error(
-                    f"Clog configuration file specified in 'module_config' of {DEFAULT_SPARK_RUN_CONFIG} "
-                    f'does not exist: {clog_config_file_path}.',
-                )
-        else:
-            log.info(
-                f"No 'file' specified for 'clog' namespace in 'module_config' of {DEFAULT_SPARK_RUN_CONFIG}. "
-                'Not loading any external file for clog via module_config.',
-            )
-
-        # Return the inline config dictionary, which might be None if not present
-        if isinstance(clog_inline_config, dict):
-            return clog_inline_config
-        elif clog_inline_config is not None:
-            log.warning(f"Inline 'config' for 'clog' namespace in {DEFAULT_SPARK_RUN_CONFIG} is not a dictionary.")
-            return None
-        else:
-            return None
-
-    except FileNotFoundError:
-        log.error(
-            f'Error: Main Spark run config file {DEFAULT_SPARK_RUN_CONFIG} not found. '
-            'Cannot process clog configurations.',
-        )
-        return None
-    except yaml.YAMLError as e_yaml:
-        log.error(f'Error parsing YAML from {DEFAULT_SPARK_RUN_CONFIG}: {e_yaml}')
-        return None
-    except Exception as e_main:
-        log.error(
-            f'An unexpected error occurred in _load_default_service_configurations_for_clog: {e_main}',
-        )
-        return None
-
-
-def get_clog_handler(
-    client_id: Optional[str] = None,
-    stream_name_override: Optional[str] = None,
-) -> Optional[MonkHandler]:
-    """
-    Configures and returns a clog MonkHandler for logging.
-
-    This utility helps in setting up a MonkHandler. It ensures the external
-    clog configuration file (if specified in DEFAULT_SPARK_RUN_CONFIG) is loaded
-    into srv_configs. It then determines the log_stream_name with the following
-    priority:
-    1. `stream_name_override` argument.
-    2. `log_stream_name` from the inline 'config' of the 'clog' module_config
-       in DEFAULT_SPARK_RUN_CONFIG.
-    3. `log_stream_name` from the 'clog' namespace in srv_configs (typically
-       loaded from the external file like /nail/srv/configs/clog.yaml).
-
-    Args:
-        client_id: Optional client identifier for the log messages.
-                   Defaults to the current OS user or 'unknown_spark_user'.
-        stream_name_override: Optional explicit clog stream name to use,
-                              overriding any configured values.
-
-    Returns:
-        A configured MonkHandler instance if successful, otherwise None.
-    """
-    # Load external file (if any) and get inline config from spark.yaml's module_config
-    inline_clog_config = _load_default_service_configurations_for_clog()
-
-    actual_client_id = client_id or os.getenv('USER') or 'unknown_spark_user'
-    final_stream_name = stream_name_override
-
-    if not final_stream_name:
-        if inline_clog_config and isinstance(inline_clog_config.get('log_stream_name'), str):
-            final_stream_name = inline_clog_config['log_stream_name']
-            log.info(
-                f"Using log_stream_name '{final_stream_name}' from inline module_config in "
-                f'{DEFAULT_SPARK_RUN_CONFIG}.',
-            )
-        else:
-            try:
-                # Fallback to srv_configs (which should have data from the external file)
-                clog_srv_configs_dict = srv_configs.get_namespace_as_dict('clog')
-                final_stream_name = clog_srv_configs_dict.get('log_stream_name')
-                if final_stream_name:
-                    log.info(
-                        f"Using log_stream_name '{final_stream_name}' from srv_configs 'clog' namespace "
-                        f'(likely from external file).',
-                    )
-            except Exception as e:
-                log.warning(
-                    f"Could not get 'clog' namespace from srv_configs or 'log_stream_name' key missing. "
-                    f'This may be okay if stream_name_override or inline config provides it. Error: {e}',
-                )
-
-    if not final_stream_name:
-        log.error(
-            'Clog stream_name could not be determined. It was not provided as an argument, '
-            'not found in the inline module_config for "clog", and not found in the '
-            '"clog" srv_configs namespace. Clog handler cannot be configured.',
-        )
-        return None
-
-    # Ensure that clog is configured to enable Monk logging.
-    # The default in clog.config.monk_disable is True.
-    clog.config.configure_from_dict({'monk_disable': False})
-    log.info('Clog has been configured to enable Monk logging (monk_disable=False).')
-
-    try:
-        handler = MonkHandler(
-            client_id=actual_client_id,
-            host=monk_host,
-            port=monk_port,
-            stream=final_stream_name,
-        )
-        return handler
-    except Exception as e:
-        log.error(f"Failed to create MonkHandler for clog with stream '{final_stream_name}'. Error: {e}")
-        return None