Scheduler.cpp 52.2 KB
Newer Older
1
/*
2
3
 * The CERN Tape Archive(CTA) project
 * Copyright(C) 2015  CERN
4
5
6
7
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 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
12
13
 * but WITHOUT ANY WARRANTY {

} without even the implied warranty of
14
15
16
17
18
19
20
 * 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, see <http://www.gnu.org/licenses/>.
 */

21

22
23
24
#include "Scheduler.hpp"
#include "ArchiveMount.hpp"
#include "RetrieveMount.hpp"
25
#include "common/dataStructures/ArchiveFileQueueCriteriaAndFileId.hpp"
26
#include "common/utils/utils.hpp"
27
#include "common/Timer.hpp"
28
#include "common/exception/NonRetryableError.hpp"
Daniele Kruse's avatar
Daniele Kruse committed
29
#include "common/exception/UserError.hpp"
30

31
#include <iostream>
32
#include <sstream>
33
34
35
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
Eric Cano's avatar
Eric Cano committed
36
#include <algorithm>
37
38
#include <random>
#include <chrono>
39

40
41
namespace cta {

42
//------------------------------------------------------------------------------
43
// constructor
44
//------------------------------------------------------------------------------
45
Scheduler::Scheduler(
46
  catalogue::Catalogue &catalogue,
47
48
49
  SchedulerDatabase &db, const uint64_t minFilesToWarrantAMount, const uint64_t minBytesToWarrantAMount): 
    m_catalogue(catalogue), m_db(db), m_minFilesToWarrantAMount(minFilesToWarrantAMount), 
    m_minBytesToWarrantAMount(minBytesToWarrantAMount) {}
50
51
52
53

//------------------------------------------------------------------------------
// destructor
//------------------------------------------------------------------------------
54
Scheduler::~Scheduler() throw() { }
55
56
57
58

//------------------------------------------------------------------------------
// ping
//------------------------------------------------------------------------------
59
60
void Scheduler::ping(log::LogContext & lc) {
  cta::utils::Timer t;
61
  m_catalogue.ping();
62
63
64
65
66
67
68
  auto catalogueTime = t.secs(cta::utils::Timer::resetCounter);
  m_db.ping();
  auto schedulerDbTime = t.secs();
  log::ScopedParamContainer spc(lc);
  spc.add("catalogueTime", catalogueTime)
     .add("schedulerDbTime", schedulerDbTime);
  lc.log(log::INFO, "In Scheduler::ping(): success.");
69
}
70

71
72
73
74
75
76
77
//------------------------------------------------------------------------------
// waitSchedulerDbThreads
//------------------------------------------------------------------------------
void Scheduler::waitSchedulerDbSubthreadsComplete() {
  m_db.waitSubthreadsComplete();
}

78
//------------------------------------------------------------------------------
Daniele Kruse's avatar
Daniele Kruse committed
79
// authorizeAdmin
80
//------------------------------------------------------------------------------
81
82
void Scheduler::authorizeAdmin(const common::dataStructures::SecurityIdentity &cliIdentity, log::LogContext & lc){
  cta::utils::Timer t;
Daniele Kruse's avatar
Daniele Kruse committed
83
84
85
  if(!(m_catalogue.isAdmin(cliIdentity))) {
    std::stringstream msg;
    msg << "User: " << cliIdentity.username << " on host: " << cliIdentity.host << " is not authorized to execute CTA admin commands";
86
    throw exception::UserError(msg.str());
Daniele Kruse's avatar
Daniele Kruse committed
87
  }
88
89
90
91
  auto catalogueTime = t.secs();
  log::ScopedParamContainer spc(lc);
  spc.add("catalogueTime", catalogueTime);
  lc.log(log::INFO, "In Scheduler::authorizeAdmin(): success.");
92
93
}

94
95
96
97
98
99
100
101
102
103
104
105
//------------------------------------------------------------------------------
// checkAndGetNextArchiveFileId
//------------------------------------------------------------------------------
uint64_t Scheduler::checkAndGetNextArchiveFileId(const std::string &instanceName,
  const std::string &storageClassName, const common::dataStructures::UserIdentity &user, log::LogContext &lc) {
  cta::utils::Timer t;
  const uint64_t archiveFileId = m_catalogue.checkAndGetNextArchiveFileId(instanceName, storageClassName, user);
  const auto catalogueTime = t.secs();
  const auto schedulerDbTime = catalogueTime;

  log::ScopedParamContainer spc(lc);
  spc.add("instanceName", instanceName)
106
107
     .add("username", user.name)
     .add("usergroup", user.group)
108
109
110
111
112
113
114
115
116
     .add("storageClass", storageClassName)
     .add("fileId", archiveFileId)
     .add("catalogueTime", catalogueTime)
     .add("schedulerDbTime", schedulerDbTime);
  lc.log(log::INFO, "Checked request and got next archive file ID");

  return archiveFileId;
}

117
118
119
120
121
122
123
124
//------------------------------------------------------------------------------
// queueArchiveWithGivenId
//------------------------------------------------------------------------------
void Scheduler::queueArchiveWithGivenId(const uint64_t archiveFileId, const std::string &instanceName,
  const cta::common::dataStructures::ArchiveRequest &request, log::LogContext &lc) {
  cta::utils::Timer t;
  using utils::postEllipsis;
  using utils::midEllipsis;
125
126
  
  if (!request.fileSize)
127
    throw cta::exception::UserError(std::string("Rejecting archive request for zero-length file: ")+request.diskFileInfo.path);
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159

  const auto queueCriteria = m_catalogue.getArchiveFileQueueCriteria(instanceName, request.storageClass,
    request.requester);
  auto catalogueTime = t.secs(cta::utils::Timer::resetCounter);

  const common::dataStructures::ArchiveFileQueueCriteriaAndFileId catalogueInfo(archiveFileId,
    queueCriteria.copyToPoolMap, queueCriteria.mountPolicy);

  m_db.queueArchive(instanceName, request, catalogueInfo, lc);
  auto schedulerDbTime = t.secs();
  log::ScopedParamContainer spc(lc);
  spc.add("instanceName", instanceName)
     .add("storageClass", request.storageClass)
     .add("diskFileID", request.diskFileID)
     .add("fileSize", request.fileSize)
     .add("fileId", catalogueInfo.fileId);
  for (auto & ctp: catalogueInfo.copyToPoolMap) {
    std::stringstream tp;
    tp << "tapePool" << ctp.first;
    spc.add(tp.str(), ctp.second);
  }
  spc.add("policyName", catalogueInfo.mountPolicy.name)
     .add("policyArchiveMinAge", catalogueInfo.mountPolicy.archiveMinRequestAge)
     .add("policyArchivePriority", catalogueInfo.mountPolicy.archivePriority)
     .add("policyMaxDrives", catalogueInfo.mountPolicy.maxDrivesAllowed)
     .add("diskFilePath", request.diskFileInfo.path)
     .add("diskFileOwner", request.diskFileInfo.owner)
     .add("diskFileGroup", request.diskFileInfo.group)
     .add("diskFileRecoveryBlob", postEllipsis(request.diskFileInfo.recoveryBlob, 20))
     .add("checksumValue", request.checksumValue)
     .add("checksumType", request.checksumType)
     .add("archiveReportURL", midEllipsis(request.archiveReportURL, 50, 15))
160
     .add("archiveErrorReportURL", midEllipsis(request.archiveErrorReportURL, 50, 15))
161
162
163
164
165
166
167
168
169
170
171
     .add("creationHost", request.creationLog.host)
     .add("creationTime", request.creationLog.time)
     .add("creationUser", request.creationLog.username)
     .add("requesterName", request.requester.name)
     .add("requesterGroup", request.requester.group)
     .add("srcURL", midEllipsis(request.srcURL, 50, 15))
     .add("catalogueTime", catalogueTime)
     .add("schedulerDbTime", schedulerDbTime);
  lc.log(log::INFO, "Queued archive request");
}

172
//------------------------------------------------------------------------------
173
// queueRetrieve
174
//------------------------------------------------------------------------------
175
void Scheduler::queueRetrieve(
176
  const std::string &instanceName,
177
178
179
180
181
  const common::dataStructures::RetrieveRequest &request,
  log::LogContext & lc) {
  using utils::postEllipsis;
  using utils::midEllipsis;
  utils::Timer t;
182
  // Get the queue criteria
183
  const common::dataStructures::RetrieveFileQueueCriteria queueCriteria =
184
    m_catalogue.prepareToRetrieveFile(instanceName, request.archiveFileID, request.requester, lc);
185
  auto catalogueTime = t.secs(cta::utils::Timer::resetCounter);
186
  std::string selectedVid = m_db.queueRetrieve(request, queueCriteria, lc);
187
188
  auto schedulerDbTime = t.secs();
  log::ScopedParamContainer spc(lc);
189
  spc.add("fileId", request.archiveFileID)
190
191
192
193
194
195
     .add("instanceName", instanceName)
     .add("diskFilePath", request.diskFileInfo.path)
     .add("diskFileOwner", request.diskFileInfo.owner)
     .add("diskFileGroup", request.diskFileInfo.group)
     .add("diskFileRecoveryBlob", postEllipsis(request.diskFileInfo.recoveryBlob, 20))
     .add("dstURL", request.dstURL)
196
     .add("errorReportURL", request.errorReportURL)
197
198
199
     .add("creationHost", request.creationLog.host)
     .add("creationTime", request.creationLog.time)
     .add("creationUser", request.creationLog.username)
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
     .add("requesterName", request.requester.name)
     .add("requesterGroup", request.requester.group)
     .add("criteriaArchiveFileId", queueCriteria.archiveFile.archiveFileID)
     .add("criteriaChecksumType", queueCriteria.archiveFile.checksumType)
     .add("criteriaChecksumValue", queueCriteria.archiveFile.checksumValue)
     .add("criteriaCreationTime", queueCriteria.archiveFile.creationTime)
     .add("criteriaDiskFileId", queueCriteria.archiveFile.diskFileId)
     .add("criteriaDiskFilePath", queueCriteria.archiveFile.diskFileInfo.path)
     .add("criteriaDiskFileOwner", queueCriteria.archiveFile.diskFileInfo.owner)
     .add("criteriaDiskRecoveryBlob", postEllipsis(queueCriteria.archiveFile.diskFileInfo.recoveryBlob, 20))
     .add("criteriaDiskInstance", queueCriteria.archiveFile.diskInstance)
     .add("criteriaFileSize", queueCriteria.archiveFile.fileSize)
     .add("reconciliationTime", queueCriteria.archiveFile.reconciliationTime)
     .add("storageClass", queueCriteria.archiveFile.storageClass);
  for (auto & tf:queueCriteria.archiveFile.tapeFiles) {
    std::stringstream tc;
    tc << "tapeCopy" << tf.first;
    spc.add(tc.str(), tf.second);
  }
  spc.add("selectedVid", selectedVid)
     .add("policyName", queueCriteria.mountPolicy.name)
     .add("policyMaxDrives", queueCriteria.mountPolicy.maxDrivesAllowed)
     .add("policyMinAge", queueCriteria.mountPolicy.retrieveMinRequestAge)
     .add("policyPriority", queueCriteria.mountPolicy.retrievePriority)
224
     .add("catalogueTime", catalogueTime)
225
226
     .add("schedulerDbTime", schedulerDbTime);
  lc.log(log::INFO, "Queued retrieve request");
227
228
229
}

//------------------------------------------------------------------------------
230
// deleteArchive
231
//------------------------------------------------------------------------------
232
void Scheduler::deleteArchive(const std::string &instanceName, const common::dataStructures::DeleteArchiveRequest &request, log::LogContext & lc) {
233
234
235
  // We have different possible scenarios here. The file can be safe in the catalogue,
  // fully queued, or partially queued.
  // First, make sure the file is not queued anymore.
236
237
238
239
240
241
242
243
244
245
246
247
// TEMPORARILY commenting out SchedulerDatabase::deleteArchiveRequest() in order
// to reduce latency.  PLEASE NOTE however that this means files "in-flight" to
// tape will not be deleted and they will appear in the CTA catalogue when they
// are finally written to tape.
//try {
//  m_db.deleteArchiveRequest(instanceName, request.archiveFileID);
//} catch (exception::Exception &dbEx) {
//  // The file was apparently not queued. If we fail to remove it from the
//  // catalogue for any reason other than it does not exist in the catalogue,
//  // then it is an error.
//  m_catalogue.deleteArchiveFile(instanceName, request.archiveFileID);
//}
248
249
250
251
252
253
  utils::Timer t;
  m_catalogue.deleteArchiveFile(instanceName, request.archiveFileID, lc);
  auto catalogueTime = t.secs(cta::utils::Timer::resetCounter);
  log::ScopedParamContainer spc(lc);
  spc.add("catalogueTime", catalogueTime);
  lc.log(log::INFO, "In Scheduler::deleteArchive(): success.");
254
255
256
}

//------------------------------------------------------------------------------
257
// cancelRetrieve
258
//------------------------------------------------------------------------------
259
260
void Scheduler::cancelRetrieve(const std::string &instanceName, const common::dataStructures::CancelRetrieveRequest &request) {
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
261
262
263
}

//------------------------------------------------------------------------------
264
// updateFileInfo
265
//------------------------------------------------------------------------------
266
267
void Scheduler::updateFileInfo(const std::string &instanceName, const common::dataStructures::UpdateFileInfoRequest &request) {
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
268
}
269

270
//------------------------------------------------------------------------------
271
272
// updateFileStorageClass
//------------------------------------------------------------------------------
273
274
void Scheduler::updateFileStorageClass(const std::string &instanceName, const common::dataStructures::UpdateFileStorageClassRequest &request) {
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
275
276
277
278
}

//------------------------------------------------------------------------------
// listStorageClass
279
//------------------------------------------------------------------------------
280
281
std::list<common::dataStructures::StorageClass> Scheduler::listStorageClass(const std::string &instanceName, const common::dataStructures::ListStorageClassRequest &request) {
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
282
283
284
}

//------------------------------------------------------------------------------
285
// labelTape
286
//------------------------------------------------------------------------------
287
void Scheduler::queueLabel(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &vid, const bool force, const bool lbp) {
288
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
289
}
290

291
292
293
//------------------------------------------------------------------------------
// repack
//------------------------------------------------------------------------------
294
void Scheduler::queueRepack(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &vid, const common::dataStructures::RepackType) {
295
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
296
297
298
299
300
}

//------------------------------------------------------------------------------
// cancelRepack
//------------------------------------------------------------------------------
301
302
void Scheduler::cancelRepack(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &vid) {
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
303
304
305
306
307
}

//------------------------------------------------------------------------------
// getRepacks
//------------------------------------------------------------------------------
308
309
std::list<common::dataStructures::RepackInfo> Scheduler::getRepacks(const common::dataStructures::SecurityIdentity &cliIdentity) {
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
310
311
312
313
314
}

//------------------------------------------------------------------------------
// getRepack
//------------------------------------------------------------------------------
315
316
common::dataStructures::RepackInfo Scheduler::getRepack(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &vid) {
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__); 
317
318
319
320
321
}

