from github.env import get_required_env_var, exit_on_error_and_write_summary from actions_logging.app_logging import logger import os import string import json from aws.client import create_client_or_exit import time from aws.msk.msk_info import get_msk_environment_name def setup_cloudwatch_log_group(logs_client, env_name: str, log_group_name: str, retention_days: int = 7) -> None: tags = { 'Department': 'DevOps', 'Environment': env_name, 'GitPath': 'devops-yarkon-pipelines', 'Temp': 'false', 'Terraform': 'false' } existing_groups = logs_client.describe_log_groups(logGroupNamePrefix=log_group_name) if not any(group['logGroupName'] == log_group_name for group in existing_groups.get('logGroups', [])): logs_client.create_log_group( logGroupName=log_group_name, tags=tags ) logger.info(f"Created log group: {log_group_name}") else: logger.info(f"Log group '{log_group_name}' already exists.") logs_client.put_retention_policy( logGroupName=log_group_name, retentionInDays=retention_days ) logger.debug(f"Set retention policy of {retention_days} days for log group: {log_group_name}") def get_missing_template_vars(connector_configuration_template: str, config_vars: dict) -> dict: template = string.Template(connector_configuration_template) required_vars = set() # Find all variables in template for match in template.pattern.finditer(connector_configuration_template): var_name = match.group('named') or match.group('braced') if var_name: required_vars.add(var_name) # Find missing variables missing_vars = required_vars - set(config_vars.keys()) # Try to get missing variables from environment for var_name in missing_vars: config_vars[var_name] = get_required_env_var(var_name) logger.debug(f"found {var_name} in environment variables") return config_vars def wait_until_connector_is_created(connector_arn: str, connector_plugin_name: str, kafka_connect_client, sleep_time: int = 10) -> None: counter = 0 while True: counter += 1 response = kafka_connect_client.describe_connector(connectorArn=connector_arn) if response['connectorState'] == 'CREATING': logger.info(f'{counter}) connector is still creating, {counter * sleep_time} seconds passed') elif response['connectorState'] == 'FAILED': exit_on_error_and_write_summary(f"Connector {connector_plugin_name} failed to start, halting") elif response['connectorState'] == 'RUNNING': logger.info(f"connector {connector_plugin_name} created successfully") return else: exit_on_error_and_write_summary(f"Connector {connector_plugin_name} is not in wanted state: {response['connectorState']}, halting") time.sleep(sleep_time) def main(): kafka_env_name = get_required_env_var("KAFKA_ENV") aws_msk_region = get_required_env_var("AWS_MSK_REGION") env_name = get_required_env_var("ENV_NAME") connector_name = get_required_env_var("CONNECTOR_NAME") connector_role_arn = get_required_env_var("KAFKA_CONNECTOR_ROLE_ARN") msk_env = get_msk_environment_name(env_name) yarkon_mongo_uri = get_required_env_var(f"ENV_{msk_env.upper()}_P81_YARKON_YARKON_MONGO_URI") yarkon_activity_log_mongo_uri = get_required_env_var(f"ENV_{msk_env.upper()}_P81_YARKON_YARKON_ACTIVITY_LOG_MONGO_URI") snowflake_prv_key = get_required_env_var(f"ENV_{msk_env.upper()}_P81_YARKON_SNOWFLAKE_PRV_KEY_SL") config_vars = { key: value for key, value in os.environ.items() if key.startswith('KAFKA_CONFIG_VAR_') } config_vars["KAFKA_VERSION"] = "2.7.1" config_vars["PLUGIN_VERSION"] = "1" config_vars["SNOWFLAKE_PRV_KEY"] = snowflake_prv_key ssm = create_client_or_exit("ssm", aws_msk_region) kafka_iam_connection_string = ssm.get_parameter(Name=f"/{kafka_env_name}/kafka/KAFKA_IAM_CONNECTION_STRING", WithDecryption=True)['Parameter']['Value'] kafka_sg = ssm.get_parameter(Name=f"/{kafka_env_name}/kafka/KAFKA_SG", WithDecryption=True)['Parameter']['Value'] kafka_subnets = ssm.get_parameter(Name=f"/{kafka_env_name}/kafka/KAFKA_SUBNETS", WithDecryption=True)['Parameter']['Value'].strip('[]') logs_group_name = f"/{msk_env}/pipelines/{connector_name}" logs = create_client_or_exit("logs", aws_msk_region) setup_cloudwatch_log_group(logs, msk_env, logs_group_name) logger.info("Connector log group setup completed.") config_vars["ENVIRONMENT"] = msk_env config_vars["YARKON_MONGO_URI"] = yarkon_mongo_uri config_vars["YARKON_ACTIVITY_LOG_MONGO_URI"] = yarkon_activity_log_mongo_uri config_vars["CONNECTOR_ROLE_ARN"] = connector_role_arn config_vars["KAFKA_IAM_CONNECTION_STRING"] = kafka_iam_connection_string config_vars["KAFKA_SUBNETS"] = kafka_subnets config_vars["LOG_GROUP"] = logs_group_name config_vars["KAFKA_SG"] = kafka_sg config_vars["DOLLAR"] = '$' config_vars["AWS_MSK_REGION"] = aws_msk_region config_vars["CONNECTOR_NAME"] = connector_name dir_to_build = os.path.join(os.getcwd(), connector_name := os.getenv("DIR_TO_BUILD", "")) template_path = os.path.join(dir_to_build, "src/main/resources/config/connectorConf.json") logger.debug(f"Reading template from {template_path}") try: with open(template_path, 'r') as f: connector_configuration_template = f.read() config_vars = get_missing_template_vars(connector_configuration_template, config_vars) connector_configuration = string.Template(connector_configuration_template).safe_substitute(config_vars) # Check for unsubstituted variables template = string.Template(connector_configuration_template) unmatched_vars = set() for match in template.pattern.finditer(connector_configuration_template): var_name = match.group('named') or match.group('braced') if var_name and var_name not in config_vars: unmatched_vars.add(var_name) if unmatched_vars: logger.error(f"Missing template variables: {', '.join(unmatched_vars)}") logger.debug("Generated configuration:") logger.debug(connector_configuration) connector_config = json.loads(connector_configuration) kafka_connector = create_client_or_exit('kafkaconnect', aws_msk_region) kafka_connector_creation_response = kafka_connector.create_connector(**connector_config) kafka_connector_arn = kafka_connector_creation_response['connectorArn'] logger.info(f"Connector added successfully. ARN: {kafka_connector_arn}") wait_until_connector_is_created(kafka_connector_arn, connector_name, kafka_connector) except FileNotFoundError: exit_on_error_and_write_summary(f"Template file not found at: {template_path}, halting") except Exception as e: exit_on_error_and_write_summary(f"Error processing template or failed to create a connector: {str(e)}") if __name__ == "__main__": main()