Moved the reporting of end of session status to the report packer so session

errors for a non-crashing session can also be recorded as a failures in the
tape log.

castor/tape/tapeserver/daemon/CatalogueTransferSession.cpp

@@ -229,7 +229,8 @@ bool castor::tape::tapeserver::daemon::CatalogueTransferSession::
 //------------------------------------------------------------------------------
 void castor::tape::tapeserver::daemon::CatalogueTransferSession::
   sessionSucceeded() {
-  m_sessionLogContext.pushOrReplace(log::Param("status","success"));
+  // If the session is happy, it could determine the operational status
+  // (was there a problem or not?) by itself. status should be set by now.
   m_sessionLogContext.log(LOG_INFO, "Tape session finished");
 }
 
@@ -238,6 +239,7 @@ void castor::tape::tapeserver::daemon::CatalogueTransferSession::
 //------------------------------------------------------------------------------
 void castor::tape::tapeserver::daemon::CatalogueTransferSession::
   sessionFailed() {
+  // In case of problem, we mark the session failed ourselves
   m_sessionLogContext.pushOrReplace(log::Param("status","failed"));
   m_sessionLogContext.log(LOG_INFO, "Tape session finished");
 }

castor/tape/tapeserver/daemon/DataTransferSession.cpp

@@ -224,6 +224,7 @@ castor::tape::tapeserver::daemon::Session::EndOfSessionAction
     // of the client
     rti.setClientType(m_volInfo.clientType);
     trst.setTaskInjector(&rti);
+    rrp.setWatchdog(rwd);
     
     // We are now ready to put everything in motion. First step is to check
     // we get any concrete job to be done from the client (via the task injector)
@@ -293,8 +294,7 @@ castor::tape::tapeserver::daemon::Session::EndOfSessionAction
     
     MigrationMemoryManager mm(m_castorConf.nbBufs,
         m_castorConf.bufsz,lc);
-    MigrationReportPacker mrp(m_clientProxy,
-        lc);
+    MigrationReportPacker mrp(m_clientProxy, lc);
     MigrationWatchDog mwd(15,60*10,m_intialProcess,m_driveConfig.unitName,lc);
     TapeWriteSingleThread twst(*drive.get(),
         m_mc,
@@ -318,6 +318,7 @@ castor::tape::tapeserver::daemon::Session::EndOfSessionAction
             m_castorConf.bulkRequestMigrationMaxFiles,lc);
     drtp.setTaskInjector(&mti);
     twst.setTaskInjector(&mti);
