putBlobToPendingFile implements ImageDestination.PutBlobWithOptions, storing stream into an on-disk file. The caller must arrange the blob to be eventually committed using s.commitLayer().
(stream io.Reader, blobinfo types.BlobInfo, options *private.PutBlobOptions)
| 257 | // putBlobToPendingFile implements ImageDestination.PutBlobWithOptions, storing stream into an on-disk file. |
| 258 | // The caller must arrange the blob to be eventually committed using s.commitLayer(). |
| 259 | func (s *storageImageDestination) putBlobToPendingFile(stream io.Reader, blobinfo types.BlobInfo, options *private.PutBlobOptions) (private.UploadedBlob, error) { |
| 260 | // Stores a layer or data blob in our temporary directory, checking that any information |
| 261 | // in the blobinfo matches the incoming data. |
| 262 | if blobinfo.Digest != "" { |
| 263 | if err := blobinfo.Digest.Validate(); err != nil { |
| 264 | return private.UploadedBlob{}, fmt.Errorf("invalid digest %#v: %w", blobinfo.Digest.String(), err) |
| 265 | } |
| 266 | } |
| 267 | |
| 268 | // Set up to digest the blob if necessary, and count its size while saving it to a file. |
| 269 | filename := s.computeNextBlobCacheFile() |
| 270 | file, err := os.OpenFile(filename, os.O_CREATE|os.O_TRUNC|os.O_WRONLY|os.O_EXCL, 0600) |
| 271 | if err != nil { |
| 272 | return private.UploadedBlob{}, fmt.Errorf("creating temporary file %q: %w", filename, err) |
| 273 | } |
| 274 | blobDigest, diffID, count, err := func() (_, _ digest.Digest, _ int64, retErr error) { // A scope for defer |
| 275 | // since we are writing to this file, make sure we handle err on Close() |
| 276 | defer func() { |
| 277 | closeErr := file.Close() |
| 278 | if retErr == nil { |
| 279 | retErr = closeErr |
| 280 | } |
| 281 | }() |
| 282 | counter := ioutils.NewWriteCounter(file) |
| 283 | stream = io.TeeReader(stream, counter) |
| 284 | digester, stream := putblobdigest.DigestIfUnknown(stream, blobinfo) |
| 285 | decompressed, err := archive.DecompressStream(stream) |
| 286 | if err != nil { |
| 287 | return "", "", 0, fmt.Errorf("setting up to decompress blob: %w", err) |
| 288 | |
| 289 | } |
| 290 | defer decompressed.Close() |
| 291 | |
| 292 | diffID := digest.Canonical.Digester() |
| 293 | // Copy the data to the file. |
| 294 | // TODO: This can take quite some time, and should ideally be cancellable using context.Context. |
| 295 | _, err = io.Copy(diffID.Hash(), decompressed) |
| 296 | if err != nil { |
| 297 | return "", "", 0, fmt.Errorf("storing blob to file %q: %w", filename, err) |
| 298 | } |
| 299 | |
| 300 | return digester.Digest(), diffID.Digest(), counter.Count, nil |
| 301 | }() |
| 302 | if err != nil { |
| 303 | return private.UploadedBlob{}, err |
| 304 | } |
| 305 | |
| 306 | // Determine blob properties, and fail if information that we were given about the blob |
| 307 | // is known to be incorrect. |
| 308 | blobSize := blobinfo.Size |
| 309 | if blobSize < 0 { |
| 310 | blobSize = count |
| 311 | } else if blobinfo.Size != count { |
| 312 | return private.UploadedBlob{}, ErrBlobSizeMismatch |
| 313 | } |
| 314 | |
| 315 | // Record information about the blob. |
| 316 | s.lock.Lock() |
no test coverage detected