//------------------------------------------------------------------------------
// shrink
//------------------------------------------------------------------------------
322
323
void Scheduler::shrink(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &tapepool) {
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
324
325
326
327
328
}

//------------------------------------------------------------------------------
// verify
//------------------------------------------------------------------------------
329
void Scheduler::queueVerify(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &vid, const optional<uint64_t> numberOfFiles) {
330
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
331
332
333
334
335
}

//------------------------------------------------------------------------------
// cancelVerify
//------------------------------------------------------------------------------
336
337
void Scheduler::cancelVerify(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &vid) {
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
338
339
340
341
342
}

//------------------------------------------------------------------------------
// getVerifys
//------------------------------------------------------------------------------
343
344
std::list<common::dataStructures::VerifyInfo> Scheduler::getVerifys(const common::dataStructures::SecurityIdentity &cliIdentity) const {
  return std::list<common::dataStructures::VerifyInfo>(); 
345
346
347
348
349
}

//------------------------------------------------------------------------------
// getVerify
//------------------------------------------------------------------------------
350
351
common::dataStructures::VerifyInfo Scheduler::getVerify(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &vid) const {
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
352
353
}

354
//------------------------------------------------------------------------------
355
356
// readTest
//------------------------------------------------------------------------------
357
common::dataStructures::ReadTestResult Scheduler::readTest(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &driveName, const std::string &vid,
358
        const uint64_t firstFSeq, const uint64_t lastFSeq, const bool checkChecksum, const std::string &output) const {
359
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
360
361
362
363
364
}

