]> git.sur5r.net Git - bacula/bacula/blobdiff - bacula/src/dird/catreq.c
Simplify the code path in migration and copy jobs
[bacula/bacula] / bacula / src / dird / catreq.c
index b1504fe86cadcc8f0c1464c4df6feb85a6edc184..ab858a90fc68a9eb9ddf3bc48203fdd9b30b7ce5 100644 (file)
@@ -1,12 +1,12 @@
 /*
    Bacula® - The Network Backup Solution
 
-   Copyright (C) 2001-2008 Free Software Foundation Europe e.V.
+   Copyright (C) 2001-2010 Free Software Foundation Europe e.V.
 
    The main author of Bacula is Kern Sibbald, with contributions from
    many others, a complete list can be found in the file AUTHORS.
    This program is Free Software; you can redistribute it and/or
-   modify it under the terms of version two of the GNU General Public
+   modify it under the terms of version three of the GNU Affero General Public
    License as published by the Free Software Foundation and included
    in the file LICENSE.
 
@@ -15,7 +15,7 @@
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
    General Public License for more details.
 
-   You should have received a copy of the GNU General Public License
+   You should have received a copy of the GNU Affero General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
    02110-1301, USA.
@@ -37,7 +37,6 @@
  *  Basic tasks done here:
  *      Handle Catalog services.
  *
- *   Version $Id$
  */
 
 #include "bacula.h"
@@ -110,7 +109,7 @@ void catalog_request(JCR *jcr, BSOCK *bs)
    int index, ok, label, writing;
    POOLMEM *omsg;
    POOL_DBR pr;
-   uint32_t Stripe;
+   uint32_t Stripe, Copy;
    uint64_t MediaId;
    utime_t VolFirstWritten;
    utime_t VolLastWritten;
@@ -118,7 +117,7 @@ void catalog_request(JCR *jcr, BSOCK *bs)
    memset(&mr, 0, sizeof(mr));
    memset(&sdmr, 0, sizeof(sdmr));
    memset(&jm, 0, sizeof(jm));
-   Dsm_check(1);      
+   Dsm_check(100);      
 
    /*
     * Request to find next appendable Volume for this Job
@@ -143,6 +142,7 @@ void catalog_request(JCR *jcr, BSOCK *bs)
       if (ok) {
          mr.PoolId = pr.PoolId;
          mr.StorageId = jcr->wstore->StorageId;
+         mr.ScratchPoolId = pr.ScratchPoolId;
          ok = find_next_volume_for_append(jcr, &mr, index, fnv_create_vol, fnv_prune);
          Dmsg3(050, "find_media ok=%d idx=%d vol=%s\n", ok, index, mr.VolumeName);
       }
@@ -282,7 +282,13 @@ void catalog_request(JCR *jcr, BSOCK *bs)
        *   the number of VolWrites has increased.
        */
       if (jcr->wstore && jcr->wstore->StorageId && sdmr.VolWrites > mr.VolWrites) {
-         mr.StorageId = jcr->wstore->StorageId;
+         Dmsg2(050, "Update StorageId old=%d new=%d\n",
+               mr.StorageId, jcr->wstore->StorageId);
+         if (jcr->wstore->StorageId == 0) {
+            Jmsg(jcr, M_ERROR, 0, _("Attempt to set StorageId to zero.\n"));
+         } else {
+            mr.StorageId = jcr->wstore->StorageId;
+         }
       }
 
       /* Copy updated values to original media record */
