Adding unit test

Object Store/func.py

@@ -7,73 +7,79 @@
 import oci
 import requests
 
+DD_SOURCE = "Oracle Cloud"  # Adding a source name.
+DD_SERVICE = "OCI Logs"  # Adding a service name.
+DD_TIMEOUT = 10 * 60  # Adding a timeout for the Datadog API call.
 
-def handler(ctx, data: io.BytesIO = None):
+logger = logging.getLogger(__name__)
+
+
+def handler(ctx, data: io.BytesIO = None) -> None:
     try:
         body = json.loads(data.getvalue())
-    except (Exception, ValueError) as ex:
-        logging.getLogger().info(str(ex))
+    except Exception as ex:
+        logger.exception(ex)
         return
 
     data = body.get("data", {})
     additional_details = data.get("additionalDetails", {})
 
     namespace = additional_details.get("namespace")
     if not namespace:
-        logging.getLogger().error("No namespace provided")
+        logger.error("No namespace provided")
         return
 
     bucket = additional_details.get("bucketName")
     if not bucket:
-        logging.getLogger().error("No bucket provided")
+        logger.error("No bucket provided")
         return
 
     resource_name = data.get("resourceName")
     if not resource_name:
-        logging.getLogger().error("No obj provided")
+        logger.error("No resource provided")
         return
 
     event_time = body.get("eventTime")
-
-    source = "Oracle Cloud"  # Adding a source name.
-    service = "OCI Logs"     # Adding a service name.
+    if not event_time:
+        logger.error("No eventTime provided")
+        return
 
     datafile = request_one_object(namespace, bucket, resource_name)
-    data = str(datafile, 'utf-8')
+    data = str(datafile, "utf-8")
 
     # Datadog endpoint URL and token to call the REST interface.
     # These are defined in the func.yaml file.
     try:
-        dd_host = os.environ['DATADOG_HOST']
-        dd_token = os.environ['DATADOG_TOKEN']
-        dd_tags = os.environ.get('DATADOG_TAGS', '')
+        dd_host = os.environ["DATADOG_HOST"]
+        dd_token = os.environ["DATADOG_TOKEN"]
+        dd_tags = os.environ.get("DATADOG_TAGS", "")
     except KeyError:
         err_msg = "Could not find environment variables, \
                    please ensure DATADOG_HOST and DATADOG_TOKEN \
                    are set as environment variables."
-        logging.getLogger().error(err_msg)
+        logger.error(err_msg)
 
     for lines in data.splitlines():
-        logging.getLogger().info("lines %s", lines)
+        logger.info("lines %s", lines)
         payload = {}
-        payload.update({"ddsource": source})
+        payload.update({"service": DD_SERVICE})
+        payload.update({"ddsource": DD_SOURCE})
         payload.update({"ddtags": dd_tags})
         payload.update({"host": resource_name})
         payload.update({"time": event_time})
-        payload.update({"service": service})
         payload.update({"event": lines})
 
     try:
-        headers = {'Content-type': 'application/json', 'DD-API-KEY': dd_token}
-        req = requests.post(dd_host, data=json.dumps(payload), headers=headers)
-    except (Exception, ValueError) as ex:
-        logging.getLogger().info(str(ex))
-        return
-
-    logging.getLogger().info(req.text)
+        headers = {"Content-type": "application/json", "DD-API-KEY": dd_token}
+        res = requests.post(dd_host, data=json.dumps(payload), headers=headers,
+                            timeout=DD_TIMEOUT)
+        logger.info(res.text)
+    except Exception:
+        logger.exception("Failed to send log to Datadog")
 
 
-def request_one_object(namespace: str, bucket: str, resource_name: str):
+def request_one_object(namespace: str, bucket: str,
+                       resource_name: str) -> bytes:
     """
     Calls OCI to request object from Object Storage Client and decompress
     """

Object Store/func.yaml

@@ -8,4 +8,4 @@ timeout: 120
 config:
   DATADOG_HOST: https://http-intake.logs.datadoghq.com/v1/input     # DD Log Intake Host
   DATADOG_TOKEN: <DATADOG_API_TOKEN>                                # DD API Token
-  # DATADOG_TAGS: prod                                                # Tags associated with logs
+  # DATADOG_TAGS: "prod:true"                                        # Tags associated with logs

README.md

@@ -1,2 +1,2 @@
-# Oracle_Logs_Integration
-Houses code for OCI's log collection pipeline.
+# Datadog Oracle Logs Integration
+This repository houses code for OCI's log collection pipeline to be forwarded to Datadog.

Service Connector Hub/func.py

@@ -5,21 +5,25 @@
 
 import requests
 
+logger = logging.getLogger(__name__)
 
