Python AWS Lambda Updates (#237)

* Move secrets and extra headers to BaseAgent

* Update options, support service env var

* Handle None service_name

* Add more Trigger lookup safeties

* Fix metrics->plugins reporting

* Span Kind, Service name updates with tests

* Add AWS Lambda layer release instructions

* A logger just for AWS

* Fargate tests are for another PR

* Remove infeasible test

* Better status code logging

* Linter updates

Author: pglombardo

RELEASE.md

@@ -1,5 +1,7 @@
 # Release Steps
 
+## PyPI
+
 _Note: To release a new Instana package, you must be a project member of the [Instana package project on Pypi](https://pypi.org/project/instana/).
 Contact [Peter Giacomo Lombardo](https://github.com/pglombardo) to be added._
 
@@ -12,3 +14,11 @@ Contact [Peter Giacomo Lombardo](https://github.com/pglombardo) to be added._
 7. Validate the new release on https://pypi.org/project/instana/
 8. Update Python documentation with latest changes: https://docs.instana.io/ecosystem/python/
 9. Publish the draft release on [Github](https://github.com/instana/python-sensor/releases)
+
+## AWS Lambda Layer
+
+To release a new AWS Lambda layer, see `bin/lambda_build_publish_layer.py`.
+
+./bin/lambda_build_publish_layer.py [-dev|-prod]
+
+This script assumes you have the AWS CLI tools installed and credentials already configured.

instana/agent.py

@@ -39,11 +39,15 @@ def __init__(self, **kwds):
 
 class BaseAgent(object):
     """ Base class for all agent flavors """
-    client = requests.Session()
+    client = None
     sensor = None
+    secrets_matcher = 'contains-ignore-case'
+    secrets_list = ['key', 'pass', 'secret']
+    extra_headers = None
+    options = None
 
     def __init__(self):
-        pass
+        self.client = requests.Session()
 
 
 class StandardAgent(BaseAgent):
@@ -68,9 +72,6 @@ class StandardAgent(BaseAgent):
     last_seen = None
     last_fork_check = None
     _boot_pid = os.getpid()
-    extra_headers = None
-    secrets_matcher = 'contains-ignore-case'
-    secrets_list = ['key', 'password', 'secret']
     should_threads_shutdown = threading.Event()
 
     def __init__(self):
@@ -147,7 +148,7 @@ def set_from(self, json_string):
         @param json_string: source identifiers
         @return: None
         """
-        if type(json_string) is bytes:
+        if isinstance(json_string, bytes):
             raw_json = json_string.decode("UTF-8")
         else:
             raw_json = json_string
@@ -353,10 +354,7 @@ def __init__(self):
         self.options = AWSLambdaOptions()
         self.report_headers = None
         self._can_send = False
-        self.extra_headers = None
-
-        if "INSTANA_EXTRA_HTTP_HEADERS" in os.environ:
-            self.extra_headers = str(os.environ["INSTANA_EXTRA_HTTP_HEADERS"]).lower().split(';')
+        self.extra_headers = self.options.extra_http_headers
 
         if self._validate_options():
             self._can_send = True
@@ -394,7 +392,7 @@ def report_data_payload(self, payload):
                 self.report_headers["X-Instana-Key"] = self.options.agent_key
                 self.report_headers["X-Instana-Time"] = str(round(time.time() * 1000))
 
-            logger.debug("using these headers: %s", self.report_headers)
+            # logger.debug("using these headers: %s", self.report_headers)
 
             if 'INSTANA_DISABLE_CA_CHECK' in os.environ:
                 ssl_verify = False
@@ -407,7 +405,10 @@ def report_data_payload(self, payload):
                                         timeout=self.options.timeout,
                                         verify=ssl_verify)
 
-            logger.debug("report_data_payload: response.status_code is %s", response.status_code)
+            if 200 <= response.status_code < 300:
+                logger.debug("report_data_payload: Instana responded with status code %s", response.status_code)
+            else:
+                logger.info("report_data_payload: Instana responded with status code %s", response.status_code)
         except Exception as e:
             logger.debug("report_data_payload: connection error (%s)", type(e))
         finally:

instana/collector.py

@@ -85,8 +85,10 @@ def collect_snapshot(self, event, context):
         self.event = event
 
         try:
-            self.snapshot_data["plugins"]["name"] = "com.instana.plugin.aws.lambda"
-            self.snapshot_data["plugins"]["entityId"] = self.context.invoked_function_arn
+            plugin_data = dict()
+            plugin_data["name"] = "com.instana.plugin.aws.lambda"
+            plugin_data["entityId"] = self.context.invoked_function_arn
+            self.snapshot_data["plugins"] = [plugin_data]
         except:
             logger.debug("collect_snapshot error", exc_info=True)
         finally:

instana/instrumentation/aws/triggers.py

@@ -35,9 +35,10 @@ def is_cloudwatch_trigger(event):
 
 
 def is_cloudwatch_logs_trigger(event):
-    if "awslogs" in event and event["awslogs"] != None:
+    if hasattr(event, 'get') and event.get("awslogs", False) is not False:
         return True
-    return False
+    else:
+        return False
 
 
 def is_s3_trigger(event):
@@ -61,19 +62,26 @@ def read_http_query_params(event):
     @param event: lambda event dict
     @return: String in the form of "a=b&c=d"
     """
-    if event is None or type(event) is not dict:
-        return ""
-
     params = []
-    if 'multiValueQueryStringParameters' in event and event['multiValueQueryStringParameters'] is not None:
-        for key in event['multiValueQueryStringParameters']:
-            params.append("%s=%s" % (key, event['multiValueQueryStringParameters'][key]))
-        return "&".join(params)
-    elif 'queryStringParameters' in event and event['queryStringParameters'] is not None:
-        for key in event['queryStringParameters']:
-            params.append("%s=%s" % (key, event['queryStringParameters'][key]))
-        return "&".join(params)
-    else:
+    try:
+        if event is None or type(event) is not dict:
+            return ""
+
+        mvqsp = event.get('multiValueQueryStringParameters', None)
+        qsp = event.get('queryStringParameters', None)
+
+        if mvqsp is not None and type(mvqsp) is dict:
+            for key in mvqsp:
+                params.append("%s=%s" % (key, mvqsp[key]))
+            return "&".join(params)
+        elif qsp is not None and type(qsp) is dict:
+            for key in qsp:
+                params.append("%s=%s" % (key, qsp[key]))
+            return "&".join(params)
+        else:
+            return ""
+    except:
+        logger.debug("read_http_query_params: ", exc_info=True)
         return ""
 
 
@@ -87,10 +95,16 @@ def capture_extra_headers(event, span, extra_headers):
     @param extra_headers: a list of http headers to capture
     @return: None
     """
-    for custom_header in extra_headers:
-        for key in event["headers"]:
-            if key.lower() == custom_header.lower():
-                span.set_tag("http.%s" % custom_header, event["headers"][key])
+    try:
+        event_headers = event.get("headers", None)
+
+        if event_headers is not None:
+            for custom_header in extra_headers:
+                for key in event_headers:
+                    if key.lower() == custom_header.lower():
+                        span.set_tag("http.%s" % custom_header, event_headers[key])
+    except:
+        logger.debug("capture_extra_headers: ", exc_info=True)
 
 
 def enrich_lambda_span(agent, span, event, context):
@@ -109,6 +123,10 @@ def enrich_lambda_span(agent, span, event, context):
         span.set_tag('lambda.name', context.function_name)
         span.set_tag('lambda.version', context.function_version)
 
+        if event is None or type(event) is not dict:
+            logger.debug("enrich_lambda_span: bad event %s", type(event))
+            return
+
         if is_api_gateway_proxy_trigger(event):
             span.set_tag('lambda.trigger', 'aws:api.gateway')
             span.set_tag('http.method', event["httpMethod"])
@@ -204,6 +222,5 @@ def enrich_lambda_span(agent, span, event, context):
                 for item in event["Records"][:3]:
                     events.append({'queue': item['eventSourceARN']})
                 span.set_tag('lambda.sqs.messages', events)
-
     except:
         logger.debug("enrich_lambda_span: ", exc_info=True)

instana/log.py

@@ -10,7 +10,7 @@ def get_standard_logger():
     """
     Retrieves and configures a standard logger for the Instana package
 
-    :return: Logger
+    @return: Logger
     """
     standard_logger = logging.getLogger("instana")
 
@@ -26,12 +26,28 @@ def get_standard_logger():
     return standard_logger
 
 
+def get_aws_lambda_logger():
+    """
+    Retrieves the preferred logger for AWS Lambda
+
+    @return: Logger
+    """
+    aws_lambda_logger = logging.getLogger()
+
+    if "INSTANA_DEBUG" in os.environ:
+        aws_lambda_logger.setLevel(logging.DEBUG)
+    else:
+        aws_lambda_logger.setLevel(logging.WARN)
+
+    return aws_lambda_logger
+
+
 def running_in_gunicorn():
     """
     Determines if we are running inside of a gunicorn process and that the gunicorn logging package
     is available.
 
-    :return:  Boolean
+    @return:  Boolean
     """
     process_check = False
     package_check = False
@@ -70,5 +86,7 @@ def running_in_gunicorn():
 
 if running_in_gunicorn():
     logger = logging.getLogger("gunicorn.error")
+elif os.environ.get("INSTANA_ENDPOINT_URL", False):
+    logger = get_aws_lambda_logger()
 else:
     logger = get_standard_logger()

instana/options.py

@@ -4,22 +4,36 @@
 
 from .util import determine_service_name
 
-class StandardOptions(object):
-    """ Configurable option bits for this package """
-    service = None
+
+class BaseOptions(object):
     service_name = None
-    agent_host = None
-    agent_port = None
+    extra_http_headers = None
     log_level = logging.WARN
     debug = None
 
+    def __init__(self, **kwds):
+        try:
+            if "INSTANA_DEBUG" in os.environ:
+                self.log_level = logging.DEBUG
+                self.debug = True
+            if "INSTANA_EXTRA_HTTP_HEADERS" in os.environ:
+                self.extra_http_headers = str(os.environ["INSTANA_EXTRA_HTTP_HEADERS"]).lower().split(';')
+        except:
+            pass
+
+        self.__dict__.update(kwds)
+
+
+class StandardOptions(BaseOptions):
+    """ Configurable option bits for this package """
     AGENT_DEFAULT_HOST = "localhost"
     AGENT_DEFAULT_PORT = 42699
 
+    agent_host = None
+    agent_port = None
+
     def __init__(self, **kwds):
-        if "INSTANA_DEBUG" in os.environ:
-            self.log_level = logging.DEBUG
-            self.debug = True
+        super(StandardOptions, self).__init__()
 
         self.service_name = determine_service_name()
         self.agent_host = os.environ.get("INSTANA_AGENT_HOST", self.AGENT_DEFAULT_HOST)
@@ -28,22 +42,15 @@ def __init__(self, **kwds):
         if type(self.agent_port) is str:
             self.agent_port = int(self.agent_port)
 
-        self.debug = os.environ.get("INSTANA_DEBUG", False)
-        self.__dict__.update(kwds)
-
 
-class AWSLambdaOptions:
+class AWSLambdaOptions(BaseOptions):
     endpoint_url = None
     agent_key = None
     extra_http_headers = None
     timeout = None
-    log_level = logging.WARN
-    debug = None
 
     def __init__(self, **kwds):
-        if "INSTANA_DEBUG" in os.environ:
-            self.log_level = logging.DEBUG
-            self.debug = True
+        super(AWSLambdaOptions, self).__init__()
 
         self.endpoint_url = os.environ.get("INSTANA_ENDPOINT_URL", None)
 
@@ -52,9 +59,7 @@ def __init__(self, **kwds):
             self.endpoint_url = self.endpoint_url[:-1]
 
         self.agent_key = os.environ.get("INSTANA_AGENT_KEY", None)
-
-        self.extra_http_headers = os.environ.get("INSTANA_EXTRA_HTTP_HEADERS", None)
+        self.service_name = os.environ.get("INSTANA_SERVICE_NAME", None)
         self.timeout = os.environ.get("INSTANA_TIMEOUT", 0.5)
         self.log_level = os.environ.get("INSTANA_LOG_LEVEL", None)
 
-        self.__dict__.update(kwds)

instana/recorder.py

@@ -104,7 +104,7 @@ def record_span(self, span):
             source = instana.singletons.agent.get_from_structure()
 
             if span.operation_name in self.REGISTERED_SPANS:
-                json_span = RegisteredSpan(span, source)
+                json_span = RegisteredSpan(span, source, None)
             else:
                 service_name = instana.singletons.agent.options.service_name
                 json_span = SDKSpan(span, source, service_name)
@@ -122,11 +122,11 @@ def record_span(self, span):
         Convert the passed BasicSpan and add it to the span queue
         """
         source = self.agent.get_from_structure()
+        service_name = self.agent.options.service_name
 
         if span.operation_name in self.REGISTERED_SPANS:
-            json_span = RegisteredSpan(span, source)
+            json_span = RegisteredSpan(span, source, service_name)
         else:
-            service_name = self.agent.options.service_name
             json_span = SDKSpan(span, source, service_name)
 
         # logger.debug("Recorded span: %s", json_span)