//------------------------------------------------------------------------------
// writeTest
//------------------------------------------------------------------------------
365
common::dataStructures::WriteTestResult Scheduler::writeTest(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &driveName, const std::string &vid,
366
        const std::string &inputFile) const {
367
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
Daniele Kruse's avatar
Daniele Kruse committed
368
369
370
371
372
}

//------------------------------------------------------------------------------
// write_autoTest
//------------------------------------------------------------------------------
373
common::dataStructures::WriteTestResult Scheduler::write_autoTest(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &driveName, const std::string &vid,
374
        const uint64_t numberOfFiles, const uint64_t fileSize, const common::dataStructures::TestSourceType testSourceType) const {
375
  throw exception::Exception(std::string("Not implemented: ") + __PRETTY_FUNCTION__);
376
377
378
}

//------------------------------------------------------------------------------
379
// getDesiredDriveState
380
//------------------------------------------------------------------------------
381
382
common::dataStructures::DesiredDriveState Scheduler::getDesiredDriveState(const std::string& driveName, log::LogContext & lc) {
  utils::Timer t;
383
  auto driveStates = m_db.getDriveStates(lc);
384
385
  for (auto & d: driveStates) {
    if (d.driveName == driveName) {
386
      auto schedulerDbTime = t.secs();
387
      log::ScopedParamContainer spc(lc);
388
389
      spc.add("drive", driveName)
         .add("schedulerDbTime", schedulerDbTime);
390
      lc.log(log::INFO, "In Scheduler::getDesiredDriveState(): success.");
391
392
393
394
395
396
397
398
399
      return d.desiredDriveState;
    }
  }
  throw NoSuchDrive ("In Scheduler::getDesiredDriveState(): no such drive");
}

//------------------------------------------------------------------------------
// setDesiredDriveState
//------------------------------------------------------------------------------
400
void Scheduler::setDesiredDriveState(const common::dataStructures::SecurityIdentity &cliIdentity, const std::string &driveName, const bool up, const bool force, log::LogContext & lc) {
401
  utils::Timer t;
402
403
404
  common::dataStructures::DesiredDriveState desiredDriveState;
  desiredDriveState.up = up;
  desiredDriveState.forceDown = force;
405
  m_db.setDesiredDriveState(driveName, desiredDriveState, lc);
406
407
  auto schedulerDbTime = t.secs();
  log::ScopedParamContainer spc(lc);
408
409
  spc.add("drive", driveName)
     .add("up", up?"up":"down")
410
411
412
     .add("force", force?"yes":"no")
     .add("schedulerDbTime", schedulerDbTime);
   lc.log(log::INFO, "In Scheduler::setDesiredDriveState(): success.");   
413
414
}

