#include <DataRetrieverMonitorCollection.h>

Inheritance diagram for smproxy::DataRetrieverMonitorCollection:

Classes
struct	DataRetrieverMQ
struct	EventMQ
struct	EventStats
class	EventTypeMqMap
struct	EventTypePerConnectionStats
struct	SummaryStats
Public Types
typedef std::map< std::string, EventStats >	ConnectionStats
enum	ConnectionStatus { CONNECTED, CONNECTION_FAILED, DISCONNECTED, UNKNOWN }
typedef std::vector < EventTypePerConnectionStats >	EventTypePerConnectionStatList
Public Member Functions
ConnectionID	addNewConnection (const stor::RegPtr)
bool	addRetrievedSample (const ConnectionID &, const unsigned int &size)
void	configureAlarms (AlarmParams const &)
	DataRetrieverMonitorCollection (const stor::utils::Duration_t &updateInterval, stor::AlarmHandlerPtr)
bool	getEventTypeStatsForConnection (const ConnectionID &, EventTypePerConnectionStats &)
void	getStatsByConnection (ConnectionStats &) const
void	getStatsByEventTypesPerConnection (EventTypePerConnectionStatList &) const
void	getSummaryStats (SummaryStats &) const
bool	receivedCorruptedEvent (const ConnectionID &)
bool	setConnectionStatus (const ConnectionID &, const ConnectionStatus &)
Private Types
typedef std::map< std::string, EventMQPtr >	ConnectionMqMap
typedef boost::shared_ptr < DataRetrieverMQ >	DataRetrieverMQPtr
typedef boost::shared_ptr < EventMQ >	EventMQPtr
typedef std::map< ConnectionID, DataRetrieverMQPtr >	RetrieverMqMap
Private Member Functions
void	checkForCorruptedEvents ()
	DataRetrieverMonitorCollection (DataRetrieverMonitorCollection const &)
virtual void	do_calculateStatistics ()
virtual void	do_reset ()
DataRetrieverMonitorCollection &	operator= (DataRetrieverMonitorCollection const &)
void	sendAlarms ()
Private Attributes
stor::AlarmHandlerPtr	alarmHandler_
AlarmParams	alarmParams_
ConnectionMqMap	connectionMqMap_
EventTypeMqMap	eventTypeMqMap_
ConnectionID	nextConnectionId_
RetrieverMqMap	retrieverMqMap_
boost::mutex	statsMutex_
EventMQ	totals_
const stor::utils::Duration_t	updateInterval_

Detailed Description

A collection of MonitoredQuantities related to data retrieval

Author:: mommsen

Revision:: 1.3

Date:: 2011/05/09 11:03:25

Definition at line 35 of file DataRetrieverMonitorCollection.h.

Member Typedef Documentation

typedef std::map<std::string, EventMQPtr> smproxy::DataRetrieverMonitorCollection::ConnectionMqMap [private]

Definition at line 167 of file DataRetrieverMonitorCollection.h.

typedef std::map<std::string, EventStats> smproxy::DataRetrieverMonitorCollection::ConnectionStats

Definition at line 58 of file DataRetrieverMonitorCollection.h.

typedef boost::shared_ptr<DataRetrieverMQ> smproxy::DataRetrieverMonitorCollection::DataRetrieverMQPtr [private]

Definition at line 163 of file DataRetrieverMonitorCollection.h.

typedef boost::shared_ptr<EventMQ> smproxy::DataRetrieverMonitorCollection::EventMQPtr [private]

Definition at line 138 of file DataRetrieverMonitorCollection.h.

typedef std::vector<EventTypePerConnectionStats> smproxy::DataRetrieverMonitorCollection::EventTypePerConnectionStatList

Definition at line 68 of file DataRetrieverMonitorCollection.h.

typedef std::map<ConnectionID, DataRetrieverMQPtr> smproxy::DataRetrieverMonitorCollection::RetrieverMqMap [private]

Definition at line 164 of file DataRetrieverMonitorCollection.h.

Member Enumeration Documentation

enum smproxy::DataRetrieverMonitorCollection::ConnectionStatus

Enumerator:

CONNECTED
CONNECTION_FAILED
DISCONNECTED
UNKNOWN

Definition at line 39 of file DataRetrieverMonitorCollection.h.

{ CONNECTED, CONNECTION_FAILED, DISCONNECTED, UNKNOWN };

Constructor & Destructor Documentation

