From 6d443e4afda61dfa1a6596179c3fe266f9c6f4e7 Mon Sep 17 00:00:00 2001 From: Anton SHEPILOV Date: Thu, 10 Oct 2024 11:09:49 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8Resilience=20strategy=20to=20read=20an?= =?UTF-8?q?d=20write=20only=20to=20one=20storage?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tdrive/backend/node/config/default.json | 1 + tdrive/backend/node/src/core/config/index.ts | 2 +- .../storage/default-storage-strategy.ts | 42 ++++ .../storage/oneof-storage-strategy.ts | 115 +++++++++++ .../services/storage/storage-service.ts | 153 +++++++++++---- .../node/test/e2e/files/load_test_config.ts | 8 + .../test/e2e/files/oneof-storage/runtime.json | 21 ++ .../node/test/e2e/files/storage.oneof.spec.ts | 71 +++++++ .../storage/oneof-storage-strategy.test.ts | 184 ++++++++++++++++++ 9 files changed, 554 insertions(+), 43 deletions(-) create mode 100644 tdrive/backend/node/src/core/platform/services/storage/default-storage-strategy.ts create mode 100644 tdrive/backend/node/src/core/platform/services/storage/oneof-storage-strategy.ts create mode 100644 tdrive/backend/node/test/e2e/files/load_test_config.ts create mode 100644 tdrive/backend/node/test/e2e/files/oneof-storage/runtime.json create mode 100644 tdrive/backend/node/test/e2e/files/storage.oneof.spec.ts create mode 100644 tdrive/backend/node/test/unit/core/services/storage/oneof-storage-strategy.test.ts diff --git a/tdrive/backend/node/config/default.json b/tdrive/backend/node/config/default.json index ab77e647..2d72b797 100644 --- a/tdrive/backend/node/config/default.json +++ b/tdrive/backend/node/config/default.json @@ -93,6 +93,7 @@ "storage": { "secret": "0ea28a329df23220fa814e005bfb671c", "iv": "1234abcd00000000", + "strategy": "local", "type": "local", "S3": { "endPoint": "play.min.io", diff --git a/tdrive/backend/node/src/core/config/index.ts b/tdrive/backend/node/src/core/config/index.ts index 2f819dc2..f9e3cd2f 100644 --- a/tdrive/backend/node/src/core/config/index.ts +++ b/tdrive/backend/node/src/core/config/index.ts @@ -1,3 +1,3 @@ import config from "config"; -export default config; +export default config; \ No newline at end of file diff --git a/tdrive/backend/node/src/core/platform/services/storage/default-storage-strategy.ts b/tdrive/backend/node/src/core/platform/services/storage/default-storage-strategy.ts new file mode 100644 index 00000000..520fa320 --- /dev/null +++ b/tdrive/backend/node/src/core/platform/services/storage/default-storage-strategy.ts @@ -0,0 +1,42 @@ +import { Readable, Stream } from "stream"; +import { ExecutionContext } from "../../framework/api/crud-service"; +import { + DeleteOptions, + ReadOptions, + StorageConnectorAPI, + WriteMetadata, + WriteOptions, +} from "../storage/provider"; + +export class DefaultStorageStrategy implements StorageConnectorAPI { + private connector: StorageConnectorAPI; + + constructor(connector: StorageConnectorAPI) { + this.connector = connector; + } + + write = ( + path: string, + stream: Stream, + options?: WriteOptions, + context?: ExecutionContext, + ): Promise => { + return this.connector.write(path, stream, options, context); + }; + + read = (path: string, options?: ReadOptions, context?: ExecutionContext): Promise => { + return this.connector.read(path, options, context); + }; + + exists = (path: string, options?: ReadOptions, context?: ExecutionContext): Promise => { + return this.connector.exists(path, options, context); + }; + + remove = ( + path: string, + options?: DeleteOptions, + context?: ExecutionContext, + ): Promise => { + return this.connector.remove(path, options, context); + }; +} diff --git a/tdrive/backend/node/src/core/platform/services/storage/oneof-storage-strategy.ts b/tdrive/backend/node/src/core/platform/services/storage/oneof-storage-strategy.ts new file mode 100644 index 00000000..aaaa9c19 --- /dev/null +++ b/tdrive/backend/node/src/core/platform/services/storage/oneof-storage-strategy.ts @@ -0,0 +1,115 @@ +import { Readable, Stream, PassThrough } from "stream"; +import { + DeleteOptions, + ReadOptions, + StorageConnectorAPI, + WriteMetadata, + WriteOptions, +} from "../storage/provider"; +import { logger } from "../../../platform/framework"; + +/** + * OneOfStorageStrategy is responsible for managing multiple storage backends. + * It writes to, reads from, removes, and checks for the existence of files + * across multiple storage systems. + */ +export class OneOfStorageStrategy implements StorageConnectorAPI { + constructor(private readonly storages: StorageConnectorAPI[]) {} + + /** + * Writes a file to all configured storages. + * The write operation is considered successful if one of the storage succeed. + * @param path - The path where the file should be written. + * @param stream - The input stream to be written to storages. + * @param options - Write option: chunk number, encryption key, etc ... + * @throws Error if the write operation fails for one or more storages. + */ + write = async (path: string, stream: Stream, options?: WriteOptions): Promise => { + logger.debug("Creating streams for all storages ..."); + const passThroughStreams = this.storages.map(() => new PassThrough()); + + // destroy all streams if there is an error in the input stream + stream.on("error", err => { + logger.error("Error in input stream, destroying all write streams:", err); + passThroughStreams.forEach(stream => stream.destroy(err)); + }); + + // Pipe the input stream to each PassThrough stream + stream + .pipe(new PassThrough()) + .on("data", chunk => { + passThroughStreams.forEach(stream => stream.write(chunk)); + }) + .on("end", () => { + passThroughStreams.forEach(stream => stream.end()); + }); + + // Write to all storages with error handling + const writeResults = await Promise.allSettled( + this.storages.map( + (storage, index) => storage.write(path, passThroughStreams[index]), + options, + ), + ); + + // Log all errors and throw if all write operations fail + const errors = writeResults.filter(result => result.status === "rejected"); + errors.forEach((error, index) => { + logger.error(`Error writing to storage ${index}:`, (error as PromiseRejectedResult).reason); + }); + if (errors.length === this.storages.length) { + throw new Error("Write operation failed for all storages"); + } + + const successResult = writeResults.filter( + result => result.status === "fulfilled", + )[0] as PromiseFulfilledResult; + return successResult.value; + }; + + /** + * Reads a file from the primary storage. If it fails, attempts to read from other storages. + * @param path - The path of the file to be read. + * @param options + * @returns A readable stream of the file. + * @throws Error if all storage read attempts fail. + */ + read = async (path: string, options?: ReadOptions): Promise => { + for (const storage of this.storages) { + try { + if (await storage.exists(path, options)) { + return await storage.read(path, options); + } + } catch (err) { + logger.error("Fallback storage read failed:", err); + } + } + throw new Error(`Error reading ${path} in all the storages.`); + }; + + /** + * Checks if a file exists in any of the configured storages. + * @param path - The path of the file to check. + * @param options + * @returns A boolean indicating whether the file exists in any storage. + */ + exists = async (path: string, options?: ReadOptions): Promise => { + for (const storage of this.storages) { + if (await storage.exists(path, options)) { + return true; + } + } + return false; + }; + + /** + * Removes a file from all configured storages. + * @param path - The path of the file to be removed. + * @param options + */ + remove = async (path: string, options?: DeleteOptions): Promise => { + return Promise.all(this.storages.map(storage => storage.remove(path, options))).then(array => { + return array.reduce((a, b) => a && b); + }); + }; +} diff --git a/tdrive/backend/node/src/core/platform/services/storage/storage-service.ts b/tdrive/backend/node/src/core/platform/services/storage/storage-service.ts index 2618b8e5..7ef194a2 100644 --- a/tdrive/backend/node/src/core/platform/services/storage/storage-service.ts +++ b/tdrive/backend/node/src/core/platform/services/storage/storage-service.ts @@ -9,7 +9,7 @@ import { TdriveServiceOptions, } from "../../framework"; import LocalConnectorService, { LocalConfiguration } from "./connectors/local/service"; -import S3ConnectorService from "./connectors/S3/s3-service"; +import S3ConnectorService, { S3Configuration } from "./connectors/S3/s3-service"; import StorageAPI, { DeleteOptions, ReadOptions, @@ -18,6 +18,9 @@ import StorageAPI, { WriteOptions, } from "./provider"; +import { OneOfStorageStrategy } from "./oneof-storage-strategy"; +import { DefaultStorageStrategy } from "./default-storage-strategy"; + type EncryptionConfiguration = { secret: string | null; iv: string | null; @@ -39,22 +42,10 @@ export default class StorageService extends TdriveService implements constructor(protected options?: TdriveServiceOptions) { super(options); - const type = this.getConnectorType(); - if (type === "S3") { - logger.info("Using 'S3' connector for storage."); - try { - this.homeDir = this.configuration.get("S3.homeDirectory"); - } catch (e) { - this.logger.warn("Home directory is not set, using S3.bucket instead"); - } - if (!this.homeDir) { - this.homeDir = this.configuration.get("S3.bucket"); - } - if (this.homeDir && this.homeDir.startsWith("/")) { - this.logger.error("For S3 connector home directory MUST NOT start with '/'"); - throw new Error("For S3 connector home directory MUST NOT start with '/'"); - } - } + //init home directory variable + this.initHomeDirectory(); + //init connector to storage + this.getConnector(); } api(): StorageAPI { @@ -67,31 +58,7 @@ export default class StorageService extends TdriveService implements getConnector(): StorageConnectorAPI { if (!this.connector) { - const type = this.getConnectorType(); - if (type === "S3") { - logger.info("Using 'S3' connector for storage."); - this.connector = new S3ConnectorService({ - bucket: this.configuration.get("S3.bucket"), - region: this.configuration.get("S3.region"), - endPoint: this.configuration.get("S3.endPoint"), - port: Number(this.configuration.get("S3.port")), - useSSL: Boolean(this.configuration.get("S3.useSSL")), - accessKey: this.configuration.get("S3.accessKey"), - secretKey: this.configuration.get("S3.secretKey"), - disableRemove: this.configuration.get("S3.disableRemove"), - }); - } else { - logger.info( - `Using 'local' connector for storage${ - type === "local" ? "" : " (no other connector recognized from configuration type: '%s')" - }.`, - type, - ); - logger.trace(`Home directory for the storage: ${this.homeDir}`); - this.connector = new LocalConnectorService( - this.configuration.get("local"), - ); - } + this.connector = this.getStorageStrategy(); } return this.connector; } @@ -216,4 +183,106 @@ export default class StorageService extends TdriveService implements return this; } + + /** + * Instantiates storage strategy with configuration. + * "storage": { + * "secret": "", + * "iv": "", + * "strategy": "oneof", + * "type": "local", + * "S3": { + * "endPoint": "play.min.io", + * "port": 9000, + * "useSSL": false, + * "accessKey": "ABCD", + * "secretKey": "x1yz", + * "disableRemove": false + * }, + * "local": { + * "path": "/tdrive" + * }, + * "oneof": [ + * { + * "type": "S3", + * "S3": {} + * }, + * { + * "type": "local", + * "local": {} + * }] + * }, + */ + getStorageStrategy(): StorageConnectorAPI { + const strategy = this.configuration.get("strategy"); + if (strategy == "oneof") { + logger.info("Creating storage with 'oneof' strategy"); + const connectors = this.configuration + .get>("oneof") + .map(c => this.createConnector(c)); + return new OneOfStorageStrategy(connectors); + } else { + logger.info("Creating storage with 'default' strategy"); + return new DefaultStorageStrategy(this.createConnectorFromConfiguration(this.configuration)); + } + } + + createConnector(config: StorageConfiguration) { + const type = config.type; + if (type === "S3") { + return this.newS3Connector(config.S3); + } else { + return this.newLocalConnector(config.local, type); + } + } + + createConnectorFromConfiguration(config: TdriveServiceConfiguration) { + const type = config.get("type"); + if (type === "S3") { + return this.newS3Connector(config.get("S3")); + } else { + return this.newLocalConnector(this.configuration.get("local"), type); + } + } + + newS3Connector(config: S3Configuration) { + logger.info("Using 'S3' connector for storage."); + return new S3ConnectorService(config); + } + + newLocalConnector(config: LocalConfiguration, type: string) { + logger.info( + `Using 'local' connector for storage${ + type === "local" ? "" : " (no other connector recognized from configuration type: '%s')" + }.`, + type, + ); + logger.trace(`Home directory for the storage: ${this.homeDir}`); + return new LocalConnectorService(config); + } + + initHomeDirectory() { + const type = this.getConnectorType(); + if (type === "S3") { + logger.info("Using 'S3' connector for storage."); + try { + this.homeDir = this.configuration.get("S3.homeDirectory"); + } catch (e) { + this.logger.warn("Home directory is not set, using S3.bucket instead"); + } + if (!this.homeDir) { + this.homeDir = this.configuration.get("S3.bucket"); + } + if (this.homeDir && this.homeDir.startsWith("/")) { + this.logger.error("For S3 connector home directory MUST NOT start with '/'"); + throw new Error("For S3 connector home directory MUST NOT start with '/'"); + } + } + } } + +type StorageConfiguration = { + type: string; + S3?: S3Configuration; + local?: LocalConfiguration; +}; diff --git a/tdrive/backend/node/test/e2e/files/load_test_config.ts b/tdrive/backend/node/test/e2e/files/load_test_config.ts new file mode 100644 index 00000000..dc6bd765 --- /dev/null +++ b/tdrive/backend/node/test/e2e/files/load_test_config.ts @@ -0,0 +1,8 @@ +// @ts-ignore +import path from "path"; +// @ts-ignore +import config from "config"; + +const ourConfigDir = path.join(__dirname, 'oneof-storage'); +let configs = config.util.loadFileConfigs(ourConfigDir); +config.util.extendDeep(config, configs); diff --git a/tdrive/backend/node/test/e2e/files/oneof-storage/runtime.json b/tdrive/backend/node/test/e2e/files/oneof-storage/runtime.json new file mode 100644 index 00000000..9cb0bc48 --- /dev/null +++ b/tdrive/backend/node/test/e2e/files/oneof-storage/runtime.json @@ -0,0 +1,21 @@ +{ + + "storage": { + "secret": "0ea28a329df23220fa814e005bfb671c", + "iv": "1234abcd00000000", + "strategy": "oneof", + "oneof": [ + { + "type": "local", + "local": { + "path": "-tmp/storage1" + } + }, + { + "type": "local", + "local": { + "path": "/tmp/storage2" + } + }] + } +} diff --git a/tdrive/backend/node/test/e2e/files/storage.oneof.spec.ts b/tdrive/backend/node/test/e2e/files/storage.oneof.spec.ts new file mode 100644 index 00000000..ae8002f1 --- /dev/null +++ b/tdrive/backend/node/test/e2e/files/storage.oneof.spec.ts @@ -0,0 +1,71 @@ +import "./load_test_config" +import "reflect-metadata"; +import { afterAll, beforeAll, afterEach, describe, expect, it } from "@jest/globals"; +import { init, TestPlatform } from "../setup"; +import UserApi from "../common/user-api"; +import LocalConnectorService from "../../../src/core/platform/services/storage/connectors/local/service" +import { StorageConnectorAPI } from "../../../src/core/platform/services/storage/provider"; + +describe("The OneOf Storage feature", () => { + const url = "/internal/services/files/v1"; + let platform: TestPlatform; + let helpers: UserApi; + + beforeAll(async () => { + platform = await init({ + services: ["webserver", "database", "storage", "files", "previews"], + }); + helpers = await UserApi.getInstance(platform); + }, 300000000); + + afterEach(async () => { + jest.restoreAllMocks(); + }); + + afterAll(async () => { + await platform?.tearDown(); + platform = null; + }); + + describe("On file upload", () => { + + it("should fail an upload POST when ALL backend storage fails", async () => { + const thrower = () => { + throw new Error(""); + }; + const writeLocalMock = jest.spyOn(LocalConnectorService.prototype, "write").mockRejectedValue("Error"); + + // expect(response.statusCode).toBe(500); + await expect(helpers.uploadRandomFile()).rejects.toThrow("Error code: 500"); + + // expect(response.statusCode).toBe(500); + expect(writeLocalMock.mock.calls.length).toEqual(2); + }); + + it("should successfully upload file when one backend storage fails", async () => { + const thrower = () => { + throw new Error(""); + }; + + const connectors = (platform.storage.getConnector() as any).storages as Array; + + const writeLocalMock = jest.spyOn(connectors[0], "write").mockRejectedValue("Error"); + + // expect(response.statusCode).toBe(200); + const filesUpload = await helpers.uploadRandomFile(); + expect(filesUpload.id).toBeTruthy(); + + // expect failed upload + expect(writeLocalMock.mock.calls.length).toEqual(1); + //expect that file can be downloaded + const fileDownloadResponse = await platform.app.inject({ + method: "GET", + url: `${url}/companies/${platform.workspace.company_id}/files/${filesUpload.id}/download`, + }); + expect(fileDownloadResponse).toBeTruthy(); + expect(fileDownloadResponse.statusCode).toBe(200); + + }); + + }); +}); diff --git a/tdrive/backend/node/test/unit/core/services/storage/oneof-storage-strategy.test.ts b/tdrive/backend/node/test/unit/core/services/storage/oneof-storage-strategy.test.ts new file mode 100644 index 00000000..3a46b272 --- /dev/null +++ b/tdrive/backend/node/test/unit/core/services/storage/oneof-storage-strategy.test.ts @@ -0,0 +1,184 @@ +import { Readable, PassThrough } from 'stream'; +import { OneOfStorageStrategy } from "../../../../../src/core/platform/services/storage/oneof-storage-strategy"; // Adjust the import path as necessary +import {StorageConnectorAPI } from "../../../../../src/core/platform/services/storage/provider"; // Adjust the import path as necessary + +function mockStream() { + const stream = new Readable(); + stream.push("Test data"); + stream.push(null); + return stream; +} + +function mockStorage(size: number) { + return { + write: jest.fn().mockResolvedValue({ size: size }), + read: jest.fn(), + exists: jest.fn(), + remove: jest.fn().mockResolvedValue(true) + } as unknown as StorageConnectorAPI; +} + +describe('OneOfStorageStrategy', () => { + let storage1: StorageConnectorAPI; + let storage2: StorageConnectorAPI; + let storage3: StorageConnectorAPI; + let oneOfStorageStrategy: OneOfStorageStrategy; + + beforeEach(() => { + // Mocking two storage backends + storage1 = mockStorage(100); + storage2 = mockStorage(200); + storage3 = mockStorage(300); + + oneOfStorageStrategy = new OneOfStorageStrategy([storage1, storage2, storage3]); + }); + + afterEach(() => { + jest.clearAllMocks(); + }); + + describe('write', () => { + it('should write to all configured storages', async () => { + const path = 'test/path'; + const stream = mockStream(); + + const result = await oneOfStorageStrategy.write(path, stream); + + expect(storage1.write).toHaveBeenCalled(); + expect(storage2.write).toHaveBeenCalled(); + expect(result.size).toBe(100); // Assuming the first storage returned this size + }); + + it('should throw an error if all storage writes fail', async () => { + storage1.write = jest.fn().mockRejectedValue(new Error('Write failed')); + storage2.write = jest.fn().mockRejectedValue(new Error('Write failed')); + storage3.write = jest.fn().mockRejectedValue(new Error('Write failed')); + + const path = 'test/path'; + const stream = mockStream(); + await expect(oneOfStorageStrategy.write(path, stream)).rejects.toThrow('Write operation failed for all storages'); + }); + + it('should succeed if one storage write fails but the other succeeds', async () => { + //given + storage1.write = jest.fn().mockRejectedValue(new Error('Write failed')); + storage2.write = jest.fn().mockResolvedValue({ size: 200 }); + storage3.write = jest.fn().mockRejectedValue(new Error('Write failed')); + + const path = 'test/path'; + const stream = mockStream(); + + //when + const result = await oneOfStorageStrategy.write(path, stream); + + //then + expect(result.size).toBe(200); // Assuming the second storage returned this size + }); + + it('should verify that all data was written to the storage stream', async () => { + const data = "Test data"; + const stream = new Readable(); + stream.push(data); + stream.push(null); + + storage1.write = jest.fn().mockImplementation((_, stream) => { + return new Promise((resolve, reject) => { + let writtenData = ''; + stream.on('data', chunk => { + writtenData += chunk; + }); + stream.on('end', () => { + if (writtenData === data) { + resolve({ size: writtenData.length }); + } else { + reject(new Error('Data mismatch')); + } + }); + }); + }); + + const result = await oneOfStorageStrategy.write("test/path", stream); + + expect(result.size).toBe(data.length); + }); + }); + + describe('read', () => { + it('should read from the first storage that has the file', async () => { + const path = 'test/path'; + const mockStream = new Readable(); + mockStream.push('Test data'); + mockStream.push(null); + storage1.read = jest.fn().mockResolvedValue(mockStream); + storage1.exists = jest.fn().mockResolvedValue(true); + + const result = await oneOfStorageStrategy.read(path); + + expect(storage1.exists).toHaveBeenCalledWith(path, undefined); + expect(storage1.read).toHaveBeenCalledWith(path, undefined); + expect(result).toBe(mockStream); + }); + + it('should fallback to the next storage if the first fails', async () => { + const path = 'test/path'; + storage1.read = jest.fn().mockRejectedValue(new Error('Read failed')); + storage1.exists = jest.fn().mockResolvedValue(true); + const mockStream = new Readable(); + mockStream.push('Test data'); + mockStream.push(null); + storage2.read = jest.fn().mockResolvedValue(mockStream); + storage2.exists = jest.fn().mockResolvedValue(true); + + const result = await oneOfStorageStrategy.read(path); + + expect(storage1.exists).toHaveBeenCalled(); + expect(storage1.read).toHaveBeenCalled(); + expect(storage2.exists).toHaveBeenCalled(); + expect(storage2.read).toHaveBeenCalled(); + expect(result).toBe(mockStream); + }); + }); + + describe('exists', () => { + it('should return true if the file exists in any storage', async () => { + storage1.exists = jest.fn().mockResolvedValue(false); + storage2.exists = jest.fn().mockResolvedValue(true); + + const result = await oneOfStorageStrategy.exists('test/path'); + + expect(result).toBe(true); + expect(storage1.exists).toHaveBeenCalled(); + expect(storage2.exists).toHaveBeenCalled(); + }); + + it('should return false if the file does not exist in any storage', async () => { + storage1.exists = jest.fn().mockResolvedValue(false); + storage2.exists = jest.fn().mockResolvedValue(false); + + const result = await oneOfStorageStrategy.exists('test/path'); + + expect(result).toBe(false); + expect(storage1.exists).toHaveBeenCalled(); + expect(storage2.exists).toHaveBeenCalled(); + }); + }); + + describe('remove', () => { + it('should remove the file from all storages', async () => { + const result = await oneOfStorageStrategy.remove('test/path'); + + expect(storage1.remove).toHaveBeenCalledWith('test/path', undefined); + expect(storage2.remove).toHaveBeenCalledWith('test/path', undefined); + expect(result).toBe(true); + }); + + it('should return false if any storage fails to remove the file', async () => { + storage1.remove = jest.fn().mockResolvedValue(true); + storage2.remove = jest.fn().mockResolvedValue(false); + + const result = await oneOfStorageStrategy.remove('test/path'); + + expect(result).toBe(false); + }); + }); +}); \ No newline at end of file