415
416
417
418
419
420
//------------------------------------------------------------------------------
// removeDrive
//------------------------------------------------------------------------------
void Scheduler::removeDrive(const common::dataStructures::SecurityIdentity &cliIdentity, 
  const std::string &driveName, log::LogContext & lc) {
  utils::Timer t;
421
  m_db.removeDrive(driveName, lc);
422
423
424
425
426
427
428
  auto schedulerDbTime = t.secs();
  log::ScopedParamContainer spc(lc);
  spc.add("drive", driveName)
     .add("schedulerDbTime", schedulerDbTime);
  lc.log(log::INFO, "In Scheduler::removeDrive(): success.");   
}

429
430
431
//------------------------------------------------------------------------------
// setDesiredDriveState
//------------------------------------------------------------------------------
432
void Scheduler::reportDriveStatus(const common::dataStructures::DriveInfo& driveInfo, common::dataStructures::MountType type, common::dataStructures::DriveStatus status, log::LogContext & lc) {
433
  // TODO: mount type should be transmitted too.
434
  utils::Timer t;
435
  m_db.reportDriveStatus(driveInfo, type, status, time(NULL), lc);
436
437
  auto schedulerDbTime = t.secs();
  log::ScopedParamContainer spc(lc);
438
439
  spc.add("drive", driveInfo.driveName)
     .add("schedulerDbTime", schedulerDbTime);
Eric Cano's avatar
Eric Cano committed
440
  lc.log(log::INFO, "In Scheduler::reportDriveStatus(): success.");
441
442
443
444
445
}

//------------------------------------------------------------------------------
// getPendingArchiveJobs
//------------------------------------------------------------------------------
Eric Cano's avatar
Eric Cano committed
446
447
448
449
450
451
452
453
std::map<std::string, std::list<common::dataStructures::ArchiveJob> > Scheduler::getPendingArchiveJobs(log::LogContext & lc) const {
  utils::Timer t;
  auto ret = m_db.getArchiveJobs();
  auto schedulerDbTime = t.secs();
  log::ScopedParamContainer spc(lc);
  spc.add("schedulerDbTime", schedulerDbTime);
  lc.log(log::INFO, "In Scheduler::getPendingArchiveJobs(): success.");
  return ret;
454
455
456
457
458
}

//------------------------------------------------------------------------------
// getPendingArchiveJobs
//------------------------------------------------------------------------------
Eric Cano's avatar
Eric Cano committed
459
460
std::list<common::dataStructures::ArchiveJob> Scheduler::getPendingArchiveJobs(const std::string &tapePoolName, log::LogContext & lc) const {
  utils::Timer t;
461
  if(!m_catalogue.tapePoolExists(tapePoolName)) {
462
    throw exception::UserError(std::string("Tape pool ") + tapePoolName + " does not exist");
463
  }
Eric Cano's avatar
Eric Cano committed
464
465
466
467
468
469
470
471
  auto catalogueTime = t.secs(utils::Timer::resetCounter);
  auto ret = m_db.getArchiveJobs(tapePoolName);
  auto schedulerDbTime = t.secs();
  log::ScopedParamContainer spc(lc);
  spc.add("catalogueTime", catalogueTime)
     .add("schedulerDbTime", schedulerDbTime);
  lc.log(log::INFO, "In Scheduler::getPendingArchiveJobs(tapePool): success.");
  return ret;
472
473
474
475
476
}

//------------------------------------------------------------------------------
// getPendingRetrieveJobs
//------------------------------------------------------------------------------
Eric Cano's avatar
Eric Cano committed
477
478
479
480
481
482
483
484
std::map<std::string, std::list<common::dataStructures::RetrieveJob> > Scheduler::getPendingRetrieveJobs(log::LogContext & lc) const {
  utils::Timer t;
  auto ret =  m_db.getRetrieveJobs();
  auto schedulerDbTime = t.secs();
  log::ScopedParamContainer spc(lc);
  spc.add("schedulerDbTime", schedulerDbTime);
  lc.log(log::INFO, "In Scheduler::getPendingRetrieveJobs(): success.");
  return ret;
485
486
487
488
489
}

//------------------------------------------------------------------------------
// getPendingRetrieveJobs
//------------------------------------------------------------------------------
Eric Cano's avatar
Eric Cano committed
490
std::list<common::dataStructures::RetrieveJob> Scheduler::getPendingRetrieveJobs(const std::string& vid, log::LogContext &lc) const {
491
492
493
494
495
496
497
  utils::Timer t;
  auto ret =  m_db.getRetrieveJobs(vid);
  auto schedulerDbTime = t.secs();
  log::ScopedParamContainer spc(lc);
  spc.add("schedulerDbTime", schedulerDbTime);
  lc.log(log::INFO, "In Scheduler::getPendingRetrieveJobs(): success.");
  return ret;
498
499
}

500
501
502
//------------------------------------------------------------------------------
// getDriveStates
//------------------------------------------------------------------------------
503
504
std::list<common::dataStructures::DriveState> Scheduler::getDriveStates(const common::dataStructures::SecurityIdentity &cliIdentity, log::LogContext & lc) const {
  utils::Timer t;
505
  auto ret = m_db.getDriveStates(lc);
506
507
508
509
510
  auto schedulerDbTime = t.secs();
  log::ScopedParamContainer spc(lc);
  spc.add("schedulerDbTime", schedulerDbTime);
  lc.log(log::INFO, "In Scheduler::getDriveStates(): success.");
  return ret;
511
}
512
513

