UBER-1085: Improve upgrade tool

dev/client-resources/src/index.ts

@@ -29,7 +29,11 @@ export default async () => {
           client = await createClient(connect)
           for (const op of migrateOperations) {
             console.log('Migrate', op[0])
-            await op[1].upgrade(client)
+            await op[1].upgrade(client, {
+              log (...data) {
+                console.log(...data)
+              }
+            })
           }
         }
         // Check if we had dev hook for client.

dev/tool/src/index.ts

@@ -29,7 +29,8 @@ import {
   replacePassword,
   setAccountAdmin,
   setRole,
-  upgradeWorkspace
+  upgradeWorkspace,
+  WorkspaceInfoOnly
 } from '@hcengineering/account'
 import { setMetadata } from '@hcengineering/platform'
 import {
@@ -40,17 +41,18 @@ import {
   restore
 } from '@hcengineering/server-backup'
 import serverToken, { decodeToken, generateToken } from '@hcengineering/server-token'
-import toolPlugin from '@hcengineering/server-tool'
+import toolPlugin, { FileModelLogger } from '@hcengineering/server-tool'
 
 import { program } from 'commander'
 import { Db, MongoClient } from 'mongodb'
 import { clearTelegramHistory } from './telegram'
 import { diffWorkspace } from './workspace'
 
-import { Data, getWorkspaceId, Tx, Version } from '@hcengineering/core'
+import { Data, getWorkspaceId, RateLimitter, Tx, Version } from '@hcengineering/core'
 import { MinioService } from '@hcengineering/minio'
 import { MigrateOperation } from '@hcengineering/model'
 import { openAIConfigDefaults } from '@hcengineering/openai'
+import path from 'path'
 import { benchmark } from './benchmark'
 import {
   cleanArchivedSpaces,
@@ -232,19 +234,47 @@ export function devTool (
   program
     .command('upgrade')
     .description('upgrade')
-    .option('-p|--parallel', 'Parallel upgrade', false)
-    .action(async (cmd: { parallel: boolean }) => {
+    .option('-p|--parallel <parallel>', 'Parallel upgrade', '0')
+    .option('-l|--logs <logs>', 'Default logs folder', './logs')
+    .option('-r|--retry <retry>', 'Number of apply retries', '0')
+    .option('-f|--force [force]', 'Force update', false)
+    .action(async (cmd: { parallel: string, logs: string, retry: string, force: boolean }) => {
       const { mongodbUri, version, txes, migrateOperations } = prepareTools()
       return await withDatabase(mongodbUri, async (db) => {
         const workspaces = await listWorkspaces(db, productId)
-        if (cmd.parallel) {
-          await Promise.all(
-            workspaces.map((ws) => upgradeWorkspace(version, txes, migrateOperations, productId, db, ws.workspace))
-          )
+        const withError: string[] = []
+
+        async function _upgradeWorkspace (ws: WorkspaceInfoOnly): Promise<void> {
+          const t = Date.now()
+          const logger = new FileModelLogger(path.join(cmd.logs, `${ws.workspace}.log`))
+          console.log('---UPGRADING----', ws.workspace, logger.file)
+          try {
+            await upgradeWorkspace(version, txes, migrateOperations, productId, db, ws.workspace, logger, cmd.force)
+            console.log('---UPGRADING-DONE----', ws.workspace, Date.now() - t)
+          } catch (err: any) {
+            withError.push(ws.workspace)
+            logger.log('error', JSON.stringify(err))
+            console.log('---UPGRADING-FAILED----', ws.workspace, Date.now() - t)
+          } finally {
+            logger.close()
+          }
+        }
+        if (cmd.parallel !== '0') {
+          const parallel = parseInt(cmd.parallel) ?? 1
+          const rateLimit = new RateLimitter(() => ({ rate: parallel }))
+          console.log('parallel upgrade', parallel, cmd.parallel)
+          for (const ws of workspaces) {
+            await rateLimit.exec(() => {
+              return _upgradeWorkspace(ws)
+            })
+          }
         } else {
+          console.log('UPGRADE write logs at:', cmd.logs)
           for (const ws of workspaces) {
-            console.log('---UPGRADING----', ws.workspace)
-            await upgradeWorkspace(version, txes, migrateOperations, productId, db, ws.workspace)
+            await _upgradeWorkspace(ws)
+          }
+          if (withError.length > 0) {
+            console.log('Failed workspaces', withError)
           }
         }
       })

packages/core/src/classes.ts

@@ -376,7 +376,7 @@ export interface MigrationState extends Doc {
 /**
  * @public
  */
-export function versionToString (version: Version): string {
+export function versionToString (version: Version | Data<Version>): string {
   return `${version?.major}.${version?.minor}.${version?.patch}`
 }
 

packages/core/src/client.ts

@@ -358,11 +358,13 @@ async function loadModel (
     return modelResponse
   }
 
-  console.log(
-    'find' + (modelResponse.full ? 'full model' : 'model diff'),
-    modelResponse.transactions.length,
-    Date.now() - t
-  )
+  if (typeof window !== 'undefined') {
+    console.log(
+      'find' + (modelResponse.full ? 'full model' : 'model diff'),
+      modelResponse.transactions.length,
+      Date.now() - t
+    )
+  }
 
   await buildModel(modelResponse, allowedPlugins, configs, hierarchy, model)
   return modelResponse

packages/model/src/migration.ts

@@ -17,6 +17,7 @@ import core, {
   TxOperations,
   Data
 } from '@hcengineering/core'
+import { ModelLogger } from './utils'
 
 /**
  * @public
@@ -95,9 +96,9 @@ export type MigrationUpgradeClient = Client
  */
 export interface MigrateOperation {
   // Perform low level migration
-  migrate: (client: MigrationClient) => Promise<void>
+  migrate: (client: MigrationClient, logger: ModelLogger) => Promise<void>
   // Perform high level upgrade operations.
-  upgrade: (client: MigrationUpgradeClient) => Promise<void>
+  upgrade: (client: MigrationUpgradeClient, logger: ModelLogger) => Promise<void>
 }
 
 /**

packages/model/src/utils.ts

@@ -51,3 +51,19 @@ export async function createOrUpdate<T extends Doc> (
     await client.createDoc<T>(_class, space, data, _id)
   }
 }
+
+/**
+ * @public
+ */
+export interface ModelLogger {
+  log: (...data: any[]) => void
+}
+
+/**
+ * @public
+ */
+export const consoleModelLogger: ModelLogger = {
+  log (...data: any[]): void {
+    console.log(...data)
+  }
+}

plugins/client-resources/src/connection.ts

@@ -78,7 +78,6 @@ class Connection implements ClientConnection {
     private readonly onUnauthorized?: () => void,
     readonly onConnect?: (event: ClientConnectEvent) => Promise<void>
   ) {
-    console.log('connection created')
     this.interval = setInterval(() => {
       // eslint-disable-next-line @typescript-eslint/no-floating-promises
 
@@ -171,7 +170,6 @@ class Connection implements ClientConnection {
           typeof sessionStorage !== 'undefined'
             ? sessionStorage.getItem('session.id.' + this.url) ?? undefined
             : undefined
-        console.log('find sessionId', this.sessionId)
         this.sessionId = this.sessionId ?? generateId()
         if (typeof sessionStorage !== 'undefined') {
           sessionStorage.setItem('session.id.' + this.url, this.sessionId)
@@ -203,7 +201,6 @@ class Connection implements ClientConnection {
             v.reconnect?.()
           }
           resolve(websocket)
-          console.log('reconnect info', (resp as HelloResponse).reconnect)
 
           void this.onConnect?.(
             (resp as HelloResponse).reconnect === true ? ClientConnectEvent.Reconnected : ClientConnectEvent.Connected
@@ -289,7 +286,6 @@ class Connection implements ClientConnection {
       websocket.onopen = () => {
         const useBinary = getMetadata(client.metadata.UseBinaryProtocol) ?? true
         const useCompression = getMetadata(client.metadata.UseProtocolCompression) ?? false
-        console.log('connection opened...', socketId, useBinary, useCompression)
         clearTimeout(dialTimer)
         const helloRequest: HelloRequest = {
           method: 'hello',

plugins/client-resources/src/index.ts

@@ -50,7 +50,7 @@ export default async () => {
         let client = createClient(
           (handler: TxHandler) => {
             const url = new URL(`/${token}`, endpoint)
-            console.log('connecting to', url.href)
+
             const upgradeHandler: TxHandler = (tx) => {
               if (tx?._class === core.class.TxWorkspaceEvent) {
                 const event = tx as TxWorkspaceEvent

server/account/src/index.ts

@@ -32,9 +32,10 @@ import core, {
   Tx,
   TxOperations,
   Version,
+  versionToString,
   WorkspaceId
 } from '@hcengineering/core'
-import { MigrateOperation } from '@hcengineering/model'
+import { consoleModelLogger, MigrateOperation, ModelLogger } from '@hcengineering/model'
 import platform, {
   getMetadata,
   Metadata,
@@ -118,6 +119,7 @@ export interface Workspace {
   accounts: ObjectId[]
   productId: string
   disabled?: boolean
+  version?: Data<Version>
 }
 
 /**
@@ -609,7 +611,9 @@ export async function upgradeWorkspace (
   migrationOperation: [string, MigrateOperation][],
   productId: string,
   db: Db,
-  workspace: string
+  workspace: string,
+  logger: ModelLogger = consoleModelLogger,
+  forceUpdate: boolean = true
 ): Promise<string> {
   const ws = await getWorkspace(db, productId, workspace)
   if (ws === null) {
@@ -620,14 +624,26 @@ export async function upgradeWorkspace (
       throw new PlatformError(new Status(Severity.ERROR, platform.status.ProductIdMismatch, { productId }))
     }
   }
+  const versionStr = versionToString(version)
+
+  const currentVersion = await db.collection<Workspace>(WORKSPACE_COLLECTION).findOne({ workspace })
+  console.log(
+    `${forceUpdate ? 'force-' : ''}upgrade from "${
+      currentVersion?.version !== undefined ? versionToString(currentVersion.version) : ''
+    }" to "${versionStr}"`
+  )
+
+  if (currentVersion?.version !== undefined && !forceUpdate && versionStr === versionToString(currentVersion.version)) {
+    return versionStr
+  }
   await db.collection(WORKSPACE_COLLECTION).updateOne(
     { workspace },
     {
       $set: { version }
     }
   )
-  await upgradeModel(getTransactor(), getWorkspaceId(workspace, productId), txes, migrationOperation)
-  return `${version.major}.${version.minor}.${version.patch}`
+  await upgradeModel(getTransactor(), getWorkspaceId(workspace, productId), txes, migrationOperation, logger)
+  return versionStr
 }
 
 /**