smproxy::DataRetrieverMonitorCollection::DataRetrieverMonitorCollection	(	const stor::utils::Duration_t &	updateInterval,
		stor::AlarmHandlerPtr	alarmHandler
	)

Definition at line 17 of file DataRetrieverMonitorCollection.cc.

    :
  MonitorCollection(updateInterval),
  updateInterval_(updateInterval),
  alarmHandler_(alarmHandler),
  totals_(updateInterval),
  eventTypeMqMap_(updateInterval)
  {}

smproxy::DataRetrieverMonitorCollection::DataRetrieverMonitorCollection ( DataRetrieverMonitorCollection const & ) [private]

Member Function Documentation

ConnectionID smproxy::DataRetrieverMonitorCollection::addNewConnection ( const stor::RegPtr regPtr )

Add a new server connection. Returns an unique connection ID.

Definition at line 30 of file DataRetrieverMonitorCollection.cc.

  {
    boost::mutex::scoped_lock sl(statsMutex_);
    ++nextConnectionId_;
    
    DataRetrieverMQPtr dataRetrieverMQ( new DataRetrieverMQ(regPtr, updateInterval_) );
    retrieverMqMap_.insert(
      RetrieverMqMap::value_type(nextConnectionId_, dataRetrieverMQ)
    );
    
    eventTypeMqMap_.insert(regPtr);
    
    connectionMqMap_.insert(ConnectionMqMap::value_type(
        regPtr->sourceURL(),
        EventMQPtr(new EventMQ(updateInterval_))
      ));
    
    return nextConnectionId_;
  }

bool smproxy::DataRetrieverMonitorCollection::addRetrievedSample	(	const ConnectionID &	connectionId,
		const unsigned int &	size
	)

Add a retrieved sample in Bytes from the given connection. Returns false if the ConnectionID is unknown.

Definition at line 87 of file DataRetrieverMonitorCollection.cc.

References findQualityFiles::size, and reco_skim_cfg_mod::sourceURL.

  {
    boost::mutex::scoped_lock sl(statsMutex_);
    
    RetrieverMqMap::const_iterator retrieverPos = retrieverMqMap_.find(connectionId);
    if ( retrieverPos == retrieverMqMap_.end() ) return false;
    
    const double sizeKB = static_cast<double>(size) / 1024;
    retrieverPos->second->eventMQ_->size_.addSample(sizeKB);
    
    const stor::RegPtr regPtr = retrieverPos->second->regPtr_;
    
    eventTypeMqMap_.addSample(regPtr, sizeKB);
    
    const std::string sourceURL = regPtr->sourceURL();
    ConnectionMqMap::const_iterator connectionPos = connectionMqMap_.find(sourceURL);
    connectionPos->second->size_.addSample(sizeKB);
    
    totals_.size_.addSample(sizeKB);
    
    return true;
  }

void smproxy::DataRetrieverMonitorCollection::checkForCorruptedEvents ( ) [private]

Definition at line 212 of file DataRetrieverMonitorCollection.cc.

References alarmHandler_, alarmParams_, smproxy::AlarmParams::corruptedEventRate_, smproxy::DataRetrieverMonitorCollection::EventStats::corruptedEventsStats, stor::AlarmHandler::ERROR, smproxy::DataRetrieverMonitorCollection::EventMQ::getStats(), stor::MonitoredQuantity::Stats::getValueRate(), runTheMatrix::msg, stor::MonitoredQuantity::RECENT, and totals_.

Referenced by sendAlarms().

  {
    const std::string alarmName = "CorruptedEvents";
    
    EventStats eventStats;
    totals_.getStats(eventStats);
    const double corruptedEventRate =
      eventStats.corruptedEventsStats.getValueRate(stor::MonitoredQuantity::RECENT);
    if ( corruptedEventRate > alarmParams_.corruptedEventRate_ )
    {
      std::ostringstream msg;
      msg << "Received " << corruptedEventRate << " Hz of corrupted events from StorageManagers.";
      XCEPT_DECLARE(exception::CorruptedEvents, ex, msg.str());
      alarmHandler_->raiseAlarm(alarmName, stor::AlarmHandler::ERROR, ex);
    }
    else if ( corruptedEventRate < (alarmParams_.corruptedEventRate_ * 0.9) )
      // avoid revoking the alarm if we're close to the limit
    {
      alarmHandler_->revokeAlarm(alarmName);
    }
  }

void smproxy::DataRetrieverMonitorCollection::configureAlarms ( AlarmParams const & alarmParams )

Configure the alarm settings

