add faas attributes

Author: Ankcorn

CHANGELOG.md

@@ -10,6 +10,12 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 - Make sure flushing is not canceled
 
  The latest layer is: `arn:aws:lambda:${your-region-here}:097948374213:layer:baselime-node:9`
+ 
+ The latest layer is: `arn:aws:lambda:${your-region-here}:374211872663:layer:baselime-node:17`
+
+ 
+ The latest layer is: `arn:aws:lambda:${your-region-here}:374211872663:layer:baselime-node:18`
+
 ## [0.1.12] 2023-06-28
 
 - publish via CI

multiRegion.json

@@ -1,16 +1,6 @@
 {
     "regions": [
-        "eu-west-1",
-        "eu-west-2",
-        "us-east-1",
-        "us-east-2",
-        "us-west-2",
-        "eu-central-1",
-        "ap-south-1",
-        "ap-southeast-1",
-        "ap-southeast-2",
-        "ap-northeast-1",
-        "ca-central-1"
+        "eu-west-2"
     ],
     "output": "md"
 }

src/index.ts

@@ -1,5 +1,5 @@
 import api, { trace, context, propagation, Context as OtelContext, ROOT_CONTEXT, Attributes } from "@opentelemetry/api";
-import { Handler } from "aws-lambda";
+import { Handler, DynamoDBStreamEvent, S3Event } from "aws-lambda";
 import { flattenObject } from './utils';
 import { Context } from 'aws-lambda';
 export * as logger from './logger';
@@ -8,29 +8,50 @@ declare const global : {
 	baselimeLambdaFlush: () => Promise<void>;
 }
 
