feat: finish transition to s3

Author: tschoffelen

packages/api/serverless.yml

@@ -22,7 +22,7 @@ custom:
     endpointType: regional
     apiType: http
   logRetentionInDays: 14
-  bucketName: ${self:service}-storage-${sls:stage} # TODO: make this a unique name
+  bucketName: ${self:service}-traces-${sls:stage}${param:TRACER_TOKEN}
 
 package:
   individually: true
@@ -120,7 +120,7 @@ functions:
     events:
       - sqs:
           arn: !GetAtt Queue.Arn
-          batchSize: 100
+          batchSize: 300
           maximumBatchingWindow: 30
 
   main:

packages/api/src/events/collector.js

@@ -33,50 +33,35 @@ export const handler = async ({ Records }) => {
   }
 
   // Check transactions cache to see if there's any transactions we can flush
+  const hourlyStats = [];
   for (const [transactionId, spans] of Object.entries(transactionCache)) {
     const invocationEndedSpan = spans.find(
       (span) =>
         span.type === "function" && span.ended && !span.id.includes("_started"),
     );
 
     if (!invocationEndedSpan) {
-      console.log(
-        `No invocation ended span found for transaction ${spans[0].transactionId}`,
-      );
-
       // TODO: if we are close to running out of time, we should flush the transaction cache anyway
-
       continue;
-    } else {
-      console.log(
-        "Flushing transaction cache for",
-        invocationEndedSpan.transactionId,
-      );
     }
 
     // save function invocation details
     await saveInvocation(invocationEndedSpan, spans);
 
-    // const duration = invocationEndedSpan.ended - invocationEndedSpan.started;
-    await saveHourlyStat(
+    const duration = invocationEndedSpan.ended - invocationEndedSpan.started;
+    hourlyStats.push(["global", "invocations", 1, "sum"]);
+    hourlyStats.push([
       invocationEndedSpan.region,
       invocationEndedSpan.name + ".invocations",
       1,
-    );
-    // await saveHourlyStat(
-    //   invocationEndedSpan.region,
-    //   invocationEndedSpan.name + ".duration",
-    //   duration,
-    // );
-    await saveHourlyStat("global", "invocations", 1);
-    if (invocationEndedSpan.error) {
-      await saveHourlyStat(
-        invocationEndedSpan.region,
-        invocationEndedSpan.name + ".errors",
-        1,
-      );
-      await saveHourlyStat("global", "errors", 1);
-    }
+      "sum",
+    ]);
+    hourlyStats.push([
+      invocationEndedSpan.region,
+      invocationEndedSpan.name + ".duration",
+      duration,
+      "avg",
+    ]);
 
     // save error
     if (invocationEndedSpan.error) {
@@ -98,14 +83,45 @@ export const handler = async ({ Records }) => {
           region: invocationEndedSpan.region,
         },
       );
-      await saveHourlyStat(
+
+      hourlyStats.push([
+        invocationEndedSpan.region,
+        invocationEndedSpan.name + ".errors",
+        1,
+        "sum",
+      ]);
+
+      hourlyStats.push(["global", "errors", 1, "sum"]);
+      hourlyStats.push([
         invocationEndedSpan.region,
         invocationEndedSpan.name + ".error." + errorKey,
         1,
-      );
+        "sum",
+      ]);
     }
 
     // Delete the transaction from the cache
     delete transactionCache[transactionId];
   }
+
+  // Save hourly stats
+  const statsToSave = hourlyStats.reduce((acc, [region, name, value, type]) => {
+    const key = `${region}#${name}`;
+    if (!acc[key]) {
+      acc[key] = [region, name, value, 0];
+    } else if (type === "sum") {
+      acc[key][2] += value;
+    } else if (type === "avg") {
+      acc[key][2] += value;
+      acc[key][3] += 1;
+    }
+    return acc;
+  }, []);
+  for (const [region, name, value, count] of statsToSave) {
+    await saveHourlyStat({
+      region,
+      name,
+      value: count > 0 ? value / count : value,
+    });
+  }
 };

packages/api/src/lib/spans.js

