{"id":384,"date":"2024-04-14T15:10:10","date_gmt":"2024-04-14T05:10:10","guid":{"rendered":"https:\/\/seanbdurkin.id.au\/pascaliburnus2\/?p=384"},"modified":"2024-04-14T16:28:55","modified_gmt":"2024-04-14T06:28:55","slug":"application-centric-cloudwatch-logging-in-aws-lambda-functions-with-python3-runtimes","status":"publish","type":"post","link":"http:\/\/seanbdurkin.id.au\/pascaliburnus2\/archives\/384","title":{"rendered":"Application centric Cloudwatch logging in AWS Lambda functions with python3+ runtimes"},"content":{"rendered":"\n
\"\"<\/a><\/figure>\n\n\n\n

Abstract<\/h2>\n\n\n\n

Application centric logging is a system where there are one or more components all directing thier log entries to a single logger. In the AWS context, this could mean an application composed of one or more AWS Lamba functions each logging to a single application-wide AWS CloudWatch log stream. By “single”, I mean single to the application, not to each function.<\/p>\n\n\n\n

In lambda functions with python runtimes, the default mode of logging is one log stream per lambda. We can do this via the print() function or the logging module. But there are sometimes situations where mulltiple lambdas are co-operating to solve a larger problem, where the co-operation is synchronous, and where it would be of value to be able to view a unified log stream for events accross multiple lambdas.<\/p>\n\n\n\n

How do we achieve this, with a simple client interface? In this post, I present a solution.<\/p>\n\n\n\n

A Solution<\/h2>\n\n\n\n

Include in the packaging of your AWS Lambda function, the following python 3 script with filename “custom_logging.py”.<\/p>\n\n\n\n

<\/p>\n\n\n

\n################################################################################\n##  CustomLogging class\n################################################################################\nimport boto3\nimport time, sys\nimport logging \n\ndef coerceLoggingType( logType):\n  if (logType is None) or (logType == ''):\n    logType = logging.INFO\n  elif isinstance( logType, str):\n    logType = getattr( logging, logType.upper(), logging.INFO)\n  return logType\n\nglobal stockFormats\nglobal defaultFormat\nglobal levelNames\n\ndefaultFormat = '#standard'\n\nstockFormats = {\n  '#standard': '{level}: {func}: {caller}: ',\n  '#short'   : '{level}: {func}: ',\n  '#simple'  : '{func}: '} \n\nlevelNames = {\n  logging.DEBUG   : 'DEBUG',\n  logging.INFO    : 'INFO',\n  logging.WARNING : 'WARNING',\n  logging.ERROR   : 'ERROR',\n  logging.CRITICAL: 'CRITICAL'}\n\nbotoLoggers = ['boto', 'boto3', 'botocore', 'urllib3']\n\ndef _json_formatter( obj):\n  """Formatter for unserialisable values."""\n  return str( obj)\n\nclass JsonFormatter( logging.Formatter):\n  """AWS Lambda Logging formatter.\n  Formats the log message as a JSON encoded string.  If the message is a\n  dict it will be used directly.  If the message can be parsed as JSON, then\n  the parse d value is used in the output record.\n  """\n  def __init__( self, **kwargs):\n    super( JsonFormatter, self).__init__()\n    self.format_dict = {\n      'timestamp': '%(asctime)s',\n      'level': '%(levelname)s',\n      'location': '%(name)s.%(funcName)s:%(lineno)d'}\n    self.format_dict.update(kwargs)\n    self.default_json_formatter = kwargs.pop( 'json_default', _json_formatter)\n\n  def format( self, record):\n    record_dict = record.__dict__.copy()\n    record_dict['asctime'] = self.formatTime( record)\n    log_dict = {\n      k: v % record_dict\n      for k, v in self.format_dict.items() if v}\n    if isinstance( record_dict['msg'], dict):\n      log_dict['message'] = record_dict['msg']\n    else:\n      log_dict['message'] = record.getMessage()\n    # Attempt to decode the message as JSON, if so, merge it with the\n    # overall message for clarity.\n    try:\n      log_dict['message'] = json.loads( log_dict['message'])\n    except ( TypeError, ValueError):\n      pass\n    if record.exc_info:\n      # Cache the traceback text to avoid converting it multiple times\n      # (it's constant anyway)\n      # from logging.Formatter:format\n      if not record.exc_text:\n        record.exc_text = self.formatException( record.exc_info)\n    if record.exc_text:\n      log_dict['exception'] = record.exc_text\n    json_record = json.dumps( log_dict, default=self.default_json_formatter)\n    if hasattr( json_record, 'decode'):  # pragma: no cover\n      json_record = json_record.decode( 'utf-8')\n    return json_record\n\ndef setupCanonicalLogLevels( logger, level, fmt, formatter_cls=JsonFormatter, boto_level=None, **kwargs):\n  if not isinstance( logger, logging.Logger):\n    raise Exception( 'Wrong class of logger passed to setupCanonicalLogLevels().')\n  if logger is not None:\n    logger.setLevel( level)\n  logging.root.setLevel( level)\n  if fmt is not None:\n    logging.basicConfig( format=fmt)\n    fmtObj = logging.Formatter( fmt)\n  else:\n    fmtObj = None\n  for handler in logging.root.handlers:\n    try:\n      if fmtObj is not None:\n        handler.setFormatter( fmtObj)\n      elif formatter_cls is not None:\n        handler.setFormatter( formatter_cls( **kwargs))\n    except:\n      pass\n  if boto_level is None:\n    boto_level = level\n  for loggerId in botoLoggers:\n    try:\n      logging.getLogger( loggerId).setLevel( boto_level)\n    except:\n      pass\n \n \nclass NullLogger():\n  def __init__( self):\n    pass\n \n  def purge( self):\n    pass\n \n  def log( self, level, msg, withPurge=False):\n    pass\n \n  def debug( self, msg, withPurge=False):\n    pass\n \n  def info( self, msg, withPurge=False):\n    pass\n \n  def warning( self, msg, withPurge=False):\n    pass\n \n  def critical( self, msg, withPurge=False):\n    pass\n \n  def error( self, msg, withPurge=False):\n    pass\n \n  def exception( self, msg, withPurge=False):\n    pass\n \n  def classCode( self):\n    return '#null'\n \n  def isPurgeable( self):\n    return False\n \nclass PrintLogger():\n  def __init__( self, threshold):\n    self.threshold = threshold\n \n  def purge( self):\n    pass\n \n  def log( self, level, msg, withPurge=False):\n    if level >= self.threshold:\n      print( msg)\n \n  def debug( self, msg, withPurge=False):\n    self.log( logging.DEBUG, msg, False)\n \n  def info( self, msg, withPurge=False):\n    self.log( logging.INFO, msg, False)\n \n  def warning( self, msg, withPurge=False):\n    self.log( logging.WARNING, msg, False)\n \n  def critical( self, msg, withPurge=False):\n    self.log( logging.CRITICAL, msg, False)\n \n  def error( self, msg, withPurge=False):\n    self.log( logging.ERROR, msg, False)\n \n  def exception( self, msg, withPurge=False):\n    self.log( logging.ERROR, msg, False)\n \n  def classCode( self):\n    return '#print'\n \n  def isPurgeable( self):\n    return False\n \ndef createPolymorphicLogger( logClass, logGroup, logStream, logLevel = logging.INFO, functionName = None, msgFormat = None):\n  if logClass == 'cloud-watch':\n    return CustomLogging( logGroup, logStream, logLevel, functionName, msgFormat)\n  elif logClass == '#print':\n    return PrintLogger( logLevel)\n  elif (logClass == '#null') or (logClass is None):\n    return NullLogger()\n  elif isinstance( logClass, dict) and ('logging' in logClass):\n    loggingParams    = logClass.get( 'logging', {})\n    cloudWatchParams = loggingParams.get( 'cloud-watch', {})\n    if msgFormat is None:\n      msgFormat = '#mini'\n    actualLogClass  = loggingParams.get( 'class')\n    logGroup     = cloudWatchParams.get( 'group'   , logGroup)\n    logStream    = cloudWatchParams.get( 'stream'  , logStream)\n    logLevel     =    loggingParams.get( 'level'   , logLevel)\n    functionName = cloudWatchParams.get( 'function', functionName)\n    msgFormat    = cloudWatchParams.get( 'format'  , msgFormat)\n    return createLogger( actualLogClass, logGroup, logStream, logLevel, functionName, msgFormat)\n  elif isinstance( logClass, dict) and ('class' in logClass):\n    canonicalLogClassRecord = {'logging': logClass}\n    return createLogger( canonicalLogClassRecord, logGroup, logStream, logLevel, functionName, msgFormat)\n  elif logClass == '#standard-logger':\n    logger = logging.getLogger( name=logStream)\n    if msgFormat is None:\n      msgFormat = '[%(levelname)s] %(message)s'\n    setupCanonicalLogLevels( logger, logLevel, msgFormat, JsonFormatter, logging.ERROR)\n    return logger\n  else:\n    raise Exception( f'Unrecognised log class {logClass}')\n \ndef getClassCode( logger):\n  code = '#null'\n  if isinstance( logger, logging.Logger):\n    code = '#standard-logger'\n  elif logger is not None:\n    try:\n      code = logger.classCode()\n    except:\n      code = '#unrecognised'\n  return code\n \ndef isLoggerPurgeable( logger):\n  result = False\n  if (not isinstance( logger, logging.Logger)) and (logger is not None):\n    try:\n      result = logger.isPurgeable()\n    except:\n      pass\n  return result\n \nclass CustomLogging:\n  def __init__( self, logGroup, logStream, logLevel = logging.INFO, functionName = None, msgFormat = None):\n    """ logGroup is the name of the CloudWatch log group. If none, the messages passes to print.\n        logStream is the name of the stream. It is required. It is a string. There is no embedded date processing.\n        logLevel is one of the logging level constants or its string equivalent. Posts below this level will be swallowed.\n        functionName is the name of the lambda.\n        msgFormat determines the logged message prefix. It is either a format string, a label or a function.\n          If it is a format string, the following substitution identifiers:\n            {level}  The message log level.\n            {func}   The passed functionName\n            {caller} The python caller function name\n          If it is a label, is one of:\n            #standard   - This is the default.\n            #short\n            #simple\n            #mini\n          If it is a function (or callable object), it must be a function that returns a prefix string with\n            the following input parameters in order:\n              level           - passed message level\n              functionName  - constructed function name\n              caller          - invoker caller name\n              logMsg          - passed message\n             \n        EXAMPLE USAGE 1:\n          import custom_logging, logging\n         \n          logger = CustomLogging( '\/aws\/ec2\/prod\/odin', '2022-06-29-MLC_DAILY-143', logging.INFO, 'CoolLambdaFunc', '#mini')\n          logger.info( 'Hello friend! This is an info')\n          logger.error( 'I broke it!')\n          logger.purge()\n       \n        \n        EXAMPLE USAGE 2:\n          import custom_logging, logging\n         \n          logger = CustomLogging( None, None, logging.DEBUG, 'CoolLambdaFunc', '#mini')\n          logger.info( 'This is the same as print')\n      \n        \n        EXAMPLE USAGE 3:\n          import custom_logging, logging\n         \n          logger = CustomLogging( None, None, logging.WARNING, 'CoolLambdaFunc', '{caller} | {level} !! {func}: ')\n          \n       \n        \n        EXAMPLE USAGE 3:\n          import custom_logging, logging\n         \n          def colourMePink( level, functionName, caller, logMsg):\n            if level == logging.DEBUG:\n              prefix = '{level}: {func}: {caller}: '.format( level = sLevel, func = functionName, caller = caller)\n            elif  level == logging.INFO:\n              prefix = ''\n            else:\n              prefix = '{level}: '.format( level = sLevel)\n            return prefix\n         \n          logger = CustomLogging( None, None, logging.INFO, None, colourMePink)\n          \n    """\n    self.logs           = boto3.client( 'logs', region_name='ap-southeast-2')\n    self.logEvents      = []\n    self.functionName = functionName\n    if self.functionName is None:\n      self.functionName = ''\n    self.logGroup       = logGroup\n    self.logStream      = logStream\n    self.msgFormat = msgFormat\n    if self.msgFormat is None:\n      self.msgFormat = defaultFormat\n    if isinstance( self.msgFormat, str) and (self.msgFormat in stockFormats):\n      self.msgFormat = stockFormats[self.msgFormat]\n    elif self.msgFormat == '#mini':\n      self.msgFormat = self._miniFormat\n    self.logLevel       = coerceLoggingType( logLevel)\n    self.sequenceToken  = None\n    self.sequenceTokenIsValid = False\n    self.maxEventsInBuffer = 20\n    self.maxBufferAgeMs = 60000 # 1 minute.\n \n  def _formatMessage( self, caller, logType, logMsg):\n    prefix = ''\n    if caller is None:\n      try:\n        caller = sys._getframe(3).f_code.co_name\n      except:\n        caller = ''\n    sLevel = levelNames.get( logType, str( logType))\n    if isinstance( self.msgFormat, str):\n      prefix = self.msgFormat.format( level = sLevel, func = self.functionName, caller = caller)\n    elif callable( self.msgFormat):\n      prefix = self.msgFormat( logType, self.functionName, caller, logMsg)\n    return prefix + str( logMsg)\n \n  def _miniFormat( self, level, functionName, caller, logMsg):\n    prefix = ''\n    if level >= logging.WARNING:\n      prefix = levelNames[ level] + ': '\n    if functionName != '':\n      prefix = prefix + functionName + ': '\n    return prefix\n \n  def _getSequenceToken( self):\n    self.sequenceToken = None\n    self.sequenceTokenIsValid = True\n    try:\n      response = self.logs.describe_log_streams( logGroupName=self.logGroup, logStreamNamePrefix=self.logStream)\n    except self.logs.exceptions.ResourceNotFoundException:\n      return 'group-not-found'\n    try:\n      if 'uploadSequenceToken' in response['logStreams'][0]:\n        self.sequenceToken = response['logStreams'][0]['uploadSequenceToken']\n      if self.sequenceToken == '':\n        self.sequenceToken = None\n    except:\n      pass\n    if self.sequenceToken is None:\n      return 'stream-not-found-or-virgin-stream'\n    else:\n      return None\n \n  def put( self, logMsg, logType = logging.INFO, withPurge=False, callFunc = None):\n    logType = coerceLoggingType( logType)\n    if self.logLevel <= logType:\n      if self.logGroup is not None:\n        timestamp = int( round( time.time() * 1000))\n        message = self._formatMessage( callFunc, logType, logMsg)\n        logEvent = {'timestamp': timestamp, 'message': message}\n        if self.logLevel == logging.DEBUG:\n         print( message)\n        self.logEvents.append( logEvent)\n        count = len( self.logEvents)\n        if withPurge or \\\n           (count >= self.maxEventsInBuffer) or \\\n           ((count >= 1) and ((timestamp - self.logEvents[0]['timestamp']) >= self.maxBufferAgeMs)):\n          self.purge()\n      else:\n        print( logMsg)\n \n  def classCode( self):\n    return 'cloud-watch'\n \n  def _primitive_put_log_events( self):\n    event_log = {\n      'logGroupName' : self.logGroup,\n      'logStreamName': self.logStream,\n      'logEvents'    : self.logEvents}\n    if self.sequenceToken is not None:\n      event_log['sequenceToken'] = self.sequenceToken\n    try:\n      response = self.logs.put_log_events( **event_log)\n      self.sequenceToken = response.get( 'nextSequenceToken')\n      self.sequenceTokenIsValid = True\n      result = None\n    except self.logs.exceptions.ResourceAlreadyExistsException:\n      self.sequenceTokenIsValid = False\n      result = None\n    except self.logs.exceptions.DataAlreadyAcceptedException:\n      self.sequenceTokenIsValid = False\n      result = None\n    except self.logs.exceptions.InvalidSequenceTokenException:\n      self.sequenceTokenIsValid = False\n      result = 'invalid-sequence-token'\n    except self.logs.exceptions.ResourceNotFoundException:\n      self.sequenceTokenIsValid = True\n      self.sequenceToken = None\n      result = 'stream-not-found'\n    return result\n \n  def _primitive_create_log_stream( self):\n    self.sequenceTokenIsValid = True\n    self.sequenceToken = None\n    try:\n      self.logs.create_log_stream( logGroupName=self.logGroup, logStreamName=self.logStream)\n      result = None\n    except self.logs.exceptions.ResourceAlreadyExistsException:\n      self.sequenceTokenIsValid = False\n      result = None\n    except self.logs.exceptions.ResourceNotFoundException:\n      result = 'group-not-found'\n    return result\n \n  def _primitive_create_log_group( self):\n   self.sequenceTokenIsValid = True\n    self.sequenceToken = None\n    try:\n      self.logs.create_log_group( logGroupName=self.logGroup)\n    except self.logs.exceptions.ResourceAlreadyExistsException:\n      pass\n \n  def _robust_put_log_events( self):\n    status = 'hungry'\n    for tryCount in range( 100):\n      if status == 'group-not-found':\n        self._primitive_create_log_group()\n        status = 'stream-not-found'\n      elif status == 'stream-not-found':\n        status = self._primitive_create_log_stream()\n        if status is None:\n          status = 'hungry'\n      elif status == 'invalid-sequence-token':\n        getSequenceResult = self._getSequenceToken()\n        # getSequenceResult == 'group-not-found' | 'stream-not-found-or-virgin-stream' | None\n        if getSequenceResult == 'group-not-found':\n          status = 'group-not-found'\n        elif getSequenceResult == 'stream-not-found-or-virgin-stream':\n          status = 'stream-not-found'\n        else:\n          status = 'ready'\n      elif status == 'hungry':\n        if not self.sequenceTokenIsValid:\n          status = 'invalid-sequence-token'\n        else:\n          status = 'ready'\n      elif status == 'ready':\n        status = self._primitive_put_log_events()\n        if status is None:\n          status = 'done'\n      if status == 'done':\n        break\n    if status != 'done':\n      raise Exception( 'Failed to post to CloudWatch Logs.')\n \n  def purge( self):\n    if len( self.logEvents) > 0:\n      try:\n        self._robust_put_log_events()\n      except Exception as ex:\n        print( self.logEvents)\n        print( ex)\n      self.logEvents = []\n \n  def log( self, level, msg, withPurge=False):\n    self.put( msg, level, withPurge, None)\n \n  def debug( self, msg, withPurge=False):\n    self.put( msg, logging.DEBUG, withPurge, None)\n \n  def info( self, msg, withPurge=False):\n    self.put( msg, logging.INFO, withPurge, None)\n \n  def warning( self, msg, withPurge=False):\n    self.put( msg, logging.WARNING, withPurge, None)\n \n  def error( self, msg, withPurge=False):\n    self.put( msg, logging.ERROR, withPurge, None)\n \n  def critical( self, msg, callFunc = None):\n    self.put( msg, logging.CRITICAL, True, callFunc)\n \n  def exception( self, msg, withPurge=True):\n    self.log( logging.ERROR, msg, True)\n \n  def isPurgeable( self):\n    return True\n \n  def __del__( self):\n    try:\n      self.purge()\n    except:\n      pass\n<\/pre><\/div>\n\n\n
How to use<\/h2>\n\n\n\n
Import custom_logging. In your lambda code, where you need application-centric logging, invoke the factory method createPolymorphicLogger()<\/code> to create a logger. Then send all your application-centric log events to this logger, instead of print().<\/p>\n\n\n\n
The logger is going to have the following public methods.<\/p>\n\n\n\n