Definition at line 198 of file DataRetrieverMonitorCollection.cc.

References alarmParams_.

  {
    alarmParams_ = alarmParams;
  }

void smproxy::DataRetrieverMonitorCollection::do_calculateStatistics ( ) [private, virtual]

Implements stor::MonitorCollection.

Definition at line 235 of file DataRetrieverMonitorCollection.cc.

References smproxy::DataRetrieverMonitorCollection::EventTypeMqMap::calculateStatistics(), smproxy::DataRetrieverMonitorCollection::EventMQ::calculateStatistics(), connectionMqMap_, eventTypeMqMap_, retrieverMqMap_, sendAlarms(), statsMutex_, and totals_.

  {
    boost::mutex::scoped_lock sl(statsMutex_);
    
    totals_.calculateStatistics();
    
    for (RetrieverMqMap::const_iterator it = retrieverMqMap_.begin(),
           itEnd = retrieverMqMap_.end(); it != itEnd; ++it)
    {
      it->second->eventMQ_->calculateStatistics();
    }
    
    for (ConnectionMqMap::const_iterator it = connectionMqMap_.begin(),
           itEnd = connectionMqMap_.end(); it != itEnd; ++it)
    {
      it->second->calculateStatistics();
    }
    
    eventTypeMqMap_.calculateStatistics();

    sendAlarms();
  }

void smproxy::DataRetrieverMonitorCollection::do_reset ( ) [private, virtual]

Implements stor::MonitorCollection.

Definition at line 259 of file DataRetrieverMonitorCollection.cc.

References smproxy::DataRetrieverMonitorCollection::EventTypeMqMap::clear(), connectionMqMap_, eventTypeMqMap_, smproxy::DataRetrieverMonitorCollection::EventMQ::reset(), retrieverMqMap_, statsMutex_, and totals_.

  {
    boost::mutex::scoped_lock sl(statsMutex_);
    totals_.reset();
    retrieverMqMap_.clear();
    connectionMqMap_.clear();
    eventTypeMqMap_.clear();
  }

bool smproxy::DataRetrieverMonitorCollection::getEventTypeStatsForConnection	(	const ConnectionID &	connectionId,
		EventTypePerConnectionStats &	stats
	)

Put the event type statistics for the given consumer ID into the passed EventTypePerConnectionStats. Return false if the connection ID is not found.

Definition at line 68 of file DataRetrieverMonitorCollection.cc.

References smproxy::DataRetrieverMonitorCollection::EventTypePerConnectionStats::connectionStatus, smproxy::DataRetrieverMonitorCollection::EventTypePerConnectionStats::eventStats, pos, and smproxy::DataRetrieverMonitorCollection::EventTypePerConnectionStats::regPtr.

  {
    boost::mutex::scoped_lock sl(statsMutex_);
    RetrieverMqMap::const_iterator pos = retrieverMqMap_.find(connectionId);
    
    if ( pos == retrieverMqMap_.end() ) return false;
    
    stats.regPtr = pos->second->regPtr_;
    stats.connectionStatus = pos->second->connectionStatus_;
    pos->second->eventMQ_->getStats(stats.eventStats);
    
    return true;
  }

void smproxy::DataRetrieverMonitorCollection::getStatsByConnection ( ConnectionStats & cs ) const

Write the data retrieval statistics for each connection into the given struct.

Definition at line 161 of file DataRetrieverMonitorCollection.cc.

References connectionMqMap_, and statsMutex_.

  {
    boost::mutex::scoped_lock sl(statsMutex_);
    cs.clear();
    
    for (ConnectionMqMap::const_iterator it = connectionMqMap_.begin(),
           itEnd = connectionMqMap_.end(); it != itEnd; ++it)
    {
      EventStats stats;
      it->second->getStats(stats);
      cs.insert(ConnectionStats::value_type(it->first, stats));
    }
  }

void smproxy::DataRetrieverMonitorCollection::getStatsByEventTypesPerConnection ( EventTypePerConnectionStatList & etsl ) const

Write the data retrieval statistics for each event type request into the given struct.

Definition at line 177 of file DataRetrieverMonitorCollection.cc.