@@ -325,7 +331,7 @@ void catalog_request(JCR *jcr, BSOCK *bs)
     */
    } else if (sscanf(bs->msg, Create_job_media, &Job,
       &jm.FirstIndex, &jm.LastIndex, &jm.StartFile, &jm.EndFile,
-      &jm.StartBlock, &jm.EndBlock, &jm.Copy, &Stripe, &MediaId) == 10) {
+      &jm.StartBlock, &jm.EndBlock, &Copy, &Stripe, &MediaId) == 10) {
 
       if (jcr->mig_jcr) {
          jm.JobId = jcr->mig_jcr->JobId;
@@ -357,91 +363,117 @@ void catalog_request(JCR *jcr, BSOCK *bs)
 }
 
 /*
- * Update File Attributes in the catalog with data
- *  sent by the Storage daemon.  Note, we receive the whole
- *  attribute record, but we select out only the stat packet,
- *  VolSessionId, VolSessionTime, FileIndex, file type, and 
- *  file name to store in the catalog.
+ * Note, we receive the whole attribute record, but we select out only the stat
+ * packet, VolSessionId, VolSessionTime, FileIndex, file type, and file name to
+ * store in the catalog.
  */
-void catalog_update(JCR *jcr, BSOCK *bs)
+static void update_attribute(JCR *jcr, char *msg, int32_t msglen)
 {
    unser_declare;
    uint32_t VolSessionId, VolSessionTime;
    int32_t Stream;
    uint32_t FileIndex;
-   uint32_t data_len;
    char *p;
-   int filetype;
    int len;
    char *fname, *attr;
    ATTR_DBR *ar = NULL;
-   POOLMEM *omsg;
-
-   Dsm_check(1);
-   if (job_canceled(jcr) || !jcr->pool->catalog_files) {
-      goto bail_out;                  /* user disabled cataloging */
-   }
-   if (!jcr->db) {
-      omsg = get_memory(bs->msglen+1);
-      pm_strcpy(omsg, bs->msg);
-      bs->fsend(_("1994 Invalid Catalog Update: %s"), omsg);    
-      Jmsg1(jcr, M_FATAL, 0, _("Invalid Catalog Update; DB not open: %s"), omsg);
-      free_memory(omsg);
-      goto bail_out;
-   }
+   uint32_t reclen;
 
    /* Start transaction allocates jcr->attr and jcr->ar if needed */
    db_start_transaction(jcr, jcr->db);     /* start transaction if not already open */
    ar = jcr->ar;      
 
-   /* Start by scanning directly in the message buffer to get Stream   
+   /*
+    * Start by scanning directly in the message buffer to get Stream   
     *  there may be a cached attr so we cannot yet write into
     *  jcr->attr or jcr->ar  
     */
-   p = bs->msg;
+   p = msg;
    skip_nonspaces(&p);                /* UpdCat */
    skip_spaces(&p);
    skip_nonspaces(&p);                /* Job=nnn */
    skip_spaces(&p);
-   skip_nonspaces(&p);                /* FileAttributes */
+   skip_nonspaces(&p);                /* "FileAttributes" */
    p += 1;
+   /* The following "SD header" fields are serialized */
    unser_begin(p, 0);
-   unser_uint32(VolSessionId);
-   unser_uint32(VolSessionTime);
-   unser_int32(FileIndex);
-   unser_int32(Stream);
-   unser_uint32(data_len);
-   p += unser_length(p);
+   unser_uint32(VolSessionId);        /* VolSessionId */
+   unser_uint32(VolSessionTime);      /* VolSessionTime */
+   unser_int32(FileIndex);            /* FileIndex */
+   unser_int32(Stream);               /* Stream */
+   unser_uint32(reclen);              /* Record length */
+   p += unser_length(p);              /* Raw record follows */
+
+   /**
+    * At this point p points to the raw record, which varies according
+    *  to what kind of a record (Stream) was sent.  Note, the integer
+    *  fields at the beginning of these "raw" records are in ASCII with
+    *  spaces between them so one can use scanf or manual scanning to
+    *  extract the fields.
+    *
+    * File Attributes
+    *   File_index
+    *   File type
+    *   Filename (full path)
+    *   Encoded attributes
+    *   Link name (if type==FT_LNK or FT_LNKSAVED)
+    *   Encoded extended-attributes (for Win32)
+    *   Delta sequence number (32 bit int)
+    *
+    * Restore Object
+    *   File_index
+    *   File_type
+    *   Object_index
+    *   Object_len (possibly compressed)
+    *   Object_full_len (not compressed)
+    *   Object_compression
+    *   Plugin_name
+    *   Object_name
+    *   Binary Object data
+    */
 
-   Dmsg1(400, "UpdCat msg=%s\n", bs->msg);
-   Dmsg5(400, "UpdCat VolSessId=%d VolSessT=%d FI=%d Strm=%d data_len=%d\n",
-      VolSessionId, VolSessionTime, FileIndex, Stream, data_len);
+   Dmsg1(400, "UpdCat msg=%s\n", msg);
+   Dmsg5(400, "UpdCat VolSessId=%d VolSessT=%d FI=%d Strm=%d reclen=%d\n",
+      VolSessionId, VolSessionTime, FileIndex, Stream, reclen);
 
    if (Stream == STREAM_UNIX_ATTRIBUTES || Stream == STREAM_UNIX_ATTRIBUTES_EX) {
       if (jcr->cached_attribute) {
          Dmsg2(400, "Cached attr. Stream=%d fname=%s\n", ar->Stream, ar->fname);
-         if (!db_create_file_attributes_record(jcr, jcr->db, ar)) {
+         if (!db_create_attributes_record(jcr, jcr->db, ar)) {
             Jmsg1(jcr, M_FATAL, 0, _("Attribute create error. %s"), db_strerror(jcr->db));
          }
+         jcr->cached_attribute = false;
       }
       /* Any cached attr is flushed so we can reuse jcr->attr and jcr->ar */
-      jcr->attr = check_pool_memory_size(jcr->attr, bs->msglen);
-      memcpy(jcr->attr, bs->msg, bs->msglen);
-      p = jcr->attr - bs->msg + p;    /* point p into jcr->attr */
-      skip_nonspaces(&p);             /* skip FileIndex */
+      jcr->attr = check_pool_memory_size(jcr->attr, msglen);
+      memcpy(jcr->attr, msg, msglen);
+      p = jcr->attr - msg + p;    /* point p into jcr->attr */
+      skip_nonspaces(&p);         /* skip FileIndex */
       skip_spaces(&p);
-      filetype = str_to_int32(p);     /* TODO: choose between unserialize and str_to_int32 */
-      skip_nonspaces(&p);             /* skip FileType */
+      ar->FileType = str_to_int32(p); 
+      skip_nonspaces(&p);         /* skip FileType */
       skip_spaces(&p);
       fname = p;
       len = strlen(fname);        /* length before attributes */
       attr = &fname[len+1];
+      ar->DeltaSeq = 0;
+      if (ar->FileType == FT_REG) {
+         p = attr + strlen(attr) + 1;  /* point to link */
+         p = p + strlen(p) + 1;        /* point to extended attributes */
+         p = p + strlen(p) + 1;        /* point to delta sequence */
+         /*
+          * Older FDs don't have a delta sequence, so check if it is there 
+          */
+         if (p - jcr->attr < msglen) {
+            ar->DeltaSeq = str_to_int32(p); /* delta_seq */
+         }
+      }
 
       Dmsg2(400, "dird<stored: stream=%d %s\n", Stream, fname);
       Dmsg1(400, "dird<stored: attr=%s\n", attr);
       ar->attr = attr;
       ar->fname = fname;
-      if (filetype == FT_DELETED) {
+      if (ar->FileType == FT_DELETED) {
          ar->FileIndex = 0;     /* special value */
       } else {
          ar->FileIndex = FileIndex;
@@ -460,6 +492,52 @@ void catalog_update(JCR *jcr, BSOCK *bs)
       Dmsg2(400, "dird<filed: stream=%d %s\n", Stream, fname);
       Dmsg1(400, "dird<filed: attr=%s\n", attr);
 
+   } else if (Stream == STREAM_RESTORE_OBJECT) {
+      ROBJECT_DBR ro;
+
+      memset(&ro, 0, sizeof(ro));
+      ro.Stream = Stream;
+      ro.FileIndex = FileIndex;
+      if (jcr->mig_jcr) {
+         ro.JobId = jcr->mig_jcr->JobId;
+      } else {
+         ro.JobId = jcr->JobId;
+      }
+
+      Dmsg1(100, "Robj=%s\n", p);
+      
+      skip_nonspaces(&p);                  /* skip FileIndex */
+      skip_spaces(&p);
+      ro.FileType = str_to_int32(p);        /* FileType */
+      skip_nonspaces(&p);
+      skip_spaces(&p);
+      ro.object_index = str_to_int32(p);    /* Object Index */
+      skip_nonspaces(&p);
+      skip_spaces(&p);
+      ro.object_len = str_to_int32(p);      /* object length possibly compressed */
+      skip_nonspaces(&p);                  
+      skip_spaces(&p);
+      ro.object_full_len = str_to_int32(p); /* uncompressed object length */
+      skip_nonspaces(&p);
+      skip_spaces(&p);
+      ro.object_compression = str_to_int32(p); /* compression */
+      skip_nonspaces(&p);
+      skip_spaces(&p);
+
+      ro.plugin_name = p;                      /* point to plugin name */
+      len = strlen(ro.plugin_name);
+      ro.object_name = &ro.plugin_name[len+1]; /* point to object name */
+      len = strlen(ro.object_name);
+      ro.object = &ro.object_name[len+1];      /* point to object */
+      ro.object[ro.object_len] = 0;            /* add zero for those who attempt printing */
+      Dmsg7(100, "oname=%s stream=%d FT=%d FI=%d JobId=%d, obj_len=%d\nobj=\"%s\"\n",
+         ro.object_name, ro.Stream, ro.FileType, ro.FileIndex, ro.JobId,
+         ro.object_len, ro.object);
+      /* Send it */
+      if (!db_create_restore_object_record(jcr, jcr->db, &ro)) {
+         Jmsg1(jcr, M_FATAL, 0, _("Restore object create error. %s"), db_strerror(jcr->db));
+      }
+
    } else if (crypto_digest_stream_type(Stream) != CRYPTO_DIGEST_NONE) {
       fname = p;
       if (ar->FileIndex != FileIndex) {
@@ -494,13 +572,18 @@ void catalog_update(JCR *jcr, BSOCK *bs)
          }
 
          bin_to_base64(digestbuf, sizeof(digestbuf), fname, len, true);
-         Dmsg3(400, "DigestLen=%d Digest=%s type=%d\n", strlen(digestbuf), digestbuf, Stream);
+         Dmsg3(400, "DigestLen=%d Digest=%s type=%d\n", strlen(digestbuf),
+               digestbuf, Stream);
          if (jcr->cached_attribute) {
             ar->Digest = digestbuf;
             ar->DigestType = type;
-            Dmsg2(400, "Cached attr with digest. Stream=%d fname=%s\n", ar->Stream, ar->fname);
-            if (!db_create_file_attributes_record(jcr, jcr->db, ar)) {
-               Jmsg1(jcr, M_FATAL, 0, _("Attribute create error. %s"), db_strerror(jcr->db));
+            Dmsg2(400, "Cached attr with digest. Stream=%d fname=%s\n",
+                  ar->Stream, ar->fname);
+
+            /* Update BaseFile table */
+            if (!db_create_attributes_record(jcr, jcr->db, ar)) {
+               Jmsg1(jcr, M_FATAL, 0, _("attribute create error. %s"),
+                        db_strerror(jcr->db));
             }
             jcr->cached_attribute = false; 
          } else {
@@ -511,8 +594,114 @@ void catalog_update(JCR *jcr, BSOCK *bs)
          }
       }
    }
+}
+
+/*
+ * Update File Attributes in the catalog with data
+ *  sent by the Storage daemon.
+ */
+void catalog_update(JCR *jcr, BSOCK *bs)
+{
+   if (!jcr->pool->catalog_files) {
+      return;                         /* user disabled cataloging */
+   }
+   if (jcr->is_job_canceled()) {
+      goto bail_out;
+   }
+   if (!jcr->db) {
+      POOLMEM *omsg = get_memory(bs->msglen+1);
+      pm_strcpy(omsg, bs->msg);
+      bs->fsend(_("1994 Invalid Catalog Update: %s"), omsg);    
+      Jmsg1(jcr, M_FATAL, 0, _("Invalid Catalog Update; DB not open: %s"), omsg);
+      free_memory(omsg);
+      goto bail_out;
+   }
+   update_attribute(jcr, bs->msg, bs->msglen);
+
+bail_out:
+   if (jcr->is_job_canceled()) {
+      cancel_storage_daemon_job(jcr);
+   }
+}
+
+/*
+ * Update File Attributes in the catalog with data read from
+ * the storage daemon spool file. We receive the filename and
+ * we try to read it.
+ */
+bool despool_attributes_from_file(JCR *jcr, const char *file)
+{
+   bool ret=false;
+   int32_t pktsiz;
+   size_t nbytes;
+   ssize_t last = 0, size = 0;
+   int count = 0;
+   int32_t msglen;                    /* message length */
+   POOLMEM *msg = get_pool_memory(PM_MESSAGE);
+   FILE *spool_fd=NULL;
+
+   Dmsg0(100, "Begin despool_attributes_from_file\n");
+
+   if (jcr->is_job_canceled() || !jcr->pool->catalog_files || !jcr->db) {
+      goto bail_out;                  /* user disabled cataloging */
+   }
+
+   spool_fd = fopen(file, "rb");
+   if (!spool_fd) {
+      Dmsg0(100, "cancel despool_attributes_from_file\n");
+      /* send an error message */
+      goto bail_out;
+   }
+#if defined(HAVE_POSIX_FADVISE) && defined(POSIX_FADV_WILLNEED)
+   posix_fadvise(fileno(spool_fd), 0, 0, POSIX_FADV_WILLNEED);
+#endif
+
+   while (fread((char *)&pktsiz, 1, sizeof(int32_t), spool_fd) ==
+          sizeof(int32_t)) {
+      size += sizeof(int32_t);
+      msglen = ntohl(pktsiz);
+      if (msglen > 0) {
+         if (msglen > (int32_t) sizeof_pool_memory(msg)) {
+            msg = realloc_pool_memory(msg, msglen + 1);
+         }
+         nbytes = fread(msg, 1, msglen, spool_fd);
+         if (nbytes != (size_t) msglen) {
+            berrno be;
+            Dmsg2(400, "nbytes=%d msglen=%d\n", nbytes, msglen);
+            Qmsg1(jcr, M_FATAL, 0, _("fread attr spool error. ERR=%s\n"),
+                  be.bstrerror());
+            goto bail_out;
+         }
+         size += nbytes;
+         if ((++count & 0x3F) == 0) {
+            last = size;
+         }
+      }
+      if (!jcr->is_job_canceled()) {
+         update_attribute(jcr, msg, msglen);
+         if (jcr->is_job_canceled()) {
+            goto bail_out;
+         }
+      }
+   }
+   if (ferror(spool_fd)) {
+      berrno be;
+      Qmsg1(jcr, M_FATAL, 0, _("fread attr spool error. ERR=%s\n"),
+            be.bstrerror());
+      goto bail_out;
+   }
+   ret = true;
+
 bail_out:
-   if (job_canceled(jcr)) {
+   if (spool_fd) {
+      fclose(spool_fd);
+   }
+
+   if (jcr->is_job_canceled()) {
       cancel_storage_daemon_job(jcr);
    }
+
+   free_pool_memory(msg);
+   Dmsg1(100, "End despool_attributes_from_file ret=%i\n", ret);
+   return ret;
 }