feat: 替换StorageService.load为StorageManager.load #2065

* feat: 修改远程Generic仓库为使用StorageManager加载缓存 #2065

* feat: 替换rpm为使用storageManager.load #2065

* feat: 替换oci为使用storageManager.load #2065

* feat: 增加loadFullArtifactInputStream方法 #2065

* feat: 替换replication的storageService为storageManager #2065

* feat: 替换maven的storageService为storageManager #2065

* feat: 替换maven的storageService为storageManager #2065

* feat: 替换pypi的storageService为storageManager #2065

* feat: 替换npm的storageService为storageManager #2065

* feat: 替换helm的storageService为storageManager #2065

* feat: 替换helm的storageService为storageManager #2065

* feat: 替换composer的storageService为storageManager #2065

* feat: 替换analyst的storageService为storageManager #2065

* feat: 替换remote-repository的storageService为storageManager #2065

* feat: 替换pypi的storageService为storageManager #2065

* feat: 合并master代码 #2065

* feat: 修复代码检查错误 #2065

* feat: 增加注释 #2065

src/backend/analysis-executor/biz-analysis-executor/src/main/kotlin/com/tencent/bkrepo/analysis/executor/component/FileLoader.kt

@@ -39,9 +39,8 @@ import com.tencent.bkrepo.common.api.exception.SystemErrorException
 import com.tencent.bkrepo.common.api.message.CommonMessageCode
 import com.tencent.bkrepo.common.api.util.readJsonString
 import com.tencent.bkrepo.common.api.util.toJsonString
+import com.tencent.bkrepo.common.artifact.manager.StorageManager
 import com.tencent.bkrepo.common.artifact.pojo.RepositoryType
-import com.tencent.bkrepo.common.artifact.stream.Range
-import com.tencent.bkrepo.common.storage.core.StorageService
 import com.tencent.bkrepo.common.storage.credentials.StorageCredentials
 import com.tencent.bkrepo.repository.api.NodeClient
 import com.tencent.bkrepo.repository.api.StorageCredentialsClient
