Add Cloudwatch integration

Author: Victoremepunto

requirements.txt

@@ -1,10 +1,9 @@
 app-common-python>=0.1.9
-aiobotocore==1.4.1
 aiohttp==3.7.4.post0
 aioitertools==0.8.0
 async-timeout==3.0.1
 attrs==21.2.0
-botocore==1.20.106
+botocore<1.22.9
 chardet==4.0.0
 fsspec==2021.8.1
 idna==3.2

setup.py

@@ -14,8 +14,10 @@
     package_dir={'': 'src'},
     packages=find_packages(where='src'),
     python_requires='>=3.6, <4',
-    install_require=[
-        'app-common-python'
+    install_requires=[
+        'app-common-python',
+        'boto3',
+        'watchtower'
     ],
     extras_require={
         'test': ['pytest'],

src/floorist/config.py

@@ -1,3 +1,5 @@
+import logging
+
 from app_common_python import LoadedConfig, ObjectBuckets
 from app_common_python import isClowderEnabled
 from os import environ, access, R_OK
@@ -20,6 +22,8 @@ class Config:
     database_password=attr.ib(default=None)
     database_name=attr.ib(default=None)
     floorplan_filename=attr.ib(default=None)
+    loglevel=attr.ib(default="INFO")
+    cloudwatch_config=attr.ib(factory=dict)
 
 
 def get_config():
@@ -28,6 +32,7 @@ def get_config():
     _set_bucket_config(config)
     _set_database_config(config)
     _set_floorist_config(config)
+    _set_logging_config(config)
     _validate_config(config)
 
     return config
@@ -48,6 +53,7 @@ def _get_bucket_url(endpoint):
     else:
         return f"https://{endpoint}"
 
+
 def get_bucket_requested_name_from_environment():
 
     name = environ.get('AWS_BUCKET')
@@ -84,6 +90,28 @@ def _set_floorist_config(config):
     config.floorplan_filename = environ.get('FLOORPLAN_FILE')
 
 
+def _set_logging_config(config):
+
+    config.loglevel = environ.get('LOGLEVEL', 'INFO').upper()
+
+    if isClowderEnabled():
+        _set_cloudwatch_config_from_clowder(config)
+    else:
+        _set_cloudwatch_config_from_environment(config)
+
+
+def _set_cloudwatch_config_from_clowder(config):
+
+    if LoadedConfig.logging:
+        config.cloudwatch_config = LoadedConfig.logging.cloudwatch
+    else:
+        logging.warning("No Cloudwatch logging config provided by Clowder!")
+
+
+def _set_cloudwatch_config_from_environment(config):
+    pass
+
+
 def _validate_config(config):
 
     if not config.floorplan_filename:

src/floorist/floorist.py

@@ -1,78 +1,114 @@
 from datetime import date
 from uuid import uuid4 as uuid
+
+import watchtower
 from s3fs import S3FileSystem as s3
 from floorist.config import get_config
-from os import environ
+import os
+from boto3.session import Session
 
 import logging
 import pandas.io.sql as sqlio
 import psycopg2
 import yaml
 
 
-def _configure_loglevel():
+def _get_logger(config):
+    logging.basicConfig(level=config.loglevel)
+    logger = logging.getLogger(__name__)
+    logger.setLevel(config.loglevel)
 
-  LOGLEVEL = environ.get('LOGLEVEL', 'INFO').upper()
-  logging.basicConfig(level=LOGLEVEL)
+    if config.cloudwatch_config:
+        logger.info("Configuring Cloudwatch logging")
+        logger.addHandler(_get_cloudwatch_handler(config.cloudwatch_config))
+    else:
+        logger.info("Cloudwatch config not found - skipping")
 
+    return logger
 
-def main():
 
-  _configure_loglevel()
-  config = get_config()
-
-  # Fails if can't connect to S3 or the bucket does not exist
-  s3(secret=config.bucket_secret_key, key=config.bucket_access_key,
-     client_kwargs={'endpoint_url': config.bucket_url }).ls(config.bucket_name)
-  logging.debug('Successfully connected to the S3 bucket')
-
-  conn = psycopg2.connect(
-    host=config.database_hostname,
-    user=config.database_username,
-    password=config.database_password,
-    database=config.database_name
-  )
-  logging.debug('Successfully connected to the database')
-
-  dump_count = 0
-  dumped_count = 0
-
-  with open(config.floorplan_filename, 'r') as stream:
-    # This try block allows us to proceed if a single SQL query fails
-    for row in yaml.safe_load(stream):
-      dump_count += 1
-
-      try:
-        logging.debug(f"Dumping #{dump_count}: {row['query']} to {row['prefix']}")
-
-        data = sqlio.read_sql_query(row['query'], conn)
-        target = '/'.join([
-          f"s3://{config.bucket_name}",
-          row['prefix'],
-          date.today().strftime('year_created=%Y/month_created=%-m/day_created=%-d'),
-          f"{uuid()}.parquet"
-        ])
-
-        data.to_parquet(
-          path=target,
-          compression='gzip',
-          index=False,
-          storage_options={
-            'secret': config.bucket_secret_key,
-             'key' : config.bucket_access_key,
-            'client_kwargs':{'endpoint_url': config.bucket_url }
-          }
-        )
-
-        logging.debug(f"Dumped #{dumped_count}: {row['query']} to {row['prefix']}")
-
-        dumped_count += 1
-      except Exception as ex:
-        logging.exception(ex)
-
-  logging.info(f'Dumped {dumped_count} from total of {dump_count}')
-
-  conn.close()
-
-  if dumped_count != dump_count:
-    exit(1)
+def _get_cloudwatch_handler(config):
+    aws_access_key_id = config.accessKeyId
+    aws_secret_access_key = config.secretAccessKey
+    aws_region_name = config.region
+    aws_log_group = config.logGroup
+    aws_log_stream = os.getenv("AWS_LOG_STREAM", _get_hostname())
+
+    logging.info(f"Configuring watchtower logging (log_group={aws_log_group}, "
+                 f"stream_name={aws_log_stream})")
+    boto3_session = Session(
+        aws_access_key_id=aws_access_key_id,
+        aws_secret_access_key=aws_secret_access_key,
+        region_name=aws_region_name,
+    )
+    handler = watchtower.CloudWatchLogHandler(boto3_session=boto3_session,
+                                              stream_name=aws_log_stream,
+                                              log_group=aws_log_group,
+                                              create_log_group=False)
+
+    return handler
+
+
+def _get_hostname():
+    return os.uname().nodename
+
+
+def main():
+    config = get_config()
+    logger = _get_logger(config)
+
+    # Fails if can't connect to S3 or the bucket does not exist
+    s3(secret=config.bucket_secret_key, key=config.bucket_access_key,
+       client_kwargs={'endpoint_url': config.bucket_url}).ls(config.bucket_name)
+    logger.debug('Successfully connected to the S3 bucket')
+
+    conn = psycopg2.connect(
+        host=config.database_hostname,
+        user=config.database_username,
+        password=config.database_password,
+        database=config.database_name
+    )
+    logger.debug('Successfully connected to the database')
+
+    dump_count = 0
+    dumped_count = 0
+
+    with open(config.floorplan_filename, 'r') as stream:
+        # This try block allows us to proceed if a single SQL query fails
+        for row in yaml.safe_load(stream):
+            dump_count += 1
+
+            try:
+                logger.debug(f"Dumping #{dump_count}: {row['query']} to {row['prefix']}")
+
+                data = sqlio.read_sql_query(row['query'], conn)
+                target = '/'.join([
+                    f"s3://{config.bucket_name}",
+                    row['prefix'],
+                    date.today().strftime('year_created=%Y/month_created=%-m/day_created=%-d'),
+                    f"{uuid()}.parquet"
+                ])
+
+                data.to_parquet(
+                    path=target,
+                    compression='gzip',
+                    index=False,
+                    storage_options={
+                        'secret': config.bucket_secret_key,
+                        'key': config.bucket_access_key,
+                        'client_kwargs': {'endpoint_url': config.bucket_url}
+                    }
+                )
+
+                logger.debug(f"Dumped #{dumped_count}: {row['query']} to {row['prefix']}")
+
+                dumped_count += 1
+            except Exception as ex:
+                logger.exception(ex)
+
+    logger.info(f'Dumped {dumped_count} from total of {dump_count}')
+
+    conn.close()
+
+    if dumped_count != dump_count:
+        exit(1)