From 72fb23f4d51db1b999a10b7d8bb50293a81af163 Mon Sep 17 00:00:00 2001 From: syuilo Date: Mon, 27 May 2019 16:54:47 +0900 Subject: Improve drive management --- src/queue/index.ts | 30 ++++++++++++++++++++++ src/queue/processors/db/delete-drive-files.ts | 2 +- src/queue/processors/object-storage/delete-file.ts | 22 ++++++++++++++++ src/queue/processors/object-storage/index.ts | 12 +++++++++ 4 files changed, 65 insertions(+), 1 deletion(-) create mode 100644 src/queue/processors/object-storage/delete-file.ts create mode 100644 src/queue/processors/object-storage/index.ts (limited to 'src/queue') diff --git a/src/queue/index.ts b/src/queue/index.ts index a010004f15..70233d8318 100644 --- a/src/queue/index.ts +++ b/src/queue/index.ts @@ -8,6 +8,7 @@ import { program } from '../argv'; import processDeliver from './processors/deliver'; import processInbox from './processors/inbox'; import processDb from './processors/db'; +import procesObjectStorage from './processors/object-storage'; import { queueLogger } from './logger'; import { DriveFile } from '../models/entities/drive-file'; @@ -34,9 +35,12 @@ function renderError(e: Error): any { export const deliverQueue = initializeQueue('deliver'); export const inboxQueue = initializeQueue('inbox'); export const dbQueue = initializeQueue('db'); +export const objectStorageQueue = initializeQueue('objectStorage'); const deliverLogger = queueLogger.createSubLogger('deliver'); const inboxLogger = queueLogger.createSubLogger('inbox'); +const dbLogger = queueLogger.createSubLogger('db'); +const objectStorageLogger = queueLogger.createSubLogger('objectStorage'); deliverQueue .on('waiting', (jobId) => deliverLogger.debug(`waiting id=${jobId}`)) @@ -54,6 +58,22 @@ inboxQueue .on('error', (job: any, err: Error) => inboxLogger.error(`error ${err}`, { job, e: renderError(err) })) .on('stalled', (job) => inboxLogger.warn(`stalled id=${job.id} activity=${job.data.activity ? job.data.activity.id : 'none'}`)); +dbQueue + .on('waiting', (jobId) => dbLogger.debug(`waiting id=${jobId}`)) + .on('active', (job) => dbLogger.debug(`active id=${job.id}`)) + .on('completed', (job, result) => dbLogger.debug(`completed(${result}) id=${job.id}`)) + .on('failed', (job, err) => dbLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) })) + .on('error', (job: any, err: Error) => dbLogger.error(`error ${err}`, { job, e: renderError(err) })) + .on('stalled', (job) => dbLogger.warn(`stalled id=${job.id}`)); + +objectStorageQueue + .on('waiting', (jobId) => objectStorageLogger.debug(`waiting id=${jobId}`)) + .on('active', (job) => objectStorageLogger.debug(`active id=${job.id}`)) + .on('completed', (job, result) => objectStorageLogger.debug(`completed(${result}) id=${job.id}`)) + .on('failed', (job, err) => objectStorageLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) })) + .on('error', (job: any, err: Error) => objectStorageLogger.error(`error ${err}`, { job, e: renderError(err) })) + .on('stalled', (job) => objectStorageLogger.warn(`stalled id=${job.id}`)); + export function deliver(user: ILocalUser, content: any, to: any) { if (content == null) return null; @@ -165,11 +185,21 @@ export function createImportUserListsJob(user: ILocalUser, fileId: DriveFile['id }); } +export function createDeleteObjectStorageFileJob(key: string) { + return objectStorageQueue.add('deleteFile', { + key: key + }, { + removeOnComplete: true, + removeOnFail: true + }); +} + export default function() { if (!program.onlyServer) { deliverQueue.process(128, processDeliver); inboxQueue.process(128, processInbox); processDb(dbQueue); + procesObjectStorage(objectStorageQueue); } } diff --git a/src/queue/processors/db/delete-drive-files.ts b/src/queue/processors/db/delete-drive-files.ts index 4e4eab86b7..5ecee9139a 100644 --- a/src/queue/processors/db/delete-drive-files.ts +++ b/src/queue/processors/db/delete-drive-files.ts @@ -1,7 +1,7 @@ import * as Bull from 'bull'; import { queueLogger } from '../../logger'; -import deleteFile from '../../../services/drive/delete-file'; +import { deleteFile } from '../../../services/drive/delete-file'; import { Users, DriveFiles } from '../../../models'; import { MoreThan } from 'typeorm'; diff --git a/src/queue/processors/object-storage/delete-file.ts b/src/queue/processors/object-storage/delete-file.ts new file mode 100644 index 0000000000..8e6b5f959e --- /dev/null +++ b/src/queue/processors/object-storage/delete-file.ts @@ -0,0 +1,22 @@ +import * as Bull from 'bull'; +import * as Minio from 'minio'; +import { fetchMeta } from '../../../misc/fetch-meta'; + +export default async (job: Bull.Job) => { + const meta = await fetchMeta(); + + const minio = new Minio.Client({ + endPoint: meta.objectStorageEndpoint!, + region: meta.objectStorageRegion ? meta.objectStorageRegion : undefined, + port: meta.objectStoragePort ? meta.objectStoragePort : undefined, + useSSL: meta.objectStorageUseSSL, + accessKey: meta.objectStorageAccessKey!, + secretKey: meta.objectStorageSecretKey!, + }); + + const key: string = job.data.key; + + await minio.removeObject(meta.objectStorageBucket!, key); + + return 'Success'; +}; diff --git a/src/queue/processors/object-storage/index.ts b/src/queue/processors/object-storage/index.ts new file mode 100644 index 0000000000..f2fc08f1ab --- /dev/null +++ b/src/queue/processors/object-storage/index.ts @@ -0,0 +1,12 @@ +import * as Bull from 'bull'; +import deleteFile from './delete-file'; + +const jobs = { + deleteFile, +} as any; + +export default function(q: Bull.Queue) { + for (const [k, v] of Object.entries(jobs)) { + q.process(k, v as any); + } +} -- cgit v1.2.3-freya