Implemented proper delegation support.

diff --git a/components/delegation/delegationcomponent.cpp b/components/delegation/delegationcomponent.cpp
index ef228e1a9c0d2ad134cebb4821bd52a3f343429d..2bd2ecdec0f08da58dd4f117cc6190fb22c147a4 100644 (file)
--- a/components/delegation/delegationcomponent.cpp
+++ b/components/delegation/delegationcomponent.cpp
@@ -18,6 +18,7 @@
  ******************************************************************************/
 
 #include "i2-delegation.h"
+#include <algorithm>
 
 using namespace icinga;
 
@@ -97,23 +98,108 @@ void DelegationComponent::RevokeServiceResponseHandler(Service& service, const E
 {
 }
 
+vector<Endpoint::Ptr> DelegationComponent::GetCheckerCandidates(const Service& service) const
+{
+       vector<Endpoint::Ptr> candidates;
+
+       EndpointManager::Iterator it;
+       for (it = GetEndpointManager()->Begin(); it != GetEndpointManager()->End(); it++)
+               candidates.push_back(it->second);
+
+       return candidates;
+}
+
 void DelegationComponent::DelegationTimerHandler(void)
 {
+       map<Endpoint::Ptr, int> histogram;
+
+       EndpointManager::Iterator eit;
+       for (eit = GetEndpointManager()->Begin(); eit != GetEndpointManager()->End(); eit++) {
+               histogram[eit->second] = 0;
+       }
+
+       /* nothing to do if we have no checkers */
+       if (histogram.size() == 0)
+               return;
+
+       vector<Service> services;
+
+       /* build "checker -> service count" histogram */
        ConfigObject::Set::Iterator it;
-       long delegated = 0;
        for (it = m_AllServices->Begin(); it != m_AllServices->End(); it++) {
                Service service = *it;
 
+               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());
+
+       long delegated = 0;
+
+       /* re-assign services */
+       vector<Service>::iterator sit;
+       for (sit = services.begin(); sit != services.end(); it++) {
+               Service service = *sit;
+
                string checker = service.GetChecker();
-               if (!checker.empty() && GetEndpointManager()->GetEndpointByIdentity(checker))
+
+               Endpoint::Ptr oldEndpoint;
+               if (!checker.empty())
+                       oldEndpoint = GetEndpointManager()->GetEndpointByIdentity(checker);
+
+               vector<Endpoint::Ptr> candidates = GetCheckerCandidates(service);
+
+               long avg_services = 0;
+               vector<Endpoint::Ptr>::iterator cit;
+               for (cit = candidates.begin(); cit != candidates.end(); cit++)
+                       avg_services += histogram[*cit];
+
+               avg_services /= candidates.size();
+               long overflow_tolerance = candidates.size() * 2;
+
+               /* don't re-assign service if the checker is still valid
+                * and doesn't have too many services */
+               if (oldEndpoint && find(candidates.begin(), candidates.end(), oldEndpoint) != candidates.end() &&
+                   histogram[oldEndpoint] <= avg_services + overflow_tolerance)
                        continue;
 
-               AssignService(service);
-               delegated++;
+               /* clear the service's current checker */
+               if (!checker.empty()) {
+                       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++;
+               }
+
+               assert(!service.GetChecker().empty());
        }
 
        stringstream msgbuf;
-       msgbuf << "Delegated " << delegated << " services";
+       msgbuf << "Re-delegated " << delegated << " services";
        Application::Log(LogInformation, "delegation", msgbuf.str());
 }
 

diff --git a/components/delegation/delegationcomponent.h b/components/delegation/delegationcomponent.h
index d2559538bbfe47d19cbfe588370e438a1cfa7437..c4d85e6c5e208664d7aa134669a36988d6fb27b6 100644 (file)
--- a/components/delegation/delegationcomponent.h
+++ b/components/delegation/delegationcomponent.h
@@ -46,6 +46,8 @@ private:
 
        void DelegationTimerHandler(void);
 
+       vector<Endpoint::Ptr> GetCheckerCandidates(const Service& service) const;
+
        void AssignService(const Service& service);
        void RevokeService(const Service& service);
 };

diff --git a/icinga/endpointmanager.cpp b/icinga/endpointmanager.cpp
index b8821863bacb0a0d33c6fcbe570a1eb8394200ef..43354ccfc0b83994f1b94f5dda16e18b9b6e73ed 100644 (file)
--- a/icinga/endpointmanager.cpp
+++ b/icinga/endpointmanager.cpp
@@ -377,3 +377,14 @@ void EndpointManager::ProcessResponseMessage(const Endpoint::Ptr& sender, const
        m_Requests.erase(it);
        RescheduleRequestTimer();
 }
+
+EndpointManager::Iterator EndpointManager::Begin(void)
+{
+       return m_Endpoints.begin();
+}
+
+EndpointManager::Iterator EndpointManager::End(void)
+{
+       return m_Endpoints.end();
+}
+

diff --git a/icinga/endpointmanager.h b/icinga/endpointmanager.h
index 016c8ca83bbeba062a3cb01ddcd0b1b29663c22b..458e46858c872e3f6dfecdbadcf0cfed833edc61 100644 (file)
--- a/icinga/endpointmanager.h
+++ b/icinga/endpointmanager.h
@@ -34,6 +34,8 @@ public:
        typedef shared_ptr<EndpointManager> Ptr;
        typedef weak_ptr<EndpointManager> WeakPtr;
 
+       typedef map<string, Endpoint::Ptr>::iterator Iterator;
+
        EndpointManager(void)
                : m_NextMessageID(0)
        { }
@@ -60,6 +62,8 @@ public:
        void ProcessResponseMessage(const Endpoint::Ptr& sender, const ResponseMessage& message);
 
        void ForEachEndpoint(function<void (const EndpointManager::Ptr&, const Endpoint::Ptr&)> callback);
+       Iterator Begin(void);
+       Iterator End(void);
 
        Endpoint::Ptr GetEndpointByIdentity(string identity) const;