store.cc

   1 /* Cumulus: Efficient Filesystem Backup to the Cloud
   2  * Copyright (C) 2008-2009 The Cumulus Developers
   3  * See the AUTHORS file for a list of contributors.
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  18  */
  19
  20 /* Backup data is stored in a collection of objects, which are grouped together
  21  * into segments for storage purposes.  This implementation of the object store
  22  * represents segments as TAR files and objects as files within them. */
  23
  24 #include <assert.h>
  25 #include <errno.h>
  26 #include <stdio.h>
  27 #include <string.h>
  28 #include <sys/types.h>
  29 #include <sys/stat.h>
  30 #include <sys/resource.h>
  31 #include <sys/wait.h>
  32 #include <unistd.h>
  33 #include <fcntl.h>
  34 #include <time.h>
  35
  36 #include <algorithm>
  37 #include <list>
  38 #include <map>
  39 #include <set>
  40 #include <string>
  41 #include <iostream>
  42
  43 #include "hash.h"
  44 #include "localdb.h"
  45 #include "store.h"
  46 #include "ref.h"
  47 #include "util.h"
  48
  49 using std::max;
  50 using std::list;
  51 using std::map;
  52 using std::pair;
  53 using std::set;
  54 using std::string;
  55
  56 /* Default filter program is bzip2 */
  57 const char *filter_program = "bzip2 -c";
  58 const char *filter_extension = ".bz2";
  59
  60 Tarfile::Tarfile(RemoteFile *file, const string &segment)
  61     : size(0),
  62       segment_name(segment)
  63 {
  64     assert(sizeof(struct tar_header) == TAR_BLOCK_SIZE);
  65
  66     this->file = file;
  67     this->filter.reset(FileFilter::New(file->get_fd(), filter_program));
  68 }
  69
  70 Tarfile::~Tarfile()
  71 {
  72     char buf[TAR_BLOCK_SIZE];
  73
  74     /* Append the EOF marker: two blocks filled with nulls. */
  75     memset(buf, 0, sizeof(buf));
  76     tar_write(buf, TAR_BLOCK_SIZE);
  77     tar_write(buf, TAR_BLOCK_SIZE);
  78
  79     if (close(filter->get_wrapped_fd()) != 0)
  80         fatal("Error closing Tarfile");
  81
  82     /* ...and wait for filter process to finish. */
  83     if (filter->wait() != 0) {
  84         fatal("Filter process error");
  85     }
  86 }
  87
  88 FileFilter::FileFilter(int raw, int wrapped, pid_t pid)
  89     : fd_raw(raw), fd_wrapped(wrapped), pid(pid) { }
  90
  91 FileFilter *FileFilter::New(int fd, const char *program)
  92 {
  93     if (program == NULL || strlen(program) == 0) {
  94         return new FileFilter(fd, fd, -1);
  95     }
  96
  97     pid_t pid;
  98     int wrapped_fd = spawn_filter(fd, program, &pid);
  99     return new FileFilter(fd, wrapped_fd, pid);
 100 }
 101
 102 int FileFilter::wait()
 103 {
 104     // No filter program was launched implies no need to wait.
 105     if (pid == -1)
 106         return 0;
 107
 108     // The raw file descriptor was held open to track the output file size, but
 109     // is not needed any longer.
 110     close(fd_raw);
 111
 112     int status;
 113     if (waitpid(pid, &status, 0) < 0) {
 114         fprintf(stderr, "Error waiting for filter process: %m\n");
 115         return -1;
 116     }
 117
 118     if (!WIFEXITED(status) || WEXITSTATUS(status) != 0) {
 119         fprintf(stderr, "Filter process error: %d\n", WEXITSTATUS(status));
 120     }
 121
 122     return status;
 123 }
 124
 125 /* Launch a child process which can act as a filter (compress, encrypt, etc.)
 126  * on the TAR output.  The file descriptor to which output should be written
 127  * must be specified; the return value is the file descriptor which will be
 128  * attached to the standard input of the filter program. */
 129 int FileFilter::spawn_filter(int fd_out, const char *program, pid_t *filter_pid)
 130 {
 131     int fds[2];
 132     pid_t pid;
 133
 134     /* Create a pipe for communicating with the filter process. */
 135     if (pipe(fds) < 0) {
 136         fatal("Unable to create pipe for filter");
 137     }
 138
 139     /* Create a child process which can exec() the filter program. */
 140     pid = fork();
 141     if (pid < 0)
 142         fatal("Unable to fork filter process");
 143
 144     if (pid > 0) {
 145         /* Parent process */
 146         close(fds[0]);
 147         cloexec(fds[1]);
 148         if (filter_pid != NULL)
 149             *filter_pid = pid;
 150     } else {
 151         /* Child process.  Rearrange file descriptors.  stdin is fds[0], stdout
 152          * is fd_out, stderr is unchanged. */
 153         close(fds[1]);
 154
 155         if (dup2(fds[0], 0) < 0)
 156             exit(1);
 157         close(fds[0]);
 158
 159         if (dup2(fd_out, 1) < 0)
 160             exit(1);
 161         close(fd_out);
 162
 163         /* Exec the filter program. */
 164         execlp("/bin/sh", "/bin/sh", "-c", program, NULL);
 165
 166         /* Should not reach here except for error cases. */
 167         fprintf(stderr, "Could not exec filter: %m\n");
 168         exit(1);
 169     }
 170
 171     return fds[1];
 172 }
 173
 174 void Tarfile::tar_write(const char *data, size_t len)
 175 {
 176     size += len;
 177
 178     while (len > 0) {
 179         int res = write(filter->get_wrapped_fd(), data, len);
 180
 181         if (res < 0) {
 182             if (errno == EINTR)
 183                 continue;
 184             fprintf(stderr, "Write error: %m\n");
 185             fatal("Write error");
 186         }
 187
 188         len -= res;
 189         data += res;
 190     }
 191 }
 192
 193 void Tarfile::write_object(int id, const char *data, size_t len)
 194 {
 195     struct tar_header header;
 196     memset(&header, 0, sizeof(header));
 197
 198     char buf[64];
 199     sprintf(buf, "%08x", id);
 200     string path = segment_name + "/" + buf;
 201
 202     assert(path.size() < 100);
 203     memcpy(header.name, path.data(), path.size());
 204     sprintf(header.mode, "%07o", 0600);
 205     sprintf(header.uid, "%07o", 0);
 206     sprintf(header.gid, "%07o", 0);
 207     sprintf(header.size, "%011o", (int)len);
 208     sprintf(header.mtime, "%011o", (int)time(NULL));
 209     header.typeflag = '0';
 210     strcpy(header.magic, "ustar  ");
 211     strcpy(header.uname, "root");
 212     strcpy(header.gname, "root");
 213
 214     memset(header.chksum, ' ', sizeof(header.chksum));
 215     int checksum = 0;
 216     for (int i = 0; i < TAR_BLOCK_SIZE; i++) {
 217         checksum += ((uint8_t *)&header)[i];
 218     }
 219     sprintf(header.chksum, "%06o", checksum);
 220
 221     tar_write((const char *)&header, TAR_BLOCK_SIZE);
 222
 223     if (len == 0)
 224         return;
 225
 226     tar_write(data, len);
 227
 228     char padbuf[TAR_BLOCK_SIZE];
 229     size_t blocks = (len + TAR_BLOCK_SIZE - 1) / TAR_BLOCK_SIZE;
 230     size_t padding = blocks * TAR_BLOCK_SIZE - len;
 231     memset(padbuf, 0, padding);
 232     tar_write(padbuf, padding);
 233 }
 234
 235 /* Estimate the size based on the size of the actual output file on disk.
 236  * However, it might be the case that the filter program is buffering all its
 237  * data, and might potentially not write a single byte until we have closed
 238  * our end of the pipe.  If we don't do so until we see data written, we have
 239  * a problem.  So, arbitrarily pick an upper bound on the compression ratio
 240  * that the filter will achieve (128:1), and return a size estimate which is
 241  * the larger of a) bytes actually seen written to disk, and b) input
 242  * bytes/128. */
 243 size_t Tarfile::size_estimate()
 244 {
 245     struct stat statbuf;
 246
 247     if (fstat(filter->get_raw_fd(), &statbuf) == 0)
 248         return max((int64_t)statbuf.st_size, (int64_t)(size / 128));
 249
 250     /* Couldn't stat the file on disk, so just return the actual number of
 251      * bytes, before compression. */
 252     return size;
 253 }
 254
 255 static const size_t SEGMENT_SIZE = 4 * 1024 * 1024;
 256
 257 /* Backup size summary: segment type -> (uncompressed size, compressed size) */
 258 static map<string, pair<int64_t, int64_t> > group_sizes;
 259
 260 ObjectReference TarSegmentStore::write_object(const char *data, size_t len,
 261                                               const std::string &group,
 262                                               const std::string &checksum,
 263                                               double age)
 264 {
 265     struct segment_info *segment;
 266
 267     // Find the segment into which the object should be written, looking up by
 268     // group.  If no segment exists yet, create one.
 269     if (segments.find(group) == segments.end()) {
 270         segment = new segment_info;
 271
 272         segment->name = generate_uuid();
 273         segment->group = group;
 274         segment->basename = segment->name + ".tar";
 275         segment->basename += filter_extension;
 276         segment->count = 0;
 277         segment->data_size = 0;
 278         segment->rf = remote->alloc_file(segment->basename,
 279                                          group == "metadata" ? "segments0"
 280                                                              : "segments1");
 281         segment->file = new Tarfile(segment->rf, segment->name);
 282
 283         segments[group] = segment;
 284     } else {
 285         segment = segments[group];
 286     }
 287
 288     int id = segment->count;
 289     char id_buf[64];
 290     sprintf(id_buf, "%08x", id);
 291
 292     segment->file->write_object(id, data, len);
 293     segment->count++;
 294     segment->data_size += len;
 295
 296     group_sizes[group].first += len;
 297
 298     ObjectReference ref(segment->name, id_buf);
 299     ref.set_range(0, len, true);
 300     if (checksum.size() > 0)
 301         ref.set_checksum(checksum);
 302     if (db != NULL)
 303         db->StoreObject(ref, age);
 304
 305     // If this segment meets or exceeds the size target, close it so that
 306     // future objects will go into a new segment.
 307     if (segment->file->size_estimate() >= SEGMENT_SIZE)
 308         close_segment(group);
 309
 310     return ref;
 311 }
 312
 313 void TarSegmentStore::sync()
 314 {
 315     while (!segments.empty())
 316         close_segment(segments.begin()->first);
 317 }
 318
 319 void TarSegmentStore::dump_stats()
 320 {
 321     printf("Data written:\n");
 322     for (map<string, pair<int64_t, int64_t> >::iterator i = group_sizes.begin();
 323          i != group_sizes.end(); ++i) {
 324         printf("    %s: %lld (%lld compressed)\n", i->first.c_str(),
 325                (long long)i->second.first, (long long)i->second.second);
 326     }
 327 }
 328
 329 void TarSegmentStore::close_segment(const string &group)
 330 {
 331     struct segment_info *segment = segments[group];
 332
 333     delete segment->file;
 334
 335     if (db != NULL) {
 336         struct stat stat_buf;
 337         int disk_size = 0;
 338         if (stat(segment->rf->get_local_path().c_str(), &stat_buf) == 0) {
 339             disk_size = stat_buf.st_size;
 340             group_sizes[segment->group].second += disk_size;
 341         }
 342
 343         string checksum
 344             = Hash::hash_file(segment->rf->get_local_path().c_str());
 345
 346         db->SetSegmentMetadata(segment->name, segment->rf->get_remote_path(),
 347                                checksum, group, segment->data_size, disk_size);
 348     }
 349
 350     segment->rf->send();
 351
 352     segments.erase(segments.find(group));
 353     delete segment;
 354 }
 355
 356 string TarSegmentStore::object_reference_to_segment(const string &object)
 357 {
 358     return object;
 359 }
 360
 361 LbsObject::LbsObject()
 362     : group(""), age(0.0), data(NULL), data_len(0), written(false)
 363 {
 364 }
 365
 366 LbsObject::~LbsObject()
 367 {
 368 }
 369
 370 void LbsObject::set_data(const char *d, size_t len, const char *checksum)
 371 {
 372     data = d;
 373     data_len = len;
 374
 375     if (checksum != NULL) {
 376         this->checksum = checksum;
 377     } else {
 378         Hash *hash = Hash::New();
 379         hash->update(data, data_len);
 380         this->checksum = hash->digest_str();
 381         delete hash;
 382     }
 383 }
 384
 385 void LbsObject::write(TarSegmentStore *store)
 386 {
 387     assert(data != NULL);
 388     assert(!written);
 389
 390     ref = store->write_object(data, data_len, group, checksum, age);
 391     written = true;
 392 }