# Learn CloudWatch Logs syslog ingestion basics using an Amazon SDK The following code example shows how to: + Deploy a CloudFormation stack that provisions a syslog VPC endpoint. + Create a log group and add a resource policy for the syslog service. + Create a syslog configuration with PutSyslogConfiguration. + List syslog configurations with and without filters. + Clean up by deleting the syslog configuration, log group, and stack. ------ #### [ Python ] **SDK for Python (Boto3)** There's more on GitHub. Find the complete example and learn how to set up and run in the [Amazon Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/cloudwatch-logs/scenarios/syslog_ingestion#code-examples). Run an interactive scenario at a command prompt. ``` class SyslogIngestionScenario: """Interactive scenario demonstrating CloudWatch Logs syslog ingestion.""" STACK_NAME = "syslog-demo-stack" def __init__( self, cloudwatch_logs_wrapper: CloudWatchLogsWrapper, cf_client: boto3.client, ) -> None: """ :param cloudwatch_logs_wrapper: A CloudWatchLogsWrapper instance. :param cf_client: A Boto3 CloudFormation client. """ self.wrapper = cloudwatch_logs_wrapper self.cf_client = cf_client self.vpc_endpoint_id: Optional[str] = None self.log_group_name: Optional[str] = None self.log_group_arn: Optional[str] = None self.stack_deployed = False # ------------------------------------------------------------------ # Setup phase # ------------------------------------------------------------------ def deploy_cfn_stack(self) -> str: """ Deploys the CloudFormation stack that provisions the VPC endpoint prerequisite. Waits for CREATE_COMPLETE and returns the VpcEndpointId output. :return: The VPC endpoint ID. """ print("\n" + "=" * 68) print("Step 1: Deploy the CloudFormation prerequisite stack") print("=" * 68) print( f"\nDeploying stack '{self.STACK_NAME}' to create a VPC, subnet, " "security group, and syslog VPC endpoint.\n" "This can take several minutes..." ) try: self.cf_client.create_stack( StackName=self.STACK_NAME, TemplateBody=_load_cfn_template(), Capabilities=["CAPABILITY_IAM"], ) # Only this run's own stack should be torn down in cleanup. Mark it # as soon as creation is initiated so a stack that fails partway # (e.g. ROLLBACK) is still cleaned up. self.stack_deployed = True except ClientError as error: if error.response["Error"]["Code"] == "AlreadyExistsException": # The stack pre-existed this run. Reuse it but do NOT mark it # for cleanup — deleting a stack we did not create could destroy # infrastructure the user set up themselves. print( f"Stack '{self.STACK_NAME}' already exists. Reusing it; it " "will not be deleted during cleanup." ) else: raise waiter = self.cf_client.get_waiter("stack_create_complete") try: waiter.wait( StackName=self.STACK_NAME, WaiterConfig={"Delay": 30, "MaxAttempts": 40}, ) except WaiterError: # The waiter fails both when the stack genuinely failed # (CREATE_FAILED, ROLLBACK_COMPLETE, ...) and, harmlessly, when the # stack already existed in a completed state before this run. # Inspect the actual status so a real failure surfaces clearly # instead of as a confusing "missing outputs" error later. status = self.cf_client.describe_stacks(StackName=self.STACK_NAME)[ "Stacks" ][0]["StackStatus"] if status not in ("CREATE_COMPLETE", "UPDATE_COMPLETE"): raise RuntimeError( f"CloudFormation stack '{self.STACK_NAME}' did not reach a " f"successful state (status: {status}). Check the stack events " "in the CloudFormation console for the failure reason." ) response = self.cf_client.describe_stacks(StackName=self.STACK_NAME) outputs = response["Stacks"][0].get("Outputs", list()) vpc_endpoint_id = None for output in outputs: if output["OutputKey"] == "VpcEndpointId": vpc_endpoint_id = output["OutputValue"] break if vpc_endpoint_id is None: raise RuntimeError( "VpcEndpointId not found in stack outputs. " "Check the CloudFormation stack for errors." ) self.vpc_endpoint_id = vpc_endpoint_id print(f"\nStack deployed. VPC Endpoint ID: {vpc_endpoint_id}") return vpc_endpoint_id def create_and_verify_log_group(self) -> None: """ Prompts the user for a log group name, creates it, and verifies it with DescribeLogGroups. """ print("\n" + "=" * 68) print("Step 2: Create and verify a log group") print("=" * 68) name_pattern = re.compile(r"^[a-zA-Z0-9_\-/.#]{1,512}$") while True: log_group_name = q.ask( "\nEnter a name for the syslog log group (e.g. /syslog/demo): " ) if name_pattern.match(log_group_name): break print("Invalid name. Use 1-512 characters: a-z, A-Z, 0-9, " "_, -, /, ., #") self.log_group_name = log_group_name self.wrapper.create_log_group(log_group_name) log_groups = self.wrapper.describe_log_groups( log_group_name_prefix=log_group_name ) for lg in log_groups: if lg.get("logGroupName") == log_group_name: creation_ms = lg.get("creationTime", 0) creation_dt = datetime.fromtimestamp( creation_ms / 1000, tz=timezone.utc ) self.log_group_arn = lg.get("logGroupArn", "") print(f"\nLog group verified:") print(f" Name: {lg['logGroupName']}") print(f" ARN: {lg.get('arn', 'N/A')}") print(f" Created: {creation_dt.strftime('%Y-%m-%d %H:%M:%S UTC')}") return print(f"\nWARNING: Log group '{log_group_name}' not found in describe results.") def add_resource_policy(self) -> None: """ Adds a resource policy to the log group so the syslog service can write to it. """ print("\n" + "=" * 68) print("Step 3: Add a resource policy for syslog ingestion") print("=" * 68) if self.log_group_arn is None or self.vpc_endpoint_id is None: print( "WARNING: Cannot add resource policy - missing log group ARN or VPC endpoint." ) return self.wrapper.put_resource_policy( log_group_arn=self.log_group_arn, vpc_endpoint_id=self.vpc_endpoint_id, ) print( "\nResource policy created. The " "syslog.logs.amazonaws.com service principal can now call " "logs:PutLogEvents and logs:CreateLogStream on the log group." ) # ------------------------------------------------------------------ # Create syslog configuration phase # ------------------------------------------------------------------ def create_syslog_configuration(self) -> None: """ Creates a syslog configuration that associates the VPC endpoint with the log group. """ print("\n" + "=" * 68) print("Step 4: Create the syslog configuration") print("=" * 68) print( "\nAssociating the VPC endpoint with the log group so that syslog " "messages arriving at the endpoint are stored in the log group." ) print( "CloudWatch Logs parses common syslog formats (RFC 5424, " "RFC 3164, Cisco FTD/ASA) and extracts fields such as " "facility, severity, hostname, and appName.\n" ) self.wrapper.put_syslog_configuration( log_group_identifier=self.log_group_name, vpc_endpoint_id=self.vpc_endpoint_id, ) print( f"\nSyslog ingestion is now enabled for log group " f"'{self.log_group_name}' through VPC endpoint " f"'{self.vpc_endpoint_id}'." ) print("\nSupported transport options:") print(" - TCP with TLS on port 6514 (recommended)") print(" - Plaintext TCP on port 1514") print(" - UDP on port 514") # ------------------------------------------------------------------ # List syslog configurations phase # ------------------------------------------------------------------ def _display_configurations(self, configs: list) -> None: """Helper to display a list of syslog configurations.""" if not configs: print(" (no configurations found)") return for cfg in configs: created_ms = cfg.get("createdAt", 0) created_dt = datetime.fromtimestamp(created_ms / 1000, tz=timezone.utc) print(f" Log Group ARN: {cfg.get('logGroupArn', 'N/A')}") print(f" VPC Endpoint ID: {cfg.get('vpcEndpointId', 'N/A')}") print(f" Source Type: {cfg.get('sourceType', 'N/A')}") print(f" Created At: {created_dt.strftime('%Y-%m-%d %H:%M:%S UTC')}") print() def list_all_configurations(self) -> None: """Lists all syslog configurations in the account (no filter).""" print("\n" + "=" * 68) print("Step 5a: List all syslog configurations (no filter)") print("=" * 68) configs = self.wrapper.list_syslog_configurations() print(f"\nAll syslog configurations ({len(configs)} found):\n") self._display_configurations(configs) def list_configurations_by_log_group(self) -> None: """Lists syslog configurations filtered by the scenario's log group.""" print("\n" + "=" * 68) print("Step 5b: List configurations filtered by log group") print("=" * 68) configs = self.wrapper.list_syslog_configurations( log_group_identifier=self.log_group_name, ) print( f"\nConfigurations for log group '{self.log_group_name}' " f"({len(configs)} found):\n" ) self._display_configurations(configs) def list_configurations_by_vpc_endpoint(self) -> None: """Lists syslog configurations filtered by VPC endpoint.""" print("\n" + "=" * 68) print("Step 5c: List configurations filtered by VPC endpoint") print("=" * 68) configs = self.wrapper.list_syslog_configurations( vpc_endpoint_id=self.vpc_endpoint_id, ) print( f"\nConfigurations for VPC endpoint '{self.vpc_endpoint_id}' " f"({len(configs)} found):\n" ) self._display_configurations(configs) # ------------------------------------------------------------------ # Cleanup phase # ------------------------------------------------------------------ def cleanup(self, prompt: bool = True) -> None: """ Cleans up the resources created during the scenario. Tolerates resources that were never created or are already gone, so it is safe to call after a partial failure. :param prompt: When True (the normal end-of-scenario path), ask the user to confirm before deleting. When False (best-effort cleanup after an error), skip the prompt and delete whatever was created. """ print("\n" + "=" * 68) print("Cleanup") print("=" * 68) if prompt: do_cleanup = q.ask( "\nDo you want to delete all resources created during this " "scenario? (y/n) ", q.is_yesno, ) # q.is_yesno converts the answer to a bool (True when the user # answered 'y'); check it directly rather than calling string methods. if not do_cleanup: print("Skipping cleanup. Resources remain in your account.") return else: print( "\nAn error occurred; attempting best-effort cleanup of any " "resources that were created..." ) # Delete syslog configuration if self.log_group_name and self.vpc_endpoint_id: print( f"\nDeleting syslog configuration for log group " f"'{self.log_group_name}'..." ) self.wrapper.delete_syslog_configuration( log_group_identifier=self.log_group_name, vpc_endpoint_id=self.vpc_endpoint_id, ) print( " After deletion, syslog data is no longer ingested through " "the VPC endpoint into this log group." ) # Verify deletion configs = self.wrapper.list_syslog_configurations( log_group_identifier=self.log_group_name, ) if not configs: print(" Verified: configuration has been removed.") else: print( f" WARNING: Still found {len(configs)} configuration(s). " "They may take a moment to be fully removed." ) # Delete log group if self.log_group_name: print(f"\nDeleting log group '{self.log_group_name}'...") self.wrapper.delete_log_group(self.log_group_name) print( " The log group and all archived log events have been " "permanently deleted." ) # Delete CloudFormation stack if self.stack_deployed: print(f"\nDeleting CloudFormation stack '{self.STACK_NAME}'...") try: self.cf_client.delete_stack(StackName=self.STACK_NAME) waiter = self.cf_client.get_waiter("stack_delete_complete") waiter.wait( StackName=self.STACK_NAME, WaiterConfig={"Delay": 30, "MaxAttempts": 40}, ) print(" Stack deleted successfully.") except (ClientError, WaiterError) as error: logger.warning("Stack deletion issue: %s", error) print( f" WARNING: Stack deletion may still be in progress. " f"Check the CloudFormation console." ) # ------------------------------------------------------------------ # Run the full scenario # ------------------------------------------------------------------ def run_scenario(self) -> None: """Runs all phases of the syslog ingestion scenario.""" print("\n" + "=" * 68) print("CloudWatch Logs Syslog Ingestion Scenario") print("=" * 68) print( "\nThis scenario demonstrates CloudWatch Logs managed syslog " "ingestion, which lets you route syslog data from firewalls, " "routers, switches, and Linux servers directly into CloudWatch " "Logs through a VPC endpoint - without running a separate " "collection tier.\n" ) succeeded = False try: # Setup self.deploy_cfn_stack() self.create_and_verify_log_group() self.add_resource_policy() # Create syslog configuration self.create_syslog_configuration() q.ask("\nPress Enter to continue to listing configurations...") # List syslog configurations self.list_all_configurations() self.list_configurations_by_log_group() self.list_configurations_by_vpc_endpoint() succeeded = True # On success, prompt before cleaning up. self.cleanup(prompt=True) finally: # On failure, run best-effort cleanup without prompting so that # partially-created resources are not left behind. if not succeeded: self.cleanup(prompt=False) print("\n" + "=" * 68) print("Scenario complete!") print("=" * 68) print( "\nYou have successfully demonstrated CloudWatch Logs managed " "syslog ingestion:\n" " - Deployed a CloudFormation stack with a syslog VPC endpoint.\n" " - Created a log group and authorized the syslog service.\n" " - Associated a syslog configuration with PutSyslogConfiguration.\n" " - Listed configurations with and without filters.\n" " - Cleaned up all resources.\n" ) ``` Create a class that wraps CloudWatch Logs operations. ``` class CloudWatchLogsWrapper: """Encapsulates Amazon CloudWatch Logs actions.""" def __init__(self, logs_client: boto3.client) -> None: """ Initializes the CloudWatchLogsWrapper with a CloudWatch Logs client. :param logs_client: A Boto3 CloudWatch Logs client. """ self.logs_client = logs_client @classmethod def from_client(cls) -> "CloudWatchLogsWrapper": """ Creates a CloudWatchLogsWrapper instance with a default CloudWatch Logs client. :return: An instance of CloudWatchLogsWrapper. """ logs_client = boto3.client("logs") return cls(logs_client) ``` + For API details, see the following topics in *Amazon SDK for Python (Boto3) API Reference*. + [CreateLogGroup](https://docs.amazonaws.cn/goto/boto3/logs-2014-03-28/CreateLogGroup) + [DeleteLogGroup](https://docs.amazonaws.cn/goto/boto3/logs-2014-03-28/DeleteLogGroup) + [DeleteSyslogConfiguration](https://docs.amazonaws.cn/goto/boto3/logs-2014-03-28/DeleteSyslogConfiguration) + [DescribeLogGroups](https://docs.amazonaws.cn/goto/boto3/logs-2014-03-28/DescribeLogGroups) + [ListSyslogConfigurations](https://docs.amazonaws.cn/goto/boto3/logs-2014-03-28/ListSyslogConfigurations) + [PutResourcePolicy](https://docs.amazonaws.cn/goto/boto3/logs-2014-03-28/PutResourcePolicy) + [PutSyslogConfiguration](https://docs.amazonaws.cn/goto/boto3/logs-2014-03-28/PutSyslogConfiguration) ------ For a complete list of Amazon SDK developer guides and code examples, see [Using CloudWatch Logs with an Amazon SDK](sdk-general-information-section.md). This topic also includes information about getting started and details about previous SDK versions.