feat: 增加存储对账功能 #2091

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

@@ -47,6 +47,8 @@ import org.springframework.retry.support.RetryTemplate
 import java.io.File
 import java.io.IOException
 import java.io.InputStream
+import java.nio.file.Path
+import java.util.stream.Stream
 
 /**
  * 文件存储抽象模板类
@@ -192,6 +194,11 @@ abstract class AbstractFileStorage<Credentials : StorageCredentials, Client> : F
         restore(path, name, days, tier, client)
     }
 
+    override fun listAll(path: String, storageCredentials: StorageCredentials): Stream<Path> {
+        val client = getClient(storageCredentials)
+        return listAll(path, client)
+    }
+
     private fun getClient(storageCredentials: StorageCredentials): Client {
         return if (storageCredentials == storageProperties.defaultStorageCredentials()) {
             defaultClient
@@ -229,6 +236,10 @@ abstract class AbstractFileStorage<Credentials : StorageCredentials, Client> : F
         throw UnsupportedOperationException("Restore operation unsupported")
     }
 
+    open fun listAll(path: String, client: Client): Stream<Path> {
+        throw UnsupportedOperationException("ListAll operation unsupported")
+    }
+
     companion object {
         private val logger = LoggerFactory.getLogger(AbstractFileStorage::class.java)
         private const val MAX_CACHE_CLIENT = 10L

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

@@ -35,6 +35,8 @@ import com.tencent.bkrepo.common.artifact.stream.Range
 import com.tencent.bkrepo.common.storage.credentials.StorageCredentials
 import java.io.File
 import java.io.InputStream
+import java.nio.file.Path
+import java.util.stream.Stream
 
 /**
  * 文件存储接口
@@ -145,4 +147,11 @@ interface FileStorage {
      * @param storageCredentials 存储凭证
      */
     fun getTempPath(storageCredentials: StorageCredentials): String = System.getProperty("java.io.tmpdir")
+
+    /**
+     * 列出指定目录下的所有文件
+     * @param path 目录路径
+     * @param storageCredentials 存储实例
+     * */
+    fun listAll(path: String, storageCredentials: StorageCredentials): Stream<Path>
 }

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/filesystem/FileSystemClient.kt

@@ -39,6 +39,7 @@ import java.nio.file.FileAlreadyExistsException
 import java.nio.file.Files
 import java.nio.file.Path
 import java.nio.file.Paths
