/*
Bacula® - The Network Backup Solution
- Copyright (C) 2000-2009 Free Software Foundation Europe e.V.
+ Copyright (C) 2000-2014 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 three of the GNU Affero General Public
- License as published by the Free Software Foundation and included
- in the file LICENSE.
+ 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 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 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.
+ 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.
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 Director -- msgchan.c -- handles the message channel
* to the Storage daemon and the File daemon.
*
- * Kern Sibbald, August MM
+ * Written by Kern Sibbald, August MM
*
* This routine runs as a thread and must be thread reentrant.
*
static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
/* Commands sent to Storage daemon */
-static char jobcmd[] = "JobId=%s job=%s job_name=%s client_name=%s "
+static char jobcmd[] = "JobId=%s job=%s job_name=%s client_name=%s "
"type=%d level=%d FileSet=%s NoAttr=%d SpoolAttr=%d FileSetMD5=%s "
"SpoolData=%d WritePartAfterJob=%d PreferMountedVols=%d SpoolSize=%s "
- "rerunning=%d VolSessionId=%d VolSessionTime=%d\n";
+ "rerunning=%d VolSessionId=%d VolSessionTime=%d sd_client=%d "
+ "Authorization=%s\n";
static char use_storage[] = "use storage=%s media_type=%s pool_name=%s "
"pool_type=%s append=%d copy=%d stripe=%d\n";
static char use_device[] = "use device=%s\n";
/* Forward referenced functions */
extern "C" void *msg_thread(void *arg);
+BSOCK *open_sd_bsock(UAContext *ua)
+{
+ STORE *store = ua->jcr->wstore;
+
+ if (!is_bsock_open(ua->jcr->store_bsock)) {
+ ua->send_msg(_("Connecting to Storage daemon %s at %s:%d ...\n"),
+ store->name(), store->address, store->SDport);
+ if (!connect_to_storage_daemon(ua->jcr, 10, SDConnectTimeout, 1)) {
+ ua->error_msg(_("Failed to connect to Storage daemon.\n"));
+ return NULL;
+ }
+ }
+ return ua->jcr->store_bsock;
+}
+
+void close_sd_bsock(UAContext *ua)
+{
+ if (ua->jcr->store_bsock) {
+ ua->jcr->store_bsock->signal(BNET_TERMINATE);
+ free_bsock(ua->jcr->store_bsock);
+ }
+}
+
/*
* Establish a message channel connection with the Storage daemon
* and perform authentication.
bool connect_to_storage_daemon(JCR *jcr, int retry_interval,
int max_retry_time, int verbose)
{
- BSOCK *sd = new_bsock();
+ BSOCK *sd = jcr->store_bsock;
STORE *store;
- utime_t heart_beat;
+ utime_t heart_beat;
- if (jcr->store_bsock) {
+ if (is_bsock_open(sd)) {
return true; /* already connected */
}
+ if (!sd) {
+ sd = new_bsock();
+ }
/* If there is a write storage use it */
if (jcr->wstore) {
if (store->heartbeat_interval) {
heart_beat = store->heartbeat_interval;
- } else {
+ } else {
heart_beat = director->heartbeat_interval;
}
/*
* Open message channel with the Storage daemon
*/
- Dmsg2(100, "bnet_connect to Storage daemon %s:%d\n", store->address,
+ Dmsg2(100, "Connect to Storage daemon %s:%d\n", store->address,
store->SDport);
sd->set_source_address(director->DIRsrc_addr);
if (!sd->connect(jcr, retry_interval, max_retry_time, heart_beat, _("Storage daemon"),
store->address, NULL, store->SDport, verbose)) {
- sd->destroy();
sd = NULL;
}
if (!authenticate_storage_daemon(jcr, store)) {
sd->close();
- jcr->store_bsock = NULL;
return false;
}
return true;
}
/*
- * Here we ask the SD to send us the info for a
+ * Here we ask the SD to send us the info for a
* particular device resource.
*/
#ifdef xxx
bool update_device_res(JCR *jcr, DEVICE *dev)
{
- POOL_MEM device_name;
+ POOL_MEM device_name;
BSOCK *sd;
if (!connect_to_storage_daemon(jcr, 5, 30, 0)) {
return false;
bool ok = true;
STORE *storage;
BSOCK *sd;
- char auth_key[100];
+ char sd_auth_key[100];
POOL_MEM store_name, device_name, pool_name, pool_type, media_type;
POOL_MEM job_name, client_name, fileset_name;
int copy = 0;
int stripe = 0;
char ed1[30], ed2[30];
+ int sd_client;
sd = jcr->store_bsock;
/*
sd->fsend("cancel Job=%s\n", jcr->Job);
while (sd->recv() >= 0)
{ }
- }
- sd->fsend(jobcmd, edit_int64(jcr->JobId, ed1), jcr->Job,
- job_name.c_str(), client_name.c_str(),
+ }
+
+ sd_client = jcr->sd_client;
+ if (jcr->sd_auth_key) {
+ bstrncpy(sd_auth_key, jcr->sd_auth_key, sizeof(sd_auth_key));
+ } else {
+ bstrncpy(sd_auth_key, "dummy", sizeof(sd_auth_key));
+ }
+
+ sd->fsend(jobcmd, edit_int64(jcr->JobId, ed1), jcr->Job,
+ job_name.c_str(), client_name.c_str(),
jcr->getJobType(), jcr->getJobLevel(),
fileset_name.c_str(), !jcr->pool->catalog_files,
- jcr->job->SpoolAttributes, jcr->fileset->MD5, jcr->spool_data,
+ jcr->job->SpoolAttributes, jcr->fileset->MD5, jcr->spool_data,
jcr->write_part_after_job, jcr->job->PreferMountedVolumes,
edit_int64(jcr->spool_size, ed2), jcr->rerunning,
- jcr->VolSessionId, jcr->VolSessionTime);
+ jcr->VolSessionId, jcr->VolSessionTime, sd_client,
+ sd_auth_key);
+
Dmsg1(100, ">stored: %s", sd->msg);
+ Dmsg2(100, "=== rstore=%p wstore=%p\n", rstore, wstore);
if (bget_dirmsg(sd) > 0) {
Dmsg1(100, "<stored: %s", sd->msg);
if (sscanf(sd->msg, OKjob, &jcr->VolSessionId,
- &jcr->VolSessionTime, &auth_key) != 3) {
+ &jcr->VolSessionTime, &sd_auth_key) != 3) {
Dmsg1(100, "BadJob=%s\n", sd->msg);
Jmsg(jcr, M_FATAL, 0, _("Storage daemon rejected Job command: %s\n"), sd->msg);
return false;
} else {
bfree_and_null(jcr->sd_auth_key);
- jcr->sd_auth_key = bstrdup(auth_key);
+ jcr->sd_auth_key = bstrdup(sd_auth_key);
Dmsg1(150, "sd_auth_key=%s\n", jcr->sd_auth_key);
}
} else {
}
/*
- * We have two loops here. The first comes from the
- * Storage = associated with the Job, and we need
+ * We have two loops here. The first comes from the
+ * Storage = associated with the Job, and we need
* to attach to each one.
* The inner loop loops over all the alternative devices
* associated with each Storage. It selects the first
Dmsg1(100, "Rstore=%s\n", storage->name());
pm_strcpy(store_name, storage->name());
bash_spaces(store_name);
- pm_strcpy(media_type, storage->media_type);
+ if (jcr->media_type) {
+ pm_strcpy(media_type, jcr->media_type); /* user override */
+ } else {
+ pm_strcpy(media_type, storage->media_type);
+ }
bash_spaces(media_type);
- sd->fsend(use_storage, store_name.c_str(), media_type.c_str(),
+ sd->fsend(use_storage, store_name.c_str(), media_type.c_str(),
pool_name.c_str(), pool_type.c_str(), 0, copy, stripe);
Dmsg1(100, "rstore >stored: %s", sd->msg);
DEVICE *dev;
bash_spaces(pool_type);
bash_spaces(pool_name);
foreach_alist(storage, wstore) {
+ Dmsg1(100, "Wstore=%s\n", storage->name());
pm_strcpy(store_name, storage->name());
bash_spaces(store_name);
pm_strcpy(media_type, storage->media_type);
bash_spaces(media_type);
- sd->fsend(use_storage, store_name.c_str(), media_type.c_str(),
+ sd->fsend(use_storage, store_name.c_str(), media_type.c_str(),
pool_name.c_str(), pool_type.c_str(), 1, copy, stripe);
Dmsg1(100, "wstore >stored: %s", sd->msg);
Jmsg(jcr, M_FATAL, 0, _("\n"
" Storage daemon didn't accept Device \"%s\" because:\n %s"),
device_name.c_str(), err_msg.c_str()/* sd->msg */);
- } else {
+ } else {
Jmsg(jcr, M_FATAL, 0, _("\n"
- " Storage daemon didn't accept Device \"%s\" command.\n"),
+ " Storage daemon didn't accept Device \"%s\" command.\n"),
device_name.c_str());
}
}
jcr->inc_use_count(); /* mark in use by msg thread */
jcr->sd_msg_thread_done = false;
jcr->SD_msg_chan = 0;
- Dmsg0(100, "Start SD msg_thread.\n");
+ Dmsg0(150, "Start SD msg_thread.\n");
if ((status=pthread_create(&thid, NULL, msg_thread, (void *)jcr)) != 0) {
berrno be;
Jmsg1(jcr, M_ABORT, 0, _("Cannot create message thread: %s\n"), be.bstrerror(status));
return false;
}
}
- Dmsg1(100, "SD msg_thread started. use=%d\n", jcr->use_count());
+ Dmsg1(150, "SD msg_thread started. use=%d\n", jcr->use_count());
return true;
}
}
Dmsg1(400, "end loop use=%d\n", jcr->use_count());
}
- if (n == BNET_HARDEOF) {
+ if (n == BNET_HARDEOF && jcr->getJobStatus() != JS_Canceled) {
/*
* This probably should be M_FATAL, but I am not 100% sure
* that this return *always* corresponds to a dropped line.
*/
- Qmsg(jcr, M_ERROR, 0, _("Director's comm line to SD dropped.\n"));
+ Qmsg(jcr, M_ERROR, 0, _("Director's connection to SD for this Job was lost.\n"));
}
- if (is_bnet_error(sd)) {
+ if (jcr->getJobStatus() == JS_Canceled) {
+ jcr->SDJobStatus = JS_Canceled;
+ } else if (sd->is_error()) {
jcr->SDJobStatus = JS_ErrorTerminated;
}
pthread_cleanup_pop(1); /* remove and execute the handler */
if (jcr->unlink_bsr) {
unlink(jcr->RestoreBootstrap);
jcr->unlink_bsr = false;
- }
+ }
return true;
}
}
}
UnlockRes();
- bnet_close(jcr->store_bsock);
- jcr->store_bsock = NULL;
+ free_bsock(jcr->store_bsock);
break;
}
projects / bacula / bacula / blobdiff