@@ -66,7 +65,7 @@ import java.security.MessageDigest
 class FileLoader(
     private val executorProperties: ScannerExecutorProperties,
     private val nodeClient: NodeClient,
-    private val storageService: StorageService,
+    private val storageManager: StorageManager,
     private val storageCredentialsClient: StorageCredentialsClient,
 ) {
     /**
@@ -84,11 +83,11 @@ class FileLoader(
 
             // 获取存储凭据
             val storageCredentials = credentialsKey?.let { storageCredentialsClient.findByKey(it).data!! }
-
+            val node = nodeClient.getNodeDetail(projectId, repoName, fullPath).data
             // 获取文件
             val file = File(tempDir, fileName(taskId, fileName(), repoType))
             val fos = DigestOutputStream(file.outputStream(), MessageDigest.getInstance("SHA-256"))
-            storageService.load(sha256, Range.full(size), storageCredentials)?.use { artifactInputStream ->
+            storageManager.loadFullArtifactInputStream(node, storageCredentials)?.use { artifactInputStream ->
                 fos.use {
                     if (repoType == RepositoryType.DOCKER.name) {
                         // 加载镜像文件
@@ -158,11 +157,11 @@ class FileLoader(
         logger.info("subtask[${task.taskId}] loading layer [$filePath]")
 
         // 加载layer
-        val size = getNodeSize(task.projectId, task.repoName, sha256)
-        storageService
-            .load(sha256, Range.full(size), storageCredentials)
-            ?.use { putArchiveEntry(filePath, size, it, tos) }
-            ?: throw SystemErrorException(CommonMessageCode.RESOURCE_NOT_FOUND, "layer not found sha256[$sha256]")
+        getNode(task.projectId, task.repoName, sha256)?.let { layerNode ->
+            storageManager
+                .loadFullArtifactInputStream(layerNode, storageCredentials)
+                ?.use { putArchiveEntry(filePath, layerNode.size, it, tos) }
+        } ?: throw SystemErrorException(CommonMessageCode.RESOURCE_NOT_FOUND, "layer not found sha256[$sha256]")
     }
 
     private fun putArchiveEntry(name: String, size: Long, inputStream: InputStream?, tos: TarArchiveOutputStream) {
@@ -173,20 +172,21 @@ class FileLoader(
         tos.closeArchiveEntry()
     }
 
-    private fun getNodeSize(projectId: String, repoName: String, sha256: String): Long {
+    private fun getNode(projectId: String, repoName: String, sha256: String): NodeDetail? {
         val nodes = nodeClient.queryWithoutCount(
             NodeQueryBuilder()
                 .projectId(projectId)
                 .repoName(repoName)
                 .sha256(sha256)
-                .select(NodeDetail::size.name)
+                .select(NodeDetail::fullPath.name)
                 .page(1, 1)
                 .build()
         )
         if (nodes.isNotOk() || nodes.data!!.records.isEmpty()) {
             throw SystemErrorException(CommonMessageCode.RESOURCE_NOT_FOUND, sha256)
         }
-        return (nodes.data!!.records[0][NodeDetail::size.name] as Number).toLong()
+        val fullPath = nodes.data!!.records[0][NodeDetail::fullPath.name].toString()
+        return nodeClient.getNodeDetail(projectId, repoName, fullPath).data
     }
 
     /**

src/backend/analysis-executor/biz-analysis-executor/src/main/kotlin/com/tencent/bkrepo/analysis/executor/trivy/TrivyScanExecutor.kt

@@ -51,13 +51,14 @@ import com.tencent.bkrepo.common.api.exception.SystemErrorException
 import com.tencent.bkrepo.common.api.message.CommonMessageCode.RESOURCE_NOT_FOUND
 import com.tencent.bkrepo.common.api.message.CommonMessageCode.SYSTEM_ERROR
 import com.tencent.bkrepo.common.artifact.hash.md5
+import com.tencent.bkrepo.common.artifact.manager.StorageManager
 import com.tencent.bkrepo.common.artifact.stream.ArtifactInputStream
-import com.tencent.bkrepo.common.artifact.stream.Range
 import com.tencent.bkrepo.common.query.model.Sort
-import com.tencent.bkrepo.common.storage.core.StorageService
 import com.tencent.bkrepo.repository.api.NodeClient
 import com.tencent.bkrepo.repository.api.RepositoryClient
-import com.tencent.bkrepo.repository.pojo.search.NodeQueryBuilder
+import com.tencent.bkrepo.repository.pojo.node.NodeDetail
+import com.tencent.bkrepo.repository.pojo.node.NodeInfo
+import com.tencent.bkrepo.repository.pojo.node.NodeListOption
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
@@ -71,7 +72,7 @@ class TrivyScanExecutor @Autowired constructor(
     dockerClient: DockerClient,
     private val scannerExecutorProperties: ScannerExecutorProperties,
     private val repositoryClient: RepositoryClient,
-    private val storageService: StorageService,
+    private val storageManager: StorageManager,
     private val nodeClient: NodeClient
 ) : CommonScanExecutor() {
 
@@ -153,7 +154,7 @@ class TrivyScanExecutor @Autowired constructor(
         if (!dbFile.parentFile.exists()) {
             dbFile.parentFile.mkdirs()
         }
-        if (!dbFile.exists() || dbFile.md5() != newestNode["md5"]) {
+        if (!dbFile.exists() || dbFile.md5() != newestNode.md5) {
             logger.info(buildLogMsg(task, "updating trivy.db"))
             dbFile.delete()
             dbFile.createNewFile()
@@ -164,40 +165,39 @@ class TrivyScanExecutor @Autowired constructor(
         }
     }
 
-    private fun getTrivyDBInputStream(dbNode: Map<String, Any?>, task: ScanExecutorTask): ArtifactInputStream {
+    private fun getTrivyDBInputStream(dbNode: NodeInfo, task: ScanExecutorTask): ArtifactInputStream {
         val scanner = task.scanner
         require(scanner is TrivyScanner)
         // 获取trivy默认仓库信息
         val repoRes = repositoryClient.getRepoDetail(scanner.vulDbConfig.projectId, scanner.vulDbConfig.repo)
         if (repoRes.isNotOk()) {
             logger.error(
                 "Get repo info failed: code[${repoRes.code}], message[${repoRes.message}]," +
-                    " projectId[${scanner.vulDbConfig.projectId}], repoName[${scanner.vulDbConfig.repo}]"
+                        " projectId[${scanner.vulDbConfig.projectId}], repoName[${scanner.vulDbConfig.repo}]"
             )
             throw SystemErrorException(SYSTEM_ERROR, repoRes.message ?: "")
         }
         val repositoryDetail = repoRes.data
             ?: throw NotFoundException(RESOURCE_NOT_FOUND, scanner.vulDbConfig.repo)
 
-        val sha256 = dbNode["sha256"] as String
-        val size = dbNode["size"].toString().toLong()
-        return storageService.load(sha256, Range.full(size), repositoryDetail.storageCredentials)
+        return storageManager.loadFullArtifactInputStream(NodeDetail(dbNode), repositoryDetail.storageCredentials)
             ?: throw SystemErrorException(SYSTEM_ERROR, "load trivy.db file failed: res: ${repoRes.message}")
     }
 
-    private fun getNewestNode(projectId: String, repo: String): Map<String, Any?> {
+    private fun getNewestNode(projectId: String, repo: String): NodeInfo {
         // 按修改时间 创建时间倒序排序，第一位则为最新的trivy.db文件
-        val queryModel = NodeQueryBuilder().projectId(projectId).repoName(repo)
-            .path("/trivy/")
-            .page(1, 1)
-            .sort(Sort.Direction.DESC, "lastModifiedDate", "createdDate")
-            .select("fullPath", "size", "sha256", "md5")
-            .build()
-        val nodeRes = nodeClient.queryWithoutCount(queryModel)
+        val option = NodeListOption(
+            pageSize = 1,
+            includeFolder = false,
+            includeMetadata = true,
+            sortProperty = listOf("lastModifiedDate", "createdDate"),
+            direction = listOf(Sort.Direction.DESC.name, Sort.Direction.DESC.name)
+        )
+        val nodeRes = nodeClient.listNodePage(projectId, repo, "/trivy", option)
         if (nodeRes.isNotOk()) {
             logger.error(
                 "Get node info failed: code[${nodeRes.code}], message[${nodeRes.message}]," +
-                    " projectId[$projectId], repoName[$repo]"
+                        " projectId[$projectId], repoName[$repo]"
             )
             throw SystemErrorException(SYSTEM_ERROR, nodeRes.message ?: "")
         }
@@ -325,6 +325,6 @@ class TrivyScanExecutor @Autowired constructor(
          */
         private const val METADATA_JSON_FILE_CONTENT =
             "{\"Version\":2,\"NextUpdate\":\"2022-07-15T12:06:50.078024068Z\"," +
-                "\"UpdatedAt\":\"2022-07-15T06:06:50.078024668Z\",\"DownloadedAt\":\"0001-01-01T00:00:00Z\"}"
+                    "\"UpdatedAt\":\"2022-07-15T06:06:50.078024668Z\",\"DownloadedAt\":\"0001-01-01T00:00:00Z\"}"
     }
 }

