diff --git a/Sources/Services/ContainerImagesService/Server/ImagesService.swift b/Sources/Services/ContainerImagesService/Server/ImagesService.swift index 21a3e5146..d02030dfa 100644 --- a/Sources/Services/ContainerImagesService/Server/ImagesService.swift +++ b/Sources/Services/ContainerImagesService/Server/ImagesService.swift @@ -32,6 +32,17 @@ public actor ImagesService { private let imageStore: ImageStore private let snapshotStore: SnapshotStore + /// One landing, removal, or sweep at a time. The store operations + /// suspend mid-flight, and the actor admits other calls at every + /// suspension, so a garbage collection overlapping a landing computes + /// its keep set without the arriving image and prunes blobs and + /// snapshots the landing is about to reference. The lock is held + /// across each whole mutating operation, the way PodsService holds + /// its lock across lifecycle operations; containerd guards the same + /// window with leases. + /// https://github.com/containerd/containerd/blob/main/docs/garbage-collection.md + private let lock = AsyncLock() + public init( contentStore: ContentStore, imageStore: ImageStore, @@ -60,6 +71,39 @@ public actor ImagesService { return exists } + /// Sweep the snapshot directories no current image claims. + /// + /// An unpacked snapshot is owned by the record that names it, so a record + /// that stops naming it leaves it owned by nobody: registering an image + /// over an existing reference replaces the record, and removing a + /// reference takes it away. Only the store sees either happen, so every + /// path that lands or removes records runs this sweep afterwards. The + /// sweep is housekeeping: its failure is loud in the log and never fails + /// the operation that carried it. + private func sweepReplacedSnapshots(after function: String = #function) async { + do { + let kept = try await self._list() + let freedSnapshotBytes = try await self.snapshotStore.clean(keepingSnapshotsFor: kept) + if freedSnapshotBytes > 0 { + self.log.debug( + "ImagesService: swept snapshots of replaced images", + metadata: [ + "func": "\(function)", + "freedBytes": "\(freedSnapshotBytes)", + ] + ) + } + } catch { + self.log.warning( + "ImagesService: snapshot sweep failed", + metadata: [ + "func": "\(function)", + "error": "\(error)", + ] + ) + } + } + public func list() async throws -> [ImageDescription] { self.log.debug( "ImagesService: enter", @@ -81,6 +125,16 @@ public actor ImagesService { public func pull(reference: String, platform: Platform?, insecure: Bool, progressUpdate: ProgressUpdateHandler?, maxConcurrentDownloads: Int = 3) async throws -> ImageDescription + { + try await lock.withLock { _ in + try await self._pull( + reference: reference, platform: platform, insecure: insecure, + progressUpdate: progressUpdate, maxConcurrentDownloads: maxConcurrentDownloads) + } + } + + private func _pull(reference: String, platform: Platform?, insecure: Bool, progressUpdate: ProgressUpdateHandler?, maxConcurrentDownloads: Int) async throws + -> ImageDescription { self.log.debug( "ImagesService: enter", @@ -111,10 +165,17 @@ public actor ImagesService { guard let img else { throw ContainerizationError(.internalError, message: "failed to pull image \(reference)") } + await self.sweepReplacedSnapshots() return img.description.fromCZ } public func push(reference: String, platform: Platform?, insecure: Bool, progressUpdate: ProgressUpdateHandler?) async throws { + try await lock.withLock { _ in + try await self._push(reference: reference, platform: platform, insecure: insecure, progressUpdate: progressUpdate) + } + } + + private func _push(reference: String, platform: Platform?, insecure: Bool, progressUpdate: ProgressUpdateHandler?) async throws { self.log.debug( "ImagesService: enter", metadata: [ @@ -142,6 +203,12 @@ public actor ImagesService { } public func tag(old: String, new: String) async throws -> ImageDescription { + try await lock.withLock { _ in + try await self._tag(old: old, new: new) + } + } + + private func _tag(old: String, new: String) async throws -> ImageDescription { self.log.debug( "ImagesService: enter", metadata: [ @@ -166,6 +233,12 @@ public actor ImagesService { } public func delete(reference: String, garbageCollect: Bool) async throws { + try await lock.withLock { _ in + try await self._delete(reference: reference, garbageCollect: garbageCollect) + } + } + + private func _delete(reference: String, garbageCollect: Bool) async throws { self.log.debug( "ImagesService: enter", metadata: [ @@ -184,9 +257,16 @@ public actor ImagesService { } try await self.imageStore.delete(reference: reference, performCleanup: garbageCollect) + await self.sweepReplacedSnapshots() } public func save(references: [String], out: URL, platform: Platform?) async throws { + try await lock.withLock { _ in + try await self._save(references: references, out: out, platform: platform) + } + } + + private func _save(references: [String], out: URL, platform: Platform?) async throws { self.log.debug( "ImagesService: enter", metadata: [ @@ -215,6 +295,12 @@ public actor ImagesService { } public func load(from tarFile: URL, force: Bool) async throws -> ([ImageDescription], [String]) { + try await lock.withLock { _ in + try await self._load(from: tarFile, force: force) + } + } + + private func _load(from tarFile: URL, force: Bool) async throws -> ([ImageDescription], [String]) { let archivePathname = tarFile.absolutePath() self.log.debug( "ImagesService: enter", @@ -244,6 +330,7 @@ public actor ImagesService { } let loaded = try await self.imageStore.load(from: tempDir) + await self.sweepReplacedSnapshots() var images: [ImageDescription] = [] for image in loaded { images.append(image.description.fromCZ) @@ -252,6 +339,12 @@ public actor ImagesService { } public func cleanUpOrphanedBlobs() async throws -> ([String], UInt64) { + try await lock.withLock { _ in + try await self._cleanUpOrphanedBlobs() + } + } + + private func _cleanUpOrphanedBlobs() async throws -> ([String], UInt64) { self.log.debug( "ImagesService: enter", metadata: [ @@ -336,6 +429,12 @@ public actor ImagesService { extension ImagesService { public func unpack(description: ImageDescription, platform: Platform?, progressUpdate: ProgressUpdateHandler?) async throws { + try await lock.withLock { _ in + try await self._unpack(description: description, platform: platform, progressUpdate: progressUpdate) + } + } + + private func _unpack(description: ImageDescription, platform: Platform?, progressUpdate: ProgressUpdateHandler?) async throws { self.log.debug( "ImagesService: enter", metadata: [ @@ -360,6 +459,12 @@ extension ImagesService { } public func deleteImageSnapshot(description: ImageDescription, platform: Platform?) async throws { + try await lock.withLock { _ in + try await self._deleteImageSnapshot(description: description, platform: platform) + } + } + + private func _deleteImageSnapshot(description: ImageDescription, platform: Platform?) async throws { self.log.debug( "ImagesService: enter", metadata: [ @@ -384,6 +489,12 @@ extension ImagesService { } public func getImageSnapshot(description: ImageDescription, platform: Platform) async throws -> Filesystem { + try await lock.withLock { _ in + try await self._getImageSnapshot(description: description, platform: platform) + } + } + + private func _getImageSnapshot(description: ImageDescription, platform: Platform) async throws -> Filesystem { self.log.debug( "ImagesService: enter", metadata: [ diff --git a/Sources/Services/ContainerImagesService/Server/SnapshotStore.swift b/Sources/Services/ContainerImagesService/Server/SnapshotStore.swift index bbb5fb507..05b28fdf1 100644 --- a/Sources/Services/ContainerImagesService/Server/SnapshotStore.swift +++ b/Sources/Services/ContainerImagesService/Server/SnapshotStore.swift @@ -168,12 +168,32 @@ public actor SnapshotStore { public func clean(keepingSnapshotsFor images: [Containerization.Image] = []) async throws -> UInt64 { var toKeep: [String] = [Self.ingestDirName] for image in images { - for manifest in try await image.index().manifests { - guard let platform = manifest.platform else { - continue + // An image is read to learn which snapshot it claims, and reading + // one whose content the store no longer holds fails. That is a + // fact about that image and not about the others, so it costs the + // sweep that image rather than the whole pass: a single record + // left behind by an interrupted pull would otherwise stop every + // snapshot from ever being reclaimed again, and the store fills + // with the unpacked filesystems of images nothing names. An image + // the store cannot read is one it cannot unpack or run either, so + // the snapshot that image would have claimed is not worth keeping + // the store full for. + do { + for manifest in try await image.index().manifests { + guard let platform = manifest.platform else { + continue + } + let desc = try await image.descriptor(for: platform) + toKeep.append(desc.digest.trimmingDigestPrefix) } - let desc = try await image.descriptor(for: platform) - toKeep.append(desc.digest.trimmingDigestPrefix) + } catch { + self.log?.warning( + "snapshot sweep read no manifests for an image; its snapshots are not kept", + metadata: [ + "image": "\(image.reference)", + "error": "\(error)", + ] + ) } } let all = try self.fm.contentsOfDirectory(at: self.path, includingPropertiesForKeys: [.totalFileAllocatedSizeKey]).map { @@ -186,8 +206,24 @@ public actor SnapshotStore { guard self.fm.fileExists(atPath: unpackedPath.absolutePath()) else { continue } - deletedBytes += self.fm.allocatedSize(of: unpackedPath) - try self.fm.removeItem(at: unpackedPath) + let size = self.fm.allocatedSize(of: unpackedPath) + // A snapshot the store cannot remove costs one snapshot's worth of + // space, and every other snapshot in the set is still owed to the + // caller, as are the bytes already freed. The removal that failed + // is what the caller needs to act on, so the error goes to the log + // with the snapshot that raised it. + do { + try self.fm.removeItem(at: unpackedPath) + deletedBytes += size + } catch { + self.log?.warning( + "snapshot sweep could not remove a snapshot", + metadata: [ + "snapshot": "\(dir)", + "error": "\(error)", + ] + ) + } } return deletedBytes }