//------------------------------------------------------------------------------
514
// sortAndGetTapesForMountInfo
515
//------------------------------------------------------------------------------
516
517
518
519
void Scheduler::sortAndGetTapesForMountInfo(std::unique_ptr<SchedulerDatabase::TapeMountDecisionInfo>& mountInfo,
    const std::string & logicalLibraryName, const std::string & driveName, utils::Timer & timer,
    std::map<tpType, uint32_t> & existingMountsSummary, std::set<std::string> & tapesInUse, std::list<catalogue::TapeForWriting> & tapeList,
    double & getTapeInfoTime, double & candidateSortingTime, double & getTapeForWriteTime, log::LogContext & lc) {
520
521
522
523
524
  // The library information is not know for the tapes involved in retrieves. We 
  // need to query the catalogue now about all those tapes.
  // Build the list of tapes.
  std::set<std::string> tapeSet;
  for (auto &m:mountInfo->potentialMounts) {
525
    if (m.type==common::dataStructures::MountType::Retrieve) tapeSet.insert(m.vid);
526
527
528
  }
  if (tapeSet.size()) {
    auto tapesInfo=m_catalogue.getTapesByVid(tapeSet);
529
    getTapeInfoTime = timer.secs(utils::Timer::resetCounter);
530
    for (auto &m:mountInfo->potentialMounts) {
531
      if (m.type==common::dataStructures::MountType::Retrieve) {
532
533
        m.logicalLibrary=tapesInfo[m.vid].logicalLibraryName;
        m.tapePool=tapesInfo[m.vid].tapePoolName;
534
535
536
537
        m.vendor = tapesInfo[m.vid].vendor;
        m.mediaType = tapesInfo[m.vid].mediaType;
        m.vo = tapesInfo[m.vid].vo;
        m.capacityInBytes = tapesInfo[m.vid].capacityInBytes;
538
539
540
541
      }
    }
  }
  
542
543
544
545
546
  // We should now filter the potential mounts to keep only the ones we are
  // compatible with (match the logical library for retrieves).
  // We also only want the potential mounts for which we still have 
  // We cannot filter the archives yet
  for (auto m = mountInfo->potentialMounts.begin(); m!= mountInfo->potentialMounts.end();) {
547
    if (m->type == common::dataStructures::MountType::Retrieve && m->logicalLibrary != logicalLibraryName) {
548
549
550
551
552
553
554
555
      m = mountInfo->potentialMounts.erase(m);
    } else {
      m++;
    }
  }
  
  // With the existing mount list, we can now populate the potential mount list
  // with the per tape pool existing mount statistics.
556
  for (auto & em: mountInfo->existingOrNextMounts) {
557
558
559
560
561
562
563
    // If a mount is still listed for our own drive, it is a leftover that we disregard.
    if (em.driveName!=driveName) {
      try {
        existingMountsSummary.at(tpType(em.tapePool, em.type))++;
      } catch (std::out_of_range &) {
        existingMountsSummary[tpType(em.tapePool, em.type)] = 1;
      }
564
565
566
      if (em.vid.size()) {
        tapesInUse.insert(em.vid);
        log::ScopedParamContainer params(lc);
567
        params.add("tapeVid", em.vid)
Eric Cano's avatar
Eric Cano committed
568
569
              .add("mountType", common::dataStructures::toString(em.type))
              .add("drive", em.driveName);
570
        lc.log(log::DEBUG,"In Scheduler::sortAndGetTapesForMountInfo(): tapeAlreadyInUse found.");
571
      }
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
    }
  }
  
  // We can now filter out the potential mounts for which their mount criteria
  // is already met, filter out the potential mounts for which the maximum mount
  // quota is already reached, and weight the remaining by how much of their quota 
  // is reached
  for (auto m = mountInfo->potentialMounts.begin(); m!= mountInfo->potentialMounts.end();) {
    // Get summary data
    uint32_t existingMounts;
    try {
      existingMounts = existingMountsSummary.at(tpType(m->tapePool, m->type));
    } catch (std::out_of_range &) {
      existingMounts = 0;
    } 
587
588
    uint32_t effectiveExistingMounts = 0;
    if (m->type == common::dataStructures::MountType::Archive) effectiveExistingMounts = existingMounts;
589
    bool mountPassesACriteria = false;
590
591
    
    if (m->bytesQueued / (1 + effectiveExistingMounts) >= m_minBytesToWarrantAMount)
592
      mountPassesACriteria = true;
593
    if (m->filesQueued / (1 + effectiveExistingMounts) >= m_minFilesToWarrantAMount)
594
      mountPassesACriteria = true;
595
    if (!effectiveExistingMounts && ((time(NULL) - m->oldestJobStartTime) > m->minRequestAge))
596
      mountPassesACriteria = true;
597
    if (!mountPassesACriteria || existingMounts >= m->maxDrivesAllowed) {
598
      log::ScopedParamContainer params(lc);
599
600
      params.add("tapepool", m->tapePool);
      if ( m->type == common::dataStructures::MountType::Retrieve) {
601
        params.add("tapeVid", m->vid);
602
603
      }
      params.add("mountType", common::dataStructures::toString(m->type))
604
605
606
607
608
609
            .add("existingMounts", existingMounts)
            .add("bytesQueued", m->bytesQueued)
            .add("minBytesToWarrantMount", m_minBytesToWarrantAMount)
            .add("filesQueued", m->filesQueued)
            .add("minFilesToWarrantMount", m_minFilesToWarrantAMount)
            .add("oldestJobAge", time(NULL) - m->oldestJobStartTime)
610
            .add("minArchiveRequestAge", m->minRequestAge)
Eric Cano's avatar
Eric Cano committed
611
612
            .add("existingMounts", existingMounts)
            .add("maxDrivesAllowed", m->maxDrivesAllowed);
613
      lc.log(log::DEBUG, "In Scheduler::sortAndGetTapesForMountInfo(): Removing potential mount not passing criteria");
614
615
616
      m = mountInfo->potentialMounts.erase(m);
    } else {
      // populate the mount with a weight 
617
      m->ratioOfMountQuotaUsed = 1.0L * existingMounts / m->maxDrivesAllowed;
Eric Cano's avatar
Eric Cano committed
618
      log::ScopedParamContainer params(lc);
619
620
      params.add("tapepool", m->tapePool);
      if ( m->type == common::dataStructures::MountType::Retrieve) {
621
        params.add("tapeVid", m->vid);
622
623
      }
      params.add("mountType", common::dataStructures::toString(m->type))
Eric Cano's avatar
Eric Cano committed
624
625
626
627
628
629
            .add("existingMounts", existingMounts)
            .add("bytesQueued", m->bytesQueued)
            .add("minBytesToWarrantMount", m_minBytesToWarrantAMount)
            .add("filesQueued", m->filesQueued)
            .add("minFilesToWarrantMount", m_minFilesToWarrantAMount)
            .add("oldestJobAge", time(NULL) - m->oldestJobStartTime)
630
            .add("minArchiveRequestAge", m->minRequestAge)
Eric Cano's avatar
Eric Cano committed
631
632
633
            .add("existingMounts", existingMounts)
            .add("maxDrivesAllowed", m->maxDrivesAllowed)
            .add("ratioOfMountQuotaUsed", m->ratioOfMountQuotaUsed);
634
      lc.log(log::DEBUG, "In Scheduler::sortAndGetTapesForMountInfo(): Will consider potential mount");
635
636
637
638
639
640
641
642
643
644
      m++;
   }
  }
  
  // We can now sort the potential mounts in decreasing priority order. 
  // The ordering is defined in operator <.
  // We want the result in descending order or priority so we reverse the vector
  std::sort(mountInfo->potentialMounts.begin(), mountInfo->potentialMounts.end());
  std::reverse(mountInfo->potentialMounts.begin(), mountInfo->potentialMounts.end());
  
645
646
  candidateSortingTime = timer.secs(utils::Timer::resetCounter);
  
647
648
649
650
  // Find out if we have any potential archive mount in the list. If so, get the
  // list of tapes from the catalogue.
  if (std::count_if(
        mountInfo->potentialMounts.cbegin(), mountInfo->potentialMounts.cend(), 
651
        [](decltype(*mountInfo->potentialMounts.cbegin())& m){ return m.type == common::dataStructures::MountType::Archive; } )) {
652
    tapeList = m_catalogue.getTapesForWriting(logicalLibraryName);
653
    getTapeForWriteTime = timer.secs(utils::Timer::resetCounter);
654
  }
655
        
656
657
658
659
660
661
662
663
664
  // Remove from the tape list the ones already or soon to be mounted
  auto t=tapeList.begin();
  while (t!=tapeList.end()) {
    if (tapesInUse.count(t->vid)) {
      t=tapeList.erase(t);
    } else {
      t++;
    }
  }
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
}

