Commit 51848dc9 authored by Daniele Kruse's avatar Daniele Kruse
Browse files

Added the Probe Session

parent c12b9b76
......@@ -35,30 +35,31 @@ enum MsgType {
/* 0 */ MSG_TYPE_NONE,
/* 1 */ MSG_TYPE_EXCEPTION,
/* 2 */ MSG_TYPE_FORKCLEANER,
/* 3 */ MSG_TYPE_FORKDATATRANSFER,
/* 4 */ MSG_TYPE_FORKLABEL,
/* 5 */ MSG_TYPE_FORKSUCCEEDED,
/* 6 */ MSG_TYPE_HEARTBEAT,
/* 7 */ MSG_TYPE_MIGRATIONJOBFROMTAPEGATEWAY,
/* 8 */ MSG_TYPE_MIGRATIONJOBFROMWRITETP,
/* 9 */ MSG_TYPE_NBFILESONTAPE,
/* 10 */ MSG_TYPE_PROCESSCRASHED,
/* 11 */ MSG_TYPE_PROCESSEXITED,
/* 12 */ MSG_TYPE_RECALLJOBFROMREADTP,
/* 13 */ MSG_TYPE_RECALLJOBFROMTAPEGATEWAY,
/* 14 */ MSG_TYPE_RETURNVALUE,
/* 15 */ MSG_TYPE_STOPPROCESSFORKER,
/* 16 */ MSG_TYPE_TAPEMOUNTEDFORMIGRATION,
/* 17 */ MSG_TYPE_TAPEMOUNTEDFORRECALL,
/* 18 */ MSG_TYPE_TAPEUNMOUNTSTARTED,
/* 19 */ MSG_TYPE_TAPEUNMOUNTED,
/* 20 */ MSG_TYPE_LABELERROR,
/* 21 */ MSG_TYPE_ACSMOUNTTAPEREADONLY,
/* 22 */ MSG_TYPE_ACSMOUNTTAPEREADWRITE,
/* 23 */ MSG_TYPE_ACSDISMOUNTTAPE,
/* 24 */ MSG_TYPE_ACSFORCEDISMOUNTTAPE,
/* 25 */ MSG_TYPE_ADDLOGPARAMS,
/* 26 */ MSG_TYPE_DELETELOGPARAMS
/* 3 */ MSG_TYPE_FORKPROBE,
/* 4 */ MSG_TYPE_FORKDATATRANSFER,
/* 5 */ MSG_TYPE_FORKLABEL,
/* 6 */ MSG_TYPE_FORKSUCCEEDED,
/* 7 */ MSG_TYPE_HEARTBEAT,
/* 8 */ MSG_TYPE_MIGRATIONJOBFROMTAPEGATEWAY,
/* 9 */ MSG_TYPE_MIGRATIONJOBFROMWRITETP,
/* 10 */ MSG_TYPE_NBFILESONTAPE,
/* 11 */ MSG_TYPE_PROCESSCRASHED,
/* 12 */ MSG_TYPE_PROCESSEXITED,
/* 13 */ MSG_TYPE_RECALLJOBFROMREADTP,
/* 14 */ MSG_TYPE_RECALLJOBFROMTAPEGATEWAY,
/* 15 */ MSG_TYPE_RETURNVALUE,
/* 16 */ MSG_TYPE_STOPPROCESSFORKER,
/* 17 */ MSG_TYPE_TAPEMOUNTEDFORMIGRATION,
/* 18 */ MSG_TYPE_TAPEMOUNTEDFORRECALL,
/* 19 */ MSG_TYPE_TAPEUNMOUNTSTARTED,
/* 20 */ MSG_TYPE_TAPEUNMOUNTED,
/* 21 */ MSG_TYPE_LABELERROR,
/* 22 */ MSG_TYPE_ACSMOUNTTAPEREADONLY,
/* 23 */ MSG_TYPE_ACSMOUNTTAPEREADWRITE,
/* 24 */ MSG_TYPE_ACSDISMOUNTTAPE,
/* 25 */ MSG_TYPE_ACSFORCEDISMOUNTTAPE,
/* 26 */ MSG_TYPE_ADDLOGPARAMS,
/* 27 */ MSG_TYPE_DELETELOGPARAMS
};
enum ProtocolVersion {
......
// This file is part of the Castor project.
// See http://castor.web.cern.ch/castor
//
// Copyright (C) 2003 CERN
// 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 2
// of the License, or (at your option) any later version.
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
//
//
//
// @author Castor Dev team, castor-dev@cern.ch
package castor.messages;
message ForkProbe {
// Description of the tape drive
required string unitname = 1;
required string dgn = 2;
required string devfilename = 3;
required string libraryslot = 4;
// Description of the probe job
required uint32 drivereadydelayinseconds = 6;
}
\ No newline at end of file
......@@ -33,6 +33,7 @@ add_library(castorTapeServerDaemon
CatalogueTransferSession.cpp
CatalogueCleanerSession.cpp
CatalogueConfig.cpp
CleanerSession.cpp
DataTransferConfig.cpp
DiskReadThreadPool.cpp
DiskReadTask.cpp
......@@ -44,11 +45,11 @@ add_library(castorTapeServerDaemon
LabelCmdAcceptHandler.cpp
LabelCmdConnectionHandler.cpp
LabelSession.cpp
CleanerSession.cpp
MigrationMemoryManager.cpp
MigrationReportPacker.cpp
MigrationTaskInjector.cpp
DataTransferSession.cpp
ProbeSession.cpp
ProcessForker.cpp
ProcessForkerConnectionHandler.cpp
ProcessForkerProxy.cpp
......
/******************************************************************************
*
* This file is part of the Castor project.
* See http://castor.web.cern.ch/castor
*
* Copyright (C) 2003 CERN
* 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 2
* of the License, or (at your option) any later version.
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
*
*
*
* @author Castor Dev team, castor-dev@cern.ch
*****************************************************************************/
#include "castor/tape/tapeserver/daemon/ProbeSession.hpp"
//------------------------------------------------------------------------------
// constructor
//------------------------------------------------------------------------------
castor::tape::tapeserver::daemon::ProbeSession::ProbeSession(
server::ProcessCap &capUtils,
castor::log::Logger &log,
const DriveConfig &driveConfig,
System::virtualWrapper &sysWrapper,
const uint32_t driveReadyDelayInSeconds):
m_capUtils(capUtils),
m_log(log),
m_driveConfig(driveConfig),
m_sysWrapper(sysWrapper),
m_driveReadyDelayInSeconds(driveReadyDelayInSeconds) {
}
//------------------------------------------------------------------------------
// execute
//------------------------------------------------------------------------------
castor::tape::tapeserver::daemon::Session::EndOfSessionAction
castor::tape::tapeserver::daemon::ProbeSession::execute() throw() {
std::string errorMessage;
try {
return exceptionThrowingExecute();
} catch(castor::exception::Exception &ex) {
errorMessage = ex.getMessage().str();
} catch(std::exception &se) {
errorMessage = se.what();
} catch(...) {
errorMessage = "Caught an unknown exception";
}
// Reaching this point means the probe failed and an exception was thrown
log::Param params[] = {
log::Param("unitName", m_driveConfig.getUnitName()),
log::Param("message", errorMessage)};
m_log(LOG_ERR, "Probe failed", params);
return MARK_DRIVE_AS_DOWN;
}
//------------------------------------------------------------------------------
// exceptionThrowingExecute
//------------------------------------------------------------------------------
castor::tape::tapeserver::daemon::Session::EndOfSessionAction
castor::tape::tapeserver::daemon::ProbeSession::exceptionThrowingExecute() {
std::list<log::Param> params;
params.push_back(log::Param("unitName", m_driveConfig.getUnitName()));
setProcessCapabilities("cap_sys_rawio+ep");
std::auto_ptr<drive::DriveInterface> drivePtr = createDrive();
drive::DriveInterface &drive = *drivePtr.get();
waitUntilDriveIsReady(drive);
if(drive.hasTapeInPlace()) {
m_log(LOG_INFO, "Probe found tape drive with a tape inside", params);
return MARK_DRIVE_AS_DOWN;
}
return MARK_DRIVE_AS_UP;
}
//------------------------------------------------------------------------------
// setProcessCapabilities
//------------------------------------------------------------------------------
void castor::tape::tapeserver::daemon::ProbeSession::setProcessCapabilities(
const std::string &capabilities) {
m_capUtils.setProcText(capabilities);
{
log::Param params[] = {
log::Param("capabilities", m_capUtils.getProcText())};
m_log(LOG_INFO, "Probe set process capabilities for using tape",
params);
}
}
//------------------------------------------------------------------------------
// createDrive
//------------------------------------------------------------------------------
std::auto_ptr<castor::tape::tapeserver::drive::DriveInterface>
castor::tape::tapeserver::daemon::ProbeSession::createDrive() {
SCSI::DeviceVector dv(m_sysWrapper);
SCSI::DeviceInfo driveInfo = dv.findBySymlink(m_driveConfig.getDevFilename());
// Instantiate the drive object
std::auto_ptr<castor::tape::tapeserver::drive::DriveInterface>
drive(drive::createDrive(driveInfo, m_sysWrapper));
if(NULL == drive.get()) {
castor::exception::Exception ex;
ex.getMessage() << "Failed to instantiate drive object";
throw ex;
}
return drive;
}
//------------------------------------------------------------------------------
// waitUntilDriveIsReady
//------------------------------------------------------------------------------
void castor::tape::tapeserver::daemon::ProbeSession::waitUntilDriveIsReady(
drive::DriveInterface &drive) {
if(0 != m_driveReadyDelayInSeconds) {
std::list<log::Param> params;
params.push_back(log::Param("unitName", m_driveConfig.getUnitName()));
params.push_back(log::Param("driveReadyDelayInSeconds",
m_driveReadyDelayInSeconds));
try {
m_log(LOG_INFO, "Probe waiting for drive to be ready", params);
drive.waitUntilReady(m_driveReadyDelayInSeconds);
m_log(LOG_INFO, "Probe detected drive is ready", params);
} catch (castor::exception::Exception &ex) {
params.push_back(log::Param("message", ex.getMessage().str()));
m_log(LOG_INFO, "Probe caught non-fatal exception whilst waiting for"
" drive to become ready", params);
}
}
}
/******************************************************************************
*
* This file is part of the Castor project.
* See http://castor.web.cern.ch/castor
*
* Copyright (C) 2003 CERN
* 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 2
* of the License, or (at your option) any later version.
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
*
*
*
* @author Castor Dev team, castor-dev@cern.ch
*****************************************************************************/
#pragma once
#include "castor/log/LogContext.hpp"
#include "castor/log/Logger.hpp"
#include "castor/mediachanger/MediaChangerFacade.hpp"
#include "castor/server/ProcessCap.hpp"
#include "castor/tape/tapeserver/daemon/DriveConfig.hpp"
#include "castor/tape/tapeserver/daemon/Session.hpp"
#include "castor/tape/tapeserver/drive/DriveInterface.hpp"
#include "castor/tape/tapeserver/file/Structures.hpp"
#include "castor/tape/tapeserver/SCSI/Device.hpp"
#include <memory>
namespace castor {
namespace tape {
namespace tapeserver {
namespace daemon {
/**
* Class responsible for probing a tape drive.
*/
class ProbeSession : public Session {
public:
/**
* Constructor
*
* @param capUtils Object providing support for UNIX capabilities.
* @param log Object representing the API to the CASTOR logging system.
* @param driveConfig Configuration of the tape drive to be probed.
* @param sysWrapper Object representing the operating system.
* @param driveReadyDelayInSeconds The maximum number of seconds to wait for
* the drive to be raedy with a tape inside of it.
*/
ProbeSession(
server::ProcessCap &capUtils,
castor::log::Logger &log,
const DriveConfig &driveConfig,
System::virtualWrapper &sysWrapper,
const uint32_t driveReadyDelayInSeconds);
/**
* Execute the session and return the type of action to be performed
* immediately after the session has completed.
*
* @return Returns the type of action to be performed after the session has
* completed.
*/
EndOfSessionAction execute() throw();
private:
/**
* Object providing support for UNIX capabilities.
*/
server::ProcessCap &m_capUtils;
/**
* The logging object
*/
castor::log::Logger & m_log;
/**
* The configuration of the tape drive to be probed.
*/
const DriveConfig m_driveConfig;
/**
* The system wrapper used to find the device and instantiate the drive object
*/
System::virtualWrapper & m_sysWrapper;
/**
* The maximum number of seconds to wait for the drive to be ready with a
* tape inside of it.
*/
const uint32_t m_driveReadyDelayInSeconds;
/**
* Execute the session and return the type of action to be performed
* immediately after the session has completed.
*
* @return Returns the type of action to be performed after the session has
* completed.
*/
EndOfSessionAction exceptionThrowingExecute();
/**
* Sets the capabilities of the process and logs the result.
*
* @param capabilities The string representation of the capabilities.
*/
void setProcessCapabilities(const std::string &capabilities);
/**
* Creates and returns the object that represents the tape drive to be
* probed.
*
* @return The tape drive.
*/
std::auto_ptr<drive::DriveInterface> createDrive();
/**
* Waits for the specified drive to be ready.
*
* @param drive The tape drive.
*/
void waitUntilDriveIsReady(drive::DriveInterface &drive);
}; // class ProbeSession
} // namespace daemon
} // namespace tapeserver
} // namespace tape
} // namespace castor
......@@ -32,6 +32,7 @@
#include "castor/messages/ForkCleaner.pb.h"
#include "castor/messages/ForkDataTransfer.pb.h"
#include "castor/messages/ForkLabel.pb.h"
#include "castor/messages/ForkProbe.pb.h"
#include "castor/messages/ForkSucceeded.pb.h"
#include "castor/messages/ProcessCrashed.pb.h"
#include "castor/messages/ProcessExited.pb.h"
......@@ -44,6 +45,7 @@
#include "castor/tape/tapeserver/daemon/DataTransferSession.hpp"
#include "castor/tape/tapeserver/daemon/DriveConfig.hpp"
#include "castor/tape/tapeserver/daemon/LabelSession.hpp"
#include "castor/tape/tapeserver/daemon/ProbeSession.hpp"
#include "castor/tape/tapeserver/daemon/ProcessForker.hpp"
#include "castor/tape/tapeserver/daemon/ProcessForkerUtils.hpp"
#include "castor/utils/SmartArrayPtr.hpp"
......@@ -256,6 +258,8 @@ castor::tape::tapeserver::daemon::ProcessForker::MsgHandlerResult
switch(frame.type) {
case messages::MSG_TYPE_FORKCLEANER:
return handleForkCleanerMsg(frame);
case messages::MSG_TYPE_FORKPROBE:
return handleForkProbeMsg(frame);
case messages::MSG_TYPE_FORKDATATRANSFER:
return handleForkDataTransferMsg(frame);
case messages::MSG_TYPE_FORKLABEL:
......@@ -329,6 +333,62 @@ castor::tape::tapeserver::daemon::ProcessForker::MsgHandlerResult
}
}
//------------------------------------------------------------------------------
// handleForkProbeMsg
//------------------------------------------------------------------------------
castor::tape::tapeserver::daemon::ProcessForker::MsgHandlerResult
castor::tape::tapeserver::daemon::ProcessForker::handleForkProbeMsg(
const ProcessForkerFrame &frame) {
// Parse the incoming request
messages::ForkProbe rqst;
ProcessForkerUtils::parsePayload(frame, rqst);
// Log the contents of the incomming request
std::list<log::Param> params;
params.push_back(log::Param("unitName", rqst.unitname()));
params.push_back(log::Param("driveReadyDelayInSeconds",
rqst.drivereadydelayinseconds()));
m_log(LOG_INFO, "ProcessForker handling ForkProbe message", params);
// Fork a label session
const pid_t forkRc = fork();
// If fork failed
if(0 > forkRc) {
return createExceptionResult(SEINTERNAL,
"Failed to fork probe session for tape drive", true);
// Else if this is the parent process
} else if(0 < forkRc) {
log::Param params[] = {log::Param("pid", forkRc)};
m_log(LOG_INFO, "ProcessForker forked probe session", params);
return createForkSucceededResult(forkRc, true);
// Else this is the child process
} else {
closeCmdReceiverSocket();
castor::utils::setProcessNameAndCmdLine(m_argv0, "probe");
try {
exit(runProbeSession(rqst));
} catch(castor::exception::Exception &ne) {
log::Param params[] = {log::Param("message", ne.getMessage().str())};
m_log(LOG_ERR, "Probe session failed", params);
} catch(std::exception &ne) {
log::Param params[] = {log::Param("message", ne.what())};
m_log(LOG_ERR, "Probe session failed", params);
} catch(...) {
log::Param params[] = {log::Param("message",
"Caught an unknown exception")};
m_log(LOG_ERR, "Probe session failed", params);
}
exit(Session::MARK_DRIVE_AS_DOWN);
}
}
//------------------------------------------------------------------------------
// handleForkDataTransferMsg
//------------------------------------------------------------------------------
......@@ -509,6 +569,47 @@ castor::tape::tapeserver::daemon::Session::EndOfSessionAction
}
}
//------------------------------------------------------------------------------
// runProbeSession
//------------------------------------------------------------------------------
castor::tape::tapeserver::daemon::Session::EndOfSessionAction
castor::tape::tapeserver::daemon::ProcessForker::runProbeSession(
const messages::ForkProbe &rqst) {
try {
server::ProcessCap capUtils;
const DriveConfig driveConfig = getDriveConfig(rqst);
std::list<log::Param> params;
params.push_back(log::Param("unitName", driveConfig.getUnitName()));
m_log(LOG_INFO, "Probe-session child-process started", params);
const int sizeOfIOThreadPoolForZMQ = 1;
messages::SmartZmqContext
zmqContext(instantiateZmqContext(sizeOfIOThreadPoolForZMQ));
messages::AcsProxyZmq acs(acs::ACS_PORT, zmqContext.get());
castor::tape::System::realWrapper sWrapper;
ProbeSession probeSession(
capUtils,
m_log,
driveConfig,
sWrapper,
rqst.drivereadydelayinseconds());
return probeSession.execute();
} catch(castor::exception::Exception &ex) {
throw ex;
} catch(std::exception &se) {
castor::exception::Exception ex;
ex.getMessage() << se.what();
throw ex;
} catch(...) {
castor::exception::Exception ex;
ex.getMessage() << "Caught an unknown exception";
throw ex;
}
}
//------------------------------------------------------------------------------
// runDataTransferSession
//------------------------------------------------------------------------------
......
......@@ -28,6 +28,7 @@
#include "castor/messages/ForkCleaner.pb.h"
#include "castor/messages/ForkDataTransfer.pb.h"
#include "castor/messages/ForkLabel.pb.h"
#include "castor/messages/ForkProbe.pb.h"
#include "castor/tape/tapeserver/daemon/DataTransferSession.hpp"
#include "castor/tape/tapeserver/daemon/ProcessForkerFrame.hpp"
#include "castor/tape/tapeserver/daemon/TapeDaemonConfig.hpp"
......@@ -219,6 +220,14 @@ private:
* @return The result of the message handler.
*/
MsgHandlerResult handleForkCleanerMsg(const ProcessForkerFrame &frame);
/**
* Handles a ForkProbe message.
*
* @param frame The frame containing the message.
* @return The result of the message handler.
*/
MsgHandlerResult handleForkProbeMsg(const ProcessForkerFrame &frame);
/**
* Runs a cleaner session. This method is to be called within the child
......@@ -228,6 +237,15 @@ private:
* @return The value to be used when exiting the child process.
*/
Session::EndOfSessionAction runCleanerSession(const messages::ForkCleaner &rqst);
/**
* Runs a probe session. This method is to be called within the child
* process responsible for running the probe session.
*
* @param rqst The ForkProbe message.
* @return The value to be used when exiting the child process.
*/
Session::EndOfSessionAction runProbeSession(const messages::ForkProbe &rqst);
/**
* Runs a data-transfer session. This method is to be called within the
......
......@@ -87,6 +87,17 @@ public:
*/
virtual pid_t forkCleaner(const DriveConfig &driveConfig,
const std::string &vid, const uint32_t driveReadyDelayInSeconds) = 0;
/**
* Forks a probe session for the specified tape drive.
*
* @param driveConfig The configuration of the tape drive.
* @param driveReadyDelayInSeconds The maximum number of seconds to wait for
* the drive to be raedy with a tape inside of it.
* @return The process identifier of the newly forked session.
*/
virtual pid_t forkProbe(const DriveConfig &driveConfig,
const uint32_t driveReadyDelayInSeconds) = 0;
}; // class ProcessForkerProxy
......
......@@ -60,3 +60,13 @@ pid_t castor::tape::tapeserver::daemon::ProcessForkerProxyDummy::
// This is a dummy method and so intentionally does nothing
return 0;
}
//------------------------------------------------------------------------------
// forkProbe
//------------------------------------------------------------------------------
pid_t castor::tape::tapeserver::daemon::ProcessForkerProxyDummy::
forkProbe(const DriveConfig &driveConfig,
const uint32_t driveReadyDelayInSeconds) {
// This is a dummy method and so intentionally does nothing
return 0;
}
......@@ -93,6 +93,17 @@ public:
*/
pid_t forkCleaner(const DriveConfig &driveConfig,
const std::string