| Directory: | cvmfs/ |
|---|---|
| File: | cvmfs/upload_local.cc |
| Date: | 2025-11-02 02:35:35 |
| Exec | Total | Coverage | |
|---|---|---|---|
| Lines: | 88 | 148 | 59.5% |
| Branches: | 82 | 255 | 32.2% |
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | /** | ||
| 2 | * This file is part of the CernVM File System. | ||
| 3 | */ | ||
| 4 | |||
| 5 | #include "upload_local.h" | ||
| 6 | |||
| 7 | #include <errno.h> | ||
| 8 | |||
| 9 | #include <string> | ||
| 10 | |||
| 11 | #include "compression/compression.h" | ||
| 12 | #include "util/logging.h" | ||
| 13 | #include "util/posix.h" | ||
| 14 | |||
| 15 | namespace upload { | ||
| 16 | |||
| 17 | 950 | LocalUploader::LocalUploader(const SpoolerDefinition &spooler_definition) | |
| 18 | : AbstractUploader(spooler_definition) | ||
| 19 | 1900 | , backend_file_mode_(default_backend_file_mode_ ^ GetUmask()) | |
| 20 |
1/2✓ Branch 1 taken 950 times.
✗ Branch 2 not taken.
|
950 | , backend_dir_mode_(default_backend_dir_mode_ ^ GetUmask()) |
| 21 |
1/2✓ Branch 1 taken 950 times.
✗ Branch 2 not taken.
|
950 | , upstream_path_(spooler_definition.spooler_configuration) |
| 22 |
2/4✓ Branch 2 taken 950 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 950 times.
✗ Branch 6 not taken.
|
1900 | , temporary_path_(spooler_definition.temporary_path) { |
| 23 |
2/4✓ Branch 1 taken 950 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 950 times.
✗ Branch 4 not taken.
|
950 | assert(spooler_definition.IsValid() |
| 24 | && spooler_definition.driver_type == SpoolerDefinition::Local); | ||
| 25 | |||
| 26 | 950 | atomic_init32(©_errors_); | |
| 27 | 950 | } | |
| 28 | |||
| 29 | 1366 | bool LocalUploader::WillHandle(const SpoolerDefinition &spooler_definition) { | |
| 30 | 1366 | return spooler_definition.driver_type == SpoolerDefinition::Local; | |
| 31 | } | ||
| 32 | |||
| 33 | 2252 | unsigned int LocalUploader::GetNumberOfErrors() const { | |
| 34 | 2252 | return atomic_read32(©_errors_); | |
| 35 | } | ||
| 36 | |||
| 37 | ✗ | bool LocalUploader::Create() { | |
| 38 | ✗ | return MakeCacheDirectories(upstream_path_ + "/data", backend_dir_mode_) | |
| 39 | ✗ | && MkdirDeep(upstream_path_ + "/stats", backend_dir_mode_, false); | |
| 40 | } | ||
| 41 | |||
| 42 | 5844 | void LocalUploader::DoUpload(const std::string &remote_path, | |
| 43 | IngestionSource *source, | ||
| 44 | const CallbackTN *callback) { | ||
| 45 |
1/2✓ Branch 1 taken 5844 times.
✗ Branch 2 not taken.
|
5844 | LogCvmfs(kLogSpooler, kLogVerboseMsg, "FileUpload call started."); |
| 46 | |||
| 47 | // create destination in backend storage temporary directory | ||
| 48 | 5844 | std::string tmp_path; | |
| 49 |
2/4✓ Branch 1 taken 5844 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 5844 times.
✗ Branch 5 not taken.
|
5844 | FILE *ftmp = CreateTempFile(temporary_path_ + "/upload", 0666, "w", |
| 50 | &tmp_path); | ||
| 51 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 5844 times.
|
5844 | if (ftmp == NULL) { |
| 52 | ✗ | LogCvmfs(kLogSpooler, kLogVerboseMsg, | |
| 53 | "failed to create temp path for " | ||
| 54 | "upload of file '%s' (errno: %d)", | ||
| 55 | ✗ | source->GetPath().c_str(), errno); | |
| 56 | ✗ | atomic_inc32(©_errors_); | |
| 57 | ✗ | Respond(callback, UploaderResults(1, source->GetPath())); | |
| 58 | ✗ | return; | |
| 59 | } | ||
| 60 | |||
| 61 | // copy file into controlled temporary directory location | ||
| 62 |
1/2✓ Branch 1 taken 5844 times.
✗ Branch 2 not taken.
|
5844 | const bool rvb = source->Open(); |
| 63 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 5844 times.
|
5844 | if (!rvb) { |
| 64 | ✗ | fclose(ftmp); | |
| 65 | ✗ | unlink(tmp_path.c_str()); | |
| 66 | ✗ | atomic_inc32(©_errors_); | |
| 67 | ✗ | Respond(callback, UploaderResults(100, source->GetPath())); | |
| 68 | ✗ | return; | |
| 69 | } | ||
| 70 | unsigned char buffer[kPageSize]; | ||
| 71 | ssize_t rbytes; | ||
| 72 | do { | ||
| 73 |
1/2✓ Branch 1 taken 2002564 times.
✗ Branch 2 not taken.
|
2002564 | rbytes = source->Read(buffer, kPageSize); |
| 74 | 2002564 | size_t wbytes = 0; | |
| 75 |
2/2✓ Branch 0 taken 1999194 times.
✓ Branch 1 taken 3370 times.
|
2002564 | if (rbytes > 0) { |
| 76 |
1/2✓ Branch 1 taken 1999194 times.
✗ Branch 2 not taken.
|
1999194 | wbytes = fwrite(buffer, 1, rbytes, ftmp); |
| 77 | } | ||
| 78 |
2/4✓ Branch 0 taken 2002564 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 2002564 times.
|
2002564 | if ((rbytes < 0) || (static_cast<size_t>(rbytes) != wbytes)) { |
| 79 | ✗ | source->Close(); | |
| 80 | ✗ | fclose(ftmp); | |
| 81 | ✗ | unlink(tmp_path.c_str()); | |
| 82 | ✗ | atomic_inc32(©_errors_); | |
| 83 | ✗ | Respond(callback, UploaderResults(100, source->GetPath())); | |
| 84 | ✗ | return; | |
| 85 | } | ||
| 86 |
2/2✓ Branch 0 taken 1996720 times.
✓ Branch 1 taken 5844 times.
|
2002564 | } while (rbytes == kPageSize); |
| 87 |
1/2✓ Branch 1 taken 5844 times.
✗ Branch 2 not taken.
|
5844 | source->Close(); |
| 88 |
1/2✓ Branch 1 taken 5844 times.
✗ Branch 2 not taken.
|
5844 | fclose(ftmp); |
| 89 | |||
| 90 | // move the file in place (atomic operation) | ||
| 91 |
1/2✓ Branch 1 taken 5844 times.
✗ Branch 2 not taken.
|
5844 | const int rvi = Move(tmp_path, remote_path); |
| 92 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 5844 times.
|
5844 | if (rvi != 0) { |
| 93 | ✗ | LogCvmfs(kLogSpooler, kLogVerboseMsg, | |
| 94 | "failed to move file '%s' from the " | ||
| 95 | "staging area to the final location: " | ||
| 96 | "'%s'", | ||
| 97 | tmp_path.c_str(), remote_path.c_str()); | ||
| 98 | ✗ | unlink(tmp_path.c_str()); | |
| 99 | ✗ | atomic_inc32(©_errors_); | |
| 100 | ✗ | Respond(callback, UploaderResults(rvi, source->GetPath())); | |
| 101 | ✗ | return; | |
| 102 | } | ||
| 103 | |||
| 104 |
3/6✓ Branch 1 taken 5844 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 5844 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 5844 times.
✗ Branch 8 not taken.
|
5844 | Respond(callback, UploaderResults(rvi, source->GetPath())); |
| 105 |
1/2✓ Branch 1 taken 5844 times.
✗ Branch 2 not taken.
|
5844 | } |
| 106 | |||
| 107 | 2716 | UploadStreamHandle *LocalUploader::InitStreamedUpload( | |
| 108 | const CallbackTN *callback) { | ||
| 109 | 2716 | std::string tmp_path; | |
| 110 |
1/2✓ Branch 1 taken 2716 times.
✗ Branch 2 not taken.
|
2716 | const int tmp_fd = CreateAndOpenTemporaryChunkFile(&tmp_path); |
| 111 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 2716 times.
|
2716 | if (tmp_fd < 0) { |
| 112 | ✗ | atomic_inc32(©_errors_); | |
| 113 | ✗ | return NULL; | |
| 114 | } | ||
| 115 | |||
| 116 |
2/4✓ Branch 1 taken 2716 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2716 times.
✗ Branch 5 not taken.
|
2716 | return new LocalStreamHandle(callback, tmp_fd, tmp_path); |
| 117 | 2716 | } | |
| 118 | |||
| 119 | 9726 | void LocalUploader::StreamedUpload(UploadStreamHandle *handle, | |
| 120 | UploadBuffer buffer, | ||
| 121 | const CallbackTN *callback) { | ||
| 122 | 9726 | LocalStreamHandle *local_handle = static_cast<LocalStreamHandle *>(handle); | |
| 123 | |||
| 124 | 9726 | const size_t bytes_written = write(local_handle->file_descriptor, buffer.data, | |
| 125 | 9726 | buffer.size); | |
| 126 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 9726 times.
|
9726 | if (bytes_written != buffer.size) { |
| 127 | ✗ | const int cpy_errno = errno; | |
| 128 | ✗ | LogCvmfs(kLogSpooler, kLogVerboseMsg, | |
| 129 | "failed to write %lu bytes to '%s' " | ||
| 130 | "(errno: %d)", | ||
| 131 | buffer.size, local_handle->temporary_path.c_str(), cpy_errno); | ||
| 132 | ✗ | atomic_inc32(©_errors_); | |
| 133 | ✗ | Respond(callback, | |
| 134 | ✗ | UploaderResults(UploaderResults::kBufferUpload, cpy_errno)); | |
| 135 | ✗ | return; | |
| 136 | } | ||
| 137 | |||
| 138 |
1/2✓ Branch 2 taken 9726 times.
✗ Branch 3 not taken.
|
9726 | Respond(callback, UploaderResults(UploaderResults::kBufferUpload, 0)); |
| 139 | } | ||
| 140 | |||
| 141 | 2716 | void LocalUploader::FinalizeStreamedUpload(UploadStreamHandle *handle, | |
| 142 | const shash::Any &content_hash) { | ||
| 143 | 2716 | int retval = 0; | |
| 144 | 2716 | LocalStreamHandle *local_handle = static_cast<LocalStreamHandle *>(handle); | |
| 145 | |||
| 146 |
1/2✓ Branch 1 taken 2716 times.
✗ Branch 2 not taken.
|
2716 | retval = close(local_handle->file_descriptor); |
| 147 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 2716 times.
|
2716 | if (retval != 0) { |
| 148 | ✗ | const int cpy_errno = errno; | |
| 149 | ✗ | LogCvmfs(kLogSpooler, kLogVerboseMsg, | |
| 150 | "failed to close temp file '%s' " | ||
| 151 | "(errno: %d)", | ||
| 152 | local_handle->temporary_path.c_str(), cpy_errno); | ||
| 153 | ✗ | atomic_inc32(©_errors_); | |
| 154 | ✗ | Respond(handle->commit_callback, | |
| 155 | ✗ | UploaderResults(UploaderResults::kChunkCommit, cpy_errno)); | |
| 156 | ✗ | return; | |
| 157 | } | ||
| 158 | |||
| 159 | 2716 | std::string final_path; | |
| 160 |
2/4✓ Branch 1 taken 2716 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 2716 times.
|
2716 | if (local_handle->remote_path != "") { |
| 161 | ✗ | final_path = local_handle->remote_path; | |
| 162 | } else { | ||
| 163 |
2/4✓ Branch 1 taken 2716 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2716 times.
✗ Branch 5 not taken.
|
2716 | final_path = "data/" + content_hash.MakePath(); |
| 164 | } | ||
| 165 |
2/4✓ Branch 1 taken 2716 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 2716 times.
✗ Branch 4 not taken.
|
2716 | if (!Peek(final_path)) { |
| 166 |
1/2✓ Branch 1 taken 2716 times.
✗ Branch 2 not taken.
|
2716 | retval = Move(local_handle->temporary_path, final_path); |
| 167 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 2716 times.
|
2716 | if (retval != 0) { |
| 168 | ✗ | const int cpy_errno = errno; | |
| 169 | ✗ | LogCvmfs(kLogSpooler, kLogVerboseMsg, | |
| 170 | "failed to move temp file '%s' to " | ||
| 171 | "final location '%s' (errno: %d)", | ||
| 172 | local_handle->temporary_path.c_str(), final_path.c_str(), | ||
| 173 | cpy_errno); | ||
| 174 | ✗ | atomic_inc32(©_errors_); | |
| 175 | ✗ | Respond(handle->commit_callback, | |
| 176 | ✗ | UploaderResults(UploaderResults::kChunkCommit, cpy_errno)); | |
| 177 | ✗ | return; | |
| 178 | } | ||
| 179 | 2716 | if (!content_hash.HasSuffix() | |
| 180 |
5/6✓ Branch 0 taken 1716 times.
✓ Branch 1 taken 1000 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 1716 times.
✓ Branch 4 taken 1000 times.
✓ Branch 5 taken 1716 times.
|
2716 | || content_hash.suffix == shash::kSuffixPartial) { |
| 181 |
1/2✓ Branch 1 taken 1000 times.
✗ Branch 2 not taken.
|
1000 | CountUploadedChunks(); |
| 182 |
4/8✓ Branch 1 taken 1000 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 1000 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 1000 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 1000 times.
✗ Branch 11 not taken.
|
1000 | CountUploadedBytes(GetFileSize(upstream_path_ + "/" + final_path)); |
| 183 |
2/2✓ Branch 0 taken 1706 times.
✓ Branch 1 taken 10 times.
|
1716 | } else if (content_hash.suffix == shash::kSuffixCatalog) { |
| 184 |
1/2✓ Branch 1 taken 1706 times.
✗ Branch 2 not taken.
|
1706 | CountUploadedCatalogs(); |
| 185 |
4/8✓ Branch 1 taken 1706 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 1706 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 1706 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 1706 times.
✗ Branch 11 not taken.
|
1706 | CountUploadedCatalogBytes(GetFileSize(upstream_path_ + "/" + final_path)); |
| 186 | } | ||
| 187 | } else { | ||
| 188 | ✗ | const int retval = unlink(local_handle->temporary_path.c_str()); | |
| 189 | ✗ | if (retval != 0) { | |
| 190 | ✗ | LogCvmfs(kLogSpooler, kLogVerboseMsg, | |
| 191 | "failed to remove temporary file '%s' (errno: %d)", | ||
| 192 | ✗ | local_handle->temporary_path.c_str(), errno); | |
| 193 | } | ||
| 194 | ✗ | CountDuplicates(); | |
| 195 | } | ||
| 196 | |||
| 197 | 2716 | const CallbackTN *callback = handle->commit_callback; | |
| 198 |
1/2✓ Branch 0 taken 2716 times.
✗ Branch 1 not taken.
|
2716 | delete local_handle; |
| 199 | |||
| 200 |
2/4✓ Branch 1 taken 2716 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2716 times.
✗ Branch 5 not taken.
|
2716 | Respond(callback, UploaderResults(UploaderResults::kChunkCommit, 0)); |
| 201 |
1/2✓ Branch 1 taken 2716 times.
✗ Branch 2 not taken.
|
2716 | } |
| 202 | |||
| 203 | /** | ||
| 204 | * TODO(jblomer): investigate if parallelism increases the GC speed on local | ||
| 205 | * disks. | ||
| 206 | */ | ||
| 207 | 10 | void LocalUploader::DoRemoveAsync(const std::string &file_to_delete) { | |
| 208 |
1/2✓ Branch 2 taken 10 times.
✗ Branch 3 not taken.
|
10 | const int retval = unlink((upstream_path_ + "/" + file_to_delete).c_str()); |
| 209 |
1/4✗ Branch 0 not taken.
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
|
10 | if ((retval != 0) && (errno != ENOENT)) |
| 210 | ✗ | atomic_inc32(©_errors_); | |
| 211 |
1/2✓ Branch 2 taken 10 times.
✗ Branch 3 not taken.
|
10 | Respond(NULL, UploaderResults()); |
| 212 | 10 | } | |
| 213 | |||
| 214 | 2766 | bool LocalUploader::Peek(const std::string &path) { | |
| 215 |
2/4✓ Branch 2 taken 2766 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2766 times.
✗ Branch 6 not taken.
|
2766 | const bool retval = FileExists(upstream_path_ + "/" + path); |
| 216 | 2766 | return retval; | |
| 217 | } | ||
| 218 | |||
| 219 | ✗ | bool LocalUploader::Mkdir(const std::string &path) { | |
| 220 | ✗ | return MkdirDeep(upstream_path_ + "/" + path, backend_dir_mode_, false); | |
| 221 | } | ||
| 222 | |||
| 223 | 10 | bool LocalUploader::PlaceBootstrappingShortcut(const shash::Any &object) { | |
| 224 |
2/4✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 10 times.
✗ Branch 5 not taken.
|
10 | const std::string src = "data/" + object.MakePath(); |
| 225 |
3/6✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 10 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 10 times.
✗ Branch 8 not taken.
|
20 | const std::string dest = upstream_path_ + "/" + object.MakeAlternativePath(); |
| 226 |
1/2✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
|
20 | return SymlinkForced(src, dest); |
| 227 | 10 | } | |
| 228 | |||
| 229 | 8560 | int LocalUploader::Move(const std::string &local_path, | |
| 230 | const std::string &remote_path) const { | ||
| 231 |
2/4✓ Branch 1 taken 8560 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 8560 times.
✗ Branch 5 not taken.
|
8560 | const std::string destination_path = upstream_path_ + "/" + remote_path; |
| 232 | |||
| 233 | // make sure the file has the right permissions | ||
| 234 | 8560 | int retval = chmod(local_path.c_str(), backend_file_mode_); | |
| 235 |
1/2✓ Branch 0 taken 8560 times.
✗ Branch 1 not taken.
|
8560 | int retcode = (retval == 0) ? 0 : 101; |
| 236 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 8560 times.
|
8560 | if (retcode != 0) { |
| 237 | ✗ | LogCvmfs(kLogSpooler, kLogVerboseMsg, | |
| 238 | "failed to set file permission '%s' " | ||
| 239 | "errno: %d", | ||
| 240 | ✗ | local_path.c_str(), errno); | |
| 241 | ✗ | return retcode; | |
| 242 | } | ||
| 243 | |||
| 244 | // move the file in place | ||
| 245 | 8560 | retval = rename(local_path.c_str(), destination_path.c_str()); | |
| 246 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 8560 times.
|
8560 | retcode = (retval == 0) ? 0 : errno; |
| 247 |
1/2✗ Branch 0 not taken.
✓ Branch 1 taken 8560 times.
|
8560 | if (retcode != 0) { |
| 248 | ✗ | LogCvmfs(kLogSpooler, kLogVerboseMsg, | |
| 249 | "failed to move file '%s' to '%s' " | ||
| 250 | "errno: %d", | ||
| 251 | ✗ | local_path.c_str(), remote_path.c_str(), errno); | |
| 252 | } | ||
| 253 | |||
| 254 | 8560 | return retcode; | |
| 255 | 8560 | } | |
| 256 | |||
| 257 | ✗ | int64_t LocalUploader::DoGetObjectSize(const std::string &file_name) { | |
| 258 | ✗ | return GetFileSize(upstream_path_ + "/" + file_name); | |
| 259 | } | ||
| 260 | |||
| 261 | } // namespace upload | ||
| 262 |