Implement HA functionality for Elasticsearch feature

diff --git a/lib/perfdata/elasticsearchwriter.cpp b/lib/perfdata/elasticsearchwriter.cpp
index 029857094cbabedb7adf88bff80e0bef2d42bef5..ec89bed01f242ab6f1fb4c44c5e7edfa46788f32 100644 (file)
--- a/lib/perfdata/elasticsearchwriter.cpp
+++ b/lib/perfdata/elasticsearchwriter.cpp
@@ -49,6 +49,15 @@ void ElasticsearchWriter::OnConfigLoaded()
        ObjectImpl<ElasticsearchWriter>::OnConfigLoaded();
 
        m_WorkQueue.SetName("ElasticsearchWriter, " + GetName());
+
+       if (!GetEnableHa()) {
+               Log(LogDebug, "ElasticsearchWriter")
+                       << "HA functionality disabled. Won't pause connection: " << GetName();
+
+               SetHAMode(HARunEverywhere);
+       } else {
+               SetHAMode(HARunOnce);
+       }
 }
 
 void ElasticsearchWriter::StatsFunc(const Dictionary::Ptr& status, const Array::Ptr& perfdata)
@@ -71,14 +80,14 @@ void ElasticsearchWriter::StatsFunc(const Dictionary::Ptr& status, const Array::
        status->Set("elasticsearchwriter", new Dictionary(std::move(nodes)));
 }
 
-void ElasticsearchWriter::Start(bool runtimeCreated)
+void ElasticsearchWriter::Resume()
 {
-       ObjectImpl<ElasticsearchWriter>::Start(runtimeCreated);
+       ObjectImpl<ElasticsearchWriter>::Resume();
 
        m_EventPrefix = "icinga2.event.";
 
        Log(LogInformation, "ElasticsearchWriter")
-               << "'" << GetName() << "' started.";
+               << "'" << GetName() << "' resumed.";
 
        m_WorkQueue.SetExceptionCallback(std::bind(&ElasticsearchWriter::ExceptionHandler, this, _1));
 
@@ -95,14 +104,14 @@ void ElasticsearchWriter::Start(bool runtimeCreated)
        Checkable::OnNotificationSentToAllUsers.connect(std::bind(&ElasticsearchWriter::NotificationSentToAllUsersHandler, this, _1, _2, _3, _4, _5, _6, _7));
 }
 
-void ElasticsearchWriter::Stop(bool runtimeRemoved)
+void ElasticsearchWriter::Pause()
 {
        Log(LogInformation, "ElasticsearchWriter")
-               << "'" << GetName() << "' stopped.";
+               << "'" << GetName() << "' paused.";
 
        m_WorkQueue.Join();
 
-       ObjectImpl<ElasticsearchWriter>::Stop(runtimeRemoved);
+       ObjectImpl<ElasticsearchWriter>::Pause();
 }
 
 void ElasticsearchWriter::AddCheckResult(const Dictionary::Ptr& fields, const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
@@ -176,6 +185,9 @@ void ElasticsearchWriter::AddCheckResult(const Dictionary::Ptr& fields, const Ch
 
 void ElasticsearchWriter::CheckResultHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr)
 {
+       if (IsPaused())
+               return;
+
        m_WorkQueue.Enqueue(std::bind(&ElasticsearchWriter::InternalCheckResultHandler, this, checkable, cr));
 }
 
@@ -230,6 +242,9 @@ void ElasticsearchWriter::InternalCheckResultHandler(const Checkable::Ptr& check
 
 void ElasticsearchWriter::StateChangeHandler(const Checkable::Ptr& checkable, const CheckResult::Ptr& cr, StateType type)
 {
+       if (IsPaused())
+               return;
+
        m_WorkQueue.Enqueue(std::bind(&ElasticsearchWriter::StateChangeHandlerInternal, this, checkable, cr, type));
 }
 
@@ -279,6 +294,9 @@ void ElasticsearchWriter::NotificationSentToAllUsersHandler(const Notification::
        const Checkable::Ptr& checkable, const std::set<User::Ptr>& users, NotificationType type,
        const CheckResult::Ptr& cr, const String& author, const String& text)
 {
+       if (IsPaused())
+               return;
+
        m_WorkQueue.Enqueue(std::bind(&ElasticsearchWriter::NotificationSentToAllUsersHandlerInternal, this,
                notification, checkable, users, type, cr, author, text));
 }

diff --git a/lib/perfdata/elasticsearchwriter.hpp b/lib/perfdata/elasticsearchwriter.hpp
index a22552ad697b00d8d817f852d44ce2a7d5f5724c..7220e8ea37770cdb5290f3bc4425059c8eecc54e 100644 (file)
--- a/lib/perfdata/elasticsearchwriter.hpp
+++ b/lib/perfdata/elasticsearchwriter.hpp
@@ -41,8 +41,8 @@ public:
 
 protected:
        void OnConfigLoaded() override;
-       void Start(bool runtimeCreated) override;
-       void Stop(bool runtimeRemoved) override;
+       void Resume() override;
+       void Pause() override;
 
 private:
        String m_EventPrefix;

diff --git a/lib/perfdata/elasticsearchwriter.ti b/lib/perfdata/elasticsearchwriter.ti
index 2bbdd9d6a0ca367d866a2ea38d892ff8ef1978b9..7cf604fb94a30455fab6260c85a47f549337834d 100644 (file)
--- a/lib/perfdata/elasticsearchwriter.ti
+++ b/lib/perfdata/elasticsearchwriter.ti
@@ -37,6 +37,9 @@ class ElasticsearchWriter : ConfigObject
        [config] int flush_threshold {
                default {{{ return 1024; }}}
        };
+       [config] bool enable_ha {
+               default {{{ return true; }}}
+       };
 };
 
 }