mirror of
https://github.com/cirruslabs/tart.git
synced 2026-09-30 03:01:12 +02:00
Remove disk v1 support (#1204)
* Remove disk v1 support * fix: address PR review feedback - add explicit error for legacy disk.v1 media type during pull - include actionable re-push guidance in runtime error 🤖 Generated with [Codex](https://chatgpt.com/codex) Co-Authored-By: Codex <codex@openai.com> * Re-use legacyDiskV1MediaType in error message --------- Co-authored-by: Codex <codex@openai.com> Co-authored-by: Nikolay Edigaryev <edigaryev@gmail.com>
This commit is contained in:
co-authored by
Codex
Nikolay Edigaryev
parent
d45ef38cf7
commit
faa40b6832
@@ -31,9 +31,6 @@ struct Push: AsyncParsableCommand {
|
|||||||
discussion: "Can be specified multiple times to attach multiple labels."))
|
discussion: "Can be specified multiple times to attach multiple labels."))
|
||||||
var labels: [String] = []
|
var labels: [String] = []
|
||||||
|
|
||||||
@Option(help: .hidden)
|
|
||||||
var diskFormat: String = "v2"
|
|
||||||
|
|
||||||
@Flag(help: ArgumentHelp("cache pushed images locally",
|
@Flag(help: ArgumentHelp("cache pushed images locally",
|
||||||
discussion: "Increases disk usage, but saves time if you're going to pull the pushed images later."))
|
discussion: "Increases disk usage, but saves time if you're going to pull the pushed images later."))
|
||||||
var populateCache: Bool = false
|
var populateCache: Bool = false
|
||||||
@@ -85,7 +82,6 @@ struct Push: AsyncParsableCommand {
|
|||||||
registry: registry,
|
registry: registry,
|
||||||
references: references,
|
references: references,
|
||||||
chunkSizeMb: chunkSize,
|
chunkSizeMb: chunkSize,
|
||||||
diskFormat: diskFormat,
|
|
||||||
concurrency: concurrency,
|
concurrency: concurrency,
|
||||||
labels: parseLabels()
|
labels: parseLabels()
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -1,75 +0,0 @@
|
|||||||
import Foundation
|
|
||||||
import Compression
|
|
||||||
|
|
||||||
class DiskV1: Disk {
|
|
||||||
private static let bufferSizeBytes = 4 * 1024 * 1024
|
|
||||||
private static let layerLimitBytes = 500 * 1000 * 1000
|
|
||||||
|
|
||||||
static func push(diskURL: URL, registry: Registry, chunkSizeMb: Int, concurrency: UInt, progress: Progress) async throws -> [OCIManifestLayer] {
|
|
||||||
var pushedLayers: [OCIManifestLayer] = []
|
|
||||||
|
|
||||||
// Open the disk file
|
|
||||||
let mappedDisk = try Data(contentsOf: diskURL, options: [.alwaysMapped])
|
|
||||||
var mappedDiskReadOffset = 0
|
|
||||||
|
|
||||||
// Compress the disk file as a single stream
|
|
||||||
let compressingFilter = try InputFilter(.compress, using: .lz4, bufferCapacity: Self.bufferSizeBytes) { (length: Int) -> Data? in
|
|
||||||
// Determine the size of the next chunk
|
|
||||||
let bytesRead = min(length, mappedDisk.count - mappedDiskReadOffset)
|
|
||||||
|
|
||||||
// Read the next uncompressed chunk
|
|
||||||
let data = mappedDisk.subdata(in: mappedDiskReadOffset ..< mappedDiskReadOffset + bytesRead)
|
|
||||||
|
|
||||||
// Advance the offset
|
|
||||||
mappedDiskReadOffset += bytesRead
|
|
||||||
|
|
||||||
// Provide the uncompressed chunk to the compressing filter
|
|
||||||
return data
|
|
||||||
}
|
|
||||||
|
|
||||||
// Cut the compressed stream into layers, each equal exactly ``Self.layerLimitBytes`` bytes,
|
|
||||||
// except for the last one, which may be smaller
|
|
||||||
while let compressedData = try compressingFilter.readData(ofLength: Self.layerLimitBytes) {
|
|
||||||
let layerDigest = try await registry.pushBlob(fromData: compressedData, chunkSizeMb: chunkSizeMb)
|
|
||||||
|
|
||||||
pushedLayers.append(OCIManifestLayer(
|
|
||||||
mediaType: diskV1MediaType,
|
|
||||||
size: compressedData.count,
|
|
||||||
digest: layerDigest
|
|
||||||
))
|
|
||||||
|
|
||||||
// Update progress using an absolute value
|
|
||||||
progress.completedUnitCount = Int64(mappedDiskReadOffset)
|
|
||||||
}
|
|
||||||
|
|
||||||
return pushedLayers
|
|
||||||
}
|
|
||||||
|
|
||||||
static func pull(registry: Registry, diskLayers: [OCIManifestLayer], diskURL: URL, concurrency: UInt, progress: Progress, localLayerCache: LocalLayerCache? = nil, deduplicate: Bool = false) async throws {
|
|
||||||
if !FileManager.default.createFile(atPath: diskURL.path, contents: nil) {
|
|
||||||
throw OCIError.FailedToCreateVmFile
|
|
||||||
}
|
|
||||||
|
|
||||||
// Open the disk file
|
|
||||||
let disk = try FileHandle(forWritingTo: diskURL)
|
|
||||||
defer { try! disk.close() }
|
|
||||||
|
|
||||||
// Decompress the layers onto the disk in a single stream
|
|
||||||
let filter = try OutputFilter(.decompress, using: .lz4, bufferCapacity: Self.bufferSizeBytes) { data in
|
|
||||||
if let data = data {
|
|
||||||
try disk.write(contentsOf: data)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
for diskLayer in diskLayers {
|
|
||||||
try await registry.pullBlob(diskLayer.digest) { data in
|
|
||||||
try filter.write(data)
|
|
||||||
|
|
||||||
// Update the progress
|
|
||||||
progress.completedUnitCount += Int64(data.count)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
try filter.finalize()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -6,7 +6,6 @@ let ociConfigMediaType = "application/vnd.oci.image.config.v1+json"
|
|||||||
|
|
||||||
// Layer media types
|
// Layer media types
|
||||||
let configMediaType = "application/vnd.cirruslabs.tart.config.v1"
|
let configMediaType = "application/vnd.cirruslabs.tart.config.v1"
|
||||||
let diskV1MediaType = "application/vnd.cirruslabs.tart.disk.v1"
|
|
||||||
let diskV2MediaType = "application/vnd.cirruslabs.tart.disk.v2"
|
let diskV2MediaType = "application/vnd.cirruslabs.tart.disk.v2"
|
||||||
let nvramMediaType = "application/vnd.cirruslabs.tart.nvram.v1"
|
let nvramMediaType = "application/vnd.cirruslabs.tart.nvram.v1"
|
||||||
|
|
||||||
|
|||||||
@@ -2,6 +2,8 @@ import Compression
|
|||||||
import Foundation
|
import Foundation
|
||||||
import OpenTelemetryApi
|
import OpenTelemetryApi
|
||||||
|
|
||||||
|
let legacyDiskV1MediaType = "application/vnd.cirruslabs.tart.disk.v1"
|
||||||
|
|
||||||
enum OCIError: Error {
|
enum OCIError: Error {
|
||||||
case ShouldBeExactlyOneLayer
|
case ShouldBeExactlyOneLayer
|
||||||
case ShouldBeAtLeastOneLayer
|
case ShouldBeAtLeastOneLayer
|
||||||
@@ -29,16 +31,12 @@ extension VMDirectory {
|
|||||||
try configFile.close()
|
try configFile.close()
|
||||||
|
|
||||||
// Pull VM's disk layers and decompress them into a disk file
|
// Pull VM's disk layers and decompress them into a disk file
|
||||||
let diskImplType: Disk.Type
|
if manifest.layers.contains(where: { $0.mediaType == legacyDiskV1MediaType }) {
|
||||||
let layers: [OCIManifestLayer]
|
throw RuntimeError.Generic("Pulling OCI images with legacy disk media type \(legacyDiskV1MediaType) is no longer supported, please re-push the image using a current Tart version")
|
||||||
|
}
|
||||||
|
|
||||||
if manifest.layers.contains(where: { $0.mediaType == diskV1MediaType }) {
|
let layers = manifest.layers.filter { $0.mediaType == diskV2MediaType }
|
||||||
diskImplType = DiskV1.self
|
if layers.isEmpty {
|
||||||
layers = manifest.layers.filter { $0.mediaType == diskV1MediaType }
|
|
||||||
} else if manifest.layers.contains(where: { $0.mediaType == diskV2MediaType }) {
|
|
||||||
diskImplType = DiskV2.self
|
|
||||||
layers = manifest.layers.filter { $0.mediaType == diskV2MediaType }
|
|
||||||
} else {
|
|
||||||
throw OCIError.ShouldBeAtLeastOneLayer
|
throw OCIError.ShouldBeAtLeastOneLayer
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -55,10 +53,10 @@ extension VMDirectory {
|
|||||||
ProgressObserver(progress).log(defaultLogger)
|
ProgressObserver(progress).log(defaultLogger)
|
||||||
|
|
||||||
do {
|
do {
|
||||||
try await diskImplType.pull(registry: registry, diskLayers: layers, diskURL: diskURL,
|
try await DiskV2.pull(registry: registry, diskLayers: layers, diskURL: diskURL,
|
||||||
concurrency: concurrency, progress: progress,
|
concurrency: concurrency, progress: progress,
|
||||||
localLayerCache: localLayerCache,
|
localLayerCache: localLayerCache,
|
||||||
deduplicate: deduplicate)
|
deduplicate: deduplicate)
|
||||||
} catch let error where error is FilterError {
|
} catch let error where error is FilterError {
|
||||||
throw RuntimeError.PullFailed("failed to decompress disk: \(error.localizedDescription)")
|
throw RuntimeError.PullFailed("failed to decompress disk: \(error.localizedDescription)")
|
||||||
}
|
}
|
||||||
@@ -90,7 +88,7 @@ extension VMDirectory {
|
|||||||
try manifest.toJSON().write(to: manifestURL)
|
try manifest.toJSON().write(to: manifestURL)
|
||||||
}
|
}
|
||||||
|
|
||||||
func pushToRegistry(registry: Registry, references: [String], chunkSizeMb: Int, diskFormat: String, concurrency: UInt, labels: [String: String] = [:]) async throws -> RemoteName {
|
func pushToRegistry(registry: Registry, references: [String], chunkSizeMb: Int, concurrency: UInt, labels: [String: String] = [:]) async throws -> RemoteName {
|
||||||
var layers = Array<OCIManifestLayer>()
|
var layers = Array<OCIManifestLayer>()
|
||||||
|
|
||||||
// Read VM's config and push it as blob
|
// Read VM's config and push it as blob
|
||||||
@@ -111,14 +109,7 @@ extension VMDirectory {
|
|||||||
let progress = Progress(totalUnitCount: diskSize)
|
let progress = Progress(totalUnitCount: diskSize)
|
||||||
ProgressObserver(progress).log(defaultLogger)
|
ProgressObserver(progress).log(defaultLogger)
|
||||||
|
|
||||||
switch diskFormat {
|
layers.append(contentsOf: try await DiskV2.push(diskURL: diskURL, registry: registry, chunkSizeMb: chunkSizeMb, concurrency: concurrency, progress: progress))
|
||||||
case "v1":
|
|
||||||
layers.append(contentsOf: try await DiskV1.push(diskURL: diskURL, registry: registry, chunkSizeMb: chunkSizeMb, concurrency: concurrency, progress: progress))
|
|
||||||
case "v2":
|
|
||||||
layers.append(contentsOf: try await DiskV2.push(diskURL: diskURL, registry: registry, chunkSizeMb: chunkSizeMb, concurrency: concurrency, progress: progress))
|
|
||||||
default:
|
|
||||||
throw RuntimeError.OCIUnsupportedDiskFormat(diskFormat)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Read VM's NVRAM and push it as blob
|
// Read VM's NVRAM and push it as blob
|
||||||
defaultLogger.appendNewLine("pushing NVRAM...")
|
defaultLogger.appendNewLine("pushing NVRAM...")
|
||||||
|
|||||||
@@ -74,7 +74,6 @@ enum RuntimeError : Error {
|
|||||||
case ImportFailed(_ message: String)
|
case ImportFailed(_ message: String)
|
||||||
case SoftnetFailed(_ message: String)
|
case SoftnetFailed(_ message: String)
|
||||||
case OCIStorageError(_ message: String)
|
case OCIStorageError(_ message: String)
|
||||||
case OCIUnsupportedDiskFormat(_ format: String)
|
|
||||||
case SuspendFailed(_ message: String)
|
case SuspendFailed(_ message: String)
|
||||||
case PullFailed(_ message: String)
|
case PullFailed(_ message: String)
|
||||||
case VirtualMachineLimitExceeded(_ hint: String)
|
case VirtualMachineLimitExceeded(_ hint: String)
|
||||||
@@ -139,8 +138,6 @@ extension RuntimeError : CustomStringConvertible {
|
|||||||
return "Softnet failed: \(message)"
|
return "Softnet failed: \(message)"
|
||||||
case .OCIStorageError(let message):
|
case .OCIStorageError(let message):
|
||||||
return "OCI storage error: \(message)"
|
return "OCI storage error: \(message)"
|
||||||
case .OCIUnsupportedDiskFormat(let format):
|
|
||||||
return "OCI disk format \(format) is not supported by this version of Tart"
|
|
||||||
case .SuspendFailed(let message):
|
case .SuspendFailed(let message):
|
||||||
return "Failed to suspend the VM: \(message)"
|
return "Failed to suspend the VM: \(message)"
|
||||||
case .PullFailed(let message):
|
case .PullFailed(let message):
|
||||||
|
|||||||
@@ -24,27 +24,6 @@ final class LayerizerTests: XCTestCase {
|
|||||||
registryRunner = nil
|
registryRunner = nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func testDiskV1() async throws {
|
|
||||||
// Original disk file to be pushed to the registry
|
|
||||||
let originalDiskFileURL = try fileWithRandomData(sizeBytes: 5 * 1024 * 1024 * 1024)
|
|
||||||
addTeardownBlock {
|
|
||||||
try FileManager.default.removeItem(at: originalDiskFileURL)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Disk file to be pulled from the registry
|
|
||||||
// and compared against the original disk file
|
|
||||||
let pulledDiskFileURL = FileManager.default.temporaryDirectory.appendingPathComponent(UUID().uuidString)
|
|
||||||
|
|
||||||
print("pushing disk...")
|
|
||||||
let diskLayers = try await DiskV1.push(diskURL: originalDiskFileURL, registry: registry, chunkSizeMb: 0, concurrency: 4, progress: Progress())
|
|
||||||
|
|
||||||
print("pulling disk...")
|
|
||||||
try await DiskV1.pull(registry: registry, diskLayers: diskLayers, diskURL: pulledDiskFileURL, concurrency: 16, progress: Progress())
|
|
||||||
|
|
||||||
print("comparing disks...")
|
|
||||||
try XCTAssertEqual(Digest.hash(originalDiskFileURL), Digest.hash(pulledDiskFileURL))
|
|
||||||
}
|
|
||||||
|
|
||||||
func testDiskV2() async throws {
|
func testDiskV2() async throws {
|
||||||
// Original disk file to be pushed to the registry
|
// Original disk file to be pushed to the registry
|
||||||
let originalDiskFileURL = try fileWithRandomData(sizeBytes: 5 * 1024 * 1024 * 1024)
|
let originalDiskFileURL = try fileWithRandomData(sizeBytes: 5 * 1024 * 1024 * 1024)
|
||||||
|
|||||||
Reference in New Issue
Block a user