@@ -1,6 +1,6 @@
 const getGroupingKey = (span, extended = false) => {
   const items = [
-    span.spanType,
+    span.type,
     span.service,
 
     span.info?.resourceName,

packages/api/src/lib/spans.test.js

@@ -4,7 +4,7 @@ const traces = [
   {
     service: "dynamodb",
     ended: 1726249792234,
-    spanType: "http",
+    type: "http",
     started: 1726249792231,
     id: "ba2eec85-d01a-9b91-4a2f-44508e0407cc",
     reporterAwsRequestId: "1c065edb-2a94-4a07-9f1b-e31014df95ac",
@@ -37,7 +37,7 @@ const traces = [
   {
     service: "dynamodb",
     ended: 1726249792239,
-    spanType: "http",
+    type: "http",
     started: 1726249792236,
     id: "46d07ca2-adcb-0061-74aa-b9c8ea79196e",
     reporterAwsRequestId: "1c065edb-2a94-4a07-9f1b-e31014df95ac",
@@ -70,7 +70,7 @@ const traces = [
   {
     service: "dynamodb",
     ended: 1726249792244,
-    spanType: "http",
+    type: "http",
     started: 1726249792240,
     id: "ed4d0e6e-a004-265f-c6bb-1e9c21e3b3d0",
     reporterAwsRequestId: "1c065edb-2a94-4a07-9f1b-e31014df95ac",
@@ -103,7 +103,7 @@ const traces = [
   {
     service: "dynamodb",
     ended: 1726249792249,
-    spanType: "http",
+    type: "http",
     started: 1726249792246,
     id: "97210e79-03dc-2a30-8479-af7db8ba5f15",
     reporterAwsRequestId: "1c065edb-2a94-4a07-9f1b-e31014df95ac",
@@ -144,7 +144,7 @@ const traces2 = [
     name: "thumbnails-tokens-production-usage--report",
     ended: 1726304700362,
     envs: '{"AWS_LAMBDA_FUNCTION_VERSION":"$LATEST","TRACER_ORIGINAL_HANDLER":"src/functions/usage/report/handler.handler","REPORTS_BUCKET":"thumbnails-tokens-production-reports","AWS_SESSION_TOKEN":"IQoJb3JpZ2luX2VjEKn//////////wEaCWV1LXdlc3QtMSJGMEQCIF+ibaB9vrPCvEriTOXk2m6w62wbUcYucW0zE6VzXeFNAiBB7hYqK4ci/WlY6uxgI1+vFyMRed0OuFHBHbr7OoNXWCqyAwjS//////////8BEAMaDDY3Mjg5NTAyNzQxMyIMBN+KMPccXfO8bTSvKoYD4WalkWegJA8Kp6DBvM/xkNyhHfmCx331sVO2p9kXTPiS/XFcGerMpBmjh/Mqvh/BhQVCKylER37QPa1/LtCfjDPvQNHPOhQFL27XvaOSk3/qjfcXnuEeYxtnQz+HVUcGR98bU67Rh496R6/nODfjy5uZBqzgr9gIOX1VeNWnVhYVmZL2NszvkzuUDbBh2mQmAGDkHoZcc8dHq+/AZEKRsdBKin9rftOwL6asATAOUIUTPS+yFmvYrsRTxaOSMCz0g8JZT17pPkN9+e1BUYHsKAa78t+vV3D6bMOOJumwD2HsdTNT/KWA64Jcm81CyxbL8EWbvj6kFzpbhJySsSMVFwFzMN0D231xftsW2jMArsP87EtCfyn6AvLPnZ8bG3yzDHldGQZOmO0UqYea6OZr0IMmZZpqUvShuTnDUChLQvvIuyxHxBvmpsegUzb7H5/SdhziWWUmak7ysY37ra4/65DPNz4y129iJQRyXo2y+YFkkpGJsUakcT/z57bTzJc9MIJHvlk2MJChlbcGOp4BiyIow2gkPoH85+KS9o1p/OgP0vOHRr51zDx5Ru52s3CpjHjk/NDH79amegKwRrilnyK5rFk6MmCo0pQf2tbmPmGlaNWgMnb1XCZ1LyeEwA80mon1SE7EHAtlYS9mqnAiuIzFPFIhhBmRNZ/o7IDihnnd/GKbNHdyPP35XnCbAgsigf7cvg5QDD00RY+qZsoJfKi19VYYYHdwPwDBa5o=","LAMBDA_TASK_ROOT":"/var/task","AWS_LAMBDA_LOG_GROUP_NAME":"/aws/lambda/thumbnails-tokens-production-usage--report","LD_LIBRARY_PATH":"/var/lang/lib:/lib64:/usr/lib64:/var/runtime:/var/runtime/lib:/var/task:/var/task/lib:/opt/lib","AWS_LAMBDA_LOG_STREAM_NAME":"2024/09/14/[$LATEST]607ff0a879de44ecbc3da9de116c9da0","AWS_LAMBDA_RUNTIME_API":"127.0.0.1:9001","AWS_EXECUTION_ENV":"AWS_Lambda_nodejs16.x","AWS_XRAY_DAEMON_ADDRESS":"169.254.79.129:2000","AWS_LAMBDA_FUNCTION_NAME":"thumbnails-tokens-production-usage--report","PATH":"/var/lang/bin:/usr/local/bin:/usr/bin/:/bin:/opt/bin","TABLE_NAME":"thumbnails-tokens-production","AWS_DEFAULT_REGION":"eu-west-1","SERVICE":"thumbnails-tokens","PWD":"/var/task","AWS_SECRET_ACCESS_KEY":"****","LAMBDA_RUNTIME_DIR":"/var/runtime","LANG":"en_US.UTF-8","AWS_LAMBDA_INITIALIZATION_TYPE":"on-demand","AWS_REGION":"eu-west-1","TZ":":UTC","NODE_PATH":"/opt/nodejs/node16/node_modules:/opt/nodejs/node_modules:/var/runtime/node_modules:/var/runtime:/var/task"}...[too long]',
-    spanType: "function",
+    type: "function",
     started: 1726304700096,
     id: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",
     event:
@@ -171,7 +171,7 @@ const traces2 = [
     instances: 1,
     service: "dynamodb",
     ended: 1726304700148,
-    spanType: "http",
+    type: "http",
     started: 1726304700102,
     id: "1480dcfa-5f14-639b-cf58-4b3c8b43123e",
     reporterAwsRequestId: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",
@@ -211,7 +211,7 @@ const traces2 = [
     instances: 1,
     service: "dynamodb",
     ended: 1726304700170,
-    spanType: "http",
+    type: "http",
     started: 1726304700151,
     id: "96984001-1f79-3297-75c8-822a85efcd5c",
     reporterAwsRequestId: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",
@@ -251,7 +251,7 @@ const traces2 = [
     instances: 1,
     service: "dynamodb",
     ended: 1726304700192,
-    spanType: "http",
+    type: "http",
     started: 1726304700172,
     id: "bd2d2343-3b92-1013-52e2-738d9f44f7de",
     reporterAwsRequestId: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",
@@ -291,7 +291,7 @@ const traces2 = [
     instances: 1,
     service: "dynamodb",
     ended: 1726304700212,
-    spanType: "http",
+    type: "http",
     started: 1726304700194,
     id: "6b355f3f-cecd-6e42-56cf-e1d7d6be0a2d",
     reporterAwsRequestId: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",
@@ -331,7 +331,7 @@ const traces2 = [
     instances: 1,
     service: "dynamodb",
     ended: 1726304700234,
-    spanType: "http",
+    type: "http",
     started: 1726304700216,
     id: "5dba5c19-eddf-e7fe-4ab9-de380db43710",
     reporterAwsRequestId: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",
@@ -371,7 +371,7 @@ const traces2 = [
     instances: 1,
     service: "dynamodb",
     ended: 1726304700254,
-    spanType: "http",
+    type: "http",
     started: 1726304700236,
     id: "323cafc2-9788-49f7-ba40-0b16df8e3e3e",
     reporterAwsRequestId: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",
@@ -411,7 +411,7 @@ const traces2 = [
     instances: 1,
     service: "dynamodb",
     ended: 1726304700276,
-    spanType: "http",
+    type: "http",
     started: 1726304700257,
     id: "08408ef2-0364-f8f1-9ac5-05839bbcc565",
     reporterAwsRequestId: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",
@@ -451,7 +451,7 @@ const traces2 = [
     instances: 1,
     service: "dynamodb",
     ended: 1726304700300,
-    spanType: "http",
+    type: "http",
     started: 1726304700278,
     id: "3edfc450-ccbf-642d-e795-0219dcd7efad",
     reporterAwsRequestId: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",
@@ -491,7 +491,7 @@ const traces2 = [
     instances: 1,
     service: "dynamodb",
     ended: 1726304700320,
-    spanType: "http",
+    type: "http",
     started: 1726304700302,
     id: "0e31cde8-18a7-f93d-f2cd-1a8237d0447c",
     reporterAwsRequestId: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",
@@ -531,7 +531,7 @@ const traces2 = [
     instances: 1,
     service: "dynamodb",
     ended: 1726304700339,
-    spanType: "http",
+    type: "http",
     started: 1726304700322,
     id: "04f4c32d-3867-eb14-c388-09354caab4d0",
     reporterAwsRequestId: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",
@@ -571,7 +571,7 @@ const traces2 = [
     instances: 1,
     service: "dynamodb",
     ended: 1726304700361,
-    spanType: "http",
+    type: "http",
     started: 1726304700341,
     id: "a46580cc-7b7c-6ffc-3f0d-b868661c49cc",
     reporterAwsRequestId: "9fd58ff6-dfae-4176-b4bd-d0ca0ca1b0e7",

packages/api/src/lib/storage.js

@@ -1,4 +1,9 @@
-import { PutObjectCommand, S3Client } from "@aws-sdk/client-s3";
+import {
+  GetObjectCommand,
+  ListObjectsV2Command,
+  PutObjectCommand,
+  S3Client,
+} from "@aws-sdk/client-s3";
 
 const s3 = new S3Client();
 
@@ -17,3 +22,61 @@ export const store = async (key, data) => {
 
   return s3.send(new PutObjectCommand(params));
 };
+
+export const listByPrefix = async (prefix) => {
+  const output = [];
+  let pageToken = undefined;
+
+  do {
+    const params = {
+      Bucket: process.env.STORAGE_BUCKET_NAME,
+      Prefix: prefix,
+      ContinuationToken: pageToken,
+    };
+
+    const command = new ListObjectsV2Command(params);
+    const { Contents, NextContinuationToken } = await s3.send(command);
+
+    output.push(...(Contents?.map((item) => item.Key) || []));
+    pageToken = NextContinuationToken;
+  } while (pageToken && output.length < 10000);
+
+  return output;
+};
+
+export const list = async (startDate, endDate, prefix) => {
+  // Get a list of all days between startDate and endDate
+  const start = new Date(startDate);
+  const end = new Date(endDate);
+  const dateList = [];
+  const currentDate = new Date(start);
+  while (currentDate <= end) {
+    dateList.push(currentDate.toISOString().split("T")[0]);
+    currentDate.setDate(currentDate.getDate() + 1);
+  }
+
+  // Get all keys for each date, and flatten the array
+  const allKeys = await Promise.all(
+    dateList.map((date) => listByPrefix(`${date}/${prefix}`)),
+  );
+
+  return allKeys.flat();
+};
+
+export const get = async (key) => {
+  const params = {
+    Bucket: process.env.STORAGE_BUCKET_NAME,
+    Key: key,
+  };
+
+  const { Body } = await s3.send(new GetObjectCommand(params));
+  const string = await Body?.transformToString();
+  return string && JSON.parse(string);
+};
+
+export const listAndRead = async (startDate, endDate, prefix) => {
+  const keys = await list(startDate, endDate, prefix);
+  const data = await Promise.all(keys.map(async (key) => get(key)));
+
+  return data;
+};