[Code] disk watermark check for clone and update (#42174) (#42383)

* [Code] disk watermark check

* apply disk check

* switch to one config

* Add i18n

* Add unit tests

* update i18n message id

package.json

@@ -140,6 +140,7 @@
     "brace": "0.11.1",
     "cache-loader": "^4.0.1",
     "chalk": "^2.4.1",
+    "check-disk-space": "^2.1.0",
     "color": "1.0.3",
     "commander": "2.20.0",
     "compare-versions": "3.4.0",

x-pack/legacy/plugins/code/index.ts

@@ -102,6 +102,10 @@ export const code = (kibana: any) =>
             .default(['https', 'git', 'ssh']),
           enableGitCertCheck: Joi.boolean().default(true),
         }).default(),
+        disk: Joi.object({
+          thresholdEnabled: Joi.bool().default(true),
+          watermarkLowMb: Joi.number().default(2048),
+        }).default(),
         maxWorkspace: Joi.number().default(5), // max workspace folder for each language server
         enableGlobalReference: Joi.boolean().default(false), // Global reference as optional feature for now
         codeNodeUrl: Joi.string(),

x-pack/legacy/plugins/code/server/__tests__/clone_worker.ts

@@ -13,6 +13,7 @@ import rimraf from 'rimraf';
 import sinon from 'sinon';
 
 import { Repository } from '../../model';
+import { DiskWatermarkService } from '../disk_watermark';
 import { GitOperations } from '../git_operations';
 import { EsClient, Esqueue } from '../lib/esqueue';
 import { Logger } from '../log';
