BackendOpenTSDB.h

/*
 * Copyright (c) 2015 Pavel Vainerman.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation, version 2.1.
 *
 * 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
 * Lesser General Lesser Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */
// -----------------------------------------------------------------------------
#ifndef _BackendOpenTSDB_H_
#define _BackendOpenTSDB_H_
// -----------------------------------------------------------------------------
#include <deque>
#include <memory>
#include <unordered_map>
#include <chrono>
#include "UObject_SK.h"
#include "SMInterface.h"
#include "SharedMemory.h"
#include "extensions/Extensions.h"
#include "UTCPStream.h"
#include "USingleProcess.h"
// --------------------------------------------------------------------------
namespace uniset
{
    // -----------------------------------------------------------------------------
    // -----------------------------------------------------------------------------
    class BackendOpenTSDB:
        private USingleProcess,
        public UObject_SK
    {
        public:
            BackendOpenTSDB( uniset::ObjectId objId, xmlNode* cnode, uniset::ObjectId shmID, const std::shared_ptr<SharedMemory>& ic = nullptr,
                             const std::string& prefix = "opentsdb" );
            virtual ~BackendOpenTSDB();
 
            static std::shared_ptr<BackendOpenTSDB> init_opendtsdb( int argc, const char* const* argv,
                    uniset::ObjectId shmID, const std::shared_ptr<SharedMemory>& ic = nullptr,
                    const std::string& prefix = "opentsdb" );
 
            static void help_print( int argc, const char* const* argv );
 
            inline std::shared_ptr<LogAgregator> getLogAggregator()
            {
                return loga;
            }
            inline std::shared_ptr<DebugStream> log()
            {
                return mylog;
            }
 
            enum Timers
            {
                tmFlushBuffer,
                tmReconnect,
                tmLastNumberOfTimer
            };
 
        protected:
            BackendOpenTSDB();
 
            // переопределяем callback, чтобы оптимизировать
            // обработку большого количества сообщений
            // и убрать не нужную в данном процессе обработку (включая sleep_msec)
            virtual void callback() noexcept override;
 
            virtual void askSensors( UniversalIO::UIOCommand cmd ) override;
            virtual void sensorInfo( const uniset::SensorMessage* sm ) override;
            virtual void timerInfo( const uniset::TimerMessage* tm ) override;
            virtual void sysCommand( const uniset::SystemMessage* sm ) override;
            virtual std::string getMonitInfo() const override;
 
            void init( xmlNode* cnode );
            bool flushBuffer();
            bool reconnect();
 
            std::shared_ptr<SMInterface> shm;
 
            struct ParamInfo
            {
                const std::string name;
                const std::string tags;
 
                ParamInfo( const std::string& _name, const std::string& _tags ):
                    name(_name), tags(_tags) {}
            };
 
            std::string tsdbMeasurement = { "sensors" }; // measurement name for InfluxDB
            std::string tsdbProject; // project tag value (e.g. "theatre")
            std::string tsdbTags; // additional tags: TAG=VAL TAG2=VAL2 ...
            std::unordered_map<uniset::ObjectId, ParamInfo> tsdbParams;
 
            timeout_t bufSyncTime = { 5000 };
            size_t bufSize = { 500 };
            size_t bufMaxSize = { 5000 }; // drop messages
            bool timerIsOn = { false };
            timeout_t reconnectTime = { 5000 };
            std::string lastError;
 
            // буфер mutex-ом можно не защищать
            // т.к. к нему идёт обращение только из основного потока обработки
            // (sensorInfo, timerInfo)
            std::deque<std::string> buf;
 
            // работа с OpenTSDB
            std::shared_ptr<UTCPStream> tcp;
            std::string host = { "localhost" };
            int port = { 4242 };
 
        private:
 
            std::string prefix;
    };
    // --------------------------------------------------------------------------
} // end of namespace uniset
// -----------------------------------------------------------------------------
#endif // _BackendOpenTSDB_H_
// -----------------------------------------------------------------------------