-
Notifications
You must be signed in to change notification settings - Fork 7
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: ucan stream consumer space size track
- Loading branch information
1 parent
7d1ff48
commit 8ef0394
Showing
17 changed files
with
1,085 additions
and
57 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| import { | ||
| S3Client, | ||
| HeadObjectCommand, | ||
| } from '@aws-sdk/client-s3' | ||
|
|
||
| /** | ||
| * Abstraction layer with Factory to perform operations on bucket storing CAR files. | ||
| * | ||
| * @param {string} region | ||
| * @param {string} bucketName | ||
| * @param {import('@aws-sdk/client-s3').ServiceInputTypes} [options] | ||
| */ | ||
| export function createCarStore(region, bucketName, options) { | ||
| const s3 = new S3Client({ | ||
| region, | ||
| ...options, | ||
| }) | ||
| return useCarStore(s3, bucketName) | ||
| } | ||
|
|
||
| /** | ||
| * @param {S3Client} s3 | ||
| * @param {string} bucketName | ||
| * @returns {import('../types').CarStoreBucket} | ||
| */ | ||
| export function useCarStore(s3, bucketName) { | ||
| return { | ||
| /** | ||
| * @param {import('multiformats').UnknownLink} link | ||
| */ | ||
| getSize: async (link) => { | ||
| const cid = link.toString() | ||
| const cmd = new HeadObjectCommand({ | ||
| Key: `${cid}/${cid}.car`, | ||
| Bucket: bucketName, | ||
| }) | ||
| let res | ||
| try { | ||
| res = await s3.send(cmd) | ||
| } catch (cause) { | ||
| // @ts-expect-error | ||
| if (cause?.$metadata?.httpStatusCode === 404) { | ||
| return 0 | ||
| } | ||
| throw new Error('Failed to check if car-store', { cause }) | ||
| } | ||
| return res.ContentLength || 0 | ||
| }, | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
45 changes: 45 additions & 0 deletions
45
ucan-invocation/functions/space-metrics-store-add-size-total.js
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| import * as Sentry from '@sentry/serverless' | ||
|
|
||
| import { createCarStore } from '../buckets/car-store.js' | ||
| import { createSpaceMetricsTable } from '../tables/space-metrics.js' | ||
| import { parseKinesisEvent } from '../utils/parse-kinesis-event.js' | ||
| import { STORE_ADD } from '../constants.js' | ||
|
|
||
| Sentry.AWSLambda.init({ | ||
| environment: process.env.SST_STAGE, | ||
| dsn: process.env.SENTRY_DSN, | ||
| tracesSampleRate: 1.0, | ||
| }) | ||
|
|
||
| const AWS_REGION = process.env.AWS_REGION || 'us-west-2' | ||
|
|
||
| /** | ||
| * @param {import('aws-lambda').KinesisStreamEvent} event | ||
| */ | ||
| async function handler(event) { | ||
| const ucanInvocations = parseKinesisEvent(event) | ||
|
|
||
| const { | ||
| TABLE_NAME: tableName = '', | ||
| STORE_BUCKET_NAME: storeBucketName = '', | ||
| } = process.env | ||
|
|
||
| await updateAddSizeTotal(ucanInvocations, { | ||
| spaceMetricsTable: createSpaceMetricsTable(AWS_REGION, tableName), | ||
| carStoreBucket: createCarStore(AWS_REGION, storeBucketName) | ||
| }) | ||
| } | ||
|
|
||
| /** | ||
| * @param {import('../types').UcanInvocation[]} ucanInvocations | ||
| * @param {import('../types').MetricsBySpaceWithBucketCtx} ctx | ||
| */ | ||
| export async function updateAddSizeTotal (ucanInvocations, ctx) { | ||
| const invocationsWithStoreAdd = ucanInvocations.filter( | ||
| inv => inv.value.att.find(a => a.can === STORE_ADD) | ||
| ).flatMap(inv => inv.value.att) | ||
|
|
||
| await ctx.spaceMetricsTable.incrementStoreAddSizeTotal(invocationsWithStoreAdd) | ||
| } | ||
|
|
||
| export const consumer = Sentry.AWSLambda.wrapHandler(handler) |
55 changes: 55 additions & 0 deletions
55
ucan-invocation/functions/space-metrics-store-remove-size-total.js
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| import * as Sentry from '@sentry/serverless' | ||
|
|
||
| import { createCarStore } from '../buckets/car-store.js' | ||
| import { createSpaceMetricsTable } from '../tables/space-metrics.js' | ||
| import { parseKinesisEvent } from '../utils/parse-kinesis-event.js' | ||
| import { STORE_REMOVE } from '../constants.js' | ||
|
|
||
| Sentry.AWSLambda.init({ | ||
| environment: process.env.SST_STAGE, | ||
| dsn: process.env.SENTRY_DSN, | ||
| tracesSampleRate: 1.0, | ||
| }) | ||
|
|
||
| const AWS_REGION = process.env.AWS_REGION || 'us-west-2' | ||
|
|
||
| /** | ||
| * @param {import('aws-lambda').KinesisStreamEvent} event | ||
| */ | ||
| async function handler(event) { | ||
| const ucanInvocations = parseKinesisEvent(event) | ||
|
|
||
| const { | ||
| TABLE_NAME: tableName = '', | ||
| STORE_BUCKET_NAME: storeBucketName = '', | ||
| } = process.env | ||
|
|
||
| await updateRemoveSizeTotal(ucanInvocations, { | ||
| spaceMetricsTable: createSpaceMetricsTable(AWS_REGION, tableName), | ||
| carStoreBucket: createCarStore(AWS_REGION, storeBucketName) | ||
| }) | ||
| } | ||
|
|
||
| /** | ||
| * @param {import('../types').UcanInvocation[]} ucanInvocations | ||
| * @param {import('../types').MetricsBySpaceWithBucketCtx} ctx | ||
| */ | ||
| export async function updateRemoveSizeTotal (ucanInvocations, ctx) { | ||
| const invocationsWithStoreRemove = ucanInvocations.filter( | ||
| inv => inv.value.att.find(a => a.can === STORE_REMOVE) | ||
| ).flatMap(inv => inv.value.att) | ||
|
|
||
| // TODO: once we have receipts for store/remove, replace this | ||
| // Temporary adaptor to set size in invocation | ||
| for (const inv of invocationsWithStoreRemove) { | ||
| // @ts-ignore remove invocations always have link | ||
| const size = await ctx.carStoreBucket.getSize(inv.nb?.link) | ||
|
|
||
| // @ts-ignore | ||
| inv.nb.size = size | ||
| } | ||
|
|
||
| await ctx.spaceMetricsTable.incrementStoreRemoveSizeTotal(invocationsWithStoreRemove) | ||
| } | ||
|
|
||
| export const consumer = Sentry.AWSLambda.wrapHandler(handler) |
Oops, something went wrong.