@@ -106,6 +107,12 @@ describe('clone_worker_tests', () => {
     cancellationService.cancelCloneJob = cancelCloneJobSpy;
     cancellationService.registerCancelableCloneJob = registerCancelableCloneJobSpy;
 
+    // Setup DiskWatermarkService
+    const isLowWatermarkSpy = sinon.stub().resolves(false);
+    const diskWatermarkService: any = {
+      isLowWatermark: isLowWatermarkSpy,
+    };
+
     const cloneWorker = new CloneWorker(
       esQueue as Esqueue,
       log,
@@ -114,7 +121,8 @@ describe('clone_worker_tests', () => {
       gitOps,
       {} as IndexWorker,
       (repoServiceFactory as any) as RepositoryServiceFactory,
-      cancellationService as CancellationSerivce
+      cancellationService as CancellationSerivce,
+      diskWatermarkService as DiskWatermarkService
     );
 
     await cloneWorker.executeJob({
@@ -125,6 +133,7 @@ describe('clone_worker_tests', () => {
       timestamp: 0,
     });
 
+    assert.ok(isLowWatermarkSpy.calledOnce);
     assert.ok(newInstanceSpy.calledOnce);
     assert.ok(cloneSpy.calledOnce);
   });
@@ -154,6 +163,12 @@ describe('clone_worker_tests', () => {
     cancellationService.cancelCloneJob = cancelCloneJobSpy;
     cancellationService.registerCancelableCloneJob = registerCancelableCloneJobSpy;
 
+    // Setup DiskWatermarkService
+    const isLowWatermarkSpy = sinon.stub().resolves(false);
+    const diskWatermarkService: any = {
+      isLowWatermark: isLowWatermarkSpy,
+    };
+
     const cloneWorker = new CloneWorker(
       esQueue as Esqueue,
       log,
@@ -162,7 +177,8 @@ describe('clone_worker_tests', () => {
       gitOps,
       (indexWorker as any) as IndexWorker,
       {} as RepositoryServiceFactory,
-      cancellationService as CancellationSerivce
+      cancellationService as CancellationSerivce,
+      diskWatermarkService as DiskWatermarkService
     );
 
     await cloneWorker.onJobCompleted(
@@ -190,6 +206,8 @@ describe('clone_worker_tests', () => {
     // Index request is issued after a 1s delay.
     await delay(1000);
     assert.ok(enqueueJobSpy.calledOnce);
+
+    assert.ok(isLowWatermarkSpy.notCalled);
   });
 
   it('On clone job completed because of cancellation', async () => {
@@ -217,6 +235,12 @@ describe('clone_worker_tests', () => {
     cancellationService.cancelCloneJob = cancelCloneJobSpy;
     cancellationService.registerCancelableCloneJob = registerCancelableCloneJobSpy;
 
+    // Setup DiskWatermarkService
+    const isLowWatermarkSpy = sinon.stub().resolves(false);
+    const diskWatermarkService: any = {
+      isLowWatermark: isLowWatermarkSpy,
+    };
+
     const cloneWorker = new CloneWorker(
       esQueue as Esqueue,
       log,
@@ -225,7 +249,8 @@ describe('clone_worker_tests', () => {
       gitOps,
       (indexWorker as any) as IndexWorker,
       {} as RepositoryServiceFactory,
-      cancellationService as CancellationSerivce
+      cancellationService as CancellationSerivce,
+      diskWatermarkService as DiskWatermarkService
     );
 
     await cloneWorker.onJobCompleted(
@@ -252,6 +277,8 @@ describe('clone_worker_tests', () => {
     // Index request should not be issued after clone request is done.
     await delay(1000);
     assert.ok(enqueueJobSpy.notCalled);
+
+    assert.ok(isLowWatermarkSpy.notCalled);
   });
 
   it('On clone job enqueued.', async () => {
@@ -272,6 +299,12 @@ describe('clone_worker_tests', () => {
     cancellationService.cancelCloneJob = cancelCloneJobSpy;
     cancellationService.registerCancelableCloneJob = registerCancelableCloneJobSpy;
 
+    // Setup DiskWatermarkService
+    const isLowWatermarkSpy = sinon.stub().resolves(false);
+    const diskWatermarkService: any = {
+      isLowWatermark: isLowWatermarkSpy,
+    };
+
     const cloneWorker = new CloneWorker(
       esQueue as Esqueue,
       log,
@@ -280,7 +313,8 @@ describe('clone_worker_tests', () => {
       gitOps,
       {} as IndexWorker,
       {} as RepositoryServiceFactory,
-      cancellationService as CancellationSerivce
+      cancellationService as CancellationSerivce,
+      diskWatermarkService as DiskWatermarkService
     );
 
     await cloneWorker.onJobEnqueued({
@@ -320,6 +354,12 @@ describe('clone_worker_tests', () => {
     cancellationService.cancelCloneJob = cancelCloneJobSpy;
     cancellationService.registerCancelableCloneJob = registerCancelableCloneJobSpy;
 
+    // Setup DiskWatermarkService
+    const isLowWatermarkSpy = sinon.stub().resolves(false);
+    const diskWatermarkService: any = {
+      isLowWatermark: isLowWatermarkSpy,
+    };
+
     const cloneWorker = new CloneWorker(
       esQueue as Esqueue,
       log,
@@ -328,7 +368,8 @@ describe('clone_worker_tests', () => {
       gitOps,
       {} as IndexWorker,
       (repoServiceFactory as any) as RepositoryServiceFactory,
-      cancellationService as CancellationSerivce
+      cancellationService as CancellationSerivce,
+      diskWatermarkService as DiskWatermarkService
     );
 
     const result1 = await cloneWorker.executeJob({
@@ -342,6 +383,7 @@ describe('clone_worker_tests', () => {
     assert.ok(result1.repo === null);
     assert.ok(newInstanceSpy.notCalled);
     assert.ok(cloneSpy.notCalled);
+    assert.ok(isLowWatermarkSpy.calledOnce);
 
     const result2 = await cloneWorker.executeJob({
       payload: {
@@ -354,5 +396,66 @@ describe('clone_worker_tests', () => {
     assert.ok(result2.repo === null);
     assert.ok(newInstanceSpy.notCalled);
     assert.ok(cloneSpy.notCalled);
+    assert.ok(isLowWatermarkSpy.calledTwice);
+  });
+
+  it('Execute clone job failed because of low disk watermark', async () => {
+    // Setup RepositoryService
+    const cloneSpy = sinon.spy();
+    const repoService = {
+      clone: emptyAsyncFunc,
+    };
+    repoService.clone = cloneSpy;
+    const repoServiceFactory = {
+      newInstance: (): void => {
+        return;
+      },
+    };
+    const newInstanceSpy = sinon.fake.returns(repoService);
+    repoServiceFactory.newInstance = newInstanceSpy;
+
+    // Setup CancellationService
+    const cancelCloneJobSpy = sinon.spy();
+    const registerCancelableCloneJobSpy = sinon.spy();
+    const cancellationService: any = {
+      cancelCloneJob: emptyAsyncFunc,
+      registerCancelableCloneJob: emptyAsyncFunc,
+    };
+    cancellationService.cancelCloneJob = cancelCloneJobSpy;
+    cancellationService.registerCancelableCloneJob = registerCancelableCloneJobSpy;
+
+    // Setup DiskWatermarkService
+    const isLowWatermarkSpy = sinon.stub().resolves(true);
+    const diskWatermarkService: any = {
+      isLowWatermark: isLowWatermarkSpy,
+    };
+
+    const cloneWorker = new CloneWorker(
+      esQueue as Esqueue,
+      log,
+      {} as EsClient,
+      serverOptions,
+      gitOps,
+      {} as IndexWorker,
+      (repoServiceFactory as any) as RepositoryServiceFactory,
+      cancellationService as CancellationSerivce,
+      diskWatermarkService as DiskWatermarkService
+    );
+
+    try {
+      await cloneWorker.executeJob({
+        payload: {
+          url: 'https://github.com/Microsoft/TypeScript-Node-Starter.git',
+        },
+        options: {},
+        timestamp: 0,
+      });
+      // This step should not be touched.
+      assert.ok(false);
+    } catch (error) {
+      assert.ok(isLowWatermarkSpy.calledOnce);
+      assert.ok(newInstanceSpy.notCalled);
+      assert.ok(cloneSpy.notCalled);
+    }
   });
 });

x-pack/legacy/plugins/code/server/disk_watermark.ts

@@ -0,0 +1,21 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+
+import checkDiskSpace from 'check-disk-space';
+
+export class DiskWatermarkService {
+  constructor(private readonly diskWatermarkLowMb: number, private readonly repoPath: string) {}
+
+  public async isLowWatermark(): Promise<boolean> {
+    try {
+      const { free } = await checkDiskSpace(this.repoPath);
+      const availableMb = free / 1024 / 1024;
+      return availableMb <= this.diskWatermarkLowMb;
+    } catch (err) {
+      return true;
+    }
+  }
+}

x-pack/legacy/plugins/code/server/init_workers.ts

@@ -5,6 +5,7 @@
  */
 
 import { Server } from 'hapi';
+import { DiskWatermarkService } from './disk_watermark';
 import { EsClient, Esqueue } from './lib/esqueue';
 import { LspService } from './lsp/lsp_service';
 import { GitOperations } from './git_operations';
@@ -43,6 +44,10 @@ export function initWorkers(
 
   const repoServiceFactory: RepositoryServiceFactory = new RepositoryServiceFactory();
 
+  const watermarkService = new DiskWatermarkService(
+    serverOptions.disk.watermarkLowMb,
+    serverOptions.repoPath
+  );
   const cloneWorker = new CloneWorker(
     queue,
     log,
@@ -51,7 +56,8 @@ export function initWorkers(
     gitOps,
     indexWorker,
     repoServiceFactory,
-    cancellationService
+    cancellationService,
+    watermarkService
   ).bind();
   const deleteWorker = new DeleteWorker(
     queue,
@@ -70,9 +76,9 @@ export function initWorkers(
     serverOptions,
     gitOps,
     repoServiceFactory,
-    cancellationService
+    cancellationService,
+    watermarkService
   ).bind();
-
   codeServices.registerHandler(
     RepositoryServiceDefinition,
     getRepositoryHandler(cloneWorker, deleteWorker, indexWorker)

x-pack/legacy/plugins/code/server/queue/abstract_git_worker.ts

@@ -4,12 +4,16 @@
  * you may not use this file except in compliance with the Elastic License.
  */
 
+import { i18n } from '@kbn/i18n';
+
 import {
   CloneProgress,
   CloneWorkerProgress,
   CloneWorkerResult,
   WorkerReservedProgress,
+  WorkerResult,
 } from '../../model';
+import { DiskWatermarkService } from '../disk_watermark';
 import { GitOperations } from '../git_operations';
 import { EsClient, Esqueue } from '../lib/esqueue';
 import { Logger } from '../log';
@@ -27,12 +31,34 @@ export abstract class AbstractGitWorker extends AbstractWorker {
     protected readonly log: Logger,
     protected readonly client: EsClient,
     protected readonly serverOptions: ServerOptions,
-    protected readonly gitOps: GitOperations
+    protected readonly gitOps: GitOperations,
+    protected readonly watermarkService: DiskWatermarkService
   ) {
     super(queue, log);
     this.objectClient = new RepositoryObjectClient(client);
   }
 
+  public async executeJob(_: Job): Promise<WorkerResult> {
+    const { thresholdEnabled, watermarkLowMb } = this.serverOptions.disk;
+    if (thresholdEnabled) {
+      const isLowWatermark = await this.watermarkService.isLowWatermark();
+      if (isLowWatermark) {
+        const msg = i18n.translate('xpack.code.git.diskWatermarkLowMessage', {
+          defaultMessage: `Disk watermark level lower than {watermarkLowMb} MB`,
+          values: {
+            watermarkLowMb,
+          },
+        });
+        this.log.error(msg);
+        throw new Error(msg);
+      }
+    }
+
+    return new Promise<WorkerResult>((resolve, reject) => {
+      resolve();
+    });
+  }
+
   public async onJobCompleted(job: Job, res: CloneWorkerResult) {
     if (res.cancelled) {
       // Skip updating job progress if the job is done because of cancellation.

x-pack/legacy/plugins/code/server/queue/clone_worker.ts

@@ -14,6 +14,7 @@ import {
   CloneWorkerResult,
   WorkerReservedProgress,
 } from '../../model';
+import { DiskWatermarkService } from '../disk_watermark';
 import { GitOperations } from '../git_operations';
 import { EsClient, Esqueue } from '../lib/esqueue';
 import { Logger } from '../log';
@@ -35,12 +36,15 @@ export class CloneWorker extends AbstractGitWorker {
     protected readonly gitOps: GitOperations,
     private readonly indexWorker: IndexWorker,
     private readonly repoServiceFactory: RepositoryServiceFactory,
-    private readonly cancellationService: CancellationSerivce
+    private readonly cancellationService: CancellationSerivce,
+    protected readonly watermarkService: DiskWatermarkService
   ) {
-    super(queue, log, client, serverOptions, gitOps);
+    super(queue, log, client, serverOptions, gitOps, watermarkService);
   }
 
   public async executeJob(job: Job) {
+    await super.executeJob(job);
+
     const { payload, cancellationToken } = job;
     const { url } = payload;
     try {

x-pack/legacy/plugins/code/server/queue/update_worker.test.ts

@@ -8,6 +8,7 @@ import sinon from 'sinon';
 
 import { EsClient, Esqueue } from '../lib/esqueue';
 import { Repository } from '../../model';
+import { DiskWatermarkService } from '../disk_watermark';
 import { GitOperations } from '../git_operations';
 import { Logger } from '../log';
 import { RepositoryServiceFactory } from '../repository_service_factory';
@@ -51,6 +52,12 @@ test('Execute update job', async () => {
   cancellationService.cancelUpdateJob = cancelUpdateJobSpy;
   cancellationService.registerCancelableUpdateJob = registerCancelableUpdateJobSpy;
 
+  // Setup DiskWatermarkService
+  const isLowWatermarkSpy = sinon.stub().resolves(false);
+  const diskWatermarkService: any = {
+    isLowWatermark: isLowWatermarkSpy,
+  };
+
   const updateWorker = new UpdateWorker(
     esQueue as Esqueue,
     log,
@@ -59,10 +66,15 @@ test('Execute update job', async () => {
       security: {
         enableGitCertCheck: true,
       },
+      disk: {
+        thresholdEnabled: true,
+        watermarkLowMb: 100,
+      },
     } as ServerOptions,
     {} as GitOperations,
     (repoServiceFactory as any) as RepositoryServiceFactory,
-    cancellationService as CancellationSerivce
+    cancellationService as CancellationSerivce,
+    diskWatermarkService as DiskWatermarkService
   );
 
   await updateWorker.executeJob({
@@ -73,6 +85,7 @@ test('Execute update job', async () => {
     timestamp: 0,
   });
 
+  expect(isLowWatermarkSpy.calledOnce).toBeTruthy();
   expect(newInstanceSpy.calledOnce).toBeTruthy();
   expect(updateSpy.calledOnce).toBeTruthy();
 });
@@ -99,10 +112,15 @@ test('On update job completed because of cancellation ', async () => {
       security: {
         enableGitCertCheck: true,
       },
+      disk: {
+        thresholdEnabled: true,
+        watermarkLowMb: 100,
+      },
     } as ServerOptions,
     {} as GitOperations,
     {} as RepositoryServiceFactory,
-    cancellationService as CancellationSerivce
+    cancellationService as CancellationSerivce,
+    {} as DiskWatermarkService
   );
 
   await updateWorker.onJobCompleted(
@@ -127,3 +145,71 @@ test('On update job completed because of cancellation ', async () => {
   // cancellation.
   expect(updateSpy.notCalled).toBeTruthy();
 });
+
+test('Execute update job failed because of low disk watermark ', async () => {
+  // Setup RepositoryService
+  const updateSpy = sinon.spy();
+  const repoService = {
+    update: emptyAsyncFunc,
+  };
+  repoService.update = updateSpy;
+  const repoServiceFactory = {
+    newInstance: (): void => {
+      return;
+    },
+  };
+  const newInstanceSpy = sinon.fake.returns(repoService);
+  repoServiceFactory.newInstance = newInstanceSpy;
+
+  // Setup CancellationService
+  const cancelUpdateJobSpy = sinon.spy();
+  const registerCancelableUpdateJobSpy = sinon.spy();
+  const cancellationService: any = {
+    cancelUpdateJob: emptyAsyncFunc,
+    registerCancelableUpdateJob: emptyAsyncFunc,
+  };
+  cancellationService.cancelUpdateJob = cancelUpdateJobSpy;
+  cancellationService.registerCancelableUpdateJob = registerCancelableUpdateJobSpy;
+
+  // Setup DiskWatermarkService
+  const isLowWatermarkSpy = sinon.stub().resolves(true);
+  const diskWatermarkService: any = {
+    isLowWatermark: isLowWatermarkSpy,
+  };
+
+  const updateWorker = new UpdateWorker(
+    esQueue as Esqueue,
+    log,
+    esClient as EsClient,
+    {
+      security: {
+        enableGitCertCheck: true,
+      },
+      disk: {
+        thresholdEnabled: true,
+        watermarkLowMb: 100,
+      },
+    } as ServerOptions,
+    {} as GitOperations,
+    {} as RepositoryServiceFactory,
+    cancellationService as CancellationSerivce,
+    diskWatermarkService as DiskWatermarkService
+  );
+
+  try {
+    await updateWorker.executeJob({
+      payload: {
+        uri: 'mockrepo',
+      },
+      options: {},
+      timestamp: 0,
+    });
+    // This step should not be touched.
+    expect(false).toBeTruthy();
+  } catch (error) {
+    // Exception should be thrown.
+    expect(isLowWatermarkSpy.calledOnce).toBeTruthy();
+    expect(newInstanceSpy.notCalled).toBeTruthy();
+    expect(updateSpy.notCalled).toBeTruthy();
+  }
+});

x-pack/legacy/plugins/code/server/queue/update_worker.ts

@@ -6,6 +6,7 @@
 
 import { CloneWorkerResult, Repository } from '../../model';
 import { EsClient, Esqueue } from '../lib/esqueue';
+import { DiskWatermarkService } from '../disk_watermark';
 import { GitOperations } from '../git_operations';
 import { Logger } from '../log';
 import { RepositoryServiceFactory } from '../repository_service_factory';
@@ -18,18 +19,21 @@ export class UpdateWorker extends AbstractGitWorker {
   public id: string = 'update';
 
   constructor(
-    queue: Esqueue,
+    protected readonly queue: Esqueue,
     protected readonly log: Logger,
     protected readonly client: EsClient,
     protected readonly serverOptions: ServerOptions,
     protected readonly gitOps: GitOperations,
     protected readonly repoServiceFactory: RepositoryServiceFactory,
-    private readonly cancellationService: CancellationSerivce
+    private readonly cancellationService: CancellationSerivce,
+    protected readonly watermarkService: DiskWatermarkService
   ) {
-    super(queue, log, client, serverOptions, gitOps);
+    super(queue, log, client, serverOptions, gitOps, watermarkService);
   }
 
   public async executeJob(job: Job) {
+    await super.executeJob(job);
+
     const { payload, cancellationToken } = job;
     const repo: Repository = payload;
     this.log.info(`Execute update job for ${repo.uri}`);

x-pack/legacy/plugins/code/server/server_options.ts

@@ -22,6 +22,11 @@ export interface SecurityOptions {
   enableGitCertCheck: boolean;
 }
 
+export interface DiskOptions {
+  thresholdEnabled: boolean;
+  watermarkLowMb: number;
+}
+
 export class ServerOptions {
   public readonly workspacePath = resolve(this.config.get('path.data'), 'code/workspace');
 
@@ -49,6 +54,8 @@ export class ServerOptions {
 
   public readonly security: SecurityOptions = this.options.security;
 
+  public readonly disk: DiskOptions = this.options.disk;
+
   public readonly repoConfigs: RepoConfigs = (this.options.repos as RepoConfig[]).reduce(
     (previous, current) => {
       previous[current.repo] = current;

x-pack/legacy/plugins/code/server/test_utils.ts

@@ -37,6 +37,10 @@ const TEST_OPTIONS = {
     enableGitCertCheck: true,
     gitProtocolWhitelist: ['ssh', 'https', 'git'],
   },
+  disk: {
+    thresholdEnabled: true,
+    watermarkLowMb: 100,
+  },
   repos: [],
   maxWorkspace: 5, // max workspace folder for each language server
 };

x-pack/plugins/translations/translations/zh-CN.json

@@ -4344,6 +4344,7 @@
     "xpack.code.adminPage.setupGuide.permissionChangesDescription": "Kibana 的角色和权限模型有新的改动，请了解这些改动对您 Code 部署的影响",
     "xpack.code.adminPage.setupGuide.permissionChangesTitle": "权限模型变动",
     "xpack.code.featureRegistry.codeFeatureName": "Code",
+    "xpack.code.git.diskWatermarkLowMessage": "存储空间低于{watermarkLowMb}MB",
     "xpack.code.gitUrlUtil.urlNotWhitelistedMessage": "Git URL 主机地址没有在白名单中。",
     "xpack.code.gitUrlUtil.protocolNotWhitelistedMessage": "Git URL 的协议没有在白名单中。",
     "xpack.code.helpMenu.codeDocumentationButtonLabel": "Code 文档",

yarn.lock

@@ -7865,6 +7865,11 @@ chardet@^0.7.0:
   resolved "https://registry.yarnpkg.com/chardet/-/chardet-0.7.0.tgz#90094849f0937f2eedc2425d0d28a9e5f0cbad9e"
   integrity sha512-mT8iDcrh03qDGRRmoA2hmBJnxpllMR+0/0qlzjqZES6NdiWDcZkCNAk4rPFZ9Q85r27unkiNNg8ZOiwZXBHwcA==
 
+check-disk-space@^2.1.0:
+  version "2.1.0"
+  resolved "https://registry.yarnpkg.com/check-disk-space/-/check-disk-space-2.1.0.tgz#2e77fe62f30d9676dc37a524ea2008f40c780295"
+  integrity sha512-f0nx9oJF/AVF8nhSYlF1EBvMNnO+CXyLwKhPvN1943iOMI9TWhQigLZm80jAf0wzQhwKkzA8XXjyvuVUeGGcVQ==
+
 check-error@^1.0.2:
   version "1.0.2"
   resolved "https://registry.yarnpkg.com/check-error/-/check-error-1.0.2.tgz#574d312edd88bb5dd8912e9286dd6c0aed4aac82"