//------------------------------------------------------------------------------
// getNextMountDryRun
//------------------------------------------------------------------------------
bool Scheduler::getNextMountDryRun(const std::string& logicalLibraryName, const std::string& driveName, log::LogContext& lc) {
  // We run the same algorithm as the actual getNextMount without the global lock
  // For this reason, we just return true as soon as valid mount has been found.
  utils::Timer timer;
  double getMountInfoTime = 0;
  double getTapeInfoTime = 0;
  double candidateSortingTime = 0;
  double getTapeForWriteTime = 0;
  double decisionTime = 0;
  double schedulerDbTime = 0;
  double catalogueTime = 0;
  std::unique_ptr<SchedulerDatabase::TapeMountDecisionInfo> mountInfo;
682
  mountInfo = m_db.getMountInfoNoLock(lc);
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
  getMountInfoTime = timer.secs(utils::Timer::resetCounter);
  std::map<tpType, uint32_t> existingMountsSummary;
  std::set<std::string> tapesInUse;
  std::list<catalogue::TapeForWriting> tapeList;
  
  sortAndGetTapesForMountInfo(mountInfo, logicalLibraryName, driveName, timer,
      existingMountsSummary, tapesInUse, tapeList,
      getTapeInfoTime, candidateSortingTime, getTapeForWriteTime, lc);
  
  // We can now simply iterate on the candidates until we manage to find a valid mount
  for (auto m = mountInfo->potentialMounts.begin(); m!=mountInfo->potentialMounts.end(); m++) {
    // If the mount is an archive, we still have to find a tape.
    if (m->type==common::dataStructures::MountType::Archive) {
      // We need to find a tape for archiving. It should be both in the right 
      // tape pool and in the drive's logical library
      // The first tape matching will go for a prototype.
      // TODO: improve to reuse already partially written tapes and randomization
      for (auto & t: tapeList) {
        if (t.tapePool == m->tapePool) {
          // We have our tape. That's enough.
          decisionTime += timer.secs(utils::Timer::resetCounter);
          schedulerDbTime = getMountInfoTime;
          catalogueTime = getTapeInfoTime + getTapeForWriteTime;
          uint32_t existingMounts = 0;
          try {
            existingMounts=existingMountsSummary.at(tpType(m->tapePool, m->type));
          } catch (...) {}
          log::ScopedParamContainer params(lc);
          params.add("tapepool", m->tapePool)
712
                .add("tapeVid", t.vid)
713
714
715
716
717
718
719
                .add("mountType", common::dataStructures::toString(m->type))
                .add("existingMounts", existingMounts)
                .add("bytesQueued", m->bytesQueued)
                .add("minBytesToWarrantMount", m_minBytesToWarrantAMount)
                .add("filesQueued", m->filesQueued)
                .add("minFilesToWarrantMount", m_minFilesToWarrantAMount)
                .add("oldestJobAge", time(NULL) - m->oldestJobStartTime)
720
                .add("minArchiveRequestAge", m->minRequestAge)
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
                .add("getMountInfoTime", getMountInfoTime)
                .add("getTapeInfoTime", getTapeInfoTime)
                .add("candidateSortingTime", candidateSortingTime)
                .add("getTapeForWriteTime", getTapeForWriteTime)
                .add("decisionTime", decisionTime)
                .add("schedulerDbTime", schedulerDbTime)
                .add("catalogueTime", catalogueTime);
          lc.log(log::DEBUG, "In Scheduler::getNextMountDryRun(): Found a potential mount (archive)");
          return true;
        }
      }
    } else if (m->type==common::dataStructures::MountType::Retrieve) {
      // We know the tape we intend to mount. We have to validate the tape is 
      // actually available to read (not mounted or about to be mounted, and pass 
      // on it if so).
      if (tapesInUse.count(m->vid)) continue;
      decisionTime += timer.secs(utils::Timer::resetCounter);
      log::ScopedParamContainer params(lc);
      uint32_t existingMounts = 0;
      try {
        existingMounts=existingMountsSummary.at(tpType(m->tapePool, m->type));
      } catch (...) {}
      schedulerDbTime = getMountInfoTime;
      catalogueTime = getTapeInfoTime + getTapeForWriteTime;
      params.add("tapepool", m->tapePool)
746
            .add("tapeVid", m->vid)
747
748
749
750
751
752
753
            .add("mountType", common::dataStructures::toString(m->type))
            .add("existingMounts", existingMounts)
            .add("bytesQueued", m->bytesQueued)
            .add("minBytesToWarrantMount", m_minBytesToWarrantAMount)
            .add("filesQueued", m->filesQueued)
            .add("minFilesToWarrantMount", m_minFilesToWarrantAMount)
            .add("oldestJobAge", time(NULL) - m->oldestJobStartTime)
754
            .add("minArchiveRequestAge", m->minRequestAge)
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
            .add("getMountInfoTime", getMountInfoTime)
            .add("getTapeInfoTime", getTapeInfoTime)
            .add("candidateSortingTime", candidateSortingTime)
            .add("getTapeForWriteTime", getTapeForWriteTime)
            .add("decisionTime", decisionTime)
            .add("schedulerDbTime", schedulerDbTime)
            .add("catalogueTime", catalogueTime);
      lc.log(log::DEBUG, "In Scheduler::getNextMountDryRun(): Found a potential mount (retrieve)");
      return true;
    }
  }
  schedulerDbTime = getMountInfoTime;
  catalogueTime = getTapeInfoTime + getTapeForWriteTime;
  decisionTime += timer.secs(utils::Timer::resetCounter);
  log::ScopedParamContainer params(lc);
  params.add("getMountInfoTime", getMountInfoTime)
        .add("getTapeInfoTime", getTapeInfoTime)
        .add("candidateSortingTime", candidateSortingTime)
        .add("getTapeForWriteTime", getTapeForWriteTime)
        .add("decisionTime", decisionTime)
        .add("schedulerDbTime", schedulerDbTime)
        .add("catalogueTime", catalogueTime);
777
  lc.log(log::DEBUG, "In Scheduler::getNextMountDryRun(): No valid mount found.");
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
  return false;
}