+import java.util.stream.Stream
 
 /**
  * 本地文件存储客户端
@@ -139,7 +140,6 @@ class FileSystemClient(val root: String) {
      * 使用channel拷贝
      * */
     private fun copyByChannel(src: Path, target: Path) {
-
         if (!Files.exists(src)) {
             throw IOException("src[$src] file not exist")
         }
@@ -308,6 +308,10 @@ class FileSystemClient(val root: String) {
         return Files.size(filePath)
     }
 
+    fun walk(path: String): Stream<Path> {
+        return Files.walk(Paths.get(root, path))
+    }
+
     private fun transfer(input: ReadableByteChannel, output: FileChannel, size: Long, append: Boolean = false) {
         val startPosition: Long = if (append) output.size() else 0L
         var bytesCopied: Long

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/filesystem/FileSystemStorage.kt

@@ -40,8 +40,10 @@ import com.tencent.bkrepo.common.storage.credentials.StorageCredentials
 import java.io.File
 import java.io.InputStream
 import java.nio.file.Files
+import java.nio.file.Path
 import java.nio.file.Paths
 import java.nio.file.StandardCopyOption
+import java.util.stream.Stream
 
 /**
  * 文件系统存储
@@ -91,4 +93,8 @@ open class FileSystemStorage : AbstractEncryptorFileStorage<FileSystemCredential
         val toFullPath = Paths.get(toClient.root, toPath, toName)
         Files.move(fromFullPath, toFullPath, StandardCopyOption.REPLACE_EXISTING)
     }
+
+    override fun listAll(path: String, client: FileSystemClient): Stream<Path> {
+        return client.walk(path).filter { Files.isRegularFile(it) }
+    }
 }

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/innercos/InnerCosFileStorage.kt

@@ -31,6 +31,7 @@
 
 package com.tencent.bkrepo.common.storage.innercos
 
+import com.tencent.bkrepo.common.api.constant.StringPool
 import com.tencent.bkrepo.common.artifact.stream.Range
 import com.tencent.bkrepo.common.storage.core.AbstractEncryptorFileStorage
 import com.tencent.bkrepo.common.storage.credentials.InnerCosCredentials
@@ -39,12 +40,16 @@ import com.tencent.bkrepo.common.storage.innercos.request.CheckObjectExistReques
 import com.tencent.bkrepo.common.storage.innercos.request.CopyObjectRequest
 import com.tencent.bkrepo.common.storage.innercos.request.DeleteObjectRequest
 import com.tencent.bkrepo.common.storage.innercos.request.GetObjectRequest
+import com.tencent.bkrepo.common.storage.innercos.request.ListObjectsRequest
 import com.tencent.bkrepo.common.storage.innercos.request.MigrateObjectRequest
 import com.tencent.bkrepo.common.storage.innercos.request.RestoreObjectRequest
 import org.slf4j.LoggerFactory
 import java.io.File
 import java.io.IOException
 import java.io.InputStream
+import java.nio.file.Path
+import java.nio.file.Paths
+import java.util.stream.Stream
 
 /**
  * 内部cos文件存储实现类
@@ -109,6 +114,12 @@ open class InnerCosFileStorage : AbstractEncryptorFileStorage<InnerCosCredential
         client.restoreObject(restoreRequest)
     }
 
+    override fun listAll(path: String, client: CosClient): Stream<Path> {
+        val keyPrefix = if (path == StringPool.ROOT) null else path
+        val listObjectsRequest = ListObjectsRequest(prefix = keyPrefix)
+        return client.listObjects(listObjectsRequest).map { Paths.get(it) }
+    }
+
     override fun onCreateClient(credentials: InnerCosCredentials): CosClient {
         require(credentials.secretId.isNotBlank())
         require(credentials.secretKey.isNotBlank())

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/innercos/client/CosClient.kt

@@ -61,6 +61,7 @@ import com.tencent.bkrepo.common.storage.innercos.request.FileCleanupChunkedFutu
 import com.tencent.bkrepo.common.storage.innercos.request.GetObjectRequest
 import com.tencent.bkrepo.common.storage.innercos.request.HeadObjectRequest
 import com.tencent.bkrepo.common.storage.innercos.request.InitiateMultipartUploadRequest
+import com.tencent.bkrepo.common.storage.innercos.request.ListObjectsRequest
 import com.tencent.bkrepo.common.storage.innercos.request.MigrateObjectRequest
 import com.tencent.bkrepo.common.storage.innercos.request.PartETag
 import com.tencent.bkrepo.common.storage.innercos.request.PutObjectRequest
@@ -71,6 +72,7 @@ import com.tencent.bkrepo.common.storage.innercos.request.UploadPartRequest
 import com.tencent.bkrepo.common.storage.innercos.request.UploadPartRequestFactory
 import com.tencent.bkrepo.common.storage.innercos.response.CopyObjectResponse
 import com.tencent.bkrepo.common.storage.innercos.response.CosObject
+import com.tencent.bkrepo.common.storage.innercos.response.ListObjectsResponse
 import com.tencent.bkrepo.common.storage.innercos.response.PutObjectResponse
 import com.tencent.bkrepo.common.storage.innercos.response.handler.CheckArchiveObjectExistResponseHandler
 import com.tencent.bkrepo.common.storage.innercos.response.handler.CheckObjectRestoreResponseHandler
@@ -79,6 +81,8 @@ import com.tencent.bkrepo.common.storage.innercos.response.handler.CopyObjectRes
 import com.tencent.bkrepo.common.storage.innercos.response.handler.GetObjectResponseHandler
 import com.tencent.bkrepo.common.storage.innercos.response.handler.HeadObjectResponseHandler
 import com.tencent.bkrepo.common.storage.innercos.response.handler.InitiateMultipartUploadResponseHandler
+import com.tencent.bkrepo.common.storage.innercos.response.handler.ListObjects0ResponseHandler
+import com.tencent.bkrepo.common.storage.innercos.response.handler.ListObjectsResponseHandler
 import com.tencent.bkrepo.common.storage.innercos.response.handler.PutObjectResponseHandler
 import com.tencent.bkrepo.common.storage.innercos.response.handler.SlowLogHandler
 import com.tencent.bkrepo.common.storage.innercos.response.handler.UploadPartResponseHandler
@@ -99,6 +103,7 @@ import java.util.concurrent.PriorityBlockingQueue
 import java.util.concurrent.ThreadPoolExecutor
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicInteger
+import java.util.stream.Stream
 import kotlin.math.ceil
 import kotlin.math.max
 
@@ -114,7 +119,6 @@ class CosClient(val credentials: InnerCosCredentials) {
      */
     private val uploadThreadPool = Executors.newFixedThreadPool(config.uploadWorkers)
 
-
     /**
      * 分块下载使用的执行器。可以为null,为null则不使用分块下载
      * */
@@ -144,11 +148,14 @@ class CosClient(val credentials: InnerCosCredentials) {
         null
     }
 
+    init {
+        this.listObjects(ListObjectsRequest())
+    }
+
     private val useChunkedLoad = (watchDog != null) && (downloadThreadPool != null)
 
     private val fastFallbackTimeout = config.timeout shr 1
 
-
     fun headObject(cosRequest: HeadObjectRequest): CosObject {
         val httpRequest = buildHttpRequest(cosRequest)
         return CosHttpClient.execute(httpRequest, HeadObjectResponseHandler())
@@ -291,7 +298,7 @@ class CosClient(val credentials: InnerCosCredentials) {
                 return putObject(PutObjectRequest(key, StringPool.EMPTY.byteInputStream(), length))
             }
             val partSize = calculateOptimalPartSize(length, true)
-            val factory = DownloadPartRequestFactory(key, partSize, 0, length-1)
+            val factory = DownloadPartRequestFactory(key, partSize, 0, length - 1)
             while (factory.hasMoreRequests()) {
                 val getObjectRequest = factory.nextDownloadPartRequest()
                 val downloadRequest = fromClient.buildHttpRequest(getObjectRequest)
@@ -320,11 +327,21 @@ class CosClient(val credentials: InnerCosCredentials) {
         }
     }
 
+    fun listObjects(cosRequest: ListObjectsRequest): Stream<String> {
+        val httpRequest = buildHttpRequest(cosRequest)
+        return CosHttpClient.execute(httpRequest, ListObjectsResponseHandler(this, cosRequest))
+    }
+
+    fun listObjects0(cosRequest: ListObjectsRequest): ListObjectsResponse {
+        val httpRequest = buildHttpRequest(cosRequest)
+        return CosHttpClient.execute(httpRequest, ListObjects0ResponseHandler())
+    }
+
     private fun multipartMigrate(
         key: String,
         uploadId: String,
         partNumber: Int,
-        downloadRequest: Request
+        downloadRequest: Request,
     ): Callable<PartETag> {
         return Callable {
             retry(RETRY_COUNT) {
@@ -335,8 +352,8 @@ class CosClient(val credentials: InnerCosCredentials) {
                             uploadId = uploadId,
                             partNumber = partNumber,
                             partSize = it.header(HttpHeaders.CONTENT_LENGTH)!!.toLong(),
-                            inputStream = it.body!!.byteStream()
-                        )
+                            inputStream = it.body!!.byteStream(),
+                        ),
                     )
                     PartETag(partNumber, CosHttpClient.execute(putObjectRequest, UploadPartResponseHandler()).eTag)
                 }

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/innercos/http/HttpResponseHandler.kt

@@ -41,10 +41,13 @@ abstract class HttpResponseHandler<T> {
     open fun keepConnection(response: Response): Boolean = false
 
     companion object {
-        private val xmlMapper: XmlMapper = XmlMapper()
+        val xmlMapper: XmlMapper = XmlMapper()
 
-        fun readXmlValue(response: Response): Map<*, *> {
-            return xmlMapper.readValue(response.body?.string(), Map::class.java)
+        fun readXmlToMap(response: Response): Map<*, *> {
+            return readXmlValue(response)
+        }
+        inline fun <reified T> readXmlValue(response: Response): T {
+            return xmlMapper.readValue(response.body?.string(), T::class.java)
         }
     }
 }

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/innercos/request/ListObjectsRequest.kt

@@ -0,0 +1,52 @@
+/*
+ * Tencent is pleased to support the open source community by making BK-CI 蓝鲸持续集成平台 available.
+ *
+ * Copyright (C) 2020 THL A29 Limited, a Tencent company.  All rights reserved.
+ *
+ * BK-CI 蓝鲸持续集成平台 is licensed under the MIT license.
+ *
+ * A copy of the MIT License is included in this file.
+ *
+ *
+ * Terms of the MIT License:
+ * ---------------------------------------------------
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package com.tencent.bkrepo.common.storage.innercos.request
+
+import com.tencent.bkrepo.common.api.constant.StringPool.ROOT
+import com.tencent.bkrepo.common.storage.innercos.http.HttpMethod
+import okhttp3.RequestBody
+
+data class ListObjectsRequest(
+    val prefix: String? = null,
+    val marker: String? = null,
+    val maxKeys: Int = 1000,
+) : CosRequest(HttpMethod.GET, ROOT) {
+
+    init {
+        check(maxKeys <= 1000)
+        prefix?.let { parameters["prefix"] = prefix }
+        marker?.let { parameters["marker"] = marker }
+        parameters["max-keys"] = maxKeys.toString()
+    }
+
+    override fun buildRequestBody(): RequestBody? = null
+}

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/innercos/response/Content.kt

@@ -0,0 +1,18 @@
+package com.tencent.bkrepo.common.storage.innercos.response
+
+import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlProperty
+
+data class Content(
+    @JacksonXmlProperty(localName = "Key")
+    var key: String = "",
+    @JacksonXmlProperty(localName = "LastModified")
+    var lastModified: String = "",
+    @JacksonXmlProperty(localName = "Created")
+    var created: String = "",
+    @JacksonXmlProperty(localName = "ETag")
+    var etag: String = "",
+    @JacksonXmlProperty(localName = "Size")
+    var size: Long = 0,
+    @JacksonXmlProperty(localName = "Forbid")
+    var forbid: Int = 0,
+)

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/innercos/response/ListObjectsResponse.kt

@@ -0,0 +1,24 @@
+package com.tencent.bkrepo.common.storage.innercos.response
+
+import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlElementWrapper
+import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlProperty
+import com.fasterxml.jackson.dataformat.xml.annotation.JacksonXmlRootElement
+
+@JacksonXmlRootElement(localName = "ListBucketResult")
+data class ListObjectsResponse(
+    @JacksonXmlProperty(localName = "Name")
+    var name: String = "",
+    @JacksonXmlProperty(localName = "Prefix")
+    var prefix: String = "",
+    @JacksonXmlProperty(localName = "Marker")
+    var marker: String = "",
+    @JacksonXmlProperty(localName = "IsTruncated")
+    var sTruncated: Boolean = false,
+    @JacksonXmlProperty(localName = "MaxKeys")
+    var maxKeys: Int = 0,
+    @JacksonXmlProperty(localName = "NextMarker")
+    var nextMarker: String = "",
+    @JacksonXmlElementWrapper(useWrapping = false)
+    @JacksonXmlProperty(localName = "Contents")
+    var contents: List<Content> = mutableListOf(),
+)

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/innercos/response/handler/CompleteMultipartUploadResponseHandler.kt

@@ -40,8 +40,8 @@ import okhttp3.Response
 class CompleteMultipartUploadResponseHandler : HttpResponseHandler<PutObjectResponse>() {
     override fun handle(response: Response): PutObjectResponse {
         return PutObjectResponse(
-            readXmlValue(response)[ETAG].toString(),
-            response.header(RESPONSE_CRC64)
+            readXmlToMap(response)[ETAG].toString(),
+            response.header(RESPONSE_CRC64),
         )
     }
 }

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/innercos/response/handler/CopyObjectResponseHandler.kt

@@ -39,7 +39,7 @@ import okhttp3.Response
 
 class CopyObjectResponseHandler : HttpResponseHandler<CopyObjectResponse>() {
     override fun handle(response: Response): CopyObjectResponse {
-        val result = readXmlValue(response)
+        val result = readXmlToMap(response)
         val eTag = (result[ETAG].toString()).trim('"')
         val lastModified = result[RESPONSE_LAST_MODIFIED].toString()
         return CopyObjectResponse(eTag, lastModified)

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/innercos/response/handler/InitiateMultipartUploadResponseHandler.kt

@@ -37,6 +37,6 @@ import okhttp3.Response
 
 class InitiateMultipartUploadResponseHandler : HttpResponseHandler<String>() {
     override fun handle(response: Response): String {
-        return readXmlValue(response)[RESPONSE_UPLOAD_ID].toString()
+        return readXmlToMap(response)[RESPONSE_UPLOAD_ID].toString()
     }
 }

src/backend/common/common-storage/storage-service/src/main/kotlin/com/tencent/bkrepo/common/storage/innercos/response/handler/ListObjects0ResponseHandler.kt

@@ -0,0 +1,11 @@
+package com.tencent.bkrepo.common.storage.innercos.response.handler
+
+import com.tencent.bkrepo.common.storage.innercos.http.HttpResponseHandler
+import com.tencent.bkrepo.common.storage.innercos.response.ListObjectsResponse
+import okhttp3.Response
+
+class ListObjects0ResponseHandler : HttpResponseHandler<ListObjectsResponse>() {
+    override fun handle(response: Response): ListObjectsResponse {
+        return readXmlValue(response)
+    }
+}