Decompress and save the contents of the decompressReader stream into the passed-in temporary file. If we successfully save all of the data, rename the file to match the digest of the data, and make notes about the relationship between the file that holds a copy of the compressed data and this new f
(wg *sync.WaitGroup, decompressReader io.ReadCloser, tempFile *os.File, compressedFilename string, compressedDigest digest.Digest, isConfig bool, alternateDigest *digest.Digest)
| 79 | // and make notes about the relationship between the file that holds a copy of the compressed data |
| 80 | // and this new file. |
| 81 | func (d *blobCacheDestination) saveStream(wg *sync.WaitGroup, decompressReader io.ReadCloser, tempFile *os.File, compressedFilename string, compressedDigest digest.Digest, isConfig bool, alternateDigest *digest.Digest) { |
| 82 | defer wg.Done() |
| 83 | defer decompressReader.Close() |
| 84 | |
| 85 | succeeded := false |
| 86 | defer func() { |
| 87 | if !succeeded { |
| 88 | // Remove the temporary file. |
| 89 | if err := os.Remove(tempFile.Name()); err != nil { |
| 90 | logrus.Debugf("error cleaning up temporary file %q for decompressed copy of blob %q: %v", tempFile.Name(), compressedDigest.String(), err) |
| 91 | } |
| 92 | } |
| 93 | }() |
| 94 | |
| 95 | digester := digest.Canonical.Digester() |
| 96 | if err := func() error { // A scope for defer |
| 97 | defer tempFile.Close() |
| 98 | |
| 99 | // Decompress from and digest the reading end of that pipe. |
| 100 | decompressed, err := archive.DecompressStream(decompressReader) |
| 101 | if err != nil { |
| 102 | // Drain the pipe to keep from stalling the PutBlob() thread. |
| 103 | if _, err2 := io.Copy(io.Discard, decompressReader); err2 != nil { |
| 104 | logrus.Debugf("error draining the pipe: %v", err2) |
| 105 | } |
| 106 | return err |
| 107 | } |
| 108 | defer decompressed.Close() |
| 109 | // Read the decompressed data through the filter over the pipe, blocking until the |
| 110 | // writing end is closed. |
| 111 | _, err = io.Copy(io.MultiWriter(tempFile, digester.Hash()), decompressed) |
| 112 | return err |
| 113 | }(); err != nil { |
| 114 | return |
| 115 | } |
| 116 | |
| 117 | // Determine the name that we should give to the uncompressed copy of the blob. |
| 118 | decompressedFilename, err := d.reference.blobPath(digester.Digest(), isConfig) |
| 119 | if err != nil { |
| 120 | return |
| 121 | } |
| 122 | // Rename the temporary file. |
| 123 | if err := os.Rename(tempFile.Name(), decompressedFilename); err != nil { |
| 124 | logrus.Debugf("error renaming new decompressed copy of blob %q into place at %q: %v", digester.Digest().String(), decompressedFilename, err) |
| 125 | return |
| 126 | } |
| 127 | succeeded = true |
| 128 | *alternateDigest = digester.Digest() |
| 129 | // Note the relationship between the two files. |
| 130 | if err := ioutils.AtomicWriteFile(decompressedFilename+compressedNote, []byte(compressedDigest.String()), 0600); err != nil { |
| 131 | logrus.Debugf("error noting that the compressed version of %q is %q: %v", digester.Digest().String(), compressedDigest.String(), err) |
| 132 | } |
| 133 | if err := ioutils.AtomicWriteFile(compressedFilename+decompressedNote, []byte(digester.Digest().String()), 0600); err != nil { |
| 134 | logrus.Debugf("error noting that the decompressed version of %q is %q: %v", compressedDigest.String(), digester.Digest().String(), err) |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | func (d *blobCacheDestination) HasThreadSafePutBlob() bool { |