+type FaasDocument = {
+  collection: string
+  operation: string
+  time: string
+  name: string
+}
+let coldstart = true;
+
 export function wrap(handler: Handler) {
   return async (event: any, lambda_context: Context) => {
     const tracer = trace.getTracer('@baselime/baselime-lambda-wrapper', '1');
+    const service = detectService(event);
+    const trigger = triggerToServiceType(service);
+    const parent = determinParent(event, service);
+    let document: FaasDocument | null = null;
+    if(trigger === 'datasource') {
+      if(service === 'dynamodb') {
+        document = getDynamodbStreamDocumentAttributes(event);
+      } 
 
-    const parent = determinParent(event);
+      if(service === 's3') {
+        document = getS3DocumentAttributes(event);
+      }
+    }
     const span = tracer.startSpan(lambda_context.functionName, {
       attributes: flattenObject({
         event,
         context: lambda_context,
         faas: {
           execution: lambda_context.awsRequestId,
-          name: lambda_context.functionName,
           runtime: 'nodejs',
+          trigger,
+          document,
+          invoked_by: service,
           id: lambda_context.invokedFunctionArn,
-          
+          coldstart,
         },
         cloud: {
-          account: {
-            id: lambda_context.invokedFunctionArn.split(":")[4],
-          }
+          resource_id: lambda_context.invokedFunctionArn,
+          account_id: lambda_context.invokedFunctionArn.split(":")[4],
         }
       }) as Attributes,
     }, parent);
+    coldstart = false;
     const ctx = trace.setSpan(context.active(), span);
 
     try {
@@ -55,16 +76,56 @@ export function wrap(handler: Handler) {
 
 function detectService(event: any) {
 	if (event.requestContext?.apiId) {
-		return "api";
+		return "api-gateway";
 	}
 
+  if(event.requestContext?.apiId && event.version === "2.0") {
+    return "api-gateway-v2";
+  }
+
 	if (event.Records && event.Records[0]?.EventSource === "aws:sns") {
 		return "sns";
 	}
 
+  if(event.Records && event.Records[0]?.eventSource === "aws:sqs") {
+    return "sqs";
+  }
+
+  if(event.Records && event.Records[0]?.eventSource === "aws:kinesis") {
+    return "kinesis";
+  }
+
+  if(event.Records && event.Records[0]?.eventSource === "aws:dynamodb") {
+    return "dynamodb";
+  }
+
+  if(event.Records && event.Records[0]?.eventSource === "aws:s3") {
+    return "s3";
+  }
+
 	return 'unknown'
 }
 
+function triggerToServiceType(service: string) {
+  switch (service) {
+    case "api":
+    case "api-gateway":
+    case "api-gateway-v2":
+    case "function-url":
+      return "http";
+    case "sns":
+    case "sqs":
+    case "kinesis":
+    case "eventbridge":
+      return "pubsub";
+    case "dynamodb":
+    case "s3":
+      return "datasource"
+    default:
+      return "other";
+  }
+}
+
 const headerGetter = {
 	keys(carrier: Object): string[] {
 		return Object.keys(carrier);
@@ -83,10 +144,10 @@ const snsGetter = {
 	},
 };
 
-function determinParent(event: any): OtelContext {
+function determinParent(event: any, service: string): OtelContext {
   let parent: OtelContext | undefined = undefined;
 
-  const extractedContext = extractContext(event);
+  const extractedContext = extractContext(event, service);
 
   if (trace.getSpan(extractedContext)?.spanContext()) {
     return extractedContext;
@@ -98,9 +159,12 @@ function determinParent(event: any): OtelContext {
   return parent;
 }
 
-function extractContext(event: any) {
-  switch (detectService(event)) {
+function extractContext(event: any, service: string) {
+  switch (service) {
     case "api":
+    case "api-gateway":
+    case "api-gateway-v2":
+    case "function-url":
       const httpHeaders = event.headers || {};
       return propagation.extract(
         api.context.active(),
@@ -116,3 +180,39 @@ function extractContext(event: any) {
   }
   return propagation.extract(api.context.active(), {}, headerGetter);
 }
+
+const DynamodbEventToDocumentOperations = {
+  INSERT: 'insert',
+  MODIFY: 'update',
+  REMOVE: 'delete',
+  default: ''
+};
+
+function getDynamodbStreamDocumentAttributes(event: DynamoDBStreamEvent): FaasDocument {
+  const unixTime = event?.Records[0]?.dynamodb?.ApproximateCreationDateTime || Date.now() / 1000;
+  return { 
+    // TODO we could do better for collection (infer from single table design patterns?)
+    collection: (event?.Records[0]?.eventSourceARN || '').split("/")[1],
+    name: (event?.Records[0]?.eventSourceARN || '').split("/")[1],
+    operation: DynamodbEventToDocumentOperations[event?.Records[0]?.eventName || "default"],
+    time: new Date(unixTime).toUTCString(),
+  }
+}
+
+function getS3DocumentAttributes(event: S3Event): FaasDocument {
+  let operation = 'unkown';
+
+  if(event.Records[0].eventName.startsWith('ObjectCreated')) {
+    operation = 'insert';
+  }
+  
+  if(event.Records[0].eventName.startsWith('ObjectRemoved')) {
+    operation = 'delete';
+  }
+  return {
+    collection: event.Records[0].s3.bucket.name,
+    name: event.Records[0].s3.object.key,
+    operation,
+    time: event.Records[0].eventTime,
+  }
+}

src/lambda-wrapper.ts

@@ -1,7 +1,7 @@
 import api, { Attributes, DiagConsoleLogger, DiagLogLevel } from "@opentelemetry/api";
 import { BatchSpanProcessor } from "@opentelemetry/sdk-trace-base";
 import {
-	OTLPTraceExporter,
+	OTLPTraceExporter, 
 } from "@opentelemetry/exporter-trace-otlp-http";
 import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
 import { Instrumentation, registerInstrumentations } from "@opentelemetry/instrumentation";
@@ -12,7 +12,7 @@ import { HttpInstrumentation } from "@opentelemetry/instrumentation-http";
 import { Resource } from "@opentelemetry/resources";
 import { flattenObject } from "./utils";
 import { existsSync } from "node:fs";
-
+import { arch } from "node:os"
 if (process.env.OTEL_LOG_LEVEL === "debug") {
 	api.diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.ALL);
 }
@@ -21,7 +21,11 @@ const provider = new NodeTracerProvider({
 	resource: new Resource({
 		"service.name": process.env.BASELIME_SERVICE,
 		"faas.name": process.env.AWS_LAMBDA_FUNCTION_NAME,
-		'aws.region': process.env.AWS_REGION || 'unknown',
+		"faas.max_memory": process.env.AWS_LAMBDA_FUNCTION_MEMORY_SIZE,
+		"faas.architecture": arch(),
+		"faas.version": process.env.AWS_LAMBDA_FUNCTION_VERSION,
+		'cloud.region': process.env.AWS_REGION || 'unknown',
+		"cloud.provider": "aws",
 	}),
 });
 
@@ -31,9 +35,13 @@ if (existsSync('/opt/extensions/baselime')) {
 	collectorURL = 'http://sandbox:4323';
 }
 
+enum CompressionAlgorithm {
+	GZIP = "gzip",
+}
 const spanProcessor = new BatchSpanProcessor(
 	new OTLPTraceExporter({
-		url: collectorURL,
+		url: `${collectorURL}/${process.env.AWS_LAMBDA_FUNCTION_NAME}`,
+		compression: CompressionAlgorithm.GZIP,
 		headers: {
 			"x-api-key": process.env.BASELIME_KEY || process.env.BASELIME_OTEL_KEY,
 		},