src/backend/analyst/biz-analyst/src/main/kotlin/com/tencent/bkrepo/analyst/service/impl/SpdxLicenseServiceImpl.kt

@@ -27,24 +27,23 @@
 
 package com.tencent.bkrepo.analyst.service.impl
 
+import com.tencent.bkrepo.analyst.dao.SpdxLicenseDao
+import com.tencent.bkrepo.analyst.exception.LicenseNotFoundException
+import com.tencent.bkrepo.analyst.model.TSpdxLicense
+import com.tencent.bkrepo.analyst.pojo.license.SpdxLicenseInfo
+import com.tencent.bkrepo.analyst.pojo.license.SpdxLicenseJsonInfo
+import com.tencent.bkrepo.analyst.service.SpdxLicenseService
 import com.tencent.bkrepo.common.api.exception.NotFoundException
 import com.tencent.bkrepo.common.api.message.CommonMessageCode
 import com.tencent.bkrepo.common.api.pojo.Page
 import com.tencent.bkrepo.common.api.util.JsonUtils
 import com.tencent.bkrepo.common.api.util.readJsonString
-import com.tencent.bkrepo.common.artifact.stream.Range
+import com.tencent.bkrepo.common.artifact.manager.StorageManager
 import com.tencent.bkrepo.common.mongo.dao.util.Pages
 import com.tencent.bkrepo.common.security.util.SecurityUtils
-import com.tencent.bkrepo.common.storage.core.StorageService
 import com.tencent.bkrepo.repository.api.NodeClient
 import com.tencent.bkrepo.repository.api.RepositoryClient
 import com.tencent.bkrepo.repository.api.StorageCredentialsClient
-import com.tencent.bkrepo.analyst.dao.SpdxLicenseDao
-import com.tencent.bkrepo.analyst.exception.LicenseNotFoundException
-import com.tencent.bkrepo.analyst.model.TSpdxLicense
-import com.tencent.bkrepo.analyst.pojo.license.SpdxLicenseInfo
-import com.tencent.bkrepo.analyst.pojo.license.SpdxLicenseJsonInfo
-import com.tencent.bkrepo.analyst.service.SpdxLicenseService
 import org.slf4j.LoggerFactory
 import org.springframework.data.domain.Sort
 import org.springframework.data.mongodb.core.query.Criteria
