CernVM-FS  2.10.0
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
sync_union_tarball.cc
Go to the documentation of this file.
1 
5 #define __STDC_FORMAT_MACROS
6 
7 #include "sync_union_tarball.h"
8 
9 #include <pthread.h>
10 #include <unistd.h>
11 
12 #include <cassert>
13 #include <cstdio>
14 #include <list>
15 #include <set>
16 #include <string>
17 #include <vector>
18 
19 #include "duplex_libarchive.h"
20 #include "fs_traversal.h"
21 #include "smalloc.h"
22 #include "sync_item.h"
23 #include "sync_item_dummy.h"
24 #include "sync_item_tar.h"
25 #include "sync_mediator.h"
26 #include "sync_union.h"
27 #include "util/exception.h"
28 #include "util/posix.h"
29 #include "util_concurrency.h"
30 
31 namespace publish {
32 
34  const std::string &rdonly_path,
35  const std::string &tarball_path,
36  const std::string &base_directory,
37  const std::string &to_delete,
38  const bool create_catalog_on_root)
39  : SyncUnion(mediator, rdonly_path, "", ""),
40  src(NULL),
41  tarball_path_(tarball_path),
42  base_directory_(base_directory),
43  to_delete_(to_delete),
44  create_catalog_on_root_(create_catalog_on_root),
45  read_archive_signal_(new Signal) {}
46 
48 
50  bool result;
51 
52  // We are just deleting entity from the repo
53  if (tarball_path_ == "") {
54  assert(NULL == src);
55  return SyncUnion::Initialize();
56  }
57 
58  src = archive_read_new();
59  assert(ARCHIVE_OK == archive_read_support_format_tar(src));
60  assert(ARCHIVE_OK == archive_read_support_format_empty(src));
61 
62  if (tarball_path_ == "-") {
63  result = archive_read_open_filename(src, NULL, kBlockSize);
64  } else {
65  std::string tarball_absolute_path = GetAbsolutePath(tarball_path_);
66  result = archive_read_open_filename(src, tarball_absolute_path.c_str(),
67  kBlockSize);
68  }
69 
70  if (result != ARCHIVE_OK) {
71  LogCvmfs(kLogUnionFs, kLogStderr, "Impossible to open the archive.");
72  return false;
73  }
74 
75  return SyncUnion::Initialize();
76 }
77 
78 /*
79  * Libarchive is not thread aware, so we need to make sure that before
80  * to read/"open" the next header in the archive the content of the
81  *
82  * present header is been consumed completely.
83  * Different thread read/"open" the header from the one that consumes
84  * it so we opted for a Signal that is backed by a conditional variable.
85  * We wait for the signal just before to read the header.
86  * Then when we have done with the header the Signal is fired.
87  * The Signal can be fired inside the main loop if we don't need to read
88  * data, or when the IngestionSource get closed, which means that we are
89  * not reading data anymore from there.
90  * This whole process is not necessary for directories since we don't
91  * actually need to read data from them.
92  *
93  * It may be needed to add a catalog as a root of the archive.
94  * A possible way to do it is by creating an virtual `.cvmfscatalog` file and
95  * push it into the usual pipeline.
96  * This operation must be done only once, and it seems like a good idea to do
97  * it at the first iteration of the loop, hence this logic is managed by the
98  * `first_iteration` boolean flag.
99  */
102  assert(this->IsInitialized());
103 
104  /*
105  * As first step we eliminate the requested directories.
106  */
107  if (to_delete_ != "") {
108  vector<std::string> to_eliminate_vec = SplitString(to_delete_, ':');
109 
110  for (vector<string>::iterator s = to_eliminate_vec.begin();
111  s != to_eliminate_vec.end(); ++s) {
112  std::string parent_path;
113  std::string filename;
114  SplitPath(*s, &parent_path, &filename);
115  if (parent_path == ".") parent_path = "";
116  SharedPtr<SyncItem> sync_entry =
117  CreateSyncItem(parent_path, filename, kItemDir);
118  mediator_->Remove(sync_entry);
119  }
120  }
121 
122  // we are simplying deleting entity from the repo
123  if (NULL == src) return;
124 
125  struct archive_entry *entry = archive_entry_new();
126  while (true) {
127  // Get the lock, wait if lock is not available yet
129 
130  int result = archive_read_next_header2(src, entry);
131 
132  switch (result) {
133  case ARCHIVE_FATAL: {
134  PANIC(kLogStderr, "Fatal error in reading the archive.\n%s\n",
135  archive_error_string(src));
136  break; // Only exit point with error
137  }
138 
139  case ARCHIVE_RETRY: {
141  "Error in reading the header, retrying.\n%s\n",
142  archive_error_string(src));
143  continue;
144  break;
145  }
146 
147  case ARCHIVE_EOF: {
148  if (create_catalog_on_root_ && (base_directory_ != "/")) {
151  ProcessFile(catalog);
153  }
154  for (set<string>::iterator dir = to_create_catalog_dirs_.begin();
155  dir != to_create_catalog_dirs_.end(); ++dir) {
156  assert(dirs_.find(*dir) != dirs_.end());
157  SharedPtr<SyncItem> to_mark = dirs_[*dir];
158  assert(to_mark->IsDirectory());
159  to_mark->SetCatalogMarker();
160  to_mark->MakePlaceholderDirectory();
161  ProcessDirectory(to_mark);
162  }
163  return; // Only successful exit point
164  break;
165  }
166 
167  case ARCHIVE_WARN: {
169  "Warning in uncompression reading, going on.\n %s",
170  archive_error_string(src));
171  // We actually want this to enter the ARCHIVE_OK case
172  }
173 
174  case ARCHIVE_OK: {
175  ProcessArchiveEntry(entry);
176  break;
177  }
178 
179  default: {
180  // We should never enter in this branch, but just for safeness we prefer
181  // to abort in case we hit a case we don't how to manage.
182  PANIC(kLogStderr, "Enter in unknow state. Aborting.\nError: %s\n",
183  result, archive_error_string(src));
184  }
185  }
186  }
187 }
188 
189 void SyncUnionTarball::ProcessArchiveEntry(struct archive_entry *entry) {
190  std::string archive_file_path(archive_entry_pathname(entry));
191  archive_file_path = SanitizePath(archive_file_path);
192 
193  std::string complete_path =
194  base_directory_ != "/"
195  ? MakeCanonicalPath(base_directory_ + "/" + archive_file_path)
196  : MakeCanonicalPath(archive_file_path);
197 
198  std::string parent_path;
199  std::string filename;
200  SplitPath(complete_path, &parent_path, &filename);
201  if (parent_path == ".") parent_path.clear();
202 
203  CreateDirectories(parent_path);
204 
206  parent_path, filename, src, entry, read_archive_signal_, this));
207 
208  if (NULL != archive_entry_hardlink(entry)) {
209  const std::string hardlink_name(
210  SanitizePath(archive_entry_hardlink(entry)));
211  const std::string hardlink = base_directory_ != "/"
212  ? base_directory_ + "/" + hardlink_name
213  : hardlink_name;
214 
215  if (hardlinks_.find(hardlink) != hardlinks_.end()) {
216  hardlinks_.find(hardlink)->second.push_back(complete_path);
217  } else {
218  std::list<std::string> to_hardlink;
219  to_hardlink.push_back(complete_path);
220  hardlinks_[hardlink] = to_hardlink;
221  }
222  if (filename == ".cvmfscatalog") {
223  // the file is created in the PostUpload phase
224  to_create_catalog_dirs_.insert(parent_path);
225  }
227  return;
228  }
229 
230  if (sync_entry->IsDirectory()) {
231  if (know_directories_.find(complete_path) != know_directories_.end()) {
232  sync_entry->MakePlaceholderDirectory();
233  }
234  ProcessUnmaterializedDirectory(sync_entry);
235  dirs_[complete_path] = sync_entry;
236  know_directories_.insert(complete_path);
237 
238  read_archive_signal_->Wakeup(); // We don't need to read data and we
239  // can read the next header
240 
241  } else if (sync_entry->IsRegularFile()) {
242  // inside the process pipeline we will wake up the signal
243  ProcessFile(sync_entry);
244  if (filename == ".cvmfscatalog") {
245  to_create_catalog_dirs_.insert(parent_path);
246  }
247 
248  } else if (sync_entry->IsSymlink() || sync_entry->IsFifo() ||
249  sync_entry->IsSocket() || sync_entry->IsCharacterDevice() ||
250  sync_entry->IsBlockDevice()) {
251  // we avoid to add an entity called as a catalog marker if it is not a
252  // regular file
253  if (filename != ".cvmfscatalog") {
254  ProcessFile(sync_entry);
255  } else {
257  "Found entity called as a catalog marker '%s' that however is "
258  "not a regular file, abort",
259  complete_path.c_str());
260  }
261 
262  // here we don't need to read data from the tar file so we can wake up
263  // immediately the signal
265 
266  } else {
267  PANIC(kLogStderr, "Fatal error found unexpected file: \n%s\n",
268  filename.c_str());
269  // if for any reason this code path change and we don't abort anymore,
270  // remember to wakeup the signal, otherwise we will be stuck in a deadlock
271  //
272  // read_archive_signal_->Wakeup();
273  }
274 }
275 
276 std::string SyncUnionTarball::SanitizePath(const std::string &path) {
277  if (path.length() >= 2) {
278  if (path[0] == '.' && path[1] == '/') {
279  return path.substr(2);
280  }
281  }
282  if (path.length() >= 1) {
283  if (path[0] == '/') {
284  return path.substr(1);
285  }
286  }
287  return path;
288 }
289 
291  std::map<const std::string, std::list<std::string> >::iterator hardlink;
292  for (hardlink = hardlinks_.begin(); hardlink != hardlinks_.end();
293  ++hardlink) {
294  std::list<std::string>::iterator entry;
295  for (entry = hardlink->second.begin(); entry != hardlink->second.end();
296  ++entry) {
297  mediator_->Clone(*entry, hardlink->first);
298  }
299  }
300 }
301 
303  SharedPtr<SyncItem> entry) const {
304  return entry->filename();
305 }
306 
308  return false;
309 }
310 
312  return false;
313 }
314 
315 /* Tar files are not necessarly traversed in order from root to leave.
316  * So it may happens that we are expanding the file `/a/b/c.txt` without
317  * having created yet the directory `/a/b/`.
318  * In order to overcome this limitation the following function create dummy
319  * directories that can be used as placeholder and that they will be overwritten
320  * as soon as the real directory is found in the tarball
321  */
322 void SyncUnionTarball::CreateDirectories(const std::string &target) {
323  if (know_directories_.find(target) != know_directories_.end()) return;
324  if (target == ".") return;
325 
326  std::string dirname = "";
327  std::string filename = "";
328  SplitPath(target, &dirname, &filename);
329  CreateDirectories(dirname);
330 
331  if (dirname == ".") dirname = "";
333  new SyncItemDummyDir(dirname, filename, this, kItemDir));
334 
336  dirs_[target] = dummy;
337  know_directories_.insert(target);
338 }
339 
340 } // namespace publish
#define LogCvmfs(source, mask,...)
Definition: logging.h:20
std::map< const std::string, std::list< std::string > > hardlinks_
virtual bool ProcessUnmaterializedDirectory(SharedPtr< SyncItem > entry)
Definition: sync_union.cc:92
bool IsOpaqueDirectory(SharedPtr< SyncItem > directory) const
std::map< std::string, SharedPtr< SyncItem > > dirs_
vector< string > SplitString(const string &str, const char delim, const unsigned max_chunks)
Definition: string.cc:288
const std::string base_directory_
#define PANIC(...)
Definition: exception.h:26
virtual bool Initialize()
Definition: sync_union.cc:24
void Wakeup()
assert((mem||(size==0))&&"Out Of Memory")
std::string UnwindWhiteoutFilename(SharedPtr< SyncItem > entry) const
void CreateDirectories(const std::string &target)
SharedPtr< SyncItem > CreateSyncItem(const std::string &relative_parent_path, const std::string &filename, const SyncItemType entry_type) const
Definition: sync_union.cc:30
const std::string tarball_path_
std::set< std::string > to_create_catalog_dirs_
std::string GetAbsolutePath(const std::string &path)
Definition: posix.cc:196
void SplitPath(const std::string &path, std::string *dirname, std::string *filename)
Definition: posix.cc:112
AbstractSyncMediator * mediator_
Definition: sync_union.h:147
bool IsInitialized() const
Definition: sync_union.h:139
std::string SanitizePath(const std::string &path)
const std::string to_delete_
entity to delete before to extract the tar
void ProcessArchiveEntry(struct archive_entry *entry)
void ProcessFile(SharedPtr< SyncItem > entry)
Definition: sync_union.cc:116
void Wait()
virtual void Clone(const std::string from, const std::string to)=0
virtual void Remove(SharedPtr< SyncItem > entry)=0
virtual bool ProcessDirectory(const std::string &parent_dir, const std::string &dir_name)
static const size_t kBlockSize
bool IsWhiteoutEntry(SharedPtr< SyncItem > entry) const
std::set< std::string > know_directories_
directory that we know already exist
std::string MakeCanonicalPath(const std::string &path)
Definition: posix.cc:96
SyncUnionTarball(AbstractSyncMediator *mediator, const std::string &rdonly_path, const std::string &tarball_path, const std::string &base_directory, const std::string &to_delete, const bool create_catalog_on_root)