🤖 TEST: Test on OSS NFS store (#24)

.github/workflows/nodejs.yml

@@ -16,7 +16,7 @@ on:
     - cron: '0 2 * * *'
 
 jobs:
-  test-mysql57:
+  test-mysql57-fs-nfs:
     runs-on: ${{ matrix.os }}
 
     services:
@@ -54,3 +54,47 @@ jobs:
       uses: codecov/codecov-action@v1
       with:
         token: ${{ secrets.CODECOV_TOKEN }}
+  test-mysql57-oss-nfs:
+    runs-on: ${{ matrix.os }}
+
+    services:
+      mysql:
+        image: mysql:5.7
+        env:
+          MYSQL_ALLOW_EMPTY_PASSWORD: true
+          MYSQL_DATABASE: cnpmcore_unittest
+        ports:
+          - 3306:3306
+        options: --health-cmd="mysqladmin ping" --health-interval=10s --health-timeout=5s --health-retries=5
+
+    strategy:
+      fail-fast: false
+      matrix:
+        node-version: [16]
+        os: [ubuntu-latest]
+
+    steps:
+    - name: Checkout Git Source
+      uses: actions/checkout@v2
+
+    - name: Use Node.js ${{ matrix.node-version }}
+      uses: actions/setup-node@v1
+      with:
+        node-version: ${{ matrix.node-version }}
+
+    - name: Install Dependencies
+      run: npm i -g npminstall && npminstall
+
+    - name: Continuous Integration
+      run: npm run ci
+      env:
+        CNPMCORE_NFS_TYPE: oss
+        CNPMCORE_NFS_OSS_BUCKET: cnpmcore-unittest-github
+        CNPMCORE_NFS_OSS_ENDPOINT: https://oss-us-west-1.aliyuncs.com
+        CNPMCORE_NFS_OSS_ID: ${{ secrets.CNPMCORE_NFS_OSS_ID }}
+        CNPMCORE_NFS_OSS_SECRET: ${{ secrets.CNPMCORE_NFS_OSS_SECRET }}
+
+    - name: Code Coverage
+      uses: codecov/codecov-action@v1
+      with:
+        token: ${{ secrets.CODECOV_TOKEN }}

LICENSE

@@ -1,6 +1,6 @@
 MIT License
 
-Copyright (c) 2021 cnpm and other contributors.
+Copyright (c) 2021-present cnpm and other contributors.
 
 Permission is hereby granted, free of charge, to any person obtaining a copy
 of this software and associated documentation files (the "Software"), to deal

app/common/adapter/NFSAdapter.ts

@@ -23,11 +23,28 @@ export class NFSAdapter {
 
   async uploadBytes(storeKey: string, bytes: Uint8Array) {
     this.logger.info('[%s:uploadBytes] key: %s, bytes: %d', INSTANCE_NAME, storeKey, bytes.length);
+    if (this.nfsClientAdapter.client.uploadBytes) {
+      return await this.nfsClientAdapter.client.uploadBytes(bytes, { key: storeKey });
+    }
     await this.nfsClientAdapter.client.uploadBuffer(bytes, { key: storeKey });
   }
 
-  async appendBytes(storeKey: string, bytes: Uint8Array) {
-    await this.nfsClientAdapter.client.appendBuffer(bytes, { key: storeKey });
+  // will return next store position
+  async appendBytes(storeKey: string, bytes: Uint8Array, position?: string, headers?: object) {
+    let result;
+    // make sure position is undefined by the first time
+    if (!position) position = undefined;
+    const options = {
+      key: storeKey,
+      position,
+      headers,
+    };
+    if (this.nfsClientAdapter.client.appendBytes) {
+      result = await this.nfsClientAdapter.client.appendBytes(bytes, options);
+    } else {
+      result = await this.nfsClientAdapter.client.appendBuffer(bytes, options);
+    }
+    if (result?.nextAppendPosition) return String(result.nextAppendPosition);
   }
 
   async uploadFile(storeKey: string, file: string) {

app/core/entity/Task.ts

@@ -12,6 +12,7 @@ interface TaskData extends EntityData {
   authorIp: string;
   data: unknown;
   logPath?: string;
+  logStorePosition?: string;
 }
 
 export type SyncPackageTaskOptions = {
@@ -30,6 +31,7 @@ export class Task extends Entity {
   authorIp: string;
   data: unknown;
   logPath: string;
+  logStorePosition: string;
 
   constructor(data: TaskData) {
     super(data);
@@ -41,6 +43,7 @@ export class Task extends Entity {
     this.authorIp = data.authorIp;
     this.data = data.data;
     this.logPath = data.logPath ?? '';
+    this.logStorePosition = data.logStorePosition ?? '';
   }
 
   private static create(data: EasyData<TaskData, 'taskId'>): Task {

app/core/service/PackageSyncerService.ts

@@ -337,14 +337,33 @@ export class PackageSyncerService extends AbstractService {
   }
 
   private async appendTaskLog(task: Task, appendLog: string) {
-    await this.nfsAdapter.appendBytes(task.logPath, Buffer.from(appendLog + '\n'));
+    const nextPosition = await this.nfsAdapter.appendBytes(
+      task.logPath,
+      Buffer.from(appendLog + '\n'),
+      task.logStorePosition,
+      {
+        'Content-Type': 'text/plain; charset=utf-8',
+      },
+    );
+    if (nextPosition) {
+      task.logStorePosition = nextPosition;
+    }
     task.updatedAt = new Date();
     await this.taskRepository.saveTask(task);
   }
 
   private async finishTask(task: Task, taskState: TaskState, appendLog: string) {
-    // console.log(appendLog);
-    await this.nfsAdapter.appendBytes(task.logPath, Buffer.from(appendLog + '\n'));
+    const nextPosition = await this.nfsAdapter.appendBytes(
+      task.logPath,
+      Buffer.from(appendLog + '\n'),
+      task.logStorePosition,
+      {
+        'Content-Type': 'text/plain; charset=utf-8',
+      },
+    );
+    if (nextPosition) {
+      task.logStorePosition = nextPosition;
+    }
     task.state = taskState;
     await this.taskRepository.saveTaskToHistory(task);
   }

app/repository/model/HistoryTask.ts

@@ -42,6 +42,9 @@ export class HistoryTask extends Bone {
   @Attribute(DataTypes.STRING(512))
   logPath: string;
 
+  @Attribute(DataTypes.STRING(10))
+  logStorePosition: string;
+
   @Attribute(DataTypes.INTEGER)
   attempts: number;
 }

app/repository/model/Task.ts

@@ -42,6 +42,9 @@ export class Task extends Bone {
   @Attribute(DataTypes.STRING(512))
   logPath: string;
 
+  @Attribute(DataTypes.STRING(10))
+  logStorePosition: string;
+
   @Attribute(DataTypes.INTEGER)
   attempts: number;
 }

config/config.default.ts

@@ -1,6 +1,8 @@
+import assert from 'assert';
 import { join } from 'path';
 import { tmpdir } from 'os';
 import { EggAppConfig, PowerPartial } from 'egg';
+import OSSClient from 'oss-cnpm';
 import { patchAjv } from '../app/port/typebox';
 
 export default (/* appInfo: EggAppConfig */) => {
@@ -58,6 +60,22 @@ export default (/* appInfo: EggAppConfig */) => {
     client: null,
     dir: join(config.dataDir, 'nfs'),
   };
+  // enable oss nfs store by env values
+  if (process.env.CNPMCORE_NFS_TYPE === 'oss') {
+    assert(process.env.CNPMCORE_NFS_OSS_BUCKET, 'require env CNPMCORE_NFS_OSS_BUCKET');
+    assert(process.env.CNPMCORE_NFS_OSS_ENDPOINT, 'require env CNPMCORE_NFS_OSS_ENDPOINT');
+    assert(process.env.CNPMCORE_NFS_OSS_ID, 'require env CNPMCORE_NFS_OSS_ID');
+    assert(process.env.CNPMCORE_NFS_OSS_SECRET, 'require env CNPMCORE_NFS_OSS_SECRET');
+    config.nfs.client = new OSSClient({
+      bucket: process.env.CNPMCORE_NFS_OSS_BUCKET,
+      endpoint: process.env.CNPMCORE_NFS_OSS_ENDPOINT,
+      accessKeyId: process.env.CNPMCORE_NFS_OSS_ID,
+      accessKeySecret: process.env.CNPMCORE_NFS_OSS_SECRET,
+      defaultHeaders: {
+        'Cache-Control': 'max-age=0, s-maxage=60',
+      },
+    });
+  }
 
   config.logger = {
     enablePerformanceTimer: true,

config/config.unittest.ts

@@ -9,5 +9,9 @@ export default () => {
   config.orm = {
     database: process.env.MYSQL_DATABASE || 'cnpmcore_unittest',
   };
+
+  config.nfs = {
+    dir: join(config.dataDir, 'nfs'),
+  };
   return config;
 };

package.json

@@ -56,6 +56,7 @@
     "leoric": "^1.15.0",
     "mysql": "^2.18.1",
     "mysql2": "^2.3.0",
+    "oss-cnpm": "^3.0.1",
     "semver": "^7.3.5",
     "ssri": "^8.0.1",
     "type-fest": "^2.5.3",

sql/init.sql

@@ -203,6 +203,7 @@ CREATE TABLE IF NOT EXISTS `tasks` (
   `author_ip` varchar(100) NOT NULL COMMENT 'create task user request ip',
   `data` json NULL COMMENT 'task params',
   `log_path` varchar(512) NOT NULL COMMENT 'access path',
+  `log_store_position` varchar(10) NOT NULL COMMENT 'cloud store disk position',
   `attempts` int unsigned DEFAULT 0 COMMENT 'task execute attempts times',
   PRIMARY KEY (`id`),
   UNIQUE KEY `uk_task_id` (`task_id`),
@@ -223,6 +224,7 @@ CREATE TABLE IF NOT EXISTS `history_tasks` (
   `author_ip` varchar(100) NOT NULL COMMENT 'create task user request ip',
   `data` json NULL COMMENT 'task params',
   `log_path` varchar(512) NOT NULL COMMENT 'access path',
+  `log_store_position` varchar(10) NOT NULL COMMENT 'cloud store disk position',
   `attempts` int unsigned DEFAULT 0 COMMENT 'task execute attempts times',
   PRIMARY KEY (`id`),
   UNIQUE KEY `uk_task_id` (`task_id`)

test/TestUtil.ts

@@ -1,4 +1,5 @@
 import * as fs from 'fs/promises';
+import { Readable } from 'stream';
 import mysql from 'mysql';
 import path from 'path';
 import crypto from 'crypto';
@@ -245,4 +246,19 @@ export class TestUtil {
       authorization: `Bearer ${token}`,
     };
   }
+
+  static async readStreamToLog(urlOrStream) {
+    let stream: Readable;
+    if (typeof urlOrStream === 'string') {
+      const { res } = await this.app.curl(urlOrStream, { streaming: true });
+      stream = res;
+    } else {
+      stream = urlOrStream;
+    }
+    const chunks: any[] = [];
+    for await (const chunk of stream) {
+      chunks.push(chunk);
+    }
+    return Buffer.concat(chunks).toString();
+  }
 }