"""
Read and parse CLI args for the Logs Command and setup the context for running the command
"""
import logging
import boto3
import botocore
from samcli.commands.exceptions import UserException
from samcli.lib.logs.fetcher import LogsFetcher
from samcli.lib.logs.formatter import LogsFormatter, LambdaLogMsgFormatters, JSONMsgFormatter, KeywordHighlighter
from samcli.lib.logs.provider import LogGroupProvider
from samcli.lib.utils.colors import Colored
from samcli.lib.utils.time import to_utc, parse_date
LOG = logging.getLogger(__name__)
class InvalidTimestampError(UserException):
pass
class LogsCommandContext:
"""
Sets up a context to run the Logs command by parsing the CLI arguments and creating necessary objects to be able
to fetch and display logs
This class **must** be used inside a ``with`` statement as follows:
with LogsCommandContext(**kwargs) as context:
context.fetcher.fetch(...)
"""
def __init__(
self, function_name, stack_name=None, filter_pattern=None, start_time=None, end_time=None, output_file=None
):
"""
Initializes the context
Parameters
----------
function_name : str
Name of the function to fetch logs for
stack_name : str
Name of the stack where the function is available
filter_pattern : str
Optional pattern to filter the logs by
start_time : str
Fetch logs starting at this time
end_time : str
Fetch logs up to this time
output_file : str
Write logs to this file instead of Terminal
"""
self._function_name = function_name
self._stack_name = stack_name
self._filter_pattern = filter_pattern
self._start_time = start_time
self._end_time = end_time
self._output_file = output_file
self._output_file_handle = None
# No colors when we write to a file. Otherwise use colors
self._must_print_colors = not self._output_file
self._logs_client = boto3.client("logs")
self._cfn_client = boto3.client("cloudformation")
def __enter__(self):
"""
Performs some basic checks and returns itself when everything is ready to invoke a Lambda function.
Returns
-------
LogsCommandContext
Returns this object
"""
self._output_file_handle = self._setup_output_file(self._output_file)
return self
def __exit__(self, *args):
"""
Cleanup any necessary opened files
"""
if self._output_file_handle:
self._output_file_handle.close()
self._output_file_handle = None
@property
def fetcher(self):
return LogsFetcher(self._logs_client)
@property
def formatter(self):
"""
Creates and returns a Formatter capable of nicely formatting Lambda function logs
Returns
-------
LogsFormatter
"""
formatter_chain = [
LambdaLogMsgFormatters.colorize_errors,
# Format JSON "before" highlighting the keywords. Otherwise, JSON will be invalid from all the
# ANSI color codes and fail to pretty print
JSONMsgFormatter.format_json,
KeywordHighlighter(self._filter_pattern).highlight_keywords,
]
return LogsFormatter(self.colored, formatter_chain)
@property
def start_time(self):
return self._parse_time(self._start_time, "start-time")
@property
def end_time(self):
return self._parse_time(self._end_time, "end-time")
@property
def log_group_name(self):
"""
Name of the AWS CloudWatch Log Group that we will be querying. It generates the name based on the
Lambda Function name and stack name provided.
Returns
-------
str
Name of the CloudWatch Log Group
"""
function_id = self._function_name
if self._stack_name:
function_id = self._get_resource_id_from_stack(self._cfn_client, self._stack_name, self._function_name)
LOG.debug(
"Function with LogicalId '%s' in stack '%s' resolves to actual physical ID '%s'",
self._function_name,
self._stack_name,
function_id,
)
return LogGroupProvider.for_lambda_function(function_id)
@property
def colored(self):
"""
Instance of Colored object to colorize strings
Returns
-------
samcli.commands.utils.colors.Colored
"""
# No colors if we are writing output to a file
return Colored(colorize=self._must_print_colors)
@property
def filter_pattern(self):
return self._filter_pattern
@property
def output_file_handle(self):
return self._output_file_handle
@staticmethod
def _setup_output_file(output_file):
"""
Open a log file if necessary and return the file handle. This will create a file if it does not exist
Parameters
----------
output_file : str
Path to a file where the logs should be written to
Returns
-------
Handle to the opened log file, if necessary. None otherwise
"""
if not output_file:
return None
return open(output_file, "wb")
@staticmethod
def _parse_time(time_str, property_name):
"""
Parse the time from the given string, convert to UTC, and return the datetime object
Parameters
----------
time_str : str
The time to parse
property_name : str
Name of the property where this time came from. Used in the exception raised if time is not parseable
Returns
-------
datetime.datetime
Parsed datetime object
Raises
------
samcli.commands.exceptions.UserException
If the string cannot be parsed as a timestamp
"""
if not time_str:
return None
parsed = parse_date(time_str)
if not parsed:
raise InvalidTimestampError("Unable to parse the time provided by '{}'".format(property_name))
return to_utc(parsed)
@staticmethod
def _get_resource_id_from_stack(cfn_client, stack_name, logical_id):
"""
Given the LogicalID of a resource, call AWS CloudFormation to get physical ID of the resource within
the specified stack.
Parameters
----------
cfn_client
CloudFormation client provided by AWS SDK
stack_name : str
Name of the stack to query
logical_id : str
LogicalId of the resource
Returns
-------
str
Physical ID of the resource
Raises
------
samcli.commands.exceptions.UserException
If the stack or resource does not exist
"""
LOG.debug(
"Getting resource's PhysicalId from AWS CloudFormation stack. StackName=%s, LogicalId=%s",
stack_name,
logical_id,
)
try:
response = cfn_client.describe_stack_resource(StackName=stack_name, LogicalResourceId=logical_id)
LOG.debug("Response from AWS CloudFormation %s", response)
return response["StackResourceDetail"]["PhysicalResourceId"]
except botocore.exceptions.ClientError as ex:
LOG.debug(
"Unable to fetch resource name from CloudFormation Stack: "
"StackName=%s, ResourceLogicalId=%s, Response=%s",
stack_name,
logical_id,
ex.response,
)
# The exception message already has a well formatted error message that we can surface to user
raise UserException(str(ex), wrapped_from=ex.response["Error"]["Code"])