Lambda improvements (elastic#1489)

* Fix service_name handling * Replace service.id with faas.id * Update transaction.name to use API resource * Respect service_version if configured explicitly * API gateway service.origin.name * faas.name and faas.version * Fix sqs/sns messageattributes -> headers * Add flushed=True to manual flushes in transport * Create the Client object and instrument lazily * Fix client creation timing * CHANGELOG

Author: basepi

CHANGELOG.asciidoc

@@ -42,6 +42,7 @@ endif::[]
 ===== Bug fixes
 
 * Fix Sanic integration to properly respect the `capture_body` config {pull}1485[#1485]
+* Lambda fixes to align with the cross-agent spec {pull}1489[#1489]
 
 
 [[release-notes-6.x]]

elasticapm/contrib/serverless/aws.py

@@ -38,7 +38,7 @@
 from typing import Optional
 
 import elasticapm
-from elasticapm.base import Client, get_client
+from elasticapm.base import Client
 from elasticapm.conf import constants
 from elasticapm.utils import encoding, get_name_from_func, nested_key
 from elasticapm.utils.disttracing import TraceParent
@@ -66,25 +66,25 @@ def handler(event, context):
  return {"statusCode": r.status_code, "body": "Success!"}
  """
 
- def __init__(self, name: Optional[str] = None, **kwargs) -> None:
+ def __init__(self, name: Optional[str] = None, elasticapm_client: Optional[Client] = None, **kwargs) -> None:
  self.name = name
  self.event = {}
  self.context = {}
  self.response = None
+ self.instrumented = False
+ self.client = elasticapm_client # elasticapm_client is intended for testing only
 
  # Disable all background threads except for transport
  kwargs["metrics_interval"] = "0ms"
  kwargs["central_config"] = False
  kwargs["cloud_provider"] = "none"
  kwargs["framework_name"] = "AWS Lambda"
- if "service_name" not in kwargs:
+ if "service_name" not in kwargs and "ELASTIC_APM_SERVICE_NAME" not in os.environ:
  kwargs["service_name"] = os.environ["AWS_LAMBDA_FUNCTION_NAME"]
+ if "service_version" not in kwargs and "ELASTIC_APM_SERVICE_VERSION" not in os.environ:
+ kwargs["service_version"] = os.environ.get("AWS_LAMBDA_FUNCTION_VERSION")
 
- self.client = get_client()
- if not self.client:
- self.client = Client(**kwargs)
- if not self.client.config.debug and self.client.config.instrument and self.client.config.enabled:
- elasticapm.instrument()
+ self.client_kwargs = kwargs
 
  def __call__(self, func):
  self.name = self.name or get_name_from_func(func)
@@ -96,6 +96,21 @@ def decorated(*args, **kwds):
  self.event, self.context = args
  else:
  self.event, self.context = {}, {}
+ # We delay client creation until the function is called, so that
+ # multiple @capture_serverless instances in the same file don't create
+ # multiple clients
+ if not self.client:
+ # Don't use get_client() as we may have a config mismatch due to **kwargs
+ self.client = Client(**self.client_kwargs)
+ if (
+ not self.instrumented
+ and not self.client.config.debug
+ and self.client.config.instrument
+ and self.client.config.enabled
+ ):
+ elasticapm.instrument()
+ self.instrumented = True
+
  if not self.client.config.debug and self.client.config.instrument and self.client.config.enabled:
  with self:
  self.response = func(*args, **kwds)
@@ -124,10 +139,21 @@ def __enter__(self):
  )
  if self.httpmethod: # API Gateway
  self.source = "api"
- if os.environ.get("AWS_LAMBDA_FUNCTION_NAME"):
- transaction_name = "{} {}".format(self.httpmethod, os.environ["AWS_LAMBDA_FUNCTION_NAME"])
+ if nested_key(self.event, "requestContext", "httpMethod"):
+ # API v1
+ resource = "/{}{}".format(
+ nested_key(self.event, "requestContext", "stage"),
+ nested_key(self.event, "requestContext", "resourcePath"),
+ )
  else:
- transaction_name = self.name
+ # API v2
+ route_key = nested_key(self.event, "requestContext", "routeKey")
+ route_key = f"/{route_key}" if route_key.startswith("$") else route_key.split(" ", 1)[-1]
+ resource = "/{}{}".format(
+ nested_key(self.event, "requestContext", "stage"),
+ route_key,
+ )
+ transaction_name = "{} {}".format(self.httpmethod, resource)
  elif "Records" in self.event and len(self.event["Records"]) == 1:
  record = self.event["Records"][0]
  if record.get("eventSource") == "aws:s3": # S3
@@ -203,21 +229,17 @@ def set_metadata_and_context(self, coldstart: bool) -> None:
  faas["coldstart"] = coldstart
  faas["trigger"] = {"type": "other"}
  faas["execution"] = self.context.aws_request_id
+ arn = self.context.invoked_function_arn
+ if len(arn.split(":")) > 7:
+ arn = ":".join(arn.split(":")[:7])
+ faas["id"] = arn
+ faas["name"] = os.environ.get("AWS_LAMBDA_FUNCTION_NAME")
+ faas["version"] = os.environ.get("AWS_LAMBDA_FUNCTION_VERSION")
 
  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.httpmethod,
- self.event["requestContext"]["stage"],
- path,
- )
- }
+ service_context["origin"] = {"name": self.event["requestContext"]["domainName"]}
  service_context["origin"]["id"] = self.event["requestContext"]["apiId"]
  service_context["origin"]["version"] = self.event.get("version", "1.0")
  cloud_context["origin"] = {}
@@ -236,13 +258,18 @@ def set_metadata_and_context(self, coldstart: bool) -> None:
  cloud_context["origin"]["region"] = record["awsRegion"]
  cloud_context["origin"]["account"] = {"id": record["eventSourceARN"].split(":")[4]}
  cloud_context["origin"]["provider"] = "aws"
- message_context["queue"] = service_context["origin"]["name"]
+ message_context["queue"] = {"name": service_context["origin"]["name"]}
  if "SentTimestamp" in record["attributes"]:
  message_context["age"] = {"ms": 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.get("messageAttributes"):
- message_context["headers"] = record["messageAttributes"]
+ headers = {}
+ for k, v in record["messageAttributes"].items():
+ if v and v.get("stringValue"):
+ headers[k] = v.get("stringValue")
+ if headers:
+ message_context["headers"] = headers
  elif self.source == "sns":
  record = self.event["Records"][0]
  faas["trigger"]["type"] = "pubsub"
@@ -256,7 +283,7 @@ def set_metadata_and_context(self, coldstart: bool) -> None:
  cloud_context["origin"]["region"] = record["Sns"]["TopicArn"].split(":")[3]
  cloud_context["origin"]["account_id"] = record["Sns"]["TopicArn"].split(":")[4]
  cloud_context["origin"]["provider"] = "aws"
- message_context["queue"] = service_context["origin"]["name"]
+ message_context["queue"] = {"name": service_context["origin"]["name"]}
  if "Timestamp" in record["Sns"]:
  message_context["age"] = {
  "ms": int(
@@ -270,7 +297,12 @@ def set_metadata_and_context(self, coldstart: bool) -> None:
  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"].get("MessageAttributes"):
- message_context["headers"] = record["Sns"]["MessageAttributes"]
+ headers = {}
+ for k, v in record["Sns"]["MessageAttributes"].items():
+ if v and v.get("Type") == "String":
+ headers[k] = v.get("Value")
+ if headers:
+ message_context["headers"] = headers
  elif self.source == "s3":
  record = self.event["Records"][0]
  faas["trigger"]["type"] = "datasource"
@@ -291,11 +323,7 @@ def set_metadata_and_context(self, coldstart: bool) -> None:
  "name": os.environ.get("AWS_EXECUTION_ENV"),
  "version": platform.python_version(),
  }
- arn = self.context.invoked_function_arn
- if len(arn.split(":")) > 7:
- arn = ":".join(arn.split(":")[:7])
- metadata["service"]["id"] = arn
- metadata["service"]["version"] = os.environ.get("AWS_LAMBDA_FUNCTION_VERSION")
+ metadata["service"]["version"] = self.client.config.service_version
  metadata["service"]["node"] = {"configured_name": os.environ.get("AWS_LAMBDA_LOG_STREAM_NAME")}
  # 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()
@@ -315,7 +343,7 @@ def set_metadata_and_context(self, coldstart: bool) -> None:
  # 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")
+ elasticapm.set_context(message_context, "message")
  self.client._transport.add_metadata(metadata)
 
 

elasticapm/transport/base.py

@@ -153,7 +153,8 @@ def _process_queue(self):
 
  queue_size = 0 if buffer.fileobj is None else buffer.fileobj.tell()
 
- if flush:
+ forced_flush = flush
+ if forced_flush:
  logger.debug("forced flush")
  elif timed_out or timeout == 0:
  # update last flush time, as we might have waited for a non trivial amount of time in
@@ -172,7 +173,12 @@ def _process_queue(self):
  flush = True
  if flush:
  if buffer_written:
- self._flush(buffer)
+ self._flush(buffer, forced_flush=forced_flush)
+ elif forced_flush and "/localhost:" in self.client.config.server_url:
+ # No data on buffer, but due to manual flush we should send
+ # an empty payload with flushed=true query param, but only
+ # to a local APM server (or lambda extension)
+ self.send(None, flushed=True)
  self._last_flush = timeit.default_timer()
  buffer = self._init_buffer()
  buffer_written = False
@@ -248,7 +254,7 @@ def _init_event_queue(self, chill_until, max_chill_time):
  else:
  return _queue.Queue(maxsize=10000)
 
- def _flush(self, buffer):
+ def _flush(self, buffer, forced_flush=False):
  """
  Flush the queue. This method should only be called from the event processing queue
  :return: None
@@ -262,7 +268,7 @@ def _flush(self, buffer):
  # StringIO on Python 2 does not have getbuffer, so we need to fall back to getvalue
  data = fileobj.getbuffer() if hasattr(fileobj, "getbuffer") else fileobj.getvalue()
  try:
- self.send(data)
+ self.send(data, forced_flush=forced_flush)
  self.handle_transport_success()
  except Exception as e:
  self.handle_transport_fail(e)
@@ -279,7 +285,7 @@ def start_thread(self, pid=None):
  except RuntimeError:
  pass
 
- def send(self, data):
+ def send(self, data, forced_flush=False):
  """
  You need to override this to do something with the actual
  data. Usually - this is sending to a server

elasticapm/transport/http.py

@@ -70,13 +70,15 @@ def __init__(self, url: str, *args, **kwargs) -> None:
  self._http = None
  self._url = url
 
- def send(self, data):
+ def send(self, data, forced_flush=False):
  response = None
 
  headers = self._headers.copy() if self._headers else {}
  headers.update(self.auth_headers)
 
  url = self._url
+ if forced_flush:
+ url = f"{url}?flushed=true"
  try:
  try:
  response = self.http.urlopen(

elasticapm/transport/http_base.py

@@ -59,7 +59,7 @@ def __init__(
  self._server_info_url = "".join((base, constants.SERVER_INFO_PATH, tail))
  super(HTTPTransportBase, self).__init__(client, compress_level=compress_level, **kwargs)
 
- def send(self, data):
+ def send(self, data, forced_flush=False):
  """
  Sends a request to a remote APM Server using HTTP POST.
 

tests/contrib/serverless/aws_sns_test_data.json

@@ -15,7 +15,20 @@
  "Signature": "YBGjMCe1m0QQ0DIWq4gZLy3/0bEyXhLPZJzeo4JYMa2P9ercshfn9s+x9nqd6HSYfO3RG0ebCmzxddgO8UCmaddXbhhMRWYjsIDv3+OvUitG8+bFqvpH/rQVHdCEWla5l+NDcye6d2cl9zuYFliTIFUsBmFcqbiroyZbIIHOczUpxNKK9oQcAXU6RgIl6y30DBgxYmzdMm4FMXPpden84v0LwVOyfqVm2gmeMnlccEOB0TRMe8sLsv7OfWLA3GBl3b14MOUZfvUz4Btb15ssCq++QVHoTQWZnbJ5dA7P3ljMauQCagub0Zefx7uUmWAlczxe/5kREJt8rEfl+pN7Mg==",
  "SigningCertUrl": "https://sns.us-east-1.amazonaws.com/SimpleNotificationService-010a507c1833636cd94bdb98bd93083a.pem",
  "UnsubscribeUrl": "https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-east-1:268121251715:basepiwstesttopic:4763d52a-56d1-4dc5-99eb-ffb4315587af",
- "MessageAttributes": {}
+ "MessageAttributes": {
+ "Greeting": {
+ "Type": "Binary",
+ "Value": "SGVsbG8sIFdvcmxkIQ=="
+ },
+ "Population": {
+ "Type": "String",
+ "Value": "1250800"
+ },
+ "City": {
+ "Type": "String",
+ "Value": "Any City"
+ }
+ }
  }
  }
  ]

tests/contrib/serverless/aws_sqs_test_data.json

@@ -10,7 +10,26 @@
  "SenderId": "268121251715",
  "ApproximateFirstReceiveTimestamp": "1626973700075"
  },
- "messageAttributes": {},
+ "messageAttributes": {
+ "Greeting": {
+ "binaryValue": "SGVsbG8sIFdvcmxkIQ==",
+ "stringListValues": [],
+ "binaryListValues": [],
+ "dataType": "Binary"
+ },
+ "Population": {
+ "stringValue": "1250800",
+ "stringListValues": [],
+ "binaryListValues": [],
+ "dataType": "Number"
+ },
+ "City": {
+ "stringValue": "Any City",
+ "stringListValues": [],
+ "binaryListValues": [],
+ "dataType": "String"
+ }
+ },
  "md5OfBody": "5eb63bbbe01eeed093cb22bb8f5acdc3",
  "eventSource": "aws:sqs",
  "eventSourceARN": "arn:aws:sqs:us-east-1:268121251715:testqueue",