diff --git a/README.md b/README.md index 149e59a46..14ab3d9bd 100644 --- a/README.md +++ b/README.md @@ -208,6 +208,8 @@ Following extension configurations are supported: - `azurite.inMemoryPersistence` Disable persisting any data to disk. If the Azurite process is terminated, all data is lost. - `azurite.extentMemoryLimit` When using in-memory persistence, limit the total size of extents (blob and queue content) to a specific number of megabytes. This does not limit blob, queue, or table metadata. Defaults to 50% of total memory. - `azurite.disableTelemetry` Disable telemetry data collection of this Azurite execution. By default, Azurite will collect telemetry data to help improve the product. +- `azurite.blobEventCapture` Enable capturing blob mutation events as Azure Event Grid-shaped JSON files into a folder for later processing, by default false. +- `azurite.blobEventCapturePath` Folder to write captured blob event JSON files to. Relative paths resolve against the workspace location. Defaults to `__blobevents__` under the workspace location when `azurite.blobEventCapture` is enabled. ### [DockerHub](https://hub.docker.com/_/microsoft-azure-storage-azurite) @@ -458,6 +460,22 @@ Optional. By default, Azurite will collect telemetry data to help improve the pr --disableTelemetry ``` +### Blob Event Capture + +Optional. Capture mutating blob operations as an [Azure Event Grid](https://learn.microsoft.com/azure/storage/blobs/storage-blob-event-overview)-shaped JSON file (one file per event) written into a folder, so the events can be processed later. Disabled by default. Enable it by: + +```cmd +--blobEventCapture +``` + +By default the files are written to a `__blobevents__` folder under the workspace location. Write them to a different folder (relative paths resolve against the workspace location) by: + +```cmd +--blobEventCapturePath path/to/folder +``` + +Each event is written to its own file named `{timestamp}-{uuid}.json`, published atomically so a consumer watching the folder only ever reads a complete file. The captured event uses the Azure Event Grid schema: `eventType` is `Microsoft.Storage.BlobCreated` or `Microsoft.Storage.BlobDeleted` (plus the Azurite convention-named `Microsoft.Storage.ContainerCreated` / `Microsoft.Storage.ContainerDeleted`), with the precise operation carried in `data.api` (for example `PutBlob`, `PutBlockList`, `AppendBlock`, `PutPage`, `DeleteBlob`, `CreateContainer`, `DeleteContainer`). Capture is fire-and-forget and never affects the outcome of a storage operation. If `--blobEventCapturePath` is supplied without `--blobEventCapture`, capture stays off and the path is ignored. + ### Use in-memory storage Optional. Disable persisting any data to disk and only store data in-memory. If the Azurite process is terminated, all diff --git a/package.json b/package.json index 2c8d44869..748be599d 100644 --- a/package.json +++ b/package.json @@ -275,6 +275,16 @@ "type": "boolean", "default": false, "description": "Disable telemetry data collection of this Azurite execution. By default, Azurite will collect telemetry data to help improve the product." + }, + "azurite.blobEventCapture": { + "type": "boolean", + "default": false, + "description": "Enable capturing blob mutation events as Azure Event Grid-shaped JSON files into a folder for later processing." + }, + "azurite.blobEventCapturePath": { + "type": "string", + "default": "", + "description": "Folder to write captured blob event JSON files to. Relative paths resolve against the workspace location. Defaults to '__blobevents__' under the workspace location when blob event capture is enabled." } } } diff --git a/src/blob/BlobConfiguration.ts b/src/blob/BlobConfiguration.ts index b77f94a4d..e478df968 100644 --- a/src/blob/BlobConfiguration.ts +++ b/src/blob/BlobConfiguration.ts @@ -45,6 +45,8 @@ export default class BlobConfiguration extends ConfigurationBase { disableProductStyleUrl: boolean = false, public readonly isMemoryPersistence: boolean = false, public readonly memoryStore?: MemoryExtentChunkStore, + public readonly enableBlobEventCapture: boolean = false, + public readonly blobEventCapturePath: string = "" ) { super( host, diff --git a/src/blob/BlobEnvironment.ts b/src/blob/BlobEnvironment.ts index d046a3773..31d93d3b2 100644 --- a/src/blob/BlobEnvironment.ts +++ b/src/blob/BlobEnvironment.ts @@ -70,6 +70,14 @@ if (!(args as any).config.name) { .option( ["", "disableTelemetry"], "Optional. Disable telemetry data collection of this Azurite execution. By default, Azurite will collect telemetry data to help improve the product." + ) + .option( + ["", "blobEventCapture"], + "Optional. Enable capturing blob mutation events as Azure Event Grid-shaped JSON files for later processing" + ) + .option( + ["", "blobEventCapturePath"], + "Optional. Folder to write captured blob event JSON files to. Defaults to '__blobevents__' under the workspace location" ); (args as any).config.name = "azurite-blob"; @@ -148,6 +156,18 @@ export default class BlobEnvironment implements IBlobEnvironment { return false; } + public blobEventCapture(): boolean { + if (this.flags.blobEventCapture !== undefined) { + return this.flags.blobEventCapture; + } + // default is false: blob event capture is opt-in + return false; + } + + public blobEventCapturePath(): string | undefined { + return this.flags.blobEventCapturePath; + } + public inMemoryPersistence(): boolean { if (this.flags.inMemoryPersistence !== undefined) { if (this.flags.location) { diff --git a/src/blob/BlobRequestListenerFactory.ts b/src/blob/BlobRequestListenerFactory.ts index 498bf4d6d..e9708ebc1 100644 --- a/src/blob/BlobRequestListenerFactory.ts +++ b/src/blob/BlobRequestListenerFactory.ts @@ -26,6 +26,7 @@ import StrictModelMiddlewareFactory, { UnsupportedHeadersBlocker, UnsupportedParametersBlocker } from "./middlewares/StrictModelMiddlewareFactory"; +import IBlobEventSink from "./events/IBlobEventSink"; import IBlobMetadataStore from "./persistence/IBlobMetadataStore"; import { DEFAULT_CONTEXT_PATH } from "./utils/constants"; @@ -56,7 +57,8 @@ export default class BlobRequestListenerFactory private readonly loose?: boolean, private readonly skipApiVersionCheck?: boolean, private readonly oauth?: OAuthLevel, - private readonly disableProductStyleUrl?: boolean + private readonly disableProductStyleUrl?: boolean, + private readonly eventSink?: IBlobEventSink ) { } public createRequestListener(): RequestListener { @@ -77,20 +79,23 @@ export default class BlobRequestListenerFactory this.metadataStore, this.extentStore, logger, - loose + loose, + this.eventSink ), blobHandler: new BlobHandler( this.metadataStore, this.extentStore, logger, loose, - pageBlobRangesManager + pageBlobRangesManager, + this.eventSink ), blockBlobHandler: new BlockBlobHandler( this.metadataStore, this.extentStore, logger, - loose + loose, + this.eventSink ), containerHandler: new ContainerHandler( this.accountDataStore, @@ -99,14 +104,16 @@ export default class BlobRequestListenerFactory this.extentStore, logger, loose, - this.disableProductStyleUrl + this.disableProductStyleUrl, + this.eventSink ), pageBlobHandler: new PageBlobHandler( this.metadataStore, this.extentStore, logger, loose, - pageBlobRangesManager + pageBlobRangesManager, + this.eventSink ), serviceHandler: new ServiceHandler( this.accountDataStore, diff --git a/src/blob/BlobServer.ts b/src/blob/BlobServer.ts index fcf5952eb..1fd712ae7 100644 --- a/src/blob/BlobServer.ts +++ b/src/blob/BlobServer.ts @@ -19,6 +19,8 @@ import { handleGCCriticalErrorClose } from "../common/GCCriticalErrorCloseHelper import ServerBase, { ServerStatus } from "../common/ServerBase"; import BlobConfiguration from "./BlobConfiguration"; import BlobRequestListenerFactory from "./BlobRequestListenerFactory"; +import FileBlobEventSink from "./events/FileBlobEventSink"; +import IBlobEventSink from "./events/IBlobEventSink"; import BlobGCManager from "./gc/BlobGCManager"; import IBlobMetadataStore from "./persistence/IBlobMetadataStore"; import LokiBlobMetadataStore from "./persistence/LokiBlobMetadataStore"; @@ -47,6 +49,7 @@ export default class BlobServer extends ServerBase implements ICleaner { private readonly extentStore: IExtentStore; private readonly accountDataStore: IAccountDataStore; private readonly gcManager: IGCManager; + private readonly eventSink?: IBlobEventSink; /** * Creates an instance of Server. @@ -107,6 +110,11 @@ export default class BlobServer extends ServerBase implements ICleaner { // We can also change the HTTP framework here by // creating a new XXXListenerFactory implementing IRequestListenerFactory interface // and replace the default Express based request listener + const eventSink: IBlobEventSink | undefined = + configuration.enableBlobEventCapture && configuration.blobEventCapturePath + ? new FileBlobEventSink(configuration.blobEventCapturePath, logger) + : undefined; + const requestListenerFactory: IRequestListenerFactory = new BlobRequestListenerFactory( metadataStore, @@ -117,7 +125,8 @@ export default class BlobServer extends ServerBase implements ICleaner { configuration.loose, configuration.skipApiVersionCheck, configuration.getOAuthLevel(), - configuration.disableProductStyleUrl + configuration.disableProductStyleUrl, + eventSink ); super(host, port, httpServer, requestListenerFactory, configuration); @@ -158,6 +167,7 @@ export default class BlobServer extends ServerBase implements ICleaner { this.extentStore = extentStore; this.accountDataStore = accountDataStore; this.gcManager = gcManager; + this.eventSink = eventSink; } /** @@ -212,6 +222,10 @@ export default class BlobServer extends ServerBase implements ICleaner { if (this.gcManager !== undefined) { await this.gcManager.start(); } + + if (this.eventSink !== undefined) { + await this.eventSink.init(); + } } protected async afterStart(): Promise { @@ -224,6 +238,10 @@ export default class BlobServer extends ServerBase implements ICleaner { } protected async afterClose(): Promise { + if (this.eventSink !== undefined) { + await this.eventSink.close(); + } + if (this.gcManager !== undefined) { await this.gcManager.close(); } diff --git a/src/blob/BlobServerFactory.ts b/src/blob/BlobServerFactory.ts index 158456476..63d58b207 100644 --- a/src/blob/BlobServerFactory.ts +++ b/src/blob/BlobServerFactory.ts @@ -1,17 +1,19 @@ import { join } from "path"; import { DEFAULT_SQL_OPTIONS } from "../common/utils/constants"; +import logger from "../common/Logger"; import BlobConfiguration from "./BlobConfiguration"; import BlobEnvironment from "./BlobEnvironment"; import BlobServer from "./BlobServer"; +import { resolveBlobEventCapturePath } from "./events/resolveBlobEventCapturePath"; import IBlobEnvironment from "./IBlobEnvironment"; import SqlBlobConfiguration from "./SqlBlobConfiguration"; import SqlBlobServer from "./SqlBlobServer"; -import { DEFAULT_BLOB_PERSISTENCE_PATH } from "./utils/constants"; import { DEFAULT_BLOB_EXTENT_LOKI_DB_PATH, DEFAULT_BLOB_LOKI_DB_PATH, - DEFAULT_BLOB_PERSISTENCE_ARRAY + DEFAULT_BLOB_PERSISTENCE_ARRAY, + DEFAULT_BLOB_PERSISTENCE_PATH } from "./utils/constants"; export class BlobServerFactory { @@ -32,6 +34,26 @@ export class BlobServerFactory { ); } + const enableBlobEventCapture = env.blobEventCapture(); + const configuredCapturePath = env.blobEventCapturePath(); + const blobEventCapturePath = resolveBlobEventCapturePath( + enableBlobEventCapture, + configuredCapturePath, + location + ); + if ( + !enableBlobEventCapture && + configuredCapturePath && + configuredCapturePath.length > 0 + ) { + // Note the empty-string check: the VS Code setting declares a "" default, + // so get() returns "" (not undefined) when unset. Warn only when + // a non-empty path was actually supplied without enabling capture. + logger.warn( + "blobEventCapturePath is configured but blobEventCapture is disabled; blob event capture is OFF and the path will be ignored." + ); + } + DEFAULT_BLOB_PERSISTENCE_ARRAY[0].locationPath = join( location, DEFAULT_BLOB_PERSISTENCE_PATH @@ -66,7 +88,9 @@ export class BlobServerFactory { env.key(), env.pwd(), env.oauth(), - env.disableProductStyleUrl() + env.disableProductStyleUrl(), + enableBlobEventCapture, + blobEventCapturePath ); return new SqlBlobServer(config); @@ -90,6 +114,9 @@ export class BlobServerFactory { env.oauth(), env.disableProductStyleUrl(), env.inMemoryPersistence(), + undefined, + enableBlobEventCapture, + blobEventCapturePath ); return new BlobServer(config); diff --git a/src/blob/IBlobEnvironment.ts b/src/blob/IBlobEnvironment.ts index a57700759..d6d60489d 100644 --- a/src/blob/IBlobEnvironment.ts +++ b/src/blob/IBlobEnvironment.ts @@ -15,4 +15,6 @@ export default interface IBlobEnvironment { inMemoryPersistence(): boolean; extentMemoryLimit(): number | undefined; disableTelemetry(): boolean; + blobEventCapture(): boolean; + blobEventCapturePath(): string | undefined; } diff --git a/src/blob/SqlBlobConfiguration.ts b/src/blob/SqlBlobConfiguration.ts index a4e785812..54f8e886b 100644 --- a/src/blob/SqlBlobConfiguration.ts +++ b/src/blob/SqlBlobConfiguration.ts @@ -37,7 +37,9 @@ export default class SqlBlobConfiguration extends ConfigurationBase { key: string = "", pwd: string = "", oauth?: string, - disableProductStyleUrl: boolean = false + disableProductStyleUrl: boolean = false, + public readonly enableBlobEventCapture: boolean = false, + public readonly blobEventCapturePath: string = "" ) { super( host, diff --git a/src/blob/SqlBlobServer.ts b/src/blob/SqlBlobServer.ts index 020a9b6a5..64eb17f73 100644 --- a/src/blob/SqlBlobServer.ts +++ b/src/blob/SqlBlobServer.ts @@ -14,6 +14,8 @@ import IExtentStore from "../common/persistence/IExtentStore"; import SqlExtentMetadataStore from "../common/persistence/SqlExtentMetadataStore"; import ServerBase, { ServerStatus } from "../common/ServerBase"; import BlobRequestListenerFactory from "./BlobRequestListenerFactory"; +import FileBlobEventSink from "./events/FileBlobEventSink"; +import IBlobEventSink from "./events/IBlobEventSink"; import BlobGCManager from "./gc/BlobGCManager"; import IBlobMetadataStore from "./persistence/IBlobMetadataStore"; import SqlBlobMetadataStore from "./persistence/SqlBlobMetadataStore"; @@ -42,6 +44,7 @@ export default class SqlBlobServer extends ServerBase { private readonly extentStore: IExtentStore; private readonly accountDataStore: IAccountDataStore; private readonly gcManager: IGCManager; + private readonly eventSink?: IBlobEventSink; /** * Creates an instance of Server. @@ -89,6 +92,11 @@ export default class SqlBlobServer extends ServerBase { // We can also change the HTTP framework here by // creating a new XXXListenerFactory implementing IRequestListenerFactory interface // and replace the default Express based request listener + const eventSink: IBlobEventSink | undefined = + configuration.enableBlobEventCapture && configuration.blobEventCapturePath + ? new FileBlobEventSink(configuration.blobEventCapturePath, logger) + : undefined; + const requestListenerFactory: IRequestListenerFactory = new BlobRequestListenerFactory( metadataStore, @@ -99,7 +107,8 @@ export default class SqlBlobServer extends ServerBase { configuration.loose, configuration.skipApiVersionCheck, configuration.getOAuthLevel(), - configuration.disableProductStyleUrl + configuration.disableProductStyleUrl, + eventSink ); super(host, port, httpServer, requestListenerFactory, configuration); @@ -130,6 +139,7 @@ export default class SqlBlobServer extends ServerBase { this.extentStore = extentStore; this.accountDataStore = accountDataStore; this.gcManager = gcManager; + this.eventSink = eventSink; } /** @@ -183,6 +193,10 @@ export default class SqlBlobServer extends ServerBase { if (this.gcManager !== undefined) { await this.gcManager.start(); } + + if (this.eventSink !== undefined) { + await this.eventSink.init(); + } } protected async afterStart(): Promise { @@ -195,6 +209,10 @@ export default class SqlBlobServer extends ServerBase { } protected async afterClose(): Promise { + if (this.eventSink !== undefined) { + await this.eventSink.close(); + } + if (this.gcManager !== undefined) { await this.gcManager.close(); } diff --git a/src/blob/events/BlobEventFactory.ts b/src/blob/events/BlobEventFactory.ts new file mode 100644 index 000000000..0cd01dda6 --- /dev/null +++ b/src/blob/events/BlobEventFactory.ts @@ -0,0 +1,81 @@ +import { randomUUID } from "crypto"; + +import BlobStorageContext from "../context/BlobStorageContext"; +import Context from "../generated/Context"; +import { BlobEventType, IBlobEvent, IBlobEventProps } from "./IBlobEvent"; + +// Synthetic emulator resource identifiers (real Azure uses ARM resource ids). +const DEV_SUBSCRIPTION_ID = "00000000-0000-0000-0000-000000000000"; +const DEV_RESOURCE_GROUP = "azurite"; + +// Process-level monotonic sequencer. Not per-blob like Azure, but adequate +// for an emulator; vary by index so ordering is observable. +let sequencerCounter = 0; +function nextSequencer(): string { + sequencerCounter += 1; + return sequencerCounter.toString(16).padStart(64, "0"); +} + +function isContainerEvent(eventType: BlobEventType): boolean { + return ( + eventType === BlobEventType.ContainerCreated || + eventType === BlobEventType.ContainerDeleted + ); +} + +/** + * Drop the query string from a request URL before it is persisted. A SAS + * request carries its credential in the query (e.g. `?...&sig=...`); writing + * that verbatim into an event file on disk would leak a signing secret. Real + * Azure Storage events also expose only the bare blob URL in `data.url`, so + * stripping the query is both safer and more faithful to the Event Grid schema. + */ +function stripQuery(url: string): string { + const q = url.indexOf("?"); + return q === -1 ? url : url.slice(0, q); +} + +/** + * Build an Azure Event Grid–shaped event from the request context and the + * operation-specific properties supplied by the handler. Pure: no I/O. + */ +export function createBlobEvent( + context: Context, + eventType: BlobEventType, + api: string, + props: IBlobEventProps +): IBlobEvent { + const blobCtx = new BlobStorageContext(context); + const account = blobCtx.account ?? ""; + const container = blobCtx.container ?? ""; + const blob = blobCtx.blob ?? ""; + const requestId = blobCtx.contextId ?? ""; + const clientRequestId = context.request?.getHeader("x-ms-client-request-id"); + const url = stripQuery(context.request?.getUrl() ?? ""); + + const subject = isContainerEvent(eventType) + ? `/blobServices/default/containers/${container}` + : `/blobServices/default/containers/${container}/blobs/${blob}`; + + return { + topic: `/subscriptions/${DEV_SUBSCRIPTION_ID}/resourceGroups/${DEV_RESOURCE_GROUP}/providers/Microsoft.Storage/storageAccounts/${account}`, + subject, + eventType, + id: randomUUID(), + eventTime: (context.startTime ?? new Date()).toISOString(), + dataVersion: "", + metadataVersion: "1", + data: { + api, + clientRequestId, + requestId, + eTag: props.eTag, + contentType: props.contentType, + contentLength: props.contentLength, + blobType: props.blobType, + url, + sequencer: nextSequencer(), + storageDiagnostics: { batchId: requestId } + } + }; +} diff --git a/src/blob/events/FileBlobEventSink.ts b/src/blob/events/FileBlobEventSink.ts new file mode 100644 index 000000000..9efacd3dc --- /dev/null +++ b/src/blob/events/FileBlobEventSink.ts @@ -0,0 +1,78 @@ +import { ensureDir } from "fs-extra"; +import { rename, writeFile } from "fs/promises"; +import { join } from "path"; + +import ILogger from "../../common/ILogger"; +import { IBlobEvent } from "./IBlobEvent"; +import IBlobEventSink from "./IBlobEventSink"; + +/** + * Replace any character that isn't alphanumeric or a hyphen. The filename is + * built from event fields; sanitizing each segment guarantees a crafted value + * cannot introduce path separators or ".." traversal that would escape the + * capture folder, regardless of how the event was constructed. + */ +function sanitizeSegment(value: string): string { + return value.replace(/[^A-Za-z0-9-]/g, "_"); +} + +/** + * Writes each captured event to its own JSON file in a folder. Async and + * fire-and-forget: write failures are logged, never surfaced to the caller. + * If the folder cannot be created at init(), the sink permanently disables + * itself so the server keeps running. + */ +export default class FileBlobEventSink implements IBlobEventSink { + private enabled = true; + private readonly pending = new Set>(); + + public constructor( + private readonly folderPath: string, + private readonly logger: ILogger + ) {} + + public async init(): Promise { + try { + await ensureDir(this.folderPath); + } catch (err) { + this.enabled = false; + this.logger.error( + `Blob event capture disabled: cannot create folder "${this.folderPath}": ${ + err instanceof Error ? err.message : String(err) + }` + ); + } + } + + public emit(event: IBlobEvent): void { + if (!this.enabled) { + return; + } + const safeTime = sanitizeSegment(event.eventTime.replace(/[:.]/g, "-")); + const safeId = sanitizeSegment(event.id); + const fileName = `${safeTime}-${safeId}.json`; + const filePath = join(this.folderPath, fileName); + // Write to a temp file then atomically rename it into place. A consumer + // watching the folder for "*.json" therefore only ever sees a complete + // file — never the empty/partial state a bare writeFile exposes between + // creating the directory entry and flushing its contents. + const tempPath = `${filePath}.tmp`; + const p = writeFile(tempPath, JSON.stringify(event, null, 2)) + .then(() => rename(tempPath, filePath)) + .catch((err) => { + this.logger.warn( + `Failed to write blob event file "${filePath}": ${ + err instanceof Error ? err.message : String(err) + }` + ); + }) + .finally(() => { + this.pending.delete(p); + }); + this.pending.add(p); + } + + public async close(): Promise { + await Promise.allSettled([...this.pending]); + } +} diff --git a/src/blob/events/IBlobEvent.ts b/src/blob/events/IBlobEvent.ts new file mode 100644 index 000000000..e1f7f6c85 --- /dev/null +++ b/src/blob/events/IBlobEvent.ts @@ -0,0 +1,46 @@ +/** + * Azure Event Grid event types for Storage blob events. + * BlobCreated / BlobDeleted match real Azure. Container* are Azurite + * convention-named (Azure Event Grid has no container-level blob events); + * the precise operation is always carried in `data.api`. + */ +export enum BlobEventType { + BlobCreated = "Microsoft.Storage.BlobCreated", + BlobDeleted = "Microsoft.Storage.BlobDeleted", + ContainerCreated = "Microsoft.Storage.ContainerCreated", + ContainerDeleted = "Microsoft.Storage.ContainerDeleted" +} + +/** Operation-specific values a handler passes when emitting an event. */ +export interface IBlobEventProps { + eTag?: string; + contentType?: string; + contentLength?: number; + blobType?: string; +} + +/** The `data` payload of an Event Grid storage event. */ +export interface IBlobEventData { + api: string; + clientRequestId?: string; + requestId: string; + eTag?: string; + contentType?: string; + contentLength?: number; + blobType?: string; + url: string; + sequencer: string; + storageDiagnostics: { batchId: string }; +} + +/** The Event Grid event envelope written to a JSON file. */ +export interface IBlobEvent { + topic: string; + subject: string; + eventType: BlobEventType; + id: string; + eventTime: string; + dataVersion: string; + metadataVersion: string; + data: IBlobEventData; +} diff --git a/src/blob/events/IBlobEventSink.ts b/src/blob/events/IBlobEventSink.ts new file mode 100644 index 000000000..dd8ed49cb --- /dev/null +++ b/src/blob/events/IBlobEventSink.ts @@ -0,0 +1,14 @@ +import { IBlobEvent } from "./IBlobEvent"; + +/** + * Destination for captured blob events. Implementations must never throw from + * emit(): a capture failure must not affect the originating storage operation. + */ +export default interface IBlobEventSink { + /** Prepare the sink (e.g. ensure the target folder exists). */ + init(): Promise; + /** Fire-and-forget: record one event. Must not throw. */ + emit(event: IBlobEvent): void; + /** Await any in-flight work so shutdown is clean. */ + close(): Promise; +} diff --git a/src/blob/events/resolveBlobEventCapturePath.ts b/src/blob/events/resolveBlobEventCapturePath.ts new file mode 100644 index 000000000..148850418 --- /dev/null +++ b/src/blob/events/resolveBlobEventCapturePath.ts @@ -0,0 +1,29 @@ +import { isAbsolute, join } from "path"; + +import { DEFAULT_BLOB_EVENT_CAPTURE_PATH } from "../utils/constants"; + +/** + * Resolve the effective folder that blob event capture writes to. Shared by + * both server entry points (the CLI `BlobServerFactory` and the VS Code + * `VSCServerManagerBlob`) so the two behave identically. + * + * Returns "" when capture is disabled — callers treat an empty path as + * "no sink". When enabled: an absolute configured path is used verbatim; a + * relative configured path resolves against `location`; an empty/omitted + * configured path falls back to the default folder under `location`. + */ +export function resolveBlobEventCapturePath( + enableCapture: boolean, + configuredPath: string | undefined, + location: string +): string { + if (!enableCapture) { + return ""; + } + if (configuredPath !== undefined && configuredPath.length > 0) { + return isAbsolute(configuredPath) + ? configuredPath + : join(location, configuredPath); + } + return join(location, DEFAULT_BLOB_EVENT_CAPTURE_PATH); +} diff --git a/src/blob/handlers/AppendBlobHandler.ts b/src/blob/handlers/AppendBlobHandler.ts index 08d24f73b..25ae2de35 100644 --- a/src/blob/handlers/AppendBlobHandler.ts +++ b/src/blob/handlers/AppendBlobHandler.ts @@ -16,6 +16,7 @@ import { MAX_APPEND_BLOB_BLOCK_SIZE } from "../utils/constants"; import { computeAndValidateTransactionalChecksums, getTagsFromString } from "../utils/utils"; +import { BlobEventType } from "../events/IBlobEvent"; import BaseHandler from "./BaseHandler"; export default class AppendBlobHandler extends BaseHandler @@ -86,6 +87,13 @@ export default class AppendBlobHandler extends BaseHandler options.modifiedAccessConditions ); + this.emitBlobEvent(context, BlobEventType.BlobCreated, "PutBlob", { + eTag: etag, + contentType, + contentLength: 0, + blobType: Models.BlobType.AppendBlob + }); + const response: Models.AppendBlobCreateResponse = { statusCode: 201, eTag: etag, @@ -192,6 +200,12 @@ export default class AppendBlobHandler extends BaseHandler options.appendPositionAccessConditions ); + this.emitBlobEvent(context, BlobEventType.BlobCreated, "AppendBlock", { + eTag: properties.etag, + contentLength, + blobType: Models.BlobType.AppendBlob + }); + const response: Models.AppendBlobAppendBlockResponse = { statusCode: 201, requestId: context.contextId, diff --git a/src/blob/handlers/BaseHandler.ts b/src/blob/handlers/BaseHandler.ts index 63e6106c6..3f9a9664d 100644 --- a/src/blob/handlers/BaseHandler.ts +++ b/src/blob/handlers/BaseHandler.ts @@ -1,4 +1,8 @@ import IExtentStore from "../../common/persistence/IExtentStore"; +import { createBlobEvent } from "../events/BlobEventFactory"; +import { BlobEventType, IBlobEventProps } from "../events/IBlobEvent"; +import IBlobEventSink from "../events/IBlobEventSink"; +import Context from "../generated/Context"; import ILogger from "../generated/utils/ILogger"; import IBlobMetadataStore from "../persistence/IBlobMetadataStore"; @@ -15,6 +19,32 @@ export default class BaseHandler { protected readonly metadataStore: IBlobMetadataStore, protected readonly extentStore: IExtentStore, protected readonly logger: ILogger, - protected readonly loose: boolean + protected readonly loose: boolean, + protected readonly eventSink?: IBlobEventSink ) {} + + /** + * Emit a captured blob event if a sink is configured. Never throws: a + * capture failure must not affect the originating storage operation. + */ + protected emitBlobEvent( + context: Context, + eventType: BlobEventType, + api: string, + props: IBlobEventProps + ): void { + if (this.eventSink === undefined) { + return; + } + try { + this.eventSink.emit(createBlobEvent(context, eventType, api, props)); + } catch (err) { + this.logger.warn( + `Failed to emit blob event (${eventType}/${api}): ${ + err instanceof Error ? err.message : String(err) + }`, + context.contextId + ); + } + } } diff --git a/src/blob/handlers/BlobHandler.ts b/src/blob/handlers/BlobHandler.ts index 1ee4b9bc6..7ce0cfa94 100644 --- a/src/blob/handlers/BlobHandler.ts +++ b/src/blob/handlers/BlobHandler.ts @@ -31,6 +31,8 @@ import { getBlobTagsCount, validateBlobTag } from "../utils/utils"; +import IBlobEventSink from "../events/IBlobEventSink"; +import { BlobEventType } from "../events/IBlobEvent"; import BaseHandler from "./BaseHandler"; import IPageBlobRangesManager from "./IPageBlobRangesManager"; @@ -48,9 +50,10 @@ export default class BlobHandler extends BaseHandler implements IBlobHandler { extentStore: IExtentStore, logger: ILogger, loose: boolean, - private readonly rangesManager: IPageBlobRangesManager + private readonly rangesManager: IPageBlobRangesManager, + eventSink?: IBlobEventSink ) { - super(metadataStore, extentStore, logger, loose); + super(metadataStore, extentStore, logger, loose, eventSink); } /** @@ -187,6 +190,8 @@ export default class BlobHandler extends BaseHandler implements IBlobHandler { options ); + this.emitBlobEvent(context, BlobEventType.BlobDeleted, "DeleteBlob", {}); + const response: Models.BlobDeleteResponse = { statusCode: 202, requestId: context.contextId, diff --git a/src/blob/handlers/BlockBlobHandler.ts b/src/blob/handlers/BlockBlobHandler.ts index 666a067f0..137b394ca 100644 --- a/src/blob/handlers/BlockBlobHandler.ts +++ b/src/blob/handlers/BlockBlobHandler.ts @@ -12,6 +12,7 @@ import IBlockBlobHandler from "../generated/handlers/IBlockBlobHandler"; import { parseXML } from "../generated/utils/xml"; import { BlobModel, BlockModel } from "../persistence/IBlobMetadataStore"; import { BLOB_API_VERSION } from "../utils/constants"; +import { BlobEventType } from "../events/IBlobEvent"; import BaseHandler from "./BaseHandler"; import { computeAndValidateTransactionalChecksums, @@ -143,6 +144,13 @@ export default class BlockBlobHandler options.modifiedAccessConditions ); + this.emitBlobEvent(context, BlobEventType.BlobCreated, "PutBlob", { + eTag: etag, + contentType, + contentLength, + blobType: Models.BlobType.BlockBlob + }); + const response: Models.BlockBlobUploadResponse = { statusCode: 201, eTag: etag, @@ -240,6 +248,11 @@ export default class BlockBlobHandler options.leaseAccessConditions ); + this.emitBlobEvent(context, BlobEventType.BlobCreated, "PutBlock", { + contentLength, + blobType: Models.BlobType.BlockBlob + }); + const response: Models.BlockBlobStageBlockResponse = { statusCode: 201, contentMD5: undefined, // TODO: Block content MD5 @@ -370,6 +383,12 @@ export default class BlockBlobHandler options.modifiedAccessConditions ); + this.emitBlobEvent(context, BlobEventType.BlobCreated, "PutBlockList", { + eTag: blob.properties.etag, + contentType, + blobType: Models.BlobType.BlockBlob + }); + const contentMD5 = await getMD5FromString(rawBody); const response: Models.BlockBlobCommitBlockListResponse = { diff --git a/src/blob/handlers/ContainerHandler.ts b/src/blob/handlers/ContainerHandler.ts index fe5ec6adb..130d67d6f 100644 --- a/src/blob/handlers/ContainerHandler.ts +++ b/src/blob/handlers/ContainerHandler.ts @@ -17,6 +17,8 @@ import { } from "../utils/constants"; import { DEFAULT_LIST_BLOBS_MAX_RESULTS } from "../utils/constants"; import { getBlobTagsCount, removeQuotationFromListBlobEtag } from "../utils/utils"; +import IBlobEventSink from "../events/IBlobEventSink"; +import { BlobEventType } from "../events/IBlobEvent"; import BaseHandler from "./BaseHandler"; import { BlobBatchHandler } from "./BlobBatchHandler"; @@ -38,9 +40,10 @@ export default class ContainerHandler extends BaseHandler extentStore: IExtentStore, logger: ILogger, loose: boolean, - disableProductStyle?: boolean + disableProductStyle?: boolean, + eventSink?: IBlobEventSink ) { - super(metadataStore, extentStore, logger, loose); + super(metadataStore, extentStore, logger, loose, eventSink); this.disableProductStyle = disableProductStyle; } @@ -82,6 +85,10 @@ export default class ContainerHandler extends BaseHandler } }); + this.emitBlobEvent(context, BlobEventType.ContainerCreated, "CreateContainer", { + eTag: etag + }); + const response: Models.ContainerCreateResponse = { statusCode: 201, requestId: blobCtx.contextId, @@ -174,6 +181,8 @@ export default class ContainerHandler extends BaseHandler options ); + this.emitBlobEvent(context, BlobEventType.ContainerDeleted, "DeleteContainer", {}); + const response: Models.ContainerDeleteResponse = { statusCode: 202, requestId: context.contextId, diff --git a/src/blob/handlers/PageBlobHandler.ts b/src/blob/handlers/PageBlobHandler.ts index 28f79d018..58224c25e 100644 --- a/src/blob/handlers/PageBlobHandler.ts +++ b/src/blob/handlers/PageBlobHandler.ts @@ -18,6 +18,8 @@ import { deserializePageBlobRangeHeader, getTagsFromString } from "../utils/utils"; +import IBlobEventSink from "../events/IBlobEventSink"; +import { BlobEventType } from "../events/IBlobEvent"; import BaseHandler from "./BaseHandler"; import IPageBlobRangesManager from "./IPageBlobRangesManager"; @@ -36,9 +38,10 @@ export default class PageBlobHandler extends BaseHandler extentStore: IExtentStore, logger: ILogger, loose: boolean, - private readonly rangesManager: IPageBlobRangesManager + private readonly rangesManager: IPageBlobRangesManager, + eventSink?: IBlobEventSink ) { - super(metadataStore, extentStore, logger, loose); + super(metadataStore, extentStore, logger, loose, eventSink); } public async uploadPagesFromURL( @@ -153,6 +156,13 @@ export default class PageBlobHandler extends BaseHandler options.modifiedAccessConditions ); + this.emitBlobEvent(context, BlobEventType.BlobCreated, "PutBlob", { + eTag: etag, + contentType, + contentLength: blobContentLength, + blobType: Models.BlobType.PageBlob + }); + const response: Models.PageBlobCreateResponse = { statusCode: 201, eTag: etag, @@ -267,6 +277,12 @@ export default class PageBlobHandler extends BaseHandler options.sequenceNumberAccessConditions ); + this.emitBlobEvent(context, BlobEventType.BlobCreated, "PutPage", { + eTag: res.etag, + contentLength, + blobType: Models.BlobType.PageBlob + }); + const response: Models.PageBlobUploadPagesResponse = { statusCode: 201, eTag: res.etag, diff --git a/src/blob/utils/constants.ts b/src/blob/utils/constants.ts index 5a93a7dcd..e2017f378 100644 --- a/src/blob/utils/constants.ts +++ b/src/blob/utils/constants.ts @@ -12,6 +12,7 @@ export const DEFAULT_BLOB_LOKI_DB_PATH = "__azurite_db_blob__.json"; export const DEFAULT_BLOB_EXTENT_LOKI_DB_PATH = "__azurite_db_blob_extent__.json"; export const DEFAULT_BLOB_PERSISTENCE_PATH = "__blobstorage__"; +export const DEFAULT_BLOB_EVENT_CAPTURE_PATH = "__blobevents__"; export const DEFAULT_DEBUG_LOG_PATH = "./debug.log"; export const DEFAULT_ENABLE_DEBUG_LOG = true; export const DEFAULT_ACCESS_LOG_PATH = "./access.log"; diff --git a/src/common/Environment.ts b/src/common/Environment.ts index 44c25c237..ed99d66c6 100644 --- a/src/common/Environment.ts +++ b/src/common/Environment.ts @@ -110,7 +110,15 @@ args ) .option( ["", "disableTelemetry"], - "Optional. Disable telemtry collection of Azurite. If not specify this parameter Azurite will collect telemetry data by default." + "Optional. Disable telemetry collection of Azurite. If not specify this parameter Azurite will collect telemetry data by default." + ) + .option( + ["", "blobEventCapture"], + "Optional. Enable capturing blob mutation events as Azure Event Grid-shaped JSON files for later processing" + ) + .option( + ["", "blobEventCapturePath"], + "Optional. Folder to write captured blob event JSON files to. Defaults to '__blobevents__' under the workspace location" ); (args as any).config.name = "azurite"; @@ -231,6 +239,18 @@ export default class Environment implements IEnvironment { return false; } + public blobEventCapture(): boolean { + if (this.flags.blobEventCapture !== undefined) { + return this.flags.blobEventCapture; + } + // default is false: blob event capture is opt-in + return false; + } + + public blobEventCapturePath(): string | undefined { + return this.flags.blobEventCapturePath; + } + public async debug(): Promise { if (typeof this.flags.debug === "string") { // Enable debug log to file diff --git a/src/common/VSCEnvironment.ts b/src/common/VSCEnvironment.ts index 0bcff08f5..8f257def8 100644 --- a/src/common/VSCEnvironment.ts +++ b/src/common/VSCEnvironment.ts @@ -135,4 +135,12 @@ export default class VSCEnvironment implements IEnvironment { this.workspaceConfiguration.get("disableTelemetry") || false ); } + + public blobEventCapture(): boolean { + return this.workspaceConfiguration.get("blobEventCapture") || false; + } + + public blobEventCapturePath(): string | undefined { + return this.workspaceConfiguration.get("blobEventCapturePath"); + } } diff --git a/src/common/VSCServerManagerBlob.ts b/src/common/VSCServerManagerBlob.ts index 1a3555ca6..405d91f6c 100644 --- a/src/common/VSCServerManagerBlob.ts +++ b/src/common/VSCServerManagerBlob.ts @@ -2,6 +2,7 @@ import { join } from "path"; import BlobConfiguration from "../blob/BlobConfiguration"; import BlobServer from "../blob/BlobServer"; +import { resolveBlobEventCapturePath } from "../blob/events/resolveBlobEventCapturePath"; import { DEFAULT_BLOB_EXTENT_LOKI_DB_PATH, DEFAULT_BLOB_LOKI_DB_PATH, @@ -74,6 +75,22 @@ export default class VSCServerManagerBlob extends VSCServerManagerBase { ); AzuriteTelemetryClient.init(DEFAULT_BLOB_PERSISTENCE_ARRAY[0].locationPath, !env.disableTelemetry(), env.workspaceConfiguration, true); + const enableBlobEventCapture = env.blobEventCapture(); + const configuredCapturePath = env.blobEventCapturePath(); + const blobEventCapturePath = resolveBlobEventCapturePath( + enableBlobEventCapture, + configuredCapturePath, + location + ); + if ( + !enableBlobEventCapture && + configuredCapturePath !== undefined && + configuredCapturePath.length > 0 + ) { + Logger.default.warn( + "azurite.blobEventCapturePath is configured but azurite.blobEventCapture is disabled; blob event capture is OFF and the path will be ignored." + ); + } // Initialize server configuration const config = new BlobConfiguration( env.blobHost(), @@ -94,6 +111,9 @@ export default class VSCServerManagerBlob extends VSCServerManagerBase { env.oauth(), env.disableProductStyleUrl(), env.inMemoryPersistence(), + undefined, + enableBlobEventCapture, + blobEventCapturePath ); return config; } diff --git a/tests/BlobTestServerFactory.ts b/tests/BlobTestServerFactory.ts index 0867b07cb..3fdedd544 100644 --- a/tests/BlobTestServerFactory.ts +++ b/tests/BlobTestServerFactory.ts @@ -24,7 +24,9 @@ export default class BlobTestServerFactory { loose: boolean = false, skipApiVersionCheck: boolean = false, https: boolean = false, - oauth?: string + oauth?: string, + enableBlobEventCapture: boolean = false, + blobEventCapturePath: string = "" ): BlobServer | SqlBlobServer | LiveModeStubServer { if (LIVE_TEST_MODE) { return new LiveModeStubServer(); @@ -68,6 +70,8 @@ export default class BlobTestServerFactory { undefined, oauth, undefined, + enableBlobEventCapture, + blobEventCapturePath ); return new SqlBlobServer(config); @@ -92,7 +96,10 @@ export default class BlobTestServerFactory { undefined, oauth, undefined, - inMemoryPersistence + inMemoryPersistence, + undefined, + enableBlobEventCapture, + blobEventCapturePath ); return new BlobServer(config); } diff --git a/tests/blob/BlobEventFactory.test.ts b/tests/blob/BlobEventFactory.test.ts new file mode 100644 index 000000000..7ffcd6ad5 --- /dev/null +++ b/tests/blob/BlobEventFactory.test.ts @@ -0,0 +1,96 @@ +import * as assert from "assert"; + +import Context from "../../src/blob/generated/Context"; +import { BlobEventType } from "../../src/blob/events/IBlobEvent"; +import { createBlobEvent } from "../../src/blob/events/BlobEventFactory"; + +// Build a minimal Context whose BlobStorageContext getters return test values, +// with a fake request exposing getUrl()/getHeader(). +function makeContext(account: string, container: string, blob?: string): Context { + const holder: any = {}; + const context = new Context(holder, "context"); + context.contextId = "req-123"; + context.startTime = new Date("2026-08-06T12:34:56.789Z"); + (context as any).context.account = account; + (context as any).context.container = container; + (context as any).context.blob = blob; + context.request = { + getUrl: () => `http://127.0.0.1:10000/${account}/${container}/${blob ?? ""}`, + getHeader: (field: string) => + field.toLowerCase() === "x-ms-client-request-id" ? "client-abc" : undefined + } as any; + return context; +} + +describe("BlobEventFactory @loki @sql", () => { + it("builds a BlobCreated Event Grid envelope", () => { + const ctx = makeContext("devstoreaccount1", "c1", "path/to/file.txt"); + const event = createBlobEvent(ctx, BlobEventType.BlobCreated, "PutBlob", { + eTag: "0x8D1", + contentType: "text/plain", + contentLength: 5, + blobType: "BlockBlob" + }); + + assert.strictEqual(event.eventType, "Microsoft.Storage.BlobCreated"); + assert.strictEqual( + event.subject, + "/blobServices/default/containers/c1/blobs/path/to/file.txt" + ); + assert.ok(event.topic.endsWith("/storageAccounts/devstoreaccount1")); + assert.strictEqual(event.metadataVersion, "1"); + assert.strictEqual(event.data.api, "PutBlob"); + assert.strictEqual(event.data.requestId, "req-123"); + assert.strictEqual(event.data.clientRequestId, "client-abc"); + assert.strictEqual(event.data.eTag, "0x8D1"); + assert.strictEqual(event.data.contentLength, 5); + assert.strictEqual(event.data.blobType, "BlockBlob"); + assert.strictEqual( + event.data.url, + "http://127.0.0.1:10000/devstoreaccount1/c1/path/to/file.txt" + ); + assert.ok(typeof event.id === "string" && event.id.length > 0); + assert.strictEqual(event.data.storageDiagnostics.batchId, "req-123"); + }); + + it("strips the query string (SAS credentials) from data.url", () => { + const ctx = makeContext("devstoreaccount1", "c1", "b"); + // Simulate a SAS-authenticated request whose signature lives in the query. + (ctx.request as any).getUrl = () => + "http://127.0.0.1:10000/devstoreaccount1/c1/b?sv=2021-08-06&sig=SECRETsignature%3D%3D&se=2026-01-01"; + const event = createBlobEvent(ctx, BlobEventType.BlobCreated, "PutBlob", {}); + + assert.strictEqual( + event.data.url, + "http://127.0.0.1:10000/devstoreaccount1/c1/b", + "data.url must not retain the SAS query string" + ); + assert.ok( + !event.data.url.includes("sig="), + "the SAS signature must never be persisted" + ); + }); + + it("uses a container-scoped subject for container events", () => { + const ctx = makeContext("devstoreaccount1", "c1"); + const event = createBlobEvent( + ctx, + BlobEventType.ContainerDeleted, + "DeleteContainer", + {} + ); + assert.strictEqual(event.eventType, "Microsoft.Storage.ContainerDeleted"); + assert.strictEqual( + event.subject, + "/blobServices/default/containers/c1" + ); + }); + + it("produces monotonically increasing 64-char hex sequencers", () => { + const ctx = makeContext("devstoreaccount1", "c1", "b"); + const a = createBlobEvent(ctx, BlobEventType.BlobCreated, "PutBlob", {}); + const b = createBlobEvent(ctx, BlobEventType.BlobCreated, "PutBlob", {}); + assert.strictEqual(a.data.sequencer.length, 64); + assert.ok(BigInt("0x" + b.data.sequencer) > BigInt("0x" + a.data.sequencer)); + }); +}); diff --git a/tests/blob/FileBlobEventSink.test.ts b/tests/blob/FileBlobEventSink.test.ts new file mode 100644 index 000000000..6b21ebcb2 --- /dev/null +++ b/tests/blob/FileBlobEventSink.test.ts @@ -0,0 +1,110 @@ +import * as assert from "assert"; +import * as fs from "fs-extra"; +import { join } from "path"; + +import { BlobEventType, IBlobEvent } from "../../src/blob/events/IBlobEvent"; +import FileBlobEventSink from "../../src/blob/events/FileBlobEventSink"; +import ILogger from "../../src/common/ILogger"; + +const noopLogger: ILogger = { + error: () => undefined, + warn: () => undefined, + info: () => undefined, + verbose: () => undefined, + debug: () => undefined +}; + +function sampleEvent(id: string): IBlobEvent { + return { + topic: "/subscriptions/x/storageAccounts/devstoreaccount1", + subject: "/blobServices/default/containers/c1/blobs/b", + eventType: BlobEventType.BlobCreated, + id, + eventTime: "2026-08-06T12:34:56.789Z", + dataVersion: "", + metadataVersion: "1", + data: { + api: "PutBlob", + requestId: "req-1", + url: "http://127.0.0.1:10000/devstoreaccount1/c1/b", + sequencer: "0".repeat(63) + "1", + storageDiagnostics: { batchId: "req-1" } + } + }; +} + +describe("FileBlobEventSink @loki @sql", () => { + const folder = "__test_blob_events__"; + + afterEach(() => { + if (fs.existsSync(folder)) { + fs.removeSync(folder); + } + }); + + it("writes one JSON file per event after init", async () => { + const sink = new FileBlobEventSink(folder, noopLogger); + await sink.init(); + sink.emit(sampleEvent("id-aaa")); + sink.emit(sampleEvent("id-bbb")); + await sink.close(); + + const files = fs.readdirSync(folder).filter((f) => f.endsWith(".json")); + assert.strictEqual(files.length, 2); + + const parsed = JSON.parse( + fs.readFileSync(join(folder, files[0]), "utf8").toString() + ); + assert.strictEqual(parsed.eventType, "Microsoft.Storage.BlobCreated"); + assert.strictEqual(parsed.data.api, "PutBlob"); + }); + + it("uses a Windows-safe, id-bearing filename", async () => { + const sink = new FileBlobEventSink(folder, noopLogger); + await sink.init(); + sink.emit(sampleEvent("id-ccc")); + await sink.close(); + + const files = fs.readdirSync(folder); + assert.strictEqual(files.length, 1); + assert.ok(files[0].includes("id-ccc")); + assert.ok(!files[0].includes(":")); + assert.ok(files[0].endsWith(".json")); + // The "." between seconds and milliseconds in eventTime must be replaced; + // the only dot allowed is the .json extension. + assert.ok( + !files[0].slice(0, -".json".length).includes("."), + "eventTime dots must be replaced in the filename" + ); + }); + + it("neutralizes path separators in event fields to prevent traversal", async () => { + const sink = new FileBlobEventSink(folder, noopLogger); + await sink.init(); + // A crafted id containing traversal sequences must not escape the folder. + sink.emit(sampleEvent("../../../../evil")); + await sink.close(); + + const files = fs.readdirSync(folder).filter((f) => f.endsWith(".json")); + assert.strictEqual(files.length, 1, "event must be written inside the folder"); + assert.ok( + !files[0].includes("/") && !files[0].includes("\\"), + "filename must contain no path separators" + ); + assert.ok( + !fs.existsSync(join(folder, "..", "evil.json")), + "nothing must be written outside the capture folder" + ); + }); + + it("self-disables and does not throw when the folder cannot be created", async () => { + // Point at a path under an existing FILE so ensureDir fails. + fs.ensureFileSync(join(folder, "afile")); + const badPath = join(folder, "afile", "subdir"); + const sink = new FileBlobEventSink(badPath, noopLogger); + await sink.init(); // must not throw + sink.emit(sampleEvent("id-ddd")); // must not throw + await sink.close(); + assert.ok(!fs.existsSync(badPath)); + }); +}); diff --git a/tests/blob/apis/eventCapture.test.ts b/tests/blob/apis/eventCapture.test.ts new file mode 100644 index 000000000..d6babe0f9 --- /dev/null +++ b/tests/blob/apis/eventCapture.test.ts @@ -0,0 +1,246 @@ +import { + StorageSharedKeyCredential, + BlobServiceClient, + newPipeline +} from "@azure/storage-blob"; +import * as assert from "assert"; +import * as fs from "fs-extra"; +import { join } from "path"; + +import { configLogger } from "../../../src/common/Logger"; +import BlobTestServerFactory from "../../BlobTestServerFactory"; +import { + EMULATOR_ACCOUNT_KEY, + EMULATOR_ACCOUNT_NAME, + getUniqueName +} from "../../testutils"; + +configLogger(false); + +describe("Blob Event Capture @loki @sql", () => { + const factory = new BlobTestServerFactory(); + const eventFolder = "__test_event_capture__"; + + // Build a server with event capture enabled, pointing at eventFolder. + const server = factory.createServer(false, false, false, undefined, true, eventFolder); + + const baseURL = `http://${server.config.host}:${server.config.port}/devstoreaccount1`; + const serviceClient = new BlobServiceClient( + baseURL, + newPipeline( + new StorageSharedKeyCredential( + EMULATOR_ACCOUNT_NAME, + EMULATOR_ACCOUNT_KEY + ), + { retryOptions: { maxTries: 1 } } + ) + ); + + function readEvents(): any[] { + if (!fs.existsSync(eventFolder)) { + return []; + } + const parsed: any[] = []; + for (const f of fs.readdirSync(eventFolder)) { + if (!f.endsWith(".json")) { + continue; + } + try { + parsed.push( + JSON.parse(fs.readFileSync(join(eventFolder, f), "utf8").toString()) + ); + } catch { + // A file that isn't valid JSON yet is one the sink hasn't finished + // publishing; skip it and let the poll retry. (Atomic rename in the + // sink makes this rare, but the guard keeps the reader robust.) + } + } + return parsed; + } + + // Event writes are fire-and-forget: the SDK call returns before the JSON + // file is on disk. Poll (rather than a single fixed sleep) until the + // expected event appears, so the test stays reliable under load / on the + // slower @sql path. Returns the full snapshot once `minCount` events match + // the predicate, or after the timeout — a genuine miss still fails the + // assertion because the returned snapshot won't contain the event. + async function waitForEvents( + predicate: (e: any) => boolean, + minCount: number = 1, + timeoutMs: number = 5000 + ): Promise { + const deadline = Date.now() + timeoutMs; + let events = readEvents(); + while ( + events.filter(predicate).length < minCount && + Date.now() < deadline + ) { + await new Promise((r) => setTimeout(r, 50)); + events = readEvents(); + } + return events; + } + + before(async () => { + if (fs.existsSync(eventFolder)) { + fs.removeSync(eventFolder); + } + await server.start(); + }); + + after(async () => { + await server.close(); + await server.clean(); + if (fs.existsSync(eventFolder)) { + fs.removeSync(eventFolder); + } + }); + + it("captures ContainerCreated and ContainerDeleted", async () => { + const containerName = getUniqueName("evt-c"); + const containerClient = serviceClient.getContainerClient(containerName); + await containerClient.create(); + + // Name-scope the filter so accumulated events from other tests cannot + // satisfy it (the shared folder is not cleared between `it`s). + const createdPred = (e: any) => + e.eventType === "Microsoft.Storage.ContainerCreated" && + e.subject.endsWith(`/containers/${containerName}`); + let events = await waitForEvents(createdPred); + const created = events.filter(createdPred); + assert.ok(created.length >= 1, "expected a ContainerCreated event"); + assert.strictEqual(created[0].data.api, "CreateContainer"); + + // Deleting the container must emit a matching ContainerDeleted event. + await containerClient.delete(); + const deletedPred = (e: any) => + e.eventType === "Microsoft.Storage.ContainerDeleted" && + e.subject.endsWith(`/containers/${containerName}`); + events = await waitForEvents(deletedPred); + const deleted = events.filter(deletedPred); + assert.ok(deleted.length >= 1, "expected a ContainerDeleted event"); + assert.strictEqual(deleted[0].data.api, "DeleteContainer"); + }); + + it("captures BlobCreated (PutBlob) and BlobDeleted for a block blob", async () => { + const containerName = getUniqueName("evt-b"); + const blobName = getUniqueName("blob"); + const containerClient = serviceClient.getContainerClient(containerName); + await containerClient.create(); + + const blockBlobClient = containerClient.getBlockBlobClient(blobName); + const body = "hello events"; + await blockBlobClient.upload(body, body.length); + await blockBlobClient.delete(); + + const createdPred = (e: any) => + e.eventType === "Microsoft.Storage.BlobCreated" && + e.data.api === "PutBlob" && + e.subject.endsWith(`/blobs/${blobName}`); + const deletedPred = (e: any) => + e.eventType === "Microsoft.Storage.BlobDeleted" && + e.subject.endsWith(`/blobs/${blobName}`); + + // Both writes are independent fire-and-forget, so wait for each in turn. + await waitForEvents(createdPred); + const events = await waitForEvents(deletedPred); + + const created = events.filter(createdPred); + assert.ok(created.length >= 1, "expected a BlobCreated PutBlob event"); + assert.strictEqual(created[0].data.contentLength, body.length); + assert.strictEqual(created[0].data.blobType, "BlockBlob"); + assert.ok(created[0].data.eTag && created[0].data.eTag.length > 0); + assert.strictEqual(created[0].metadataVersion, "1"); + // The persisted URL must never carry a query string (no SAS/secret leak). + assert.ok( + typeof created[0].data.url === "string" && + !created[0].data.url.includes("?"), + "data.url must be present and query-free" + ); + + const deleted = events.filter(deletedPred); + assert.ok(deleted.length >= 1, "expected a BlobDeleted event"); + assert.strictEqual(deleted[0].data.api, "DeleteBlob"); + + await containerClient.delete(); + }); + + it("captures BlobCreated (PutBlock then PutBlockList) on commit", async () => { + const containerName = getUniqueName("evt-bl"); + const blobName = getUniqueName("blob"); + const containerClient = serviceClient.getContainerClient(containerName); + await containerClient.create(); + + const blockBlobClient = containerClient.getBlockBlobClient(blobName); + const b64 = (s: string) => Buffer.from(s).toString("base64"); + await blockBlobClient.stageBlock(b64("id1"), "part1", 5); + await blockBlobClient.commitBlockList([b64("id1")]); + + const forThisBlob = (e: any) => e.subject.endsWith(`/blobs/${blobName}`); + // PutBlock and PutBlockList are separate fire-and-forget writes and may + // land in either order; wait for both before asserting. + await waitForEvents((e) => forThisBlob(e) && e.data.api === "PutBlock"); + const events = await waitForEvents( + (e) => forThisBlob(e) && e.data.api === "PutBlockList" + ); + + const apis = events.filter(forThisBlob).map((e) => e.data.api); + assert.ok(apis.includes("PutBlock"), "expected a PutBlock event"); + assert.ok(apis.includes("PutBlockList"), "expected a PutBlockList event"); + + await containerClient.delete(); + }); +}); + +describe("Blob Event Capture disabled by default @loki @sql", () => { + const factory = new BlobTestServerFactory(); + const eventFolder = "__test_event_capture_off__"; + const server = factory.createServer(); // no capture args -> disabled + + const baseURL = `http://${server.config.host}:${server.config.port}/devstoreaccount1`; + const serviceClient = new BlobServiceClient( + baseURL, + newPipeline( + new StorageSharedKeyCredential( + EMULATOR_ACCOUNT_NAME, + EMULATOR_ACCOUNT_KEY + ), + { retryOptions: { maxTries: 1 } } + ) + ); + + before(async () => { + if (fs.existsSync(eventFolder)) { + fs.removeSync(eventFolder); + } + await server.start(); + }); + + after(async () => { + await server.close(); + await server.clean(); + if (fs.existsSync(eventFolder)) { + fs.removeSync(eventFolder); + } + }); + + it("writes no event files when capture is off", async () => { + const containerName = getUniqueName("evt-off"); + const containerClient = serviceClient.getContainerClient(containerName); + await containerClient.create(); + const blockBlobClient = containerClient.getBlockBlobClient(getUniqueName("b")); + await blockBlobClient.upload("x", 1); + + // Best-effort wait: give any (erroneous) write a chance to happen before + // asserting the folder was never created. Polling can't prove a negative, + // so a short fixed wait is the right tool here. + await new Promise((r) => setTimeout(r, 300)); + + assert.ok( + !fs.existsSync(eventFolder), + "no event folder should be created when capture is disabled" + ); + + await containerClient.delete(); + }); +}); diff --git a/tests/blob/resolveBlobEventCapturePath.test.ts b/tests/blob/resolveBlobEventCapturePath.test.ts new file mode 100644 index 000000000..6b81d3c1c --- /dev/null +++ b/tests/blob/resolveBlobEventCapturePath.test.ts @@ -0,0 +1,43 @@ +import * as assert from "assert"; +import { isAbsolute, join } from "path"; + +import { resolveBlobEventCapturePath } from "../../src/blob/events/resolveBlobEventCapturePath"; +import { DEFAULT_BLOB_EVENT_CAPTURE_PATH } from "../../src/blob/utils/constants"; + +describe("resolveBlobEventCapturePath @loki @sql", () => { + const location = join("some", "workspace"); + + it("returns empty string when capture is disabled", () => { + assert.strictEqual( + resolveBlobEventCapturePath(false, "anything", location), + "" + ); + assert.strictEqual(resolveBlobEventCapturePath(false, "", location), ""); + assert.strictEqual( + resolveBlobEventCapturePath(false, undefined, location), + "" + ); + }); + + it("falls back to the default folder under location when no path is configured", () => { + const expected = join(location, DEFAULT_BLOB_EVENT_CAPTURE_PATH); + assert.strictEqual(resolveBlobEventCapturePath(true, "", location), expected); + assert.strictEqual( + resolveBlobEventCapturePath(true, undefined, location), + expected + ); + }); + + it("resolves a relative configured path against location", () => { + assert.strictEqual( + resolveBlobEventCapturePath(true, "events", location), + join(location, "events") + ); + }); + + it("uses an absolute configured path verbatim", () => { + const abs = join(process.cwd(), "abs-events"); + assert.ok(isAbsolute(abs), "test fixture must be an absolute path"); + assert.strictEqual(resolveBlobEventCapturePath(true, abs, location), abs); + }); +});