icinga2/components/delegation/delegationcomponent.cpp

257 lines
8.2 KiB
C++
Raw Normal View History

2012-06-14 13:21:40 +02:00
/******************************************************************************
* Icinga 2 *
* Copyright (C) 2012 Icinga Development Team (http://www.icinga.org/) *
* *
* 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., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA. *
******************************************************************************/
#include "i2-delegation.h"
2012-06-20 16:52:56 +02:00
#include <algorithm>
2012-06-14 13:21:40 +02:00
using namespace icinga;
string DelegationComponent::GetName(void) const
{
return "delegation";
}
void DelegationComponent::Start(void)
{
2012-06-15 19:32:41 +02:00
m_AllServices = boost::make_shared<ConfigObject::Set>(ConfigObject::GetAllObjects(), ConfigObject::MakeTypePredicate("service"));
2012-06-14 13:21:40 +02:00
m_AllServices->Start();
2012-06-15 19:32:41 +02:00
m_DelegationTimer = boost::make_shared<Timer>();
2012-06-14 16:31:38 +02:00
m_DelegationTimer->SetInterval(30);
2012-06-15 19:32:41 +02:00
m_DelegationTimer->OnTimerExpired.connect(boost::bind(&DelegationComponent::DelegationTimerHandler, this));
2012-06-14 16:31:38 +02:00
m_DelegationTimer->Start();
2012-06-21 00:10:10 +02:00
m_DelegationTimer->Reschedule(0);
2012-06-14 16:31:38 +02:00
2012-06-15 19:32:41 +02:00
m_DelegationEndpoint = boost::make_shared<VirtualEndpoint>();
2012-06-14 13:21:40 +02:00
m_DelegationEndpoint->RegisterPublication("checker::AssignService");
2012-06-21 12:51:50 +02:00
m_DelegationEndpoint->RegisterPublication("checker::ClearServices");
2012-06-14 13:21:40 +02:00
GetEndpointManager()->RegisterEndpoint(m_DelegationEndpoint);
2012-06-21 17:39:16 +02:00
GetEndpointManager()->OnNewEndpoint.connect(bind(&DelegationComponent::NewEndpointHandler, this, _2));
2012-06-14 13:21:40 +02:00
}
void DelegationComponent::Stop(void)
{
EndpointManager::Ptr mgr = GetEndpointManager();
if (mgr)
mgr->UnregisterEndpoint(m_DelegationEndpoint);
}
2012-06-21 00:10:10 +02:00
void DelegationComponent::AssignService(const Endpoint::Ptr& checker, const Service& service)
2012-06-14 13:21:40 +02:00
{
RequestMessage request;
request.SetMethod("checker::AssignService");
MessagePart params;
params.SetProperty("service", service.GetConfigObject()->GetProperties());
2012-06-14 13:21:40 +02:00
request.SetParams(params);
2012-06-20 10:46:18 +02:00
Application::Log(LogDebug, "delegation", "Trying to delegate service '" + service.GetName() + "'");
2012-06-14 16:09:04 +02:00
2012-06-22 23:19:10 +02:00
GetEndpointManager()->SendUnicastMessage(m_DelegationEndpoint, checker, request);
2012-06-14 13:21:40 +02:00
}
2012-06-21 00:10:10 +02:00
void DelegationComponent::ClearServices(const Endpoint::Ptr& checker)
2012-06-14 13:21:40 +02:00
{
2012-06-21 13:08:26 +02:00
RequestMessage request;
request.SetMethod("checker::ClearServices");
2012-06-14 13:21:40 +02:00
2012-06-21 13:08:26 +02:00
MessagePart params;
request.SetParams(params);
2012-06-14 13:21:40 +02:00
2012-06-21 13:08:26 +02:00
GetEndpointManager()->SendUnicastMessage(m_DelegationEndpoint, checker, request);
2012-06-14 13:21:40 +02:00
}
2012-06-20 16:52:56 +02:00
vector<Endpoint::Ptr> DelegationComponent::GetCheckerCandidates(const Service& service) const
{
vector<Endpoint::Ptr> candidates;
EndpointManager::Iterator it;
2012-06-21 00:10:10 +02:00
for (it = GetEndpointManager()->Begin(); it != GetEndpointManager()->End(); it++) {
Endpoint::Ptr endpoint = it->second;
/* ignore disconnected endpoints */
if (!endpoint->IsConnected())
continue;
/* ignore endpoints that aren't running the checker component */
2012-06-21 00:10:10 +02:00
if (!endpoint->HasSubscription("checker::AssignService"))
continue;
candidates.push_back(endpoint);
}
2012-06-20 16:52:56 +02:00
return candidates;
}
2012-06-21 17:39:16 +02:00
void DelegationComponent::NewEndpointHandler(const Endpoint::Ptr& endpoint)
{
endpoint->OnSessionEstablished.connect(bind(&DelegationComponent::SessionEstablishedHandler, this, _1));
}
void DelegationComponent::SessionEstablishedHandler(const Endpoint::Ptr& endpoint)
{
stringstream msgbuf;
msgbuf << "Clearing assigned services for endpoint '" << endpoint->GetIdentity() << "'";
Application::Log(LogInformation, "delegation", msgbuf.str());
/* locally clear checker for all services that previously belonged to this endpoint */
ConfigObject::Set::Iterator it;
for (it = m_AllServices->Begin(); it != m_AllServices->End(); it++) {
Service service = *it;
if (service.GetChecker() == endpoint->GetIdentity())
service.SetChecker("");
}
/* remotely clear services for this endpoint */
ClearServices(endpoint);
}
2012-06-15 19:32:41 +02:00
void DelegationComponent::DelegationTimerHandler(void)
2012-06-14 16:31:38 +02:00
{
2012-06-20 16:52:56 +02:00
map<Endpoint::Ptr, int> histogram;
EndpointManager::Iterator eit;
2012-06-21 00:10:10 +02:00
for (eit = GetEndpointManager()->Begin(); eit != GetEndpointManager()->End(); eit++)
2012-06-20 16:52:56 +02:00
histogram[eit->second] = 0;
vector<Service> services;
/* build "checker -> service count" histogram */
2012-06-14 16:31:38 +02:00
ConfigObject::Set::Iterator it;
for (it = m_AllServices->Begin(); it != m_AllServices->End(); it++) {
Service service = *it;
2012-06-14 16:31:38 +02:00
2012-06-20 16:52:56 +02:00
services.push_back(service);
string checker = service.GetChecker();
if (checker.empty())
continue;
Endpoint::Ptr endpoint = GetEndpointManager()->GetEndpointByIdentity(checker);
if (!endpoint)
continue;
histogram[endpoint]++;
}
std::random_shuffle(services.begin(), services.end());
2012-06-21 13:31:58 +02:00
bool need_clear = false;
2012-06-21 00:10:10 +02:00
int delegated = 0;
2012-06-20 16:52:56 +02:00
/* re-assign services */
vector<Service>::iterator sit;
2012-06-21 00:10:10 +02:00
for (sit = services.begin(); sit != services.end(); sit++) {
2012-06-20 16:52:56 +02:00
Service service = *sit;
string checker = service.GetChecker();
2012-06-20 16:52:56 +02:00
Endpoint::Ptr oldEndpoint;
if (!checker.empty())
oldEndpoint = GetEndpointManager()->GetEndpointByIdentity(checker);
vector<Endpoint::Ptr> candidates = GetCheckerCandidates(service);
2012-06-21 12:51:50 +02:00
int avg_services = 0, overflow_tolerance = 0;
2012-06-20 16:52:56 +02:00
vector<Endpoint::Ptr>::iterator cit;
2012-06-21 12:51:50 +02:00
if (candidates.size() > 0) {
std::random_shuffle(candidates.begin(), candidates.end());
stringstream msgbuf;
msgbuf << "Service: " << service.GetName() << ", candidates: " << candidates.size();
Application::Log(LogDebug, "delegation", msgbuf.str());
for (cit = candidates.begin(); cit != candidates.end(); cit++)
avg_services += histogram[*cit];
avg_services /= candidates.size();
overflow_tolerance = candidates.size() * 2;
}
2012-06-20 16:52:56 +02:00
/* don't re-assign service if the checker is still valid
* and doesn't have too many services */
2012-06-22 08:30:36 +02:00
if (oldEndpoint && oldEndpoint->IsConnected() &&
find(candidates.begin(), candidates.end(), oldEndpoint) != candidates.end() &&
2012-06-20 16:52:56 +02:00
histogram[oldEndpoint] <= avg_services + overflow_tolerance)
2012-06-14 16:31:38 +02:00
continue;
2012-06-20 16:52:56 +02:00
/* clear the service's current checker */
if (!checker.empty()) {
2012-06-21 13:31:58 +02:00
need_clear = true;
2012-06-20 16:52:56 +02:00
service.SetChecker("");
if (oldEndpoint)
histogram[oldEndpoint]--;
}
/* find a new checker for the service */
for (cit = candidates.begin(); cit != candidates.end(); cit++) {
Endpoint::Ptr newEndpoint = *cit;
/* does this checker already have too many services */
if (histogram[newEndpoint] > avg_services)
continue;
service.SetChecker(newEndpoint->GetIdentity());
histogram[newEndpoint]++;
delegated++;
2012-06-21 13:12:16 +02:00
break;
2012-06-20 16:52:56 +02:00
}
2012-06-21 12:51:50 +02:00
assert(candidates.size() == 0 || !service.GetChecker().empty());
2012-06-14 16:31:38 +02:00
}
2012-06-19 12:23:52 +02:00
map<Endpoint::Ptr, int>::iterator hit;
for (hit = histogram.begin(); hit != histogram.end(); hit++) {
stringstream msgbuf;
msgbuf << "histogram: " << hit->first->GetIdentity() << " - " << hit->second;
Application::Log(LogInformation, "delegation", msgbuf.str());
}
2012-06-21 00:10:10 +02:00
if (delegated > 0) {
2012-06-21 13:31:58 +02:00
if (need_clear) {
map<Endpoint::Ptr, int>::iterator hit;
for (hit = histogram.begin(); hit != histogram.end(); hit++) {
ClearServices(hit->first);
}
2012-06-21 00:10:10 +02:00
}
for (sit = services.begin(); sit != services.end(); sit++) {
string checker = sit->GetChecker();
Endpoint::Ptr endpoint = GetEndpointManager()->GetEndpointByIdentity(checker);
if (!endpoint)
continue;
AssignService(endpoint, *sit);
}
}
2012-06-20 10:46:18 +02:00
stringstream msgbuf;
2012-06-21 17:39:16 +02:00
msgbuf << "Updated delegations for " << delegated << " services";
2012-06-20 10:46:18 +02:00
Application::Log(LogInformation, "delegation", msgbuf.str());
2012-06-14 13:21:40 +02:00
}
EXPORT_COMPONENT(delegation, DelegationComponent);