-def process(body: dict):
+DD_SOURCE = "Oracle Cloud"  # Adding a source name.
+DD_SERVICE = "OCI Logs"  # Adding a service name.
+DD_TIMEOUT = 10 * 60  # Adding a timeout for the Datadog API call.
+
+
+def process(body: dict) -> None:
     data = body.get("data", {})
     source = body.get("source")
     time = body.get("time")
-    dd_source = "oracle_cloud"
-    service = "OCI Logs"
 
     # Get json data, time, and source information
     payload = {}
     payload.update({"source": source})
     payload.update({"time": time})
     payload.update({"data": data})
-    payload.update({"ddsource": dd_source})
-    payload.update({"service": service})
+    payload.update({"ddsource": DD_SERVICE})
+    payload.update({"service": DD_SERVICE})
 
     # Datadog endpoint URL and token to call the REST interface.
     # These are defined in the func.yaml file.
@@ -31,7 +35,7 @@ def process(body: dict):
         err_msg = "Could not find environment variables, \
                    please ensure DATADOG_HOST and DATADOG_TOKEN \
                    are set as environment variables."
-        logging.getLogger().error(err_msg)
+        logger.error(err_msg)
     if dd_tags:
         payload.update({'ddtags': dd_tags})
 
@@ -40,13 +44,14 @@ def process(body: dict):
     # this will be ingested at once.
     try:
         headers = {'Content-type': 'application/json', 'DD-API-KEY': dd_token}
-        x = requests.post(dd_host, data=json.dumps(payload), headers=headers)
-        logging.getLogger().info(x.text)
-    except (Exception, ValueError) as ex:
-        logging.getLogger().error(str(ex))
+        res = requests.post(dd_host, data=json.dumps(payload), headers=headers,
+                            timeout=DD_TIMEOUT)
+        logger.info(res.text)
+    except Exception as ex:
+        logger.exception(ex)
 
 
-def handler(ctx, data: io.BytesIO = None):
+def handler(ctx, data: io.BytesIO = None) -> None:
     """
     This function receives the logging json and invokes the Datadog endpoint
     for ingesting logs. https://docs.cloud.oracle.com/en-us/iaas/Content/Logging/Reference/top_level_logging_format.htm#top_level_logging_format
@@ -62,5 +67,5 @@ def handler(ctx, data: io.BytesIO = None):
         else:
             # Single CloudEvent
             process(body)
-    except (Exception, ValueError) as ex:
-        logging.getLogger().error(str(ex))
+    except Exception as ex:
+        logger.exception(ex)

Service Connector Hub/func.yaml

@@ -7,4 +7,5 @@ memory: 1024
 timeout: 120
 config:
   DATADOG_HOST: https://http-intake.logs.datadoghq.com/v1/input 
-  DATADOG_TOKEN: <DATADOG_API_TOKEN>
+  DATADOG_TOKEN: <DATADOG_API_TOKEN>
+  # DATADOG_TAGS: "prod:true"                                        # Tags associated with logs

Service Connector Hub/tests/test_func.py

@@ -1,4 +1,5 @@
 import os
+
 from io import BytesIO
 from func import handler
 from unittest import TestCase, mock
@@ -47,6 +48,36 @@ def testSimpleData(self, mock_post, ):
                          '"oracle_cloud", "service": "OCI Logs"}'
                          )
 
+    @mock.patch("requests.post")
+    def testSimpleDataTags(self, mock_post, ):
+        """ Test single CloudEvent payload with Tags enabled """
+
+        payload = """
+        {
+            "specversion" : "1.0",
+            "type" : "com.example.someevent",
+            "source" : "/mycontext",
+            "id" : "C234-1234-1234",
+            "time" : "2018-04-05T17:31:00Z",
+            "comexampleextension1" : "value",
+            "comexampleothervalue" : 5,
+            "datacontenttype" : "application/json",
+            "data" : {
+                "appinfoA" : "abc",
+                "appinfoB" : 123,
+                "appinfoC" : true
+            }
+        }
+        """
+        os.environ['DATADOG_TAGS'] = "prod:true"
+        handler(ctx=None, data=to_BytesIO(payload))
+        mock_post.assert_called_once()
+        self.assertEqual(mock_post.mock_calls[0].kwargs['data'],
+                         '{"source": "/mycontext", "time": "2018-04-05T17:31:00Z", "data": '
+                         '{"appinfoA": "abc", "appinfoB": 123, "appinfoC": true}, "ddsource": '
+                         '"oracle_cloud", "service": "OCI Logs", "ddtags": "prod:true"}'
+                         )
+
     @mock.patch("requests.post")
     def testBatchFormat(self, mock_post):
         """ Test batch format case, where we get an array of 'CloudEvents' """