@@ -62,7 +61,7 @@ class SpdxLicenseServiceImpl(
     private val nodeClient: NodeClient,
     private val repositoryClient: RepositoryClient,
     private val storageCredentialsClient: StorageCredentialsClient,
-    private val storageService: StorageService
+    private val storageManager: StorageManager,
 ) : SpdxLicenseService {
     override fun importLicense(path: String): Boolean {
         val licenseJsonInfo = try {
@@ -84,7 +83,7 @@ class SpdxLicenseServiceImpl(
         val storageCredentials = repo.storageCredentialsKey?.let { storageCredentialsClient.findByKey(it).data }
         val node = nodeClient.getNodeDetail(projectId, repoName, fullPath).data
             ?: throw NotFoundException(CommonMessageCode.RESOURCE_NOT_FOUND, projectId, repoName, fullPath)
-        storageService.load(node.sha256!!, Range.full(node.size), storageCredentials)?.use {
+        storageManager.loadFullArtifactInputStream(node, storageCredentials)?.use {
             importLicense(it.readJsonString<SpdxLicenseJsonInfo>())
         } ?: throw NotFoundException(CommonMessageCode.RESOURCE_NOT_FOUND, projectId, repoName, fullPath)
         return true

src/backend/analyst/biz-analyst/src/main/kotlin/com/tencent/bkrepo/analyst/service/impl/TemporaryScanTokenServiceImpl.kt

@@ -50,12 +50,11 @@ import com.tencent.bkrepo.common.api.exception.SystemErrorException
 import com.tencent.bkrepo.common.api.message.CommonMessageCode.RESOURCE_NOT_FOUND
 import com.tencent.bkrepo.common.api.message.CommonMessageCode.SYSTEM_ERROR
 import com.tencent.bkrepo.common.api.util.StreamUtils.readText
+import com.tencent.bkrepo.common.artifact.manager.StorageManager
 import com.tencent.bkrepo.common.artifact.pojo.RepositoryType
-import com.tencent.bkrepo.common.artifact.stream.Range
 import com.tencent.bkrepo.common.query.enums.OperationType
 import com.tencent.bkrepo.common.security.exception.AuthenticationException
 import com.tencent.bkrepo.common.service.util.HttpContextHolder
-import com.tencent.bkrepo.common.storage.core.StorageService
 import com.tencent.bkrepo.oci.util.OciUtils
 import com.tencent.bkrepo.repository.api.NodeClient
 import com.tencent.bkrepo.repository.api.StorageCredentialsClient
@@ -75,7 +74,7 @@ class TemporaryScanTokenServiceImpl(
     private val temporaryTokenClient: ServiceTemporaryTokenClient,
     private val redisTemplate: RedisTemplate<String, String>,
     private val scannerProperties: ScannerProperties,
-    private val storageService: StorageService,
+    private val storageManager: StorageManager,
     private val storageCredentialsClient: StorageCredentialsClient,
     private val nodeClient: NodeClient
 ) : TemporaryScanTokenService {
@@ -189,9 +188,7 @@ class TemporaryScanTokenServiceImpl(
 
     private fun getFullPaths(subtask: SubScanTask): Map<String, FileUrl> = with(subtask) {
         return if (repoType == RepositoryType.DOCKER.name) {
-            val storageCredentials = credentialsKey?.let { storageCredentialsClient.findByKey(it).data!! }
-            val manifestContent = storageService.load(sha256, Range.full(size), storageCredentials)?.readText()
-                ?: throw ErrorCodeException(RESOURCE_NOT_FOUND, "file [$projectId:$repoName:$fullPath] not found")
+            val manifestContent = readManifest(projectId, repoName, sha256, credentialsKey)
             val schemeVersion = OciUtils.schemeVersion(manifestContent)
             val fullPaths = LinkedHashMap<String, FileUrl>()
             // 将manifest下载链接加入fullPaths列表，需要保证map第一项是manifest文件
@@ -222,6 +219,26 @@ class TemporaryScanTokenServiceImpl(
         }
     }
 
+    private fun readManifest(projectId: String, repoName: String, sha256: String, credentialsKey: String?): String {
+        val storageCredentials = credentialsKey?.let { storageCredentialsClient.findByKey(it).data!! }
+        val nodes = nodeClient.queryWithoutCount(
+            NodeQueryBuilder()
+                .projectId(projectId)
+                .repoName(repoName)
+                .sha256(sha256)
+                .select(NodeDetail::fullPath.name)
+                .page(1, 1)
+                .build()
+        )
+        if (nodes.isNotOk() || nodes.data!!.records.isEmpty()) {
+            throw SystemErrorException(RESOURCE_NOT_FOUND, sha256)
+        }
+        val fullPath = nodes.data!!.records[0][NodeDetail::fullPath.name].toString()
+        return nodeClient.getNodeDetail(projectId, repoName, fullPath).data?.let { node ->
+            storageManager.loadFullArtifactInputStream(node, storageCredentials)?.readText()
+        } ?: throw ErrorCodeException(RESOURCE_NOT_FOUND, "file [$projectId:$repoName:$fullPath] not found")
+    }
+
     private fun tokenKey(subtaskId: String) = "scanner:token:$subtaskId"
 
     private fun getNodes(projectId: String, repoName: String, sha256: List<String>): List<Map<String, Any?>> {

src/backend/common/common-artifact/artifact-service/src/main/kotlin/com/tencent/bkrepo/common/artifact/manager/StorageManager.kt

@@ -152,6 +152,18 @@ class StorageManager(
         return loadArtifactInputStream(load.nodeInfo, storageCredentials)
     }
 
+    /**
+     * 加载[node]对应的完整ArtifactInputStream
+     */
+    fun loadFullArtifactInputStream(node: NodeDetail?, storageCredentials: StorageCredentials?): ArtifactInputStream? {
+        if (node == null || node.folder) {
+            return null
+        }
+        val range = Range.full(node.size)
+        val nodeResource = nodeResourceFactory.getNodeResource(node.nodeInfo, range, storageCredentials)
+        return nodeResource.getArtifactInputStream()
+    }
+
     companion object {
         private val logger = LoggerFactory.getLogger(StorageManager::class.java)
     }

src/backend/common/common-artifact/artifact-service/src/main/kotlin/com/tencent/bkrepo/common/artifact/repository/remote/RemoteRepository.kt

@@ -119,7 +119,7 @@ abstract class RemoteRepository : AbstractArtifactRepository() {
      * 加载要返回的资源
      */
     open fun loadArtifactResource(cacheNode: NodeDetail, context: ArtifactDownloadContext): ArtifactResource? {
-        return storageService.load(cacheNode.sha256!!, Range.full(cacheNode.size), context.storageCredentials)?.run {
+        return storageManager.loadFullArtifactInputStream(cacheNode, context.storageCredentials)?.run {
             if (logger.isDebugEnabled) {
                 logger.debug("Cached remote artifact[${context.artifactInfo}] is hit.")
             }

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/core/StorageService.kt

@@ -66,6 +66,9 @@ interface StorageService :
     /**
      * 在存储实例[storageCredentials]上加载摘要为[digest]的文件
      * 当文件未找到时，会尝试去默认存储实例上查找文件
+     *
+     * 注意：该方法只会从指定存储[storageCredentials]中加载文件，如果文件正在迁移中还在旧存储或者存在于其他集群该方法会加载失败，
+     * 此时需要考虑使用[com.tencent.bkrepo.common.artifact.manager.StorageManager]中加载方法
      */
     fun load(digest: String, range: Range, storageCredentials: StorageCredentials?): ArtifactInputStream?
 

src/backend/composer/biz-composer/src/main/kotlin/com/tencent/bkrepo/composer/artifact/repository/ComposerLocalRepository.kt

@@ -40,7 +40,6 @@ import com.tencent.bkrepo.common.artifact.repository.local.LocalRepository
 import com.tencent.bkrepo.common.artifact.resolve.file.ArtifactFileFactory
 import com.tencent.bkrepo.common.artifact.resolve.response.ArtifactChannel
 import com.tencent.bkrepo.common.artifact.resolve.response.ArtifactResource
-import com.tencent.bkrepo.common.artifact.stream.Range
 import com.tencent.bkrepo.common.artifact.stream.closeQuietly
 import com.tencent.bkrepo.common.artifact.util.PackageKeys
 import com.tencent.bkrepo.common.service.util.HttpContextHolder
@@ -440,11 +439,10 @@ class ComposerLocalRepository(private val stageClient: StageClient) : LocalRepos
     }
 
     private fun nodeToJson(node: NodeDetail): String {
-        val inputStream = storageService.load(
-            node.sha256!!,
-            Range.full(node.size),
-            null
-        ) ?: throw RuntimeException("load ${node.projectId} | ${node.repoName} | ${node.fullPath} error")
+        val repoId = ArtifactContextHolder.RepositoryId(node.projectId, node.repoName)
+        val storageCredentials = ArtifactContextHolder.getRepoDetail(repoId).storageCredentials
+        val inputStream = storageManager.loadFullArtifactInputStream(node, storageCredentials)
+            ?: throw RuntimeException("load ${node.projectId} | ${node.repoName} | ${node.fullPath} error")
         val stringBuilder = StringBuilder()
         var line: String?
         try {