X-Git-Url: https://git.sur5r.net/?a=blobdiff_plain;f=bacula%2Fsrc%2Fdird%2Fmigrate.c;h=0fe24e0a6738e2abe09d35318d370bf1c3fb4535;hb=44566f589dd96e4414e38ec4bf7d76b22fbcd9aa;hp=95e94414167d2d70c42efb314c40d43813e137e2;hpb=6e71af05e603ae93e3def9e0c30f25359340e480;p=bacula%2Fbacula diff --git a/bacula/src/dird/migrate.c b/bacula/src/dird/migrate.c index 95e9441416..0fe24e0a67 100644 --- a/bacula/src/dird/migrate.c +++ b/bacula/src/dird/migrate.c @@ -38,11 +38,18 @@ #include #endif +static const int dbglevel = 100; + static char OKbootstrap[] = "3000 OK bootstrap\n"; static bool get_job_to_migrate(JCR *jcr); -struct jobitems; -static bool regex_find_jobids(JCR *jcr, jobitems *ji, const char *query1, +struct idpkt; +static bool regex_find_jobids(JCR *jcr, idpkt *ids, const char *query1, const char *query2, const char *type); +static bool find_mediaid_then_jobids(JCR *jcr, idpkt *ids, const char *query1, + const char *type); +static bool find_jobids_from_mediaid_list(JCR *jcr, idpkt *ids, const char *type); +static void start_migration_job(JCR *jcr); +static int get_next_dbid_from_list(char **p, DBId_t *DBId); /* * Called here before the job is run to do the job @@ -50,9 +57,7 @@ static bool regex_find_jobids(JCR *jcr, jobitems *ji, const char *query1, */ bool do_migration_init(JCR *jcr) { - POOL_DBR pr; - - /* If we find a job to migrate it is previous_jr.JobId */ + /* If we find a job or jobs to migrate it is previous_jr.JobId */ if (!get_job_to_migrate(jcr)) { return false; } @@ -65,52 +70,22 @@ bool do_migration_init(JCR *jcr) return false; } - /* - * Get the Pool record -- first apply any level defined pools - */ - switch (jcr->previous_jr.JobLevel) { - case L_FULL: - if (jcr->full_pool) { - jcr->pool = jcr->full_pool; - } - break; - case L_INCREMENTAL: - if (jcr->inc_pool) { - jcr->pool = jcr->inc_pool; - } - break; - case L_DIFFERENTIAL: - if (jcr->dif_pool) { - jcr->pool = jcr->dif_pool; - } - break; - } - memset(&pr, 0, sizeof(pr)); - bstrncpy(pr.Name, jcr->pool->hdr.name, sizeof(pr.Name)); + apply_pool_overrides(jcr); - while (!db_get_pool_record(jcr, jcr->db, &pr)) { /* get by Name */ - /* Try to create the pool */ - if (create_pool(jcr, jcr->db, jcr->pool, POOL_OP_CREATE) < 0) { - Jmsg(jcr, M_FATAL, 0, _("Pool %s not in database. %s"), pr.Name, - db_strerror(jcr->db)); - return false; - } else { - Jmsg(jcr, M_INFO, 0, _("Pool %s created in database.\n"), pr.Name); - } + jcr->jr.PoolId = get_or_create_pool_record(jcr, jcr->pool->hdr.name); + if (jcr->jr.PoolId == 0) { + return false; } - jcr->jr.PoolId = pr.PoolId; /* If pool storage specified, use it instead of job storage */ - copy_storage(jcr, jcr->pool->storage); + copy_wstorage(jcr, jcr->pool->storage, _("Pool resource")); - if (!jcr->storage) { + if (!jcr->wstorage) { Jmsg(jcr, M_FATAL, 0, _("No Storage specification found in Job or Pool.\n")); return false; } - if (!create_restore_bootstrap_file(jcr)) { - return false; - } + create_restore_bootstrap_file(jcr); return true; } @@ -127,19 +102,35 @@ bool do_migration(JCR *jcr) char ed1[100]; BSOCK *sd; JOB *job, *prev_job; - JCR *prev_jcr; + JCR *mig_jcr; /* newly migrated job */ - if (jcr->previous_jr.JobId == 0) { + /* + * previous_jr refers to the job DB record of the Job that is + * going to be migrated. + * prev_job refers to the job resource of the Job that is + * going to be migrated. + * jcr is the jcr for the current "migration" job. It is a + * control job that is put in the DB as a migration job, which + * means that this job migrated a previous job to a new job. + * No Volume or File data is associated with this control + * job. + * mig_jcr refers to the newly migrated job that is run by + * the current jcr. It is a backup job that moves (migrates) the + * data written for the previous_jr into the new pool. This + * job (mig_jcr) becomes the new backup job that replaces + * the original backup job. + */ + if (jcr->previous_jr.JobId == 0 || jcr->ExpectedFiles == 0) { set_jcr_job_status(jcr, JS_Terminated); migration_cleanup(jcr, jcr->JobStatus); return true; /* no work */ } - Dmsg4(000, "Previous:: Name=%s JobId=%d Type=%c Level=%c\n", + Dmsg4(dbglevel, "Previous: Name=%s JobId=%d Type=%c Level=%c\n", jcr->previous_jr.Name, jcr->previous_jr.JobId, jcr->previous_jr.JobType, jcr->previous_jr.JobLevel); - Dmsg4(000, "Current: Name=%s JobId=%d Type=%c Level=%c\n", + Dmsg4(dbglevel, "Current: Name=%s JobId=%d Type=%c Level=%c\n", jcr->jr.Name, jcr->jr.JobId, jcr->jr.JobType, jcr->jr.JobLevel); @@ -151,31 +142,36 @@ bool do_migration(JCR *jcr) return false; } - /* - * prev_jcr is the new Job that corresponds to the original - * job. It "runs" at the same time as the current - * migration job and becomes a new backup job that replaces - * the original backup job. Most operations on the current - * migration jcr are also done on the prev_jcr. - */ - prev_jcr = jcr->previous_jcr = new_jcr(sizeof(JCR), dird_free_jcr); - memcpy(&prev_jcr->previous_jr, &jcr->previous_jr, sizeof(prev_jcr->previous_jr)); + /* Create a migation jcr */ + mig_jcr = jcr->mig_jcr = new_jcr(sizeof(JCR), dird_free_jcr); + memcpy(&mig_jcr->previous_jr, &jcr->previous_jr, sizeof(mig_jcr->previous_jr)); - /* Turn the prev_jcr into a "real" job */ - set_jcr_defaults(prev_jcr, prev_job); - if (!setup_job(prev_jcr)) { + /* + * Turn the mig_jcr into a "real" job that takes on the aspects of + * the previous backup job "prev_job". + */ + set_jcr_defaults(mig_jcr, prev_job); + if (!setup_job(mig_jcr)) { return false; } - /* Set output PoolId and FileSetId. */ - prev_jcr->jr.PoolId = jcr->jr.PoolId; - prev_jcr->jr.FileSetId = jcr->jr.FileSetId; + + /* Now reset the job record from the previous job */ + memcpy(&mig_jcr->jr, &jcr->previous_jr, sizeof(mig_jcr->jr)); + /* Update the jr to reflect the new values of PoolId, FileSetId, and JobId. */ + mig_jcr->jr.PoolId = jcr->jr.PoolId; + mig_jcr->jr.FileSetId = jcr->jr.FileSetId; + mig_jcr->jr.JobId = mig_jcr->JobId; + + Dmsg4(dbglevel, "mig_jcr: Name=%s JobId=%d Type=%c Level=%c\n", + mig_jcr->jr.Name, mig_jcr->jr.JobId, + mig_jcr->jr.JobType, mig_jcr->jr.JobLevel); /* * Get the PoolId used with the original job. Then * find the pool name from the database record. */ memset(&pr, 0, sizeof(pr)); - pr.PoolId = prev_jcr->previous_jr.PoolId; + pr.PoolId = mig_jcr->previous_jr.PoolId; if (!db_get_pool_record(jcr, jcr->db, &pr)) { Jmsg(jcr, M_FATAL, 0, _("Pool for JobId %s not in database. ERR=%s\n"), edit_int64(pr.PoolId, ed1), db_strerror(jcr->db)); @@ -188,54 +184,53 @@ bool do_migration(JCR *jcr) return false; } - /* Check Migration time and High/Low water marks */ - /* ***FIXME*** */ - /* If pool storage specified, use it for restore */ - copy_storage(prev_jcr, pool->storage); + copy_rstorage(mig_jcr, pool->storage, _("Pool resource")); + copy_rstorage(jcr, pool->storage, _("Pool resource")); - /* If the original backup pool has a NextPool, make sure a - * record exists in the database. + /* + * If the original backup pool has a NextPool, make sure a + * record exists in the database. Note, in this case, we + * will be migrating from pool to pool->NextPool. */ if (pool->NextPool) { - memset(&pr, 0, sizeof(pr)); - bstrncpy(pr.Name, pool->NextPool->hdr.name, sizeof(pr.Name)); - - while (!db_get_pool_record(jcr, jcr->db, &pr)) { /* get by Name */ - /* Try to create the pool */ - if (create_pool(jcr, jcr->db, pool->NextPool, POOL_OP_CREATE) < 0) { - Jmsg(jcr, M_FATAL, 0, _("Pool \"%s\" not in database. %s"), pr.Name, - db_strerror(jcr->db)); - return false; - } else { - Jmsg(jcr, M_INFO, 0, _("Pool \"%s\" created in database.\n"), pr.Name); - } + jcr->jr.PoolId = get_or_create_pool_record(jcr, pool->NextPool->hdr.name); + if (jcr->jr.PoolId == 0) { + return false; } /* * put the "NextPool" resource pointer in our jcr so that we * can pull the Storage reference from it. */ - prev_jcr->pool = jcr->pool = pool->NextPool; - prev_jcr->jr.PoolId = jcr->jr.PoolId = pr.PoolId; + mig_jcr->pool = jcr->pool = pool->NextPool; + mig_jcr->jr.PoolId = jcr->jr.PoolId; + pm_strcpy(jcr->pool_source, _("NextPool in Pool resource")); } /* If pool storage specified, use it instead of job storage for backup */ - copy_storage(jcr, jcr->pool->storage); + copy_wstorage(jcr, jcr->pool->storage, _("Pool resource")); /* Print Job Start message */ Jmsg(jcr, M_INFO, 0, _("Start Migration JobId %s, Job=%s\n"), edit_uint64(jcr->JobId, ed1), jcr->Job); set_jcr_job_status(jcr, JS_Running); - set_jcr_job_status(prev_jcr, JS_Running); - Dmsg2(000, "JobId=%d JobLevel=%c\n", jcr->jr.JobId, jcr->jr.JobLevel); + set_jcr_job_status(mig_jcr, JS_Running); + Dmsg2(dbglevel, "JobId=%d JobLevel=%c\n", jcr->jr.JobId, jcr->jr.JobLevel); + + /* Update job start record for this migration control job */ if (!db_update_job_start_record(jcr, jcr->db, &jcr->jr)) { Jmsg(jcr, M_FATAL, 0, "%s", db_strerror(jcr->db)); return false; } - if (!db_update_job_start_record(prev_jcr, prev_jcr->db, &prev_jcr->jr)) { - Jmsg(jcr, M_FATAL, 0, "%s", db_strerror(prev_jcr->db)); + Dmsg4(dbglevel, "mig_jcr: Name=%s JobId=%d Type=%c Level=%c\n", + mig_jcr->jr.Name, mig_jcr->jr.JobId, + mig_jcr->jr.JobType, mig_jcr->jr.JobLevel); + + /* Update job start record for the real migration backup job */ + if (!db_update_job_start_record(mig_jcr, mig_jcr->db, &mig_jcr->jr)) { + Jmsg(jcr, M_FATAL, 0, "%s", db_strerror(mig_jcr->db)); return false; } @@ -248,7 +243,7 @@ bool do_migration(JCR *jcr) */ Dmsg0(110, "Open connection with storage daemon\n"); set_jcr_job_status(jcr, JS_WaitSD); - set_jcr_job_status(prev_jcr, JS_WaitSD); + set_jcr_job_status(mig_jcr, JS_WaitSD); /* * Start conversation with Storage daemon */ @@ -259,10 +254,10 @@ bool do_migration(JCR *jcr) /* * Now start a job with the Storage daemon */ - Dmsg2(000, "Read store=%s, write store=%s\n", - ((STORE *)prev_jcr->storage->first())->hdr.name, - ((STORE *)jcr->storage->first())->hdr.name); - if (!start_storage_daemon_job(jcr, prev_jcr->storage, jcr->storage)) { + Dmsg2(dbglevel, "Read store=%s, write store=%s\n", + ((STORE *)jcr->rstorage->first())->name(), + ((STORE *)jcr->wstorage->first())->name()); + if (!start_storage_daemon_job(jcr, jcr->rstorage, jcr->wstorage)) { return false; } Dmsg0(150, "Storage daemon connection OK\n"); @@ -272,6 +267,10 @@ bool do_migration(JCR *jcr) return false; } + if (!bnet_fsend(sd, "run")) { + return false; + } + /* * Now start a Storage daemon message thread */ @@ -279,44 +278,47 @@ bool do_migration(JCR *jcr) return false; } - if (!bnet_fsend(sd, "run")) { - return false; - } set_jcr_job_status(jcr, JS_Running); - set_jcr_job_status(prev_jcr, JS_Running); + set_jcr_job_status(mig_jcr, JS_Running); /* Pickup Job termination data */ /* Note, the SD stores in jcr->JobFiles/ReadBytes/JobBytes/Errors */ wait_for_storage_daemon_termination(jcr); set_jcr_job_status(jcr, jcr->SDJobStatus); - if (jcr->JobStatus == JS_Terminated) { - migration_cleanup(jcr, jcr->JobStatus); - return true; + if (jcr->JobStatus != JS_Terminated) { + return false; } - return false; + migration_cleanup(jcr, jcr->JobStatus); + if (mig_jcr) { + UAContext *ua = new_ua_context(jcr); + purge_files_from_job(ua, jcr->previous_jr.JobId); + free_ua_context(ua); + } + return true; } -struct jobitems { - POOLMEM *JobIds; +struct idpkt { + POOLMEM *list; uint32_t count; }; /* - * Callback handler make list of JobIds + * Callback handler make list of DB Ids */ -static int jobid_handler(void *ctx, int num_fields, char **row) +static int dbid_handler(void *ctx, int num_fields, char **row) { - jobitems *ji = (jobitems *)ctx; + idpkt *ids = (idpkt *)ctx; - if (ji->count == 0) { - ji->JobIds[0] = 0; + Dmsg3(dbglevel, "count=%d Ids=%p %s\n", ids->count, ids->list, ids->list); + if (ids->count == 0) { + ids->list[0] = 0; } else { - pm_strcat(ji->JobIds, ","); + pm_strcat(ids->list, ","); } - pm_strcat(ji->JobIds, row[0]); - ji->count++; + pm_strcat(ids->list, row[0]); + ids->count++; return 0; } @@ -342,7 +344,7 @@ static int unique_name_handler(void *ctx, int num_fields, char **row) memset(new_item, 0, sizeof(uitem)); new_item->item = bstrdup(row[0]); - Dmsg1(000, "Item=%s\n", row[0]); + Dmsg1(dbglevel, "Item=%s\n", row[0]); item = (uitem *)list->binary_insert((void *)new_item, item_compare); if (item != new_item) { /* already in list */ free(new_item->item); @@ -359,7 +361,7 @@ const char *sql_job = /* Get JobIds from regex'ed Job names */ const char *sql_jobids_from_job = - "SELECT DISTINCT Job.JobId FROM Job,Pool" + "SELECT DISTINCT Job.JobId,Job.StartTime FROM Job,Pool" " WHERE Job.Name='%s' AND Pool.Name='%s' AND Job.PoolId=Pool.PoolId" " ORDER by Job.StartTime"; @@ -371,7 +373,7 @@ const char *sql_client = /* Get JobIds from regex'ed Client names */ const char *sql_jobids_from_client = - "SELECT DISTINCT Job.JobId FROM Job,Pool" + "SELECT DISTINCT Job.JobId,Job.StartTime FROM Job,Pool,Client" " WHERE Client.Name='%s' AND Pool.Name='%s' AND Job.PoolId=Pool.PoolId" " AND Job.ClientId=Client.ClientId " " ORDER by Job.StartTime"; @@ -379,57 +381,82 @@ const char *sql_jobids_from_client = /* Get Volume names in Pool */ const char *sql_vol = "SELECT DISTINCT VolumeName FROM Media,Pool WHERE" - " VolStatus in ('Full','Used','Error') AND" + " VolStatus in ('Full','Used','Error') AND Media.Enabled=1 AND" " Media.PoolId=Pool.PoolId AND Pool.Name='%s'"; /* Get JobIds from regex'ed Volume names */ const char *sql_jobids_from_vol = - "SELECT DISTINCT Job.JobId FROM Media,JobMedia,Job" + "SELECT DISTINCT Job.JobId,Job.StartTime FROM Media,JobMedia,Job" " WHERE Media.VolumeName='%s' AND Media.MediaId=JobMedia.MediaId" " AND JobMedia.JobId=Job.JobId" " ORDER by Job.StartTime"; - - - const char *sql_smallest_vol = "SELECT MediaId FROM Media,Pool WHERE" - " VolStatus in ('Full','Used','Error') AND" + " VolStatus in ('Full','Used','Error') AND Media.Enabled=1 AND" " Media.PoolId=Pool.PoolId AND Pool.Name='%s'" " ORDER BY VolBytes ASC LIMIT 1"; const char *sql_oldest_vol = "SELECT MediaId FROM Media,Pool WHERE" - " VolStatus in ('Full','Used','Error') AND" + " VolStatus in ('Full','Used','Error') AND Media.Enabled=1 AND" " Media.PoolId=Pool.PoolId AND Pool.Name='%s'" " ORDER BY LastWritten ASC LIMIT 1"; +/* Get JobIds when we have selected MediaId */ const char *sql_jobids_from_mediaid = - "SELECT DISTINCT Job.JobId FROM JobMedia,Job" + "SELECT DISTINCT Job.JobId,Job.StartTime FROM JobMedia,Job" " WHERE JobMedia.JobId=Job.JobId AND JobMedia.MediaId=%s" " ORDER by Job.StartTime"; +/* Get tne number of bytes in the pool */ const char *sql_pool_bytes = "SELECT SUM(VolBytes) FROM Media,Pool WHERE" - " VolStatus in ('Full','Used','Error','Append') AND" + " VolStatus in ('Full','Used','Error','Append') AND Media.Enabled=1 AND" " Media.PoolId=Pool.PoolId AND Pool.Name='%s'"; -const char *sql_vol_bytes = +/* Get tne number of bytes in the Jobs */ +const char *sql_job_bytes = + "SELECT SUM(JobBytes) FROM Job WHERE JobId IN (%s)"; + + +/* Get Media Ids in Pool */ +const char *sql_mediaids = "SELECT MediaId FROM Media,Pool WHERE" - " VolStatus in ('Full','Used','Error') AND" - " Media.PoolId=Pool.PoolId AND Pool.Name='%s' AND" - " VolBytes<%s ORDER BY LastWritten ASC LIMIT 1"; + " VolStatus in ('Full','Used','Error') AND Media.Enabled=1 AND" + " Media.PoolId=Pool.PoolId AND Pool.Name='%s' ORDER BY LastWritten ASC"; +/* Get JobIds in Pool longer than specified time */ +const char *sql_pool_time = + "SELECT DISTINCT Job.JobId from Pool,Job,Media,JobMedia WHERE" + " Pool.Name='%s' AND Media.PoolId=Pool.PoolId AND" + " VolStatus in ('Full','Used','Error') AND Media.Enabled=1 AND" + " JobMedia.JobId=Job.JobId AND Job.PoolId=Media.PoolId" + " AND Job.RealEndTime<='%s'"; -const char *sql_ujobid = - "SELECT DISTINCT Job.Job from Client,Pool,Media,Job,JobMedia " - " WHERE Media.PoolId=Pool.PoolId AND Pool.Name='%s' AND" - " JobMedia.JobId=Job.JobId AND Job.PoolId=Media.PoolId"; +/* +* const char *sql_ujobid = +* "SELECT DISTINCT Job.Job from Client,Pool,Media,Job,JobMedia " +* " WHERE Media.PoolId=Pool.PoolId AND Pool.Name='%s' AND" +* " JobMedia.JobId=Job.JobId AND Job.PoolId=Media.PoolId"; +*/ /* + * + * This is the central piece of code that finds a job or jobs + * actually JobIds to migrate. It first looks to see if one + * has been "manually" specified in jcr->MigrateJobId, and if + * so, it returns that JobId to be run. Otherwise, it + * examines the Selection Type to see what kind of migration + * we are doing (Volume, Job, Client, ...) and applies any + * Selection Pattern if appropriate to obtain a list of JobIds. + * Finally, it will loop over all the JobIds found, except the last + * one starting a new job with MigrationJobId set to that JobId, and + * finally, it returns the last JobId to the caller. + * * Returns: false on error * true if OK and jcr->previous_jr filled in */ @@ -438,32 +465,51 @@ static bool get_job_to_migrate(JCR *jcr) char ed1[30]; POOL_MEM query(PM_MESSAGE); JobId_t JobId; + DBId_t MediaId = 0; int stat; char *p; - jobitems ji; + idpkt ids, mid, jids; + db_int64_ctx ctx; + int64_t pool_bytes; + bool ok; + time_t ttime; + struct tm tm; + char dt[MAX_TIME_LENGTH]; + + ids.list = get_pool_memory(PM_MESSAGE); + ids.list[0] = 0; + ids.count = 0; + mid.list = get_pool_memory(PM_MESSAGE); + mid.list[0] = 0; + mid.count = 0; + jids.list = get_pool_memory(PM_MESSAGE); + jids.list[0] = 0; + jids.count = 0; - ji.JobIds = get_pool_memory(PM_MESSAGE); - ji.count = 0; + /* + * If MigrateJobId is set, then we migrate only that Job, + * otherwise, we go through the full selection of jobs to + * migrate. + */ if (jcr->MigrateJobId != 0) { - jcr->previous_jr.JobId = jcr->MigrateJobId; - Dmsg1(000, "previous jobid=%u\n", jcr->MigrateJobId); + Dmsg1(dbglevel, "At Job start previous jobid=%u\n", jcr->MigrateJobId); + edit_uint64(jcr->MigrateJobId, ids.list); + ids.count = 1; } else { switch (jcr->job->selection_type) { case MT_JOB: - if (!regex_find_jobids(jcr, &ji, sql_job, sql_jobids_from_job, "Job")) { + if (!regex_find_jobids(jcr, &ids, sql_job, sql_jobids_from_job, "Job")) { goto bail_out; } break; case MT_CLIENT: - if (!regex_find_jobids(jcr, &ji, sql_client, - sql_jobids_from_client, "Client")) { + if (!regex_find_jobids(jcr, &ids, sql_client, sql_jobids_from_client, "Client")) { goto bail_out; } break; case MT_VOLUME: - if (!regex_find_jobids(jcr, &ji, sql_vol, - sql_jobids_from_vol, "Volume")) { + if (!regex_find_jobids(jcr, &ids, sql_vol, sql_jobids_from_vol, "Volume")) { goto bail_out; } break; @@ -472,44 +518,160 @@ static bool get_job_to_migrate(JCR *jcr) Jmsg(jcr, M_FATAL, 0, _("No Migration SQL selection pattern specified.\n")); goto bail_out; } - Dmsg1(000, "SQL=%s\n", jcr->job->selection_pattern); + Dmsg1(dbglevel, "SQL=%s\n", jcr->job->selection_pattern); if (!db_sql_query(jcr->db, jcr->job->selection_pattern, - jobid_handler, (void *)&ji)) { + dbid_handler, (void *)&ids)) { Jmsg(jcr, M_FATAL, 0, _("SQL failed. ERR=%s\n"), db_strerror(jcr->db)); goto bail_out; } break; - - -/***** Below not implemented yet *********/ case MT_SMALLEST_VOL: - Mmsg(query, sql_smallest_vol, jcr->pool->hdr.name); -// Mmsg(query2, sql_jobids_from_mediaid, JobIds); -// Dmsg1(000, "Smallest Vol Jobids=%s\n", JobIds); + if (!find_mediaid_then_jobids(jcr, &ids, sql_smallest_vol, "Smallest Volume")) { + goto bail_out; + } break; case MT_OLDEST_VOL: - Mmsg(query, sql_oldest_vol, jcr->pool->hdr.name); -// Mmsg(query2, sql_jobids_from_mediaid, JobIds); -// Dmsg1(000, "Oldest Vol Jobids=%s\n", JobIds); + if (!find_mediaid_then_jobids(jcr, &ids, sql_oldest_vol, "Oldest Volume")) { + goto bail_out; + } break; + case MT_POOL_OCCUPANCY: + ctx.count = 0; + /* Find count of bytes in pool */ Mmsg(query, sql_pool_bytes, jcr->pool->hdr.name); -// Dmsg1(000, "Pool Occupancy Jobids=%s\n", JobIds); + if (!db_sql_query(jcr->db, query.c_str(), db_int64_handler, (void *)&ctx)) { + Jmsg(jcr, M_FATAL, 0, _("SQL failed. ERR=%s\n"), db_strerror(jcr->db)); + goto bail_out; + } + if (ctx.count == 0) { + Jmsg(jcr, M_INFO, 0, _("No Volumes found to migrate.\n")); + goto ok_out; + } + pool_bytes = ctx.value; + Dmsg2(dbglevel, "highbytes=%d pool=%d\n", (int)jcr->pool->MigrationHighBytes, + (int)pool_bytes); + if (pool_bytes < (int64_t)jcr->pool->MigrationHighBytes) { + Jmsg(jcr, M_INFO, 0, _("No Volumes found to migrate.\n")); + goto ok_out; + } + Dmsg0(dbglevel, "We should do Occupation migration.\n"); + + ids.count = 0; + /* Find a list of MediaIds that could be migrated */ + Mmsg(query, sql_mediaids, jcr->pool->hdr.name); +// Dmsg1(dbglevel, "query=%s\n", query.c_str()); + if (!db_sql_query(jcr->db, query.c_str(), dbid_handler, (void *)&ids)) { + Jmsg(jcr, M_FATAL, 0, _("SQL failed. ERR=%s\n"), db_strerror(jcr->db)); + goto bail_out; + } + if (ids.count == 0) { + Jmsg(jcr, M_INFO, 0, _("No Volumes found to migrate.\n")); + goto ok_out; + } + Dmsg2(dbglevel, "Pool Occupancy ids=%d MediaIds=%s\n", ids.count, ids.list); + + /* + * Now loop over MediaIds getting more JobIds to migrate until + * we reduce the pool occupancy below the low water mark. + */ + p = ids.list; + for (int i=0; i < (int)ids.count; i++) { + stat = get_next_dbid_from_list(&p, &MediaId); + Dmsg2(dbglevel, "get_next_dbid stat=%d MediaId=%u\n", stat, MediaId); + if (stat < 0) { + Jmsg(jcr, M_FATAL, 0, _("Invalid MediaId found.\n")); + goto bail_out; + } else if (stat == 0) { + break; + } + mid.count = 1; + Mmsg(mid.list, "%s", edit_int64(MediaId, ed1)); + ok = find_jobids_from_mediaid_list(jcr, &mid, "Volumes"); + if (!ok) { + continue; + } + if (i != 0) { + pm_strcat(jids.list, ","); + } + pm_strcat(jids.list, mid.list); + jids.count += mid.count; + + /* Now get the count of bytes added */ + ctx.count = 0; + /* Find count of bytes from Jobs */ + Mmsg(query, sql_job_bytes, mid.list); + if (!db_sql_query(jcr->db, query.c_str(), db_int64_handler, (void *)&ctx)) { + Jmsg(jcr, M_FATAL, 0, _("SQL failed. ERR=%s\n"), db_strerror(jcr->db)); + goto bail_out; + } + pool_bytes -= ctx.value; + Dmsg1(dbglevel, "Job bytes=%d\n", (int)ctx.value); + Dmsg2(dbglevel, "lowbytes=%d pool=%d\n", (int)jcr->pool->MigrationLowBytes, + (int)pool_bytes); + if (pool_bytes <= (int64_t)jcr->pool->MigrationLowBytes) { + Dmsg0(dbglevel, "We should be done.\n"); + break; + } + + } + Dmsg2(dbglevel, "Pool Occupancy ids=%d JobIds=%s\n", jids.count, jids.list); + break; + case MT_POOL_TIME: - Dmsg0(000, "Pool time not implemented\n"); + ttime = time(NULL) - (time_t)jcr->pool->MigrationTime; + (void)localtime_r(&ttime, &tm); + strftime(dt, sizeof(dt), "%Y-%m-%d %H:%M:%S", &tm); + + ids.count = 0; + Mmsg(query, sql_pool_time, jcr->pool->hdr.name, dt); +// Dmsg1(000, "query=%s\n", query.c_str()); + if (!db_sql_query(jcr->db, query.c_str(), dbid_handler, (void *)&ids)) { + Jmsg(jcr, M_FATAL, 0, _("SQL failed. ERR=%s\n"), db_strerror(jcr->db)); + goto bail_out; + } + if (ids.count == 0) { + Jmsg(jcr, M_INFO, 0, _("No Volumes found to migrate.\n")); + goto ok_out; + } + Dmsg2(dbglevel, "PoolTime ids=%d JobIds=%s\n", ids.count, ids.list); break; + default: Jmsg(jcr, M_FATAL, 0, _("Unknown Migration Selection Type.\n")); goto bail_out; } } - p = ji.JobIds; + /* + * Loop over all jobids except the last one, sending + * them to start_migration_job(), which will start a job + * for each of them. For the last JobId, we handle it below. + */ + p = ids.list; + Jmsg(jcr, M_INFO, 0, _("The following %u JobIds will be migrated: %s\n"), + ids.count, ids.list); + for (int i=1; i < (int)ids.count; i++) { + JobId = 0; + stat = get_next_jobid_from_list(&p, &JobId); + Dmsg2(dbglevel, "get_next_jobid stat=%d JobId=%u\n", stat, JobId); + jcr->MigrateJobId = JobId; + start_migration_job(jcr); + if (stat < 0) { + Jmsg(jcr, M_FATAL, 0, _("Invalid JobId found.\n")); + goto bail_out; + } else if (stat == 0) { + Jmsg(jcr, M_INFO, 0, _("No JobIds found to migrate.\n")); + goto ok_out; + } + } + + /* Now get the last JobId and handle it in the current job */ JobId = 0; stat = get_next_jobid_from_list(&p, &JobId); - Dmsg2(000, "get_next_jobid stat=%d JobId=%u\n", stat, JobId); + Dmsg2(dbglevel, "Last get_next_jobid stat=%d JobId=%u\n", stat, JobId); if (stat < 0) { Jmsg(jcr, M_FATAL, 0, _("Invalid JobId found.\n")); goto bail_out; @@ -517,9 +679,9 @@ static bool get_job_to_migrate(JCR *jcr) Jmsg(jcr, M_INFO, 0, _("No JobIds found to migrate.\n")); goto ok_out; } - + jcr->previous_jr.JobId = JobId; - Dmsg1(000, "Last jobid=%d\n", jcr->previous_jr.JobId); + Dmsg1(100, "Previous jobid=%d\n", jcr->previous_jr.JobId); if (!db_get_job_record(jcr, jcr->db, &jcr->previous_jr)) { Jmsg(jcr, M_FATAL, 0, _("Could not get job record for JobId %s to migrate. ERR=%s"), @@ -531,17 +693,87 @@ static bool get_job_to_migrate(JCR *jcr) jcr->previous_jr.JobId, jcr->previous_jr.Job); ok_out: - free_pool_memory(ji.JobIds); + free_pool_memory(ids.list); + free_pool_memory(mid.list); + free_pool_memory(jids.list); return true; bail_out: - free_pool_memory(ji.JobIds); + free_pool_memory(ids.list); + free_pool_memory(mid.list); + free_pool_memory(jids.list); return false; } +static void start_migration_job(JCR *jcr) +{ + UAContext *ua = new_ua_context(jcr); + char ed1[50]; + ua->batch = true; + Mmsg(ua->cmd, "run %s jobid=%s", jcr->job->hdr.name, + edit_uint64(jcr->MigrateJobId, ed1)); + Dmsg1(dbglevel, "=============== Migration cmd=%s\n", ua->cmd); + parse_ua_args(ua); /* parse command */ + int stat = run_cmd(ua, ua->cmd); + if (stat == 0) { + Jmsg(jcr, M_ERROR, 0, _("Could not start migration job.\n")); + } else { + Jmsg(jcr, M_INFO, 0, _("Migration JobId %d started.\n"), stat); + } + free_ua_context(ua); +} + +static bool find_mediaid_then_jobids(JCR *jcr, idpkt *ids, const char *query1, + const char *type) +{ + bool ok = false; + POOL_MEM query(PM_MESSAGE); + + ids->count = 0; + /* Basic query for MediaId */ + Mmsg(query, query1, jcr->pool->hdr.name); + if (!db_sql_query(jcr->db, query.c_str(), dbid_handler, (void *)ids)) { + Jmsg(jcr, M_FATAL, 0, _("SQL failed. ERR=%s\n"), db_strerror(jcr->db)); + goto bail_out; + } + if (ids->count == 0) { + Jmsg(jcr, M_INFO, 0, _("No %ss found to migrate.\n"), type); + } + if (ids->count != 1) { + Jmsg(jcr, M_FATAL, 0, _("SQL logic error. Count should be 1 but is %d\n"), + ids->count); + goto bail_out; + } + Dmsg1(dbglevel, "Smallest Vol Jobids=%s\n", ids->list); + + ok = find_jobids_from_mediaid_list(jcr, ids, type); + +bail_out: + return ok; +} + +static bool find_jobids_from_mediaid_list(JCR *jcr, idpkt *ids, const char *type) +{ + bool ok = false; + POOL_MEM query(PM_MESSAGE); + + Mmsg(query, sql_jobids_from_mediaid, ids->list); + ids->count = 0; + if (!db_sql_query(jcr->db, query.c_str(), dbid_handler, (void *)ids)) { + Jmsg(jcr, M_FATAL, 0, _("SQL failed. ERR=%s\n"), db_strerror(jcr->db)); + goto bail_out; + } + if (ids->count == 0) { + Jmsg(jcr, M_INFO, 0, _("No %ss found to migrate.\n"), type); + } + ok = true; +bail_out: + return ok; +} -static bool regex_find_jobids(JCR *jcr, jobitems *ji, const char *query1, - const char *query2, const char *type) { +static bool regex_find_jobids(JCR *jcr, idpkt *ids, const char *query1, + const char *query2, const char *type) +{ dlist *item_chain; uitem *item = NULL; uitem *last_item = NULL; @@ -557,7 +789,7 @@ static bool regex_find_jobids(JCR *jcr, jobitems *ji, const char *query1, type); goto bail_out; } - Dmsg1(000, "regex=%s\n", jcr->job->selection_pattern); + Dmsg1(dbglevel, "regex=%s\n", jcr->job->selection_pattern); /* Compile regex expression */ rc = regcomp(&preg, jcr->job->selection_pattern, REG_EXTENDED); if (rc != 0) { @@ -568,6 +800,7 @@ static bool regex_find_jobids(JCR *jcr, jobitems *ji, const char *query1, } /* Basic query for names */ Mmsg(query, query1, jcr->pool->hdr.name); + Dmsg1(dbglevel, "query1=%s\n", query.c_str()); if (!db_sql_query(jcr->db, query.c_str(), unique_name_handler, (void *)item_chain)) { Jmsg(jcr, M_FATAL, 0, @@ -579,11 +812,11 @@ static bool regex_find_jobids(JCR *jcr, jobitems *ji, const char *query1, const int nmatch = 30; regmatch_t pmatch[nmatch]; if (last_item) { - Dmsg1(000, "Remove item %s\n", last_item->item); + Dmsg1(dbglevel, "Remove item %s\n", last_item->item); free(last_item->item); item_chain->remove(last_item); } - Dmsg1(000, "Jobitem=%s\n", item->item); + Dmsg1(dbglevel, "Item=%s\n", item->item); rc = regexec(&preg, item->item, nmatch, pmatch, 0); if (rc == 0) { last_item = NULL; /* keep this one */ @@ -593,7 +826,7 @@ static bool regex_find_jobids(JCR *jcr, jobitems *ji, const char *query1, } if (last_item) { free(last_item->item); - Dmsg1(000, "Remove item %s\n", last_item->item); + Dmsg1(dbglevel, "Remove item %s\n", last_item->item); item_chain->remove(last_item); } regfree(&preg); @@ -602,23 +835,25 @@ static bool regex_find_jobids(JCR *jcr, jobitems *ji, const char *query1, * that have been matched by the regex, so now we need * to look up their jobids. */ - ji->count = 0; + ids->count = 0; foreach_dlist(item, item_chain) { - Dmsg1(000, "Got Job: %s\n", item->item); + Dmsg2(dbglevel, "Got %s: %s\n", type, item->item); Mmsg(query, query2, item->item, jcr->pool->hdr.name); - if (!db_sql_query(jcr->db, query.c_str(), jobid_handler, (void *)&ji)) { + Dmsg1(dbglevel, "query2=%s\n", query.c_str()); + if (!db_sql_query(jcr->db, query.c_str(), dbid_handler, (void *)ids)) { Jmsg(jcr, M_FATAL, 0, _("SQL failed. ERR=%s\n"), db_strerror(jcr->db)); goto bail_out; } } - if (ji->count == 0) { + if (ids->count == 0) { Jmsg(jcr, M_INFO, 0, _("No %ss found to migrate.\n"), type); - ok = true; } + ok = true; bail_out: - Dmsg1(000, "Job Jobids=%s\n", ji->JobIds); + Dmsg2(dbglevel, "Count=%d Jobids=%s\n", ids->count, ids->list); delete item_chain; + Dmsg0(dbglevel, "After delete item_chain\n"); return ok; } @@ -630,13 +865,14 @@ void migration_cleanup(JCR *jcr, int TermCode) { char sdt[MAX_TIME_LENGTH], edt[MAX_TIME_LENGTH]; char ec1[30], ec2[30], ec3[30], ec4[30], ec5[30], elapsed[50]; + char ec6[50], ec7[50], ec8[50]; char term_code[100], sd_term_msg[100]; const char *term_msg; int msg_type; MEDIA_DBR mr; double kbps; utime_t RunTime; - JCR *prev_jcr = jcr->previous_jcr; + JCR *mig_jcr = jcr->mig_jcr; POOL_MEM query(PM_MESSAGE); Dmsg2(100, "Enter migrate_cleanup %d %c\n", TermCode, TermCode); @@ -645,23 +881,35 @@ void migration_cleanup(JCR *jcr, int TermCode) set_jcr_job_status(jcr, TermCode); update_job_end_record(jcr); /* update database */ - /* Check if we actually did something */ - if (prev_jcr) { - prev_jcr->JobFiles = jcr->JobFiles = jcr->SDJobFiles; - prev_jcr->JobBytes = jcr->JobBytes = jcr->SDJobBytes; - prev_jcr->VolSessionId = jcr->VolSessionId; - prev_jcr->VolSessionTime = jcr->VolSessionTime; - - set_jcr_job_status(prev_jcr, TermCode); - - update_job_end_record(prev_jcr); - + /* + * Check if we actually did something. + * mig_jcr is jcr of the newly migrated job. + */ + if (mig_jcr) { + mig_jcr->JobFiles = jcr->JobFiles = jcr->SDJobFiles; + mig_jcr->JobBytes = jcr->JobBytes = jcr->SDJobBytes; + mig_jcr->VolSessionId = jcr->VolSessionId; + mig_jcr->VolSessionTime = jcr->VolSessionTime; + mig_jcr->jr.RealEndTime = 0; + mig_jcr->jr.PriorJobId = jcr->previous_jr.JobId; + + set_jcr_job_status(mig_jcr, TermCode); + + + update_job_end_record(mig_jcr); + + /* Update final items to set them to the previous job's values */ Mmsg(query, "UPDATE Job SET StartTime='%s',EndTime='%s'," "JobTDate=%s WHERE JobId=%s", jcr->previous_jr.cStartTime, jcr->previous_jr.cEndTime, edit_uint64(jcr->previous_jr.JobTDate, ec1), - edit_uint64(prev_jcr->jr.JobId, ec2)); - db_sql_query(prev_jcr->db, query.c_str(), NULL, NULL); + edit_uint64(mig_jcr->jr.JobId, ec2)); + db_sql_query(mig_jcr->db, query.c_str(), NULL, NULL); + + /* Now marke the previous job as migrated */ + Mmsg(query, "UPDATE Job SET Type='%c' WHERE JobId=%s", + (char)JT_MIGRATED_JOB, edit_uint64(jcr->previous_jr.JobId, ec1)); + db_sql_query(mig_jcr->db, query.c_str(), NULL, NULL); if (!db_get_job_record(jcr, jcr->db, &jcr->jr)) { Jmsg(jcr, M_WARNING, 0, _("Error getting job record for stats: %s"), @@ -676,9 +924,9 @@ void migration_cleanup(JCR *jcr, int TermCode) set_jcr_job_status(jcr, JS_ErrorTerminated); } - update_bootstrap_file(prev_jcr); + update_bootstrap_file(mig_jcr); - if (!db_get_job_volume_names(prev_jcr, prev_jcr->db, prev_jcr->jr.JobId, &prev_jcr->VolumeName)) { + if (!db_get_job_volume_names(mig_jcr, mig_jcr->db, mig_jcr->jr.JobId, &mig_jcr->VolumeName)) { /* * Note, if the job has erred, most likely it did not write any * tape, so suppress this "error" message since in that case @@ -686,44 +934,44 @@ void migration_cleanup(JCR *jcr, int TermCode) * normal exit should we complain about this error. */ if (jcr->JobStatus == JS_Terminated && jcr->jr.JobBytes) { - Jmsg(jcr, M_ERROR, 0, "%s", db_strerror(prev_jcr->db)); + Jmsg(jcr, M_ERROR, 0, "%s", db_strerror(mig_jcr->db)); } - prev_jcr->VolumeName[0] = 0; /* none */ + mig_jcr->VolumeName[0] = 0; /* none */ } } msg_type = M_INFO; /* by default INFO message */ switch (jcr->JobStatus) { - case JS_Terminated: - if (jcr->Errors || jcr->SDErrors) { - term_msg = _("%s OK -- with warnings"); - } else { - term_msg = _("%s OK"); - } - break; - case JS_FatalError: - case JS_ErrorTerminated: - term_msg = _("*** %s Error ***"); - msg_type = M_ERROR; /* Generate error message */ - if (jcr->store_bsock) { - bnet_sig(jcr->store_bsock, BNET_TERMINATE); - if (jcr->SD_msg_chan) { - pthread_cancel(jcr->SD_msg_chan); - } + case JS_Terminated: + if (jcr->Errors || jcr->SDErrors) { + term_msg = _("%s OK -- with warnings"); + } else { + term_msg = _("%s OK"); + } + break; + case JS_FatalError: + case JS_ErrorTerminated: + term_msg = _("*** %s Error ***"); + msg_type = M_ERROR; /* Generate error message */ + if (jcr->store_bsock) { + bnet_sig(jcr->store_bsock, BNET_TERMINATE); + if (jcr->SD_msg_chan) { + pthread_cancel(jcr->SD_msg_chan); } - break; - case JS_Canceled: - term_msg = _("%s Canceled"); - if (jcr->store_bsock) { - bnet_sig(jcr->store_bsock, BNET_TERMINATE); - if (jcr->SD_msg_chan) { - pthread_cancel(jcr->SD_msg_chan); - } + } + break; + case JS_Canceled: + term_msg = _("%s Canceled"); + if (jcr->store_bsock) { + bnet_sig(jcr->store_bsock, BNET_TERMINATE); + if (jcr->SD_msg_chan) { + pthread_cancel(jcr->SD_msg_chan); } - break; - default: - term_msg = _("Inappropriate %s term code"); - break; + } + break; + default: + term_msg = _("Inappropriate %s term code"); + break; } bsnprintf(term_code, sizeof(term_code), term_msg, "Migration"); bstrftimes(sdt, sizeof(sdt), jcr->jr.StartTime); @@ -738,17 +986,16 @@ void migration_cleanup(JCR *jcr, int TermCode) jobstatus_to_ascii(jcr->SDJobStatus, sd_term_msg, sizeof(sd_term_msg)); -// bmicrosleep(15, 0); /* for debugging SIGHUP */ - Jmsg(jcr, msg_type, 0, _("Bacula %s (%s): %s\n" -" Old Backup JobId: %u\n" -" New Backup JobId: %u\n" -" JobId: %u\n" -" Job: %s\n" +" Prev Backup JobId: %s\n" +" New Backup JobId: %s\n" +" Migration JobId: %s\n" +" Migration Job: %s\n" " Backup Level: %s%s\n" " Client: %s\n" " FileSet: \"%s\" %s\n" -" Pool: \"%s\"\n" +" Pool: \"%s\" (From %s)\n" +" Storage: \"%s\" (From %s)\n" " Start time: %s\n" " End time: %s\n" " Elapsed time: %s\n" @@ -766,14 +1013,15 @@ void migration_cleanup(JCR *jcr, int TermCode) VERSION, LSMDATE, edt, - prev_jcr ? jcr->previous_jr.JobId : 0, - prev_jcr ? prev_jcr->jr.JobId : 0, - jcr->jr.JobId, + mig_jcr ? edit_uint64(jcr->previous_jr.JobId, ec6) : "0", + mig_jcr ? edit_uint64(mig_jcr->jr.JobId, ec7) : "0", + edit_uint64(jcr->jr.JobId, ec8), jcr->jr.Job, level_to_str(jcr->JobLevel), jcr->since, - jcr->client->hdr.name, - jcr->fileset->hdr.name, jcr->FSCreateTime, - jcr->pool->hdr.name, + jcr->client->name(), + jcr->fileset->name(), jcr->FSCreateTime, + jcr->pool->name(), jcr->pool_source, + jcr->wstore->name(), jcr->storage_source, sdt, edt, edit_utime(RunTime, elapsed, sizeof(elapsed)), @@ -782,7 +1030,7 @@ void migration_cleanup(JCR *jcr, int TermCode) edit_uint64_with_commas(jcr->SDJobBytes, ec2), edit_uint64_with_suffix(jcr->SDJobBytes, ec3), (float)kbps, - prev_jcr ? prev_jcr->VolumeName : "", + mig_jcr ? mig_jcr->VolumeName : "", jcr->VolSessionId, jcr->VolSessionTime, edit_uint64_with_commas(mr.VolBytes, ec4), @@ -791,10 +1039,44 @@ void migration_cleanup(JCR *jcr, int TermCode) sd_term_msg, term_code); - Dmsg1(000, "migrate_cleanup() previous_jcr=0x%x\n", jcr->previous_jcr); - if (jcr->previous_jcr) { -// free_jcr(jcr->previous_jcr); -// jcr->previous_jcr = NULL; + Dmsg1(100, "migrate_cleanup() mig_jcr=0x%x\n", jcr->mig_jcr); + if (jcr->mig_jcr) { + free_jcr(jcr->mig_jcr); + jcr->mig_jcr = NULL; + } + Dmsg0(100, "Leave migrate_cleanup()\n"); +} + +/* + * Return next DBId from comma separated list + * + * Returns: + * 1 if next DBId returned + * 0 if no more DBIds are in list + * -1 there is an error + */ +static int get_next_dbid_from_list(char **p, DBId_t *DBId) +{ + char id[30]; + char *q = *p; + + id[0] = 0; + for (int i=0; i<(int)sizeof(id); i++) { + if (*q == 0) { + break; + } else if (*q == ',') { + q++; + break; + } + id[i] = *q++; + id[i+1] = 0; + } + if (id[0] == 0) { + return 0; + } else if (!is_a_number(id)) { + return -1; /* error */ } - Dmsg0(000, "Leave migrate_cleanup()\n"); + *p = q; + *DBId = str_to_int64(id); + return 1; }