[WIP] More lambda improvements + docs (elastic#1368)

* Set transaction result for error cases in Lambda * Only use HTTP result/outcome for API Gateway * Enable (and fix) server version fetching * Add lambda docs (+elasticapm.get_client docs) * Move faas from metadata to top-level on the transaction * Fix for possibly-missing message attributes (used as headers) * Add API v2 sample payload and update data collection to match * Use nested_key and add docstring * CHANGELOG + experimental tag in docs

Author: basepi

CHANGELOG.asciidoc

@@ -28,6 +28,18 @@ endif::[]
 //
 //[float]
 //===== Bug fixes
+=== Unreleased
+
+// Unreleased changes go here
+// When the next release happens, nest these changes under the "Python Agent version 6.x" heading
+//[float]
+//===== Features
+
+
+[float]
+===== Bug fixes
+
+* Fix some context fields and metadata handling in AWS Lambda support {pull}1368[#1368]
 
 
 [[release-notes-6.x]]

docs/api.asciidoc

@@ -34,6 +34,21 @@ client = Client({'SERVICE_NAME': 'example'}, **defaults)
 NOTE: framework integrations like <<django-support, Django>> and <<flask-support, Flask>>
 instantiate the client automatically.
 
+[float]
+[[api-get-client]]
+===== `elasticapm.get_client()`
+
+[small]#Added in v6.1.0.
+
+Retrieves the `Client` singleton. This is useful for many framework integrations,
+where the client is instantiated automatically.
+
+[source,python]
+----
+client = elasticapm.get_client()
+client.capture_message('foo')
+----
+
 [float]
 [[error-api]]
 ==== Errors

docs/serverless.asciidoc

@@ -0,0 +1,65 @@
+[[lambda-support]]
+=== AWS Lambda Support
+
+experimental::[]
+
+Incorporating Elastic APM into your AWS Lambda functions is easy!
+
+[float]
+[[lambda-installation]]
+==== Installation
+
+First, you need to add `elastic-apm` as a dependency for your python function.
+Depending on your deployment strategy, this could be as easy as adding
+`elastic-apm` to your `requirements.txt` file, or installing it in the directory
+you plan to deploy using pip:
+
+[source,bash]
+----
+$ pip install -t <target_dir> elastic-apm
+----
+
+You should also add the
+https://github.com/elastic/apm-aws-lambda[Elastic AWS Lambda Extension layer]
+to your function.
+
+[float]
+[[lambda-setup]]
+==== Setup
+
+Once the library is included as a dependency in your function, you must
+import the `capture_serverless` decorator and apply it to your handler:
+
+[source,python]
+----
+from elasticapm import capture_serverless
+
+@capture_serverless()
+def handler(event, context):
+ return {"statusCode": r.status_code, "body": "Success!"}
+----
+
+The agent uses environment variables for <<configuration,configuration>>
+
+[source]
+----
+ELASTIC_APM_LAMBDA_APM_SERVER=<apm-server url>
+ELASTIC_APM_SECRET_TOKEN=<apm-server token>
+ELASTIC_APM_SERVICE_NAME=my-awesome-service
+----
+
+Note that the above configuration assumes you're using the Elastic Lambda
+Extension. The agent will automatically send data to the extension at `localhost`,
+and the extension will then send to the APM Server as specified with
+`ELASTIC_APM_LAMBDA_APM_SERVER`.
+
+[float]
+[[lambda-usage]]
+==== Usage
+
+Once the agent is installed and working, spans will be captured for
+<<supported-technologies,supported technologies>>. You can also use
+<<api-capture-span,`capture_span`>> to capture custom spans, and
+you can retrieve the `Client` object for capturing exceptions/messages
+using <<api-get-client,`get_client`>>.
+

docs/set-up.asciidoc

@@ -8,6 +8,7 @@ To get you off the ground, we’ve prepared guides for setting up the Agent with
  * <<aiohttp-server-support,aiohttp>>
  * <<tornado-support,Tornado>>
  * <<starlette-support,Starlette/FastAPI>>
+ * <<lambda-support,AWS Lambda>>
 
 For custom instrumentation, see <<instrumenting-custom-code, Instrumenting Custom Code>>.
 
@@ -19,4 +20,6 @@ include::./aiohttp-server.asciidoc[]
 
 include::./tornado.asciidoc[]
 
-include::./starlette.asciidoc[]
+include::./starlette.asciidoc[]
+
+include::./serverless.asciidoc[]

elasticapm/conf/constants.py

@@ -59,7 +59,7 @@ def _starmatch_to_regex(pattern):
 
 EVENTS_API_PATH = "intake/v2/events"
 AGENT_CONFIG_PATH = "config/v1/agents"
-SERVER_INFO_PATH = "/"
+SERVER_INFO_PATH = ""
 
 TRACE_CONTEXT_VERSION = 0
 TRACEPARENT_HEADER_NAME = "traceparent"

elasticapm/contrib/serverless/aws.py

@@ -39,7 +39,7 @@
 import elasticapm
 from elasticapm.base import Client, get_client
 from elasticapm.conf import constants
-from elasticapm.utils import compat, encoding, get_name_from_func
+from elasticapm.utils import compat, encoding, get_name_from_func, nested_key
 from elasticapm.utils.disttracing import TraceParent
 from elasticapm.utils.logging import get_logger
 
@@ -63,9 +63,6 @@ class capture_serverless(object):
  @capture_serverless()
  def handler(event, context):
  return {"statusCode": r.status_code, "body": "Success!"}
-
- Note: This is an experimental feature, and we may introduce breaking
- changes in the future.
  """
 
  def __init__(self, name=None, **kwargs):
@@ -79,8 +76,6 @@ def __init__(self, name=None, **kwargs):
  kwargs["central_config"] = False
  kwargs["cloud_provider"] = "none"
  kwargs["framework_name"] = "AWS Lambda"
- # TODO this can probably be removed once the extension proxies the serverinfo endpoint
- kwargs["server_version"] = (8, 0, 0)
  if "service_name" not in kwargs:
  kwargs["service_name"] = os.environ["AWS_LAMBDA_FUNCTION_NAME"]
 
@@ -123,10 +118,13 @@ def __enter__(self):
  transaction_type = "request"
  transaction_name = os.environ.get("AWS_LAMBDA_FUNCTION_NAME", self.name)
 
- if "httpMethod" in self.event: # API Gateway
+ self.httpmethod = nested_key(self.event, "requestContext", "httpMethod") or nested_key(
+ self.event, "requestContext", "http", "method"
+ )
+ if self.httpmethod: # API Gateway
  self.source = "api"
  if os.environ.get("AWS_LAMBDA_FUNCTION_NAME"):
- transaction_name = "{} {}".format(self.event["httpMethod"], os.environ["AWS_LAMBDA_FUNCTION_NAME"])
+ transaction_name = "{} {}".format(self.httpmethod, os.environ["AWS_LAMBDA_FUNCTION_NAME"])
  else:
  transaction_name = self.name
  elif "Records" in self.event and len(self.event["Records"]) == 1:
@@ -160,9 +158,6 @@ def __exit__(self, exc_type, exc_val, exc_tb):
  """
  Transaction teardown
  """
- if exc_val:
- self.client.capture_exception(exc_info=(exc_type, exc_val, exc_tb), handled=False)
-
  if self.response and isinstance(self.response, dict):
  elasticapm.set_context(
  lambda: get_data_from_response(self.response, capture_headers=self.client.config.capture_headers),
@@ -180,6 +175,16 @@ def __exit__(self, exc_type, exc_val, exc_tb):
  result = "HTTP {}xx".format(int(status_code) // 100)
  elasticapm.set_transaction_result(result, override=False)
 
+ if exc_val:
+ self.client.capture_exception(exc_info=(exc_type, exc_val, exc_tb), handled=False)
+ if self.source == "api":
+ elasticapm.set_transaction_result("HTTP 5xx", override=False)
+ elasticapm.set_transaction_outcome(http_status_code=500, override=False)
+ elasticapm.set_context({"status_code": 500}, "response")
+ else:
+ elasticapm.set_transaction_result("failure", override=False)
+ elasticapm.set_transaction_outcome(outcome="failure", override=False)
+
  self.client.end_transaction()
 
  try:
@@ -206,15 +211,19 @@ def set_metadata_and_context(self, coldstart):
  if self.source == "api":
  faas["trigger"]["type"] = "http"
  faas["trigger"]["request_id"] = self.event["requestContext"]["requestId"]
+ path = (
+ self.event["requestContext"].get("resourcePath")
+ or self.event["requestContext"]["http"]["path"].split(self.event["requestContext"]["stage"])[-1]
+ )
  service_context["origin"] = {
  "name": "{} {}/{}".format(
- self.event["requestContext"]["httpMethod"],
- self.event["requestContext"]["resourcePath"],
+ self.httpmethod,
+ path,
  self.event["requestContext"]["stage"],
  )
  }
  service_context["origin"]["id"] = self.event["requestContext"]["apiId"]
- service_context["origin"]["version"] = "2.0" if self.event["headers"]["Via"].startswith("2.0") else "1.0"
+ service_context["origin"]["version"] = self.event.get("version", "1.0")
  cloud_context["origin"] = {}
  cloud_context["origin"]["service"] = {"name": "api gateway"}
  cloud_context["origin"]["account"] = {"id": self.event["requestContext"]["accountId"]}
@@ -236,7 +245,7 @@ def set_metadata_and_context(self, coldstart):
  message_context["age"] = int((time.time() * 1000) - int(record["attributes"]["SentTimestamp"]))
  if self.client.config.capture_body in ("transactions", "all") and "body" in record:
  message_context["body"] = record["body"]
- if self.client.config.capture_headers and record["messageAttributes"]:
+ if self.client.config.capture_headers and record.get("messageAttributes"):
  message_context["headers"] = record["messageAttributes"]
  elif self.source == "sns":
  record = self.event["Records"][0]
@@ -262,7 +271,7 @@ def set_metadata_and_context(self, coldstart):
  )
  if self.client.config.capture_body in ("transactions", "all") and "Message" in record["Sns"]:
  message_context["body"] = record["Sns"]["Message"]
- if self.client.config.capture_headers and record["Sns"]["MessageAttributes"]:
+ if self.client.config.capture_headers and record["Sns"].get("MessageAttributes"):
  message_context["headers"] = record["Sns"]["MessageAttributes"]
  elif self.source == "s3":
  record = self.event["Records"][0]
@@ -277,8 +286,6 @@ def set_metadata_and_context(self, coldstart):
  cloud_context["origin"]["region"] = record["awsRegion"]
  cloud_context["origin"]["provider"] = "aws"
 
- metadata["faas"] = faas
-
  metadata["service"] = {}
  metadata["service"]["name"] = os.environ.get("AWS_LAMBDA_FUNCTION_NAME")
  metadata["service"]["framework"] = {"name": "AWS Lambda"}
@@ -295,7 +302,7 @@ def set_metadata_and_context(self, coldstart):
  # This is the one piece of metadata that requires deep merging. We add it manually
  # here to avoid having to deep merge in _transport.add_metadata()
  if self.client._transport._metadata:
- node_name = self.client._transport._metadata.get("service", {}).get("node", {}).get("name")
+ node_name = nested_key(self.client._transport._metadata, "service", "node", "name")
  if node_name:
  metadata["service"]["node"]["name"] = node_name
 
@@ -307,6 +314,8 @@ def set_metadata_and_context(self, coldstart):
 
  elasticapm.set_context(cloud_context, "cloud")
  elasticapm.set_context(service_context, "service")
+ # faas doesn't actually belong in context, but we handle this in to_dict
+ elasticapm.set_context(faas, "faas")
  if message_context:
  elasticapm.set_context(service_context, "message")
  self.client._transport.add_metadata(metadata)
@@ -319,12 +328,13 @@ def get_data_from_request(event, capture_body=False, capture_headers=True):
  result = {}
  if capture_headers and "headers" in event:
  result["headers"] = event["headers"]
- if "httpMethod" not in event:
+ method = nested_key(event, "requestContext", "httpMethod") or nested_key(event, "requestContext", "http", "method")
+ if not method:
  # Not API Gateway
  return result
 
- result["method"] = event["httpMethod"]
- if event["httpMethod"] in constants.HTTP_WITH_BODY and "body" in event:
+ result["method"] = method
+ if method in constants.HTTP_WITH_BODY and "body" in event:
  body = event["body"]
  if capture_body:
  if event.get("isBase64Encoded"):
@@ -362,21 +372,23 @@ def get_url_dict(event):
  Reconstruct URL from API Gateway
  """
  headers = event.get("headers", {})
- proto = headers.get("X-Forwarded-Proto", "https")
- host = headers.get("Host", "")
- path = event.get("path", "")
- port = headers.get("X-Forwarded-Port")
- stage = "/" + event.get("requestContext", {}).get("stage", "")
+ protocol = headers.get("X-Forwarded-Proto", headers.get("x-forwarded-proto", "https"))
+ host = headers.get("Host", headers.get("host", ""))
+ stage = "/" + (nested_key(event, "requestContext", "stage") or "")
+ path = event.get("path", event.get("rawPath", "").split(stage)[-1])
+ port = headers.get("X-Forwarded-Port", headers.get("x-forwarded-port"))
  query = ""
- if event.get("queryStringParameters"):
+ if "rawQueryString" in event:
+ query = event["rawQueryString"]
+ elif event.get("queryStringParameters"):
  query = "?"
  for k, v in compat.iteritems(event["queryStringParameters"]):
  query += "{}={}".format(k, v)
- url = proto + "://" + host + stage + path + query
+ url = protocol + "://" + host + stage + path + query
 
  url_dict = {
  "full": encoding.keyword_field(url),
- "protocol": proto,
+ "protocol": protocol,
  "hostname": encoding.keyword_field(host),
  "pathname": encoding.keyword_field(stage + path),
  }

elasticapm/traces.py

@@ -393,6 +393,9 @@ def to_dict(self) -> dict:
  # only set parent_id if this transaction isn't the root
  if self.trace_parent.span_id and self.trace_parent.span_id != self.id:
  result["parent_id"] = self.trace_parent.span_id
+ # faas context belongs top-level on the transaction
+ if "faas" in self.context:
+ result["faas"] = self.context.pop("faas")
  if self.is_sampled:
  result["context"] = self.context
  return result
@@ -954,7 +957,7 @@ def set_transaction_name(name, override=True):
 def set_transaction_result(result, override=True):
  """
  Sets the result of the transaction. The result could be e.g. the HTTP status class (e.g "HTTP 5xx") for
- HTTP requests, or "success"/"fail" for background tasks.
+ HTTP requests, or "success"/"failure" for background tasks.
 
  :param name: the name of the transaction
  :param override: if set to False, the name is only set if no name has been set before

elasticapm/utils/__init__.py

@@ -198,6 +198,19 @@ def starmatch_to_regex(pattern: str) -> Pattern:
 
 
 def nested_key(d: dict, *args):
+ """
+ Traverses a dictionary for nested keys. Returns `None` if the at any point
+ in the traversal a key cannot be found.
+
+ Example:
+
+ >>> from elasticapm.utils import nested_key
+ >>> d = {"a": {"b": {"c": 0}}}
+ >>> nested_key(d, "a", "b", "c")
+ 0
+ >>> nested_key(d, "a", "b", "d")
+ None
+ """
  for arg in args:
  try:
  d = d[arg]

tests/contrib/serverless/aws_api2_test_data.json

@@ -0,0 +1,35 @@
+{
+ "version": "2.0",
+ "routeKey": "ANY /fetch_all",
+ "rawPath": "/dev/fetch_all",
+ "rawQueryString": "",
+ "headers": {
+ "accept": "*/*",
+ "content-length": "0",
+ "host": "02plqthge2.execute-api.us-east-1.amazonaws.com",
+ "user-agent": "curl/7.64.1",
+ "x-amzn-trace-id": "Root=1-618018c5-763ade2b18f5734547c93e98",
+ "x-forwarded-for": "67.171.184.49",
+ "x-forwarded-port": "443",
+ "x-forwarded-proto": "https"
+ },
+ "requestContext": {
+ "accountId": "627286350134",
+ "apiId": "02plqthge2",
+ "domainName": "02plqthge2.execute-api.us-east-1.amazonaws.com",
+ "domainPrefix": "02plqthge2",
+ "http": {
+ "method": "GET",
+ "path": "/dev/fetch_all",
+ "protocol": "HTTP/1.1",
+ "sourceIp": "67.171.184.49",
+ "userAgent": "curl/7.64.1"
+ },
+ "requestId": "IIjO5hs7PHcEPIA=",
+ "routeKey": "ANY /fetch_all",
+ "stage": "dev",
+ "time": "01/Nov/2021:16:41:41 +0000",
+ "timeEpoch": 1635784901594
+ },
+ "isBase64Encoded": false
+}