X-Git-Url: https://git.sur5r.net/?a=blobdiff_plain;f=bacula%2Fsrc%2Fdird%2Fcatreq.c;h=88ddd4729680c75937990e21f396aa26731733dc;hb=de901692a6f4c84d6e8cebe3a5a38b711c60207f;hp=00af56fa38d25241265fab2d47e41aaf2fe9228e;hpb=83d32d30375f7dae0e999dcaf3f06393a5c70ff6;p=bacula%2Fbacula diff --git a/bacula/src/dird/catreq.c b/bacula/src/dird/catreq.c index 00af56fa38..88ddd47296 100644 --- a/bacula/src/dird/catreq.c +++ b/bacula/src/dird/catreq.c @@ -1,32 +1,22 @@ /* - Bacula® - The Network Backup Solution - - Copyright (C) 2001-2008 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 - License as published by the Free Software Foundation and included - in the file LICENSE. - - This program is distributed in the hope that it will be useful, but - WITHOUT ANY WARRANTY; without even the implied warranty of - 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 - along with this program; if not, write to the Free Software - Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA - 02110-1301, USA. - - Bacula® is a registered trademark of Kern Sibbald. - The licensor of Bacula is the Free Software Foundation Europe - (FSFE), Fiduciary Program, Sumatrastrasse 25, 8006 Zürich, - Switzerland, email:ftf@fsfeurope.org. + Bacula(R) - The Network Backup Solution + + Copyright (C) 2000-2017 Kern Sibbald + + The original author of Bacula is Kern Sibbald, with contributions + from many others, a complete list can be found in the file AUTHORS. + + You may use this file and others of this release according to the + license defined in the LICENSE file, which includes the Affero General + Public License, v3.0 ("AGPLv3") and some additional permissions and + terms pursuant to its AGPLv3 Section 7. + + This notice must be preserved when any source code is + conveyed and/or propagated. + + Bacula(R) is a registered trademark of Kern Sibbald. */ /* - * * Bacula Director -- catreq.c -- handles the message channel * catalog request from the Storage daemon. * @@ -36,8 +26,6 @@ * * Basic tasks done here: * Handle Catalog services. - * - * Version $Id$ */ #include "bacula.h" @@ -50,84 +38,110 @@ */ /* Requests from the Storage daemon */ -static char Find_media[] = "CatReq Job=%127s FindMedia=%d pool_name=%127s media_type=%127s\n"; -static char Get_Vol_Info[] = "CatReq Job=%127s GetVolInfo VolName=%127s write=%d\n"; +static char Find_media[] = "CatReq JobId=%ld FindMedia=%d pool_name=%127s media_type=%127s vol_type=%d\n"; +static char Get_Vol_Info[] = "CatReq JobId=%ld GetVolInfo VolName=%127s write=%d\n"; -static char Update_media[] = "CatReq Job=%127s UpdateMedia VolName=%s" - " VolJobs=%u VolFiles=%u VolBlocks=%u VolBytes=%lld VolMounts=%u" - " VolErrors=%u VolWrites=%u MaxVolBytes=%lld EndTime=%lld VolStatus=%10s" +static char Update_media[] = "CatReq JobId=%ld UpdateMedia VolName=%s" + " VolJobs=%u VolFiles=%u VolBlocks=%u VolBytes=%lld VolABytes=%lld" + " VolHoleBytes=%lld VolHoles=%u VolMounts=%u" + " VolErrors=%u VolWrites=%lld MaxVolBytes=%lld EndTime=%lld VolStatus=%10s" " Slot=%d relabel=%d InChanger=%d VolReadTime=%lld VolWriteTime=%lld" - " VolFirstWritten=%lld VolParts=%u\n"; - -static char Create_job_media[] = "CatReq Job=%127s CreateJobMedia " - " FirstIndex=%u LastIndex=%u StartFile=%u EndFile=%u " - " StartBlock=%u EndBlock=%u Copy=%d Strip=%d MediaId=%" lld "\n"; + " VolFirstWritten=%lld VolType=%u VolParts=%d VolCloudParts=%d" + " LastPartBytes=%lld Enabled=%d\n"; +static char Create_jobmedia[] = "CatReq JobId=%ld CreateJobMedia\n"; /* Responses sent to Storage daemon */ static char OK_media[] = "1000 OK VolName=%s VolJobs=%u VolFiles=%u" - " VolBlocks=%u VolBytes=%s VolMounts=%u VolErrors=%u VolWrites=%u" + " VolBlocks=%u VolBytes=%s VolABytes=%s VolHoleBytes=%s VolHoles=%u" + " VolMounts=%u VolErrors=%u VolWrites=%s" " MaxVolBytes=%s VolCapacityBytes=%s VolStatus=%s Slot=%d" " MaxVolJobs=%u MaxVolFiles=%u InChanger=%d VolReadTime=%s" - " VolWriteTime=%s EndFile=%u EndBlock=%u VolParts=%u LabelType=%d" - " MediaId=%s\n"; + " VolWriteTime=%s EndFile=%u EndBlock=%u VolType=%u LabelType=%d" + " MediaId=%s ScratchPoolId=%s VolParts=%d VolCloudParts=%d" + " LastPartBytes=%lld Enabled=%d\n"; static char OK_create[] = "1000 OK CreateJobMedia\n"; +void remove_dummy_jobmedia_records(JCR *jcr) +{ + if (jcr->dummy_jobmedia) { + char ec1[30]; + POOL_MEM buf; + Mmsg(buf, "DELETE FROM JobMedia WHERE JobId=%s AND FirstIndex=0 AND LastIndex=0", + edit_int64(jcr->JobId, ec1)); + Dmsg1(150, "Delete dummy: %s\n", buf.c_str()); + db_sql_query(jcr->db, buf.c_str(), NULL, NULL); + jcr->dummy_jobmedia = false; + } +} + static int send_volume_info_to_storage_daemon(JCR *jcr, BSOCK *sd, MEDIA_DBR *mr) { int stat; - char ed1[50], ed2[50], ed3[50], ed4[50], ed5[50], ed6[50]; + char ed1[50], ed2[50], ed3[50], ed4[50], ed5[50], ed6[50], ed7[50], ed8[50], + ed9[50], ed10[50]; jcr->MediaId = mr->MediaId; pm_strcpy(jcr->VolumeName, mr->VolumeName); bash_spaces(mr->VolumeName); stat = sd->fsend(OK_media, mr->VolumeName, mr->VolJobs, mr->VolFiles, mr->VolBlocks, edit_uint64(mr->VolBytes, ed1), - mr->VolMounts, mr->VolErrors, mr->VolWrites, - edit_uint64(mr->MaxVolBytes, ed2), - edit_uint64(mr->VolCapacityBytes, ed3), + edit_uint64(mr->VolABytes, ed2), + edit_uint64(mr->VolHoleBytes, ed3), + mr->VolHoles, mr->VolMounts, mr->VolErrors, + edit_uint64(mr->VolWrites, ed4), + edit_uint64(mr->MaxVolBytes, ed5), + edit_uint64(mr->VolCapacityBytes, ed6), mr->VolStatus, mr->Slot, mr->MaxVolJobs, mr->MaxVolFiles, mr->InChanger, - edit_int64(mr->VolReadTime, ed4), - edit_int64(mr->VolWriteTime, ed5), + edit_int64(mr->VolReadTime, ed7), + edit_int64(mr->VolWriteTime, ed8), mr->EndFile, mr->EndBlock, - mr->VolParts, + mr->VolType, mr->LabelType, - edit_uint64(mr->MediaId, ed6)); + edit_uint64(mr->MediaId, ed9), + edit_uint64(mr->ScratchPoolId, ed10), + mr->VolParts, + mr->VolCloudParts, + mr->LastPartBytes, + mr->Enabled); unbash_spaces(mr->VolumeName); Dmsg2(100, "Vol Info for %s: %s", jcr->Job, sd->msg); return stat; } +/* TODO: See if we want to let the FD do all kind + * of catalog request/update + */ void catalog_request(JCR *jcr, BSOCK *bs) { MEDIA_DBR mr, sdmr; JOBMEDIA_DBR jm; - char Job[MAX_NAME_LENGTH]; char pool_name[MAX_NAME_LENGTH]; int index, ok, label, writing; POOLMEM *omsg; POOL_DBR pr; - uint32_t Stripe; uint64_t MediaId; utime_t VolFirstWritten; utime_t VolLastWritten; + int n; + int Enabled; + JobId_t JobId = 0; - memset(&mr, 0, sizeof(mr)); memset(&sdmr, 0, sizeof(sdmr)); memset(&jm, 0, sizeof(jm)); - Dsm_check(100); + Dsm_check(100); /* * Request to find next appendable Volume for this Job */ - Dmsg1(100, "catreq %s", bs->msg); + Dmsg1(200, "catreq %s", bs->msg); if (!jcr->db) { omsg = get_memory(bs->msglen+1); pm_strcpy(omsg, bs->msg); - bs->fsend(_("1990 Invalid Catalog Request: %s"), omsg); + bs->fsend(_("1990 Invalid Catalog Request: %s"), omsg); Jmsg1(jcr, M_FATAL, 0, _("Invalid Catalog request; DB not open: %s"), omsg); free_memory(omsg); return; @@ -135,17 +149,22 @@ void catalog_request(JCR *jcr, BSOCK *bs) /* * Find next appendable medium for SD */ - if (sscanf(bs->msg, Find_media, &Job, &index, &pool_name, &mr.MediaType) == 4) { + n = sscanf(bs->msg, Find_media, &JobId, &index, &pool_name, &mr.MediaType, &mr.VolType); + if (n == 5) { memset(&pr, 0, sizeof(pr)); bstrncpy(pr.Name, pool_name, sizeof(pr.Name)); unbash_spaces(pr.Name); ok = db_get_pool_record(jcr, jcr->db, &pr); if (ok) { mr.PoolId = pr.PoolId; - mr.StorageId = jcr->wstore->StorageId; + set_storageid_in_mr(jcr->wstore, &mr); 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); + } else { + /* Report problem finding pool */ + Jmsg1(jcr, M_WARNING, 0, _("Pool \"%s\" not found for SD find media request.\n"), + pr.Name); } /* * Send Find Media response to Storage daemon @@ -156,11 +175,15 @@ void catalog_request(JCR *jcr, BSOCK *bs) bs->fsend(_("1901 No Media.\n")); Dmsg0(500, "1901 No Media.\n"); } + goto ok_out; + } + Dmsg1(1000, "Tried find_media. fields wanted=4, got=%d\n", n); /* * Request to find specific Volume information */ - } else if (sscanf(bs->msg, Get_Vol_Info, &Job, &mr.VolumeName, &writing) == 3) { + n = sscanf(bs->msg, Get_Vol_Info, &JobId, &mr.VolumeName, &writing); + if (n == 3) { Dmsg1(100, "CatReq GetVolInfo Vol=%s\n", mr.VolumeName); /* * Find the Volume @@ -202,7 +225,7 @@ void catalog_request(JCR *jcr, BSOCK *bs) send_volume_info_to_storage_daemon(jcr, bs, &mr); } else { /* Not suitable volume */ - bs->fsend(_("1998 Volume \"%s\" status is %s, %s.\n"), mr.VolumeName, + bs->fsend(_("1998 Volume \"%s\" catalog status is %s, %s.\n"), mr.VolumeName, mr.VolStatus, reason); } @@ -210,19 +233,25 @@ void catalog_request(JCR *jcr, BSOCK *bs) bs->fsend(_("1997 Volume \"%s\" not in catalog.\n"), mr.VolumeName); Dmsg1(100, "1997 Volume \"%s\" not in catalog.\n", mr.VolumeName); } + goto ok_out; + } + Dmsg1(1000, "Tried get_vol_info. fields wanted=3, got=%d\n", n); + /* * Request to update Media record. Comes typically at the end * of a Storage daemon Job Session, when labeling/relabeling a * Volume, or when an EOF mark is written. */ - } else if (sscanf(bs->msg, Update_media, &Job, &sdmr.VolumeName, + n = sscanf(bs->msg, Update_media, &JobId, &sdmr.VolumeName, &sdmr.VolJobs, &sdmr.VolFiles, &sdmr.VolBlocks, &sdmr.VolBytes, + &sdmr.VolABytes, &sdmr.VolHoleBytes, &sdmr.VolHoles, &sdmr.VolMounts, &sdmr.VolErrors, &sdmr.VolWrites, &sdmr.MaxVolBytes, &VolLastWritten, &sdmr.VolStatus, &sdmr.Slot, &label, &sdmr.InChanger, &sdmr.VolReadTime, &sdmr.VolWriteTime, &VolFirstWritten, - &sdmr.VolParts) == 19) { - + &sdmr.VolType, &sdmr.VolParts, &sdmr.VolCloudParts, + &sdmr.LastPartBytes, &Enabled); + if (n == 26) { db_lock(jcr->db); Dmsg3(400, "Update media %s oldStat=%s newStat=%s\n", sdmr.VolumeName, mr.VolStatus, sdmr.VolStatus); @@ -258,19 +287,16 @@ void catalog_request(JCR *jcr, BSOCK *bs) * Insanity check for VolFiles get set to a smaller value */ if (sdmr.VolFiles < mr.VolFiles) { - Jmsg(jcr, M_FATAL, 0, _("Volume Files at %u being set to %u" - " for Volume \"%s\". This is incorrect.\n"), + Jmsg(jcr, M_INFO, 0, _("Attempt to set Volume Files from %u to %u" + " for Volume \"%s\". Ignored.\n"), mr.VolFiles, sdmr.VolFiles, mr.VolumeName); - bs->fsend(_("1992 Update Media error. VolFiles=%u, CatFiles=%u\n"), - sdmr.VolFiles, mr.VolFiles); - db_unlock(jcr->db); - return; + sdmr.VolFiles = mr.VolFiles; /* keep orginal value */ } } Dmsg2(400, "Update media: BefVolJobs=%u After=%u\n", mr.VolJobs, sdmr.VolJobs); /* - * Check if the volume has been written by the job, + * Check if the volume has been written by the job, * and update the LastWritten field if needed. */ if (mr.VolBlocks != sdmr.VolBlocks && VolLastWritten != 0) { @@ -282,23 +308,36 @@ void catalog_request(JCR *jcr, BSOCK *bs) * However, do so only if we are writing the tape, i.e. * the number of VolWrites has increased. */ - if (jcr->wstore && jcr->wstore->StorageId && sdmr.VolWrites > mr.VolWrites) { - mr.StorageId = jcr->wstore->StorageId; + if (jcr->wstore && sdmr.VolWrites > mr.VolWrites) { + Dmsg2(050, "Update StorageId old=%d new=%d\n", + mr.StorageId, jcr->wstore->StorageId); + /* Update StorageId after write */ + set_storageid_in_mr(jcr->wstore, &mr); + } else { + /* Nothing written, reset same StorageId */ + set_storageid_in_mr(NULL, &mr); } /* Copy updated values to original media record */ - mr.VolJobs = sdmr.VolJobs; - mr.VolFiles = sdmr.VolFiles; - mr.VolBlocks = sdmr.VolBlocks; - mr.VolBytes = sdmr.VolBytes; - mr.VolMounts = sdmr.VolMounts; - mr.VolErrors = sdmr.VolErrors; - mr.VolWrites = sdmr.VolWrites; - mr.Slot = sdmr.Slot; - mr.InChanger = sdmr.InChanger; - mr.VolParts = sdmr.VolParts; + mr.VolJobs = sdmr.VolJobs; + mr.VolFiles = sdmr.VolFiles; + mr.VolBlocks = sdmr.VolBlocks; + mr.VolBytes = sdmr.VolBytes; + mr.VolABytes = sdmr.VolABytes; + mr.VolHoleBytes = sdmr.VolHoleBytes; + mr.VolHoles = sdmr.VolHoles; + mr.VolMounts = sdmr.VolMounts; + mr.VolErrors = sdmr.VolErrors; + mr.VolWrites = sdmr.VolWrites; + mr.Slot = sdmr.Slot; + mr.InChanger = sdmr.InChanger; + mr.VolType = sdmr.VolType; + mr.VolParts = sdmr.VolParts; + mr.VolCloudParts = sdmr.VolCloudParts; + mr.LastPartBytes = sdmr.LastPartBytes; + mr.Enabled = Enabled; /* byte assignment */ bstrncpy(mr.VolStatus, sdmr.VolStatus, sizeof(mr.VolStatus)); - if (sdmr.VolReadTime >= 0) { + if (sdmr.VolReadTime >= 0) { mr.VolReadTime = sdmr.VolReadTime; } if (sdmr.VolWriteTime >= 0) { @@ -314,44 +353,74 @@ void catalog_request(JCR *jcr, BSOCK *bs) Jmsg(jcr, M_FATAL, 0, _("Catalog error updating Media record. %s"), db_strerror(jcr->db)); bs->fsend(_("1993 Update Media error\n")); - Dmsg0(400, "send error\n"); + Pmsg0(000, "1993 Update Media error\n"); } else { (void)has_volume_expired(jcr, &mr); send_volume_info_to_storage_daemon(jcr, bs, &mr); } db_unlock(jcr->db); + goto ok_out; + } + Dmsg1(1000, "Tried update_media. fields wanted=25, got=%d\n", n); /* * Request to create a JobMedia record */ - } 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) { - - if (jcr->mig_jcr) { - jm.JobId = jcr->mig_jcr->JobId; + if (sscanf(bs->msg, Create_jobmedia, &JobId) == 1) { + if (jcr->wjcr) { + jm.JobId = jcr->wjcr->JobId; } else { jm.JobId = jcr->JobId; } - jm.MediaId = MediaId; - Dmsg6(400, "create_jobmedia JobId=%d MediaId=%d SF=%d EF=%d FI=%d LI=%d\n", - jm.JobId, jm.MediaId, jm.StartFile, jm.EndFile, jm.FirstIndex, jm.LastIndex); - if (!db_create_jobmedia_record(jcr, jcr->db, &jm)) { + ok = true; + db_lock(jcr->db); + db_start_transaction(jcr, jcr->db); + while (bs->recv() >= 0) { + if (ok && sscanf(bs->msg, "%u %u %u %u %u %u %lld\n", + &jm.FirstIndex, &jm.LastIndex, &jm.StartFile, &jm.EndFile, + &jm.StartBlock, &jm.EndBlock, &MediaId) != 7) { + ok = false; + continue; + } + if (ok) { + jm.MediaId = MediaId; + Dmsg6(400, "create_jobmedia JobId=%ld MediaId=%lu SF=%lu EF=%lu FI=%lu LI=%lu\n", + jm.JobId, jm.MediaId, jm.StartFile, jm.EndFile, jm.FirstIndex, jm.LastIndex); + ok = db_create_jobmedia_record(jcr, jcr->db, &jm); + if (jm.FirstIndex == 0 && jm.LastIndex == 0) { + jcr->dummy_jobmedia = true; + } + } + } + db_end_transaction(jcr, jcr->db); + if (!ok) { Jmsg(jcr, M_FATAL, 0, _("Catalog error creating JobMedia record. %s"), db_strerror(jcr->db)); + db_unlock(jcr->db); bs->fsend(_("1992 Create JobMedia error\n")); - } else { - Dmsg0(400, "JobMedia record created\n"); - bs->fsend(OK_create); + goto ok_out; } + db_unlock(jcr->db); + Dmsg0(400, "JobMedia record created\n"); + bs->fsend(OK_create); + goto ok_out; + } - } else { - omsg = get_memory(bs->msglen+1); - pm_strcpy(omsg, bs->msg); - bs->fsend(_("1990 Invalid Catalog Request: %s"), omsg); - Jmsg1(jcr, M_FATAL, 0, _("Invalid Catalog request: %s"), omsg); - free_memory(omsg); + /* Handle snapshot catalog request */ + if (snapshot_catreq(jcr, bs)) { + goto ok_out; } + + Dmsg1(1000, "Tried create_jobmedia. fields wanted=10, got=%d\n", n); + + /* Everything failed. Send error message. */ + omsg = get_memory(bs->msglen+1); + pm_strcpy(omsg, bs->msg); + bs->fsend(_("1990 Invalid Catalog Request: %s"), omsg); + Jmsg1(jcr, M_FATAL, 0, _("Invalid Catalog request: %s"), omsg); + free_memory(omsg); + +ok_out: Dmsg1(400, ">CatReq response: %s", bs->msg); Dmsg1(400, "Leave catreq jcr 0x%x\n", jcr); return; @@ -368,58 +437,101 @@ static void update_attribute(JCR *jcr, char *msg, int32_t msglen) uint32_t VolSessionId, VolSessionTime; int32_t Stream; uint32_t FileIndex; - uint32_t data_len; char *p; int len; char *fname, *attr; ATTR_DBR *ar = NULL; + 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; + 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 + * jcr->attr or jcr->ar */ 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", msg); - Dmsg5(400, "UpdCat VolSessId=%d VolSessT=%d FI=%d Strm=%d data_len=%d\n", - VolSessionId, VolSessionTime, FileIndex, Stream, data_len); + 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_attributes_record(jcr, jcr->db, ar)) { - Jmsg1(jcr, M_FATAL, 0, _("Attribute create error. %s"), db_strerror(jcr->db)); + Jmsg1(jcr, M_FATAL, 0, _("Attribute create error: ERR=%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, msglen); memcpy(jcr->attr, msg, msglen); p = jcr->attr - msg + p; /* point p into jcr->attr */ - skip_nonspaces(&p); /* skip FileIndex */ + skip_nonspaces(&p); /* skip FileIndex */ skip_spaces(&p); - ar->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, "dirdStream = Stream; ar->link = NULL; - if (jcr->mig_jcr) { - ar->JobId = jcr->mig_jcr->JobId; + if (jcr->wjcr) { + ar->JobId = jcr->wjcr->JobId; + Dmsg1(100, "=== set JobId=%d\n", ar->JobId); } else { ar->JobId = jcr->JobId; } @@ -444,10 +557,58 @@ static void update_attribute(JCR *jcr, char *msg, int32_t msglen) Dmsg2(400, "dirdwjcr) { + ro.JobId = jcr->wjcr->JobId; + Dmsg1(100, "=== set JobId=%ld\n", ar->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=%ld, 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) { - Jmsg(jcr, M_WARNING, 0, _("Got %s but not same File as attributes\n"), stream_to_ascii(Stream)); + Jmsg3(jcr, M_WARNING, 0, _("%s not same FileIndex=%d as attributes FI=%d\n"), + stream_to_ascii(Stream), FileIndex, ar->FileIndex); } else { /* Update digest in catalog */ char digestbuf[BASE64_SIZE(CRYPTO_DIGEST_MAX_SIZE)]; @@ -477,9 +638,13 @@ static void update_attribute(JCR *jcr, char *msg, int32_t msglen) Stream); } - bin_to_base64(digestbuf, sizeof(digestbuf), fname, len, true); - Dmsg3(400, "DigestLen=%d Digest=%s type=%d\n", strlen(digestbuf), - digestbuf, Stream); + if (len != 0) { + bin_to_base64(digestbuf, sizeof(digestbuf), fname, len, true); + Dmsg3(400, "DigestLen=%d Digest=%s type=%d\n", strlen(digestbuf), + digestbuf, Stream); + } else { + digestbuf[0] = 0; + } if (jcr->cached_attribute) { ar->Digest = digestbuf; ar->DigestType = type; @@ -488,15 +653,18 @@ static void update_attribute(JCR *jcr, char *msg, int32_t msglen) /* Update BaseFile table */ if (!db_create_attributes_record(jcr, jcr->db, ar)) { - Jmsg1(jcr, M_FATAL, 0, _("attribute create error. %s"), + Jmsg1(jcr, M_FATAL, 0, _("attribute create error. ERR=%s"), db_strerror(jcr->db)); } - jcr->cached_attribute = false; - } else { + jcr->cached_attribute = false; + } else if (ar->FileId != 0) { if (!db_add_digest_to_file_record(jcr, jcr->db, ar->FileId, digestbuf, type)) { Jmsg(jcr, M_ERROR, 0, _("Catalog error updating file digest. %s"), db_strerror(jcr->db)); } + } else { /* Something is wrong FileId == 0 */ + Jmsg(jcr, M_WARNING, 0, "Illegal FileId in update attribute: FileId=0 Stream=%d fname=%s\n", + ar->Stream, ar->fname); } } } @@ -508,25 +676,25 @@ static void update_attribute(JCR *jcr, char *msg, int32_t msglen) */ void catalog_update(JCR *jcr, BSOCK *bs) { - POOLMEM *omsg; - - if (job_canceled(jcr) || !jcr->pool->catalog_files) { - goto bail_out; /* user disabled cataloging */ + if (!jcr->pool->catalog_files) { + return; /* user disabled cataloging */ + } + if (jcr->is_job_canceled()) { + goto bail_out; } if (!jcr->db) { - omsg = get_memory(bs->msglen+1); + POOLMEM *omsg = get_memory(bs->msglen+1); pm_strcpy(omsg, bs->msg); - bs->fsend(_("1994 Invalid Catalog Update: %s"), omsg); + 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 (job_canceled(jcr)) { + if (jcr->is_job_canceled()) { + jcr->cached_attribute = false; cancel_storage_daemon_job(jcr); } } @@ -540,20 +708,21 @@ 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; + ssize_t nbytes; + ssize_t size = 0; int32_t msglen; /* message length */ POOLMEM *msg = get_pool_memory(PM_MESSAGE); FILE *spool_fd=NULL; + int32_t recnum = 0; - Dmsg0(100, "Begin despool_attributes_from_file\n"); + Dmsg1(100, "Begin despool_attributes_from_file\n", file); - if (job_canceled(jcr) || !jcr->pool->catalog_files || !jcr->db) { + if (jcr->is_job_canceled() || !jcr->pool->catalog_files || !jcr->db) { goto bail_out; /* user disabled cataloging */ } - spool_fd = fopen(file, "rb"); + spool_fd = bfopen(file, "rb"); + //Dmsg1(000, "Open attr read file=%s\n", file); if (!spool_fd) { Dmsg0(100, "cancel despool_attributes_from_file\n"); /* send an error message */ @@ -563,43 +732,72 @@ bool despool_attributes_from_file(JCR *jcr, const char *file) posix_fadvise(fileno(spool_fd), 0, 0, POSIX_FADV_WILLNEED); #endif - while (fread((char *)&pktsiz, 1, sizeof(int32_t), spool_fd) == - sizeof(int32_t)) { + /* + * We read the attributes file or stream from the SD. It should + * be in the following format: + * + * 1. 4 bytes representing the record length + * 2. An attribute string starting with: UpdCat Job=nnn FileAttributes ... + */ + for ( ;; ) { + nbytes = fread((char *)&pktsiz, 1, sizeof(int32_t), spool_fd); + if (nbytes == 0) { /* EOF */ + break; + } + if (nbytes != sizeof(int32_t)) { + Dmsg2(000, "Error: attr read status=%lld addr=%lld\n", nbytes, ftello(spool_fd)); + break; + } size += sizeof(int32_t); msglen = ntohl(pktsiz); + if (msglen > 10000000) { + Qmsg1(jcr, M_FATAL, 0, _("fread attr spool error. Wanted %ld bytes, maximum permitted 10000000 bytes\n"), msglen); + goto bail_out; + } if (msglen > 0) { - if (msglen > (int32_t) sizeof_pool_memory(msg)) { + 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) { + recnum++; + if (nbytes > 0 && strncmp(msg, "UpdCat Job", 10) != 0) { + Dmsg3(000, "Error: recnum=%ld nbytes=%lld msg=%s\n", recnum, nbytes, msg); + } + if (nbytes != (ssize_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()); + boffset_t size; + size = ftello(spool_fd); + Dmsg4(000, "Error at size=%lld record %ld: got nbytes=%lld, want msglen=%ld\n", size, recnum, (int32_t)nbytes, msglen); + Qmsg3(jcr, M_FATAL, 0, _("fread attr spool error. Wanted %ld bytes but got %lld ERR=%s\n"), + msglen, nbytes, 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() || (jcr->wjcr && jcr->wjcr->is_job_canceled())) { + goto bail_out; } } - update_attribute(jcr, msg, msglen); } if (ferror(spool_fd)) { berrno be; Qmsg1(jcr, M_FATAL, 0, _("fread attr spool error. ERR=%s\n"), be.bstrerror()); + Dmsg1(050, "fread attr spool error. ERR=%s\n", be.bstrerror()); goto bail_out; } ret = true; bail_out: if (spool_fd) { + //Dmsg1(000, "Close attr read file=%s\n", file); fclose(spool_fd); } - if (job_canceled(jcr)) { + if (jcr->is_job_canceled()) { + jcr->cached_attribute = false; cancel_storage_daemon_job(jcr); }