+    mrp.setWatchdog(mwd);
     castor::utils::Timer timer;
     if (mti.synchronousInjection()) {
       const uint64_t firstFseqFromClient = mti.firstFseqToWrite();

castor/tape/tapeserver/daemon/MigrationReportPacker.cpp

@@ -24,6 +24,7 @@
 #include <memory>
 
 #include "castor/tape/tapeserver/daemon/MigrationReportPacker.hpp"
+#include "castor/tape/tapeserver/daemon/TaskWatchDog.hpp"
 #include "castor/tape/tapegateway/FileErrorReportStruct.hpp"
 #include "castor/tape/tapegateway/FileMigratedNotificationStruct.hpp"
 #include "castor/tape/tapeserver/drive/DriveInterface.hpp"
@@ -45,7 +46,8 @@ namespace daemon {
 //------------------------------------------------------------------------------
 //Constructor
 //------------------------------------------------------------------------------
-MigrationReportPacker::MigrationReportPacker(client::ClientInterface & tg,castor::log::LogContext lc):
+MigrationReportPacker::MigrationReportPacker(client::ClientInterface & tg,
+  castor::log::LogContext lc):
 ReportPackerInterface<detail::Migration>(tg,lc),
 m_workerThread(*this),m_errorHappened(false),m_continue(true) {
 }
@@ -245,6 +247,9 @@ void MigrationReportPacker::ReportEndofSession::execute(MigrationReportPacker& r
       .add("sendRecvDuration", chrono.sendRecvDuration)
       .add("transactionId", chrono.transactionId);
     reportPacker.m_lc.log(LOG_INFO,"Reported end of session to client");
+    if(reportPacker.m_watchdog) {
+      reportPacker.m_watchdog->addParameter(log::Param("status","failure"));
+    }
   }
   else {
     reportFileErrors(reportPacker);
@@ -257,6 +262,9 @@ void MigrationReportPacker::ReportEndofSession::execute(MigrationReportPacker& r
       .add("sendRecvDuration", chrono.sendRecvDuration)
       .add("transactionId", chrono.transactionId);
     reportPacker.m_lc.log(LOG_ERR,"Reported end of session with error to client due to previous file errors");
+    if(reportPacker.m_watchdog) {
+      reportPacker.m_watchdog->addParameter(log::Param("status","success"));
+    }
   }
   reportPacker.m_continue=false;
 }
@@ -286,6 +294,9 @@ void MigrationReportPacker::ReportEndofSessionWithErrors::execute(MigrationRepor
     reportPacker.m_client.reportEndOfSessionWithError(msg,m_errorCode,chrono); 
     reportPacker.m_lc.log(LOG_INFO,msg);
   }
+  if(reportPacker.m_watchdog) {
+    reportPacker.m_watchdog->addParameter(log::Param("status","failure"));
+  }
   reportPacker.m_continue=false;
 }
 //------------------------------------------------------------------------------

castor/tape/tapeserver/daemon/MigrationReportPacker.hpp

@@ -42,7 +42,7 @@ public:
    * @param tg The client who is asking for a migration of his files 
    * and to whom we have to report to the status of the operations.
    */
-  MigrationReportPacker(client::ClientInterface & tg,log::LogContext lc);
+  MigrationReportPacker(client::ClientInterface & tg, log::LogContext lc);
   
   ~MigrationReportPacker();
     

castor/tape/tapeserver/daemon/RecallReportPacker.cpp

@@ -22,6 +22,7 @@
  *****************************************************************************/
 
 #include "castor/tape/tapeserver/daemon/RecallReportPacker.hpp"
+#include "castor/tape/tapeserver/daemon/TaskWatchDog.hpp"
 #include "castor/tape/tapegateway/FileRecalledNotificationStruct.hpp"
 #include "castor/tape/tapegateway/FileRecalledNotificationStruct.hpp"
 #include "castor/log/Logger.hpp"
@@ -44,8 +45,8 @@ namespace daemon {
 //------------------------------------------------------------------------------
 //Constructor
 //------------------------------------------------------------------------------
-RecallReportPacker::RecallReportPacker(client::ClientInterface & tg,
-        unsigned int reportFilePeriod,log::LogContext lc):
+RecallReportPacker::RecallReportPacker(client::ClientInterface & tg, 
+    unsigned int reportFilePeriod,log::LogContext lc):
 ReportPackerInterface<detail::Recall>(tg,lc),
         m_workerThread(*this),m_reportFilePeriod(reportFilePeriod),m_errorHappened(false){
 
@@ -152,6 +153,9 @@ void RecallReportPacker::ReportEndofSession::execute(RecallReportPacker& parent)
     if(!parent.m_errorHappened){
       parent.m_client.reportEndOfSession(chrono);
       parent.logRequestReport(chrono,"Nominal RecallReportPacker::EndofSession has been reported",LOG_INFO);
+      if (parent.m_watchdog) {
+        parent.m_watchdog->addParameter(log::Param("status","failure"));
+      }
     }
     else {
       const std::string& msg ="RecallReportPacker::EndofSession has been reported  but an error happened somewhere in the process";

castor/tape/tapeserver/daemon/RecallReportPacker.hpp

@@ -42,8 +42,8 @@ public:
    * @param reportFilePeriod how often  do we report to the client
    * @param lc log context, copied du to threads
    */
-  RecallReportPacker(client::ClientInterface & tg,unsigned int reportFilePeriod,
-          log::LogContext lc);
+  RecallReportPacker(client::ClientInterface & tg, unsigned int reportFilePeriod, 
+    log::LogContext lc);
   
   virtual ~RecallReportPacker();
   

castor/tape/tapeserver/daemon/ReportPackerInterface.hpp

@@ -34,13 +34,14 @@
 #include "castor/tape/tapegateway/FilesToRecallList.hpp"
 #include "castor/tape/tapegateway/FilesToMigrateList.hpp"
 
+
 #include <memory>
 
 namespace castor {
 namespace tape {
 namespace tapeserver {
 namespace daemon {
-    
+  
 namespace detail{
   //nameholder
   struct Recall{};
@@ -74,7 +75,10 @@ namespace detail{
     ReportByFile
   };
 }
- 
+
+// Forward declaration to avoid circular inclusions.
+class TaskWatchDog;
+
 /**
  * Utility class that should be inherited privately/protectedly 
  * the type PlaceHolder is either detail::Recall or detail::Migration
@@ -86,12 +90,17 @@ template <class PlaceHolder> class ReportPackerInterface{
   typedef typename detail::HelperTrait<PlaceHolder>::FileStruct FileStruct;
   typedef typename detail::HelperTrait<PlaceHolder>::FileSuccessStruct FileSuccessStruct;
   typedef typename detail::HelperTrait<PlaceHolder>::FileErrorStruct FileErrorStruct;
+  
+  // Pass a reference to the watchdog for initial process reporting.
+  void setWatchdog(TaskWatchDog & wd) {
+    m_watchdog = &wd;
+  }
 
   protected:
     virtual ~ReportPackerInterface() {}
-    ReportPackerInterface(client::ClientInterface & tg,log::LogContext lc):
+    ReportPackerInterface(client::ClientInterface & tg, log::LogContext lc):
     m_client(tg),m_lc(lc),m_listReports(new FileReportList),
-    m_reportBatching(detail::ReportInBulk) {}
+    m_reportBatching(detail::ReportInBulk),m_watchdog(NULL) {}
   
   /**
    * Log a set of files independently of the success/failure 
@@ -172,6 +181,13 @@ template <class PlaceHolder> class ReportPackerInterface{
    * This is used for recalls driven by read_tp.
    */
   virtual void disableBulk() { m_reportBatching = detail::ReportByFile; }
+  
+  /**
+   * Pointer to the watchdog, so we can communicate communication errors
+   * and end of session results to the initial process
+   */
+  TaskWatchDog * m_watchdog;
+  
 };
 
 }}}}

castor/tape/tapeserver/daemon/TaskWatchDogTest.cpp

@@ -36,7 +36,7 @@ namespace unitTests {
   public tapeserver::daemon::ReportPackerInterface<tapeserver::daemon::detail::Recall> {
     // TODO MOCK_METHOD1(reportStuckOn, void(FileStruct& file));
     
-    MockReportPacker(tapeserver::client::ClientInterface & tg,castor::log::LogContext lc):
+    MockReportPacker(tapeserver::client::ClientInterface & tg, castor::log::LogContext lc):
     tapeserver::daemon::ReportPackerInterface<castor::tape::tapeserver::daemon::detail::Recall>(tg,lc){
     
     }