//------------------------------------------------------------------------------
// getNextMount
//------------------------------------------------------------------------------
std::unique_ptr<TapeMount> Scheduler::getNextMount(const std::string &logicalLibraryName, const std::string &driveName, log::LogContext & lc) {
  // In order to decide the next mount to do, we have to take a global lock on 
  // the scheduling, retrieve a list of all running mounts, queues sizes for 
  // tapes and tape pools, order the candidates by priority
  // below threshold, and pick one at a time, we then attempt to get a tape 
  // from the catalogue (for the archive mounts), and walk the list until we
  // mount or find nothing to do.
  // We then skip to the next candidate, until we find a suitable one and
  // return the mount, or exhaust all of them an 
  // Many steps for this logic are not specific for the database and are hence
  // implemented in the scheduler itself.
  // First, get the mount-related info from the DB
  utils::Timer timer;
  double getMountInfoTime = 0;
  double queueTrimingTime = 0;
  double getTapeInfoTime = 0;
  double candidateSortingTime = 0;
  double getTapeForWriteTime = 0;
  double decisionTime = 0;
  double mountCreationTime = 0;
  double driveStatusSetTime = 0;
  double schedulerDbTime = 0;
  double catalogueTime = 0;
  std::unique_ptr<SchedulerDatabase::TapeMountDecisionInfo> mountInfo;
  mountInfo = m_db.getMountInfo(lc);
  getMountInfoTime = timer.secs(utils::Timer::resetCounter);
  if (mountInfo->queueTrimRequired) {
    m_db.trimEmptyQueues(lc);
    queueTrimingTime = timer.secs(utils::Timer::resetCounter);
  }
  __attribute__((unused)) SchedulerDatabase::TapeMountDecisionInfo & debugMountInfo = *mountInfo;
  
  std::map<tpType, uint32_t> existingMountsSummary;
  std::set<std::string> tapesInUse;
  std::list<catalogue::TapeForWriting> tapeList;
  
  sortAndGetTapesForMountInfo(mountInfo, logicalLibraryName, driveName, timer,
      existingMountsSummary, tapesInUse, tapeList,
      getTapeInfoTime, candidateSortingTime, getTapeForWriteTime, lc);
824

825
826
827
828
  // We can now simply iterate on the candidates until we manage to create a
  // mount for one of them
  for (auto m = mountInfo->potentialMounts.begin(); m!=mountInfo->potentialMounts.end(); m++) {
    // If the mount is an archive, we still have to find a tape.
829
    if (m->type==common::dataStructures::MountType::Archive) {
830
831
832
      // We need to find a tape for archiving. It should be both in the right 
      // tape pool and in the drive's logical library
      // The first tape matching will go for a prototype.
833
834
835
      // TODO: improve to reuse already partially written tapes and randomization
      for (auto & t: tapeList) {
        if (t.tapePool == m->tapePool) {
836
837
          // We have our tape. Try to create the session. Prepare a return value
          // for it.
838
          std::unique_ptr<ArchiveMount> internalRet(new ArchiveMount(m_catalogue));
839
840
          // Get the db side of the session
          try {
841
            decisionTime += timer.secs(utils::Timer::resetCounter);
Eric Cano's avatar
Eric Cano committed
842
            internalRet->m_dbMount.reset(mountInfo->createArchiveMount(t,
843
844
                driveName, 
                logicalLibraryName, 
845
846
847
848
849
                utils::getShortHostname(),
                t.vo,
                t.mediaType,
                t.vendor,
                t.capacityInBytes,
850
                time(NULL)).release());
851
            mountCreationTime += timer.secs(utils::Timer::resetCounter);
852
            internalRet->m_sessionRunning = true;
853
            internalRet->setDriveStatus(common::dataStructures::DriveStatus::Starting);
854
            driveStatusSetTime += timer.secs(utils::Timer::resetCounter);
855
856
857
858
859
            log::ScopedParamContainer params(lc);
            uint32_t existingMounts = 0;
            try {
              existingMounts=existingMountsSummary.at(tpType(m->tapePool, m->type));
            } catch (...) {}
860
861
            schedulerDbTime = getMountInfoTime + queueTrimingTime + mountCreationTime + driveStatusSetTime;
            catalogueTime = getTapeInfoTime + getTapeForWriteTime;
862
            
863
            params.add("tapepool", m->tapePool)
864
                  .add("tapeVid", t.vid)
865
866
867
                  .add("vo",t.vo)
                  .add("mediaType",t.mediaType)
                  .add("vendor",t.vendor)
868
869
870
871
872
873
874
                  .add("mountType", common::dataStructures::toString(m->type))
                  .add("existingMounts", existingMounts)
                  .add("bytesQueued", m->bytesQueued)
                  .add("minBytesToWarrantMount", m_minBytesToWarrantAMount)
                  .add("filesQueued", m->filesQueued)
                  .add("minFilesToWarrantMount", m_minFilesToWarrantAMount)
                  .add("oldestJobAge", time(NULL) - m->oldestJobStartTime)
875
                  .add("minArchiveRequestAge", m->minRequestAge)
876
877
878
879
880
881
882
883
884
885
                  .add("getMountInfoTime", getMountInfoTime)
                  .add("queueTrimingTime", queueTrimingTime)
                  .add("getTapeInfoTime", getTapeInfoTime)
                  .add("candidateSortingTime", candidateSortingTime)
                  .add("getTapeForWriteTime", getTapeForWriteTime)
                  .add("decisionTime", decisionTime)
                  .add("mountCreationTime", mountCreationTime)
                  .add("driveStatusSetTime", driveStatusSetTime)
                  .add("schedulerDbTime", schedulerDbTime)
                  .add("catalogueTime", catalogueTime);
886
            lc.log(log::DEBUG, "In Scheduler::getNextMount(): Selected next mount (archive)");
887
            return std::unique_ptr<TapeMount> (internalRet.release());
888
889
          } catch (cta::exception::Exception & ex) {
            log::ScopedParamContainer params(lc);
890
            params.add("Message", ex.getMessage().str());
891
            lc.log(log::WARNING, "In Scheduler::getNextMount(): got an exception trying to schedule an archive mount. Trying others.");
892
893
894
895
            continue;
          }
        }
      }
896
    } else if (m->type==common::dataStructures::MountType::Retrieve) {
897
      // We know the tape we intend to mount. We have to validate the tape is 
898
899
900
901
902
      // actually available to read (not mounted or about to be mounted, and pass 
      // on it if so).
      if (tapesInUse.count(m->vid)) continue;
      try {
        // create the mount, and populate its DB side.
903
        decisionTime += timer.secs(utils::Timer::resetCounter);
904
905
906
907
908
909
        std::unique_ptr<RetrieveMount> internalRet (
          new RetrieveMount(mountInfo->createRetrieveMount(m->vid, 
            m->tapePool,
            driveName,
            logicalLibraryName, 
            utils::getShortHostname(), 
910
911
912
913
            m->vo,
            m->mediaType,
            m->vendor,
            m->capacityInBytes,
914
            time(NULL))));
915
        mountCreationTime += timer.secs(utils::Timer::resetCounter);
916
917
918
919
        internalRet->m_sessionRunning = true;
        internalRet->m_diskRunning = true;
        internalRet->m_tapeRunning = true;
        internalRet->setDriveStatus(common::dataStructures::DriveStatus::Starting);
920
        driveStatusSetTime += timer.secs(utils::Timer::resetCounter);
921
922
        log::ScopedParamContainer params(lc);
        uint32_t existingMounts = 0;
923
        try {
924
925
          existingMounts=existingMountsSummary.at(tpType(m->tapePool, m->type));
        } catch (...) {}
926
927
        schedulerDbTime = getMountInfoTime + queueTrimingTime + mountCreationTime + driveStatusSetTime;
        catalogueTime = getTapeInfoTime + getTapeForWriteTime;
928
        params.add("tapepool", m->tapePool)
929
              .add("tapeVid", m->vid)
930
931
932
              .add("vo",m->vo)
              .add("mediaType",m->mediaType)
              .add("vendor",m->vendor)
933
934
935
936
937
938
939
              .add("mountType", common::dataStructures::toString(m->type))
              .add("existingMounts", existingMounts)
              .add("bytesQueued", m->bytesQueued)
              .add("minBytesToWarrantMount", m_minBytesToWarrantAMount)
              .add("filesQueued", m->filesQueued)
              .add("minFilesToWarrantMount", m_minFilesToWarrantAMount)
              .add("oldestJobAge", time(NULL) - m->oldestJobStartTime)
940
              .add("minArchiveRequestAge", m->minRequestAge)
941
942
943
944
945
946
947
948
949
950
              .add("getMountInfoTime", getMountInfoTime)
              .add("queueTrimingTime", queueTrimingTime)
              .add("getTapeInfoTime", getTapeInfoTime)
              .add("candidateSortingTime", candidateSortingTime)
              .add("getTapeForWriteTime", getTapeForWriteTime)
              .add("decisionTime", decisionTime)
              .add("mountCreationTime", mountCreationTime)
              .add("driveStatusSetTime", driveStatusSetTime)
              .add("schedulerDbTime", schedulerDbTime)
              .add("catalogueTime", catalogueTime);
951
952
953
954
        lc.log(log::DEBUG, "In Scheduler::getNextMount(): Selected next mount (retrieve)");
        return std::unique_ptr<TapeMount> (internalRet.release()); 
      } catch (exception::Exception & ex) {
        log::ScopedParamContainer params(lc);
955
        params.add("Message", ex.getMessage().str());
956
957
        lc.log(log::WARNING, "In Scheduler::getNextMount(): got an exception trying to schedule a retrieve mount. Trying others.");
        continue;
958
959
960
961
962
      }
    } else {
      throw std::runtime_error("In Scheduler::getNextMount unexpected mount type");
    }
  }
963
964
  schedulerDbTime = getMountInfoTime + queueTrimingTime + mountCreationTime + driveStatusSetTime;
  catalogueTime = getTapeInfoTime + getTapeForWriteTime;
965
  decisionTime += timer.secs(utils::Timer::resetCounter);
966
967
968
969
970
971
972
973
974
975
976
  log::ScopedParamContainer params(lc);
  params.add("getMountInfoTime", getMountInfoTime)
        .add("queueTrimingTime", queueTrimingTime)
        .add("getTapeInfoTime", getTapeInfoTime)
        .add("candidateSortingTime", candidateSortingTime)
        .add("getTapeForWriteTime", getTapeForWriteTime)
        .add("decisionTime", decisionTime)
        .add("mountCreationTime", mountCreationTime)
        .add("driveStatusSetTime", driveStatusSetTime)
        .add("schedulerDbTime", schedulerDbTime)
        .add("catalogueTime", catalogueTime);
977
  lc.log(log::DEBUG, "In Scheduler::getNextMount(): No valid mount found.");
978
  return std::unique_ptr<TapeMount>();
979
}
980

981
982
983
//------------------------------------------------------------------------------
// getQueuesAndMountSummaries
//------------------------------------------------------------------------------
984
985
std::list<common::dataStructures::QueueAndMountSummary> Scheduler::getQueuesAndMountSummaries(log::LogContext& lc) {
  std::list<common::dataStructures::QueueAndMountSummary> ret;
986
987
988
  // Obtain a map of vids to tape info from the catalogue
  auto vid_to_tapeinfo = m_catalogue.getAllTapes();

989
  // Extract relevant information from the object store.
990
  utils::Timer t;
991
  auto mountDecisionInfo=m_db.getMountInfoNoLock(lc);
992
  auto schedulerDbTime = t.secs(utils::Timer::resetCounter);
993
  auto & mdi __attribute__((unused)) = *mountDecisionInfo;
994

995
996
  for (auto & pm: mountDecisionInfo->potentialMounts) {
    // Find or create the relevant entry.
997
    auto &summary = common::dataStructures::QueueAndMountSummary::getOrCreateEntry(ret, pm.type, pm.tapePool, pm.vid, vid_to_tapeinfo);
998
999
1000
    switch (pm.type) {
    case common::dataStructures::MountType::Archive:
      summary.mountPolicy.archivePriority = pm.priority;