References smproxy::DataRetrieverMonitorCollection::EventTypePerConnectionStats::connectionStatus, smproxy::DataRetrieverMonitorCollection::EventTypePerConnectionStats::eventStats, smproxy::DataRetrieverMonitorCollection::EventTypePerConnectionStats::regPtr, and python::multivaluedict::sort().

  {
    boost::mutex::scoped_lock sl(statsMutex_);
    etsl.clear();
    
    for (RetrieverMqMap::const_iterator it = retrieverMqMap_.begin(),
           itEnd = retrieverMqMap_.end(); it != itEnd; ++it)
    {
      const DataRetrieverMQPtr mq = it->second;
      EventTypePerConnectionStats stats;
      stats.regPtr = mq->regPtr_;
      stats.connectionStatus = mq->connectionStatus_;
      mq->eventMQ_->getStats(stats.eventStats);
      etsl.push_back(stats);
    }
    std::sort(etsl.begin(), etsl.end());
  }

void smproxy::DataRetrieverMonitorCollection::getSummaryStats ( SummaryStats & stats ) const

Write the data retrieval summary statistics into the given struct.

Definition at line 140 of file DataRetrieverMonitorCollection.cc.

References smproxy::DataRetrieverMonitorCollection::SummaryStats::activeSMs, CONNECTED, eventTypeMqMap_, smproxy::DataRetrieverMonitorCollection::SummaryStats::eventTypeStats, smproxy::DataRetrieverMonitorCollection::EventTypeMqMap::getStats(), smproxy::DataRetrieverMonitorCollection::EventMQ::getStats(), smproxy::DataRetrieverMonitorCollection::SummaryStats::registeredSMs, retrieverMqMap_, statsMutex_, smproxy::DataRetrieverMonitorCollection::SummaryStats::totals, and totals_.

  {
    boost::mutex::scoped_lock sl(statsMutex_);
    
    stats.registeredSMs = 0;
    stats.activeSMs = 0;
    
    for (RetrieverMqMap::const_iterator it = retrieverMqMap_.begin(),
           itEnd = retrieverMqMap_.end(); it != itEnd; ++it)
    {
      ++stats.registeredSMs;
      if ( it->second->connectionStatus_ == CONNECTED )
        ++stats.activeSMs;
    }
    
    eventTypeMqMap_.getStats(stats.eventTypeStats);
    
    totals_.getStats(stats.totals);
  }

DataRetrieverMonitorCollection& smproxy::DataRetrieverMonitorCollection::operator= ( DataRetrieverMonitorCollection const & ) [private]

bool smproxy::DataRetrieverMonitorCollection::receivedCorruptedEvent ( const ConnectionID & connectionId )

Increment number of corrupted events received from the given connection. Returns false if the ConnectionID is unknown.

Definition at line 115 of file DataRetrieverMonitorCollection.cc.

References reco_skim_cfg_mod::sourceURL.

Referenced by smproxy::DataRetrieverMonitorCollection::EventTypeMqMap::receivedCorruptedEvent().

  {
    boost::mutex::scoped_lock sl(statsMutex_);
    
    RetrieverMqMap::const_iterator retrieverPos = retrieverMqMap_.find(connectionId);
    if ( retrieverPos == retrieverMqMap_.end() ) return false;
    
    retrieverPos->second->eventMQ_->corruptedEvents_.addSample(1);
    
    const stor::RegPtr regPtr = retrieverPos->second->regPtr_;
    
    eventTypeMqMap_.receivedCorruptedEvent(regPtr);
    
    const std::string sourceURL = regPtr->sourceURL();
    ConnectionMqMap::const_iterator connectionPos = connectionMqMap_.find(sourceURL);
    connectionPos->second->corruptedEvents_.addSample(1);
    
    totals_.corruptedEvents_.addSample(1);

    return true;
  }

void smproxy::DataRetrieverMonitorCollection::sendAlarms ( ) [private]

Definition at line 204 of file DataRetrieverMonitorCollection.cc.

References alarmParams_, checkForCorruptedEvents(), and smproxy::AlarmParams::sendAlarms_.

Referenced by do_calculateStatistics().

  {
    if ( ! alarmParams_.sendAlarms_ ) return;
    
    checkForCorruptedEvents();
  }

bool smproxy::DataRetrieverMonitorCollection::setConnectionStatus	(	const ConnectionID &	connectionId,
		const ConnectionStatus &	status
	)

Set status of given connection. Returns false if the ConnectionID is unknown.

Definition at line 54 of file DataRetrieverMonitorCollection.cc.

References pos, and ntuplemaker::status.

  {
    boost::mutex::scoped_lock sl(statsMutex_);
    RetrieverMqMap::const_iterator pos = retrieverMqMap_.find(connectionId);
    if ( pos == retrieverMqMap_.end() ) return false;
    pos->second->connectionStatus_ = status;
    return true;
  }