src/syncer.cpp

   1 #include <sys/select.h>
   2
   3 #include <curl/curl.h>
   4 #include <curl/types.h>
   5 #include <curl/easy.h>
   6
   7 #include <cassert>
   8 #include <algorithm>
   9 #include <exception>
  10
  11 #include <boost/bind.hpp>
  12
  13 #include "snmp_pp/snmp_pp.h"
  14
  15 #include "syncer.h"
  16 #include "settings.h"
  17 #include "switch.h"
  18 #include "subscriber.h"
  19 #include "datatypes.h"
  20 #include "logger.h"
  21
  22 using GTS::Syncer;
  23 using GTS::Timer;
  24 using GTS::Switch;
  25 using GTS::Subscriber;
  26 using GTS::Lines;
  27
  28 Timer::Timer(boost::function<void ()> callback, time_t interval)
  29     : _interval(interval),
  30       _lastFire(0),
  31       _callback(callback)
  32 {
  33 }
  34
  35 Timer::~Timer()
  36 {
  37 }
  38
  39 int Timer::getTimeout() const
  40 {
  41     double delta = difftime(time(NULL), _lastFire);
  42     return _interval - delta;
  43 }
  44
  45 void Timer::fire()
  46 {
  47     _callback();
  48     time(&_lastFire);
  49 }
  50
  51 Syncer::Syncer(SettingsParser & sp,
  52                Snmp & snmp)
  53     : _settingsParser(sp),
  54       _snmp(snmp)
  55 {
  56     // 1 db syncer
  57     _timers.push_back(Timer(boost::bind(&Syncer::syncInfo, this),
  58                             _settingsParser.settings().infoSyncInterval()));
  59 }
  60
  61 Syncer::~Syncer()
  62 {
  63 }
  64
  65 void Syncer::run(const bool & running, bool & reload)
  66 {
  67     logger << "Syncer::run()" << std::endl;
  68     while (running) {
  69         if (wait()) {
  70             logger << "Syncer::run() - wait stopped by signal" << std::endl;
  71             if (!running)
  72                 break;
  73             if (reload) {
  74                 logger << "Syncer::run() - reload" << std::endl;
  75                 try {
  76                     _settingsParser.reloadConfig();
  77                 }
  78                 catch (std::exception & ex) {
  79                     logger << "Syncer::run() - exception: " << ex.what() << std::endl;
  80                 }
  81                 reload = false;
  82             }
  83         }
  84     }
  85 }
  86
  87 bool Syncer::wait()
  88 {
  89     Timer & timer(getNextTimer());
  90
  91     if (timer.getTimeout() > 0) {
  92         fd_set rfds;
  93
  94         FD_ZERO(&rfds);
  95
  96         struct timeval tv;
  97         tv.tv_sec = timer.getTimeout();
  98         tv.tv_usec = 0;
  99
 100         int retval = select(1, &rfds, NULL, NULL, &tv);
 101
 102         if (retval == -1) {
 103             return true;
 104         }
 105     }
 106
 107     timer.fire();
 108
 109     return false;
 110 }
 111
 112 Timer & Syncer::getNextTimer()
 113 {
 114     assert(_timers.size() && "Timer list must not be empty!");
 115     int timeout = _timers.begin()->getTimeout();
 116     std::list<Timer>::iterator it(_timers.begin());
 117     std::list<Timer>::iterator pos(_timers.begin());
 118     ++it;
 119     while (it != _timers.end()) {
 120         if (it->getTimeout() < timeout) {
 121             pos = it;
 122             timeout = pos->getTimeout();
 123         }
 124         ++it;
 125     }
 126     return *pos;
 127 }
 128
 129 void Syncer::syncInfo()
 130 {
 131     std::map<std::string, Switch> switches;
 132     if (!getSwitchesState(switches)) {
 133         logger << "Syncer::syncInfo() - failed to get new switch states" << std::endl;
 134         return;
 135     }
 136     std::list<TimedSwitch>::iterator it(_switches.begin());
 137     while (it != _switches.end()) {
 138         _timers.erase(it->second);
 139         _switches.erase(it++);
 140     }
 141     std::map<std::string, Switch>::const_iterator sit;
 142     for (sit = switches.begin(); sit != switches.end(); ++sit) {
 143         // Insert switch with no timer
 144         _switches.push_back(std::make_pair(sit->second, _timers.end()));
 145         // Insert timer for this switch
 146         TimerIterator tit = _timers.insert(
 147                 _timers.end(),
 148                 Timer(boost::bind(&Switch::sync, &_switches.back().first),
 149                       _settingsParser.settings().switchSyncInterval()));
 150         // Set timer iterator for this switch
 151         _switches.back().second = tit;
 152     }
 153     logger << "Syncer::syncInfo() - data synchronization successfull, switches: " << _switches.size() << std::endl;
 154 }
 155
 156 size_t curlWriteFunction(void * ptr, size_t size, size_t nmemb, void * userdata)
 157 {
 158     char * data = static_cast<char *>(ptr);
 159     std::string * dest = static_cast<std::string *>(userdata);
 160     dest->append(data, size * nmemb);
 161     return size * nmemb;
 162 }
 163
 164 bool Syncer::getDBData(std::string & data) const
 165 {
 166     CURL * handle = curl_easy_init();
 167     if (handle) {
 168         char errorBuffer[CURL_ERROR_SIZE];
 169         curl_easy_setopt(handle, CURLOPT_SSL_VERIFYPEER, 0); // Accept self-signed certs
 170         curl_easy_setopt(handle, CURLOPT_LOW_SPEED_LIMIT, 1); // Less than 1 bps
 171         curl_easy_setopt(handle, CURLOPT_LOW_SPEED_TIME, 60); // During 60 secs
 172         curl_easy_setopt(handle, CURLOPT_URL, _settingsParser.settings().dataURL().c_str()); // Our URL
 173         curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, curlWriteFunction); // Our write callback
 174         curl_easy_setopt(handle, CURLOPT_WRITEDATA, &data); // Our callback data
 175         curl_easy_setopt(handle, CURLOPT_ERRORBUFFER, errorBuffer); // Buffer for an error messages
 176         CURLcode res = curl_easy_perform(handle);
 177         if (res) {
 178             logger << "Syncer::getDBData() - DB communication error: '" << errorBuffer << "'" << std::endl;
 179             curl_easy_cleanup(handle);
 180             return false;
 181         }
 182         curl_easy_cleanup(handle);
 183         return true;
 184     }
 185     logger << "Syncer::getDBData() - failed to init CURL library" << std::endl;
 186     return false;
 187 }
 188
 189 bool Syncer::getSwitchesState(std::map<std::string, Switch> & switches)
 190 {
 191     if (_settingsParser.settings().dataURL().empty()) {
 192         logger << "Switch::getSwitchesState() - data URL is empty" << std::endl;
 193         return false;
 194     }
 195     std::string data;
 196     if (!getDBData(data)) {
 197         logger << "Syncer::getSwitchesState() - failed to fetch data from the URL: '" << _settingsParser.settings().dataURL() << "'" << std::endl;
 198         return false;
 199     }
 200     Lines lines;
 201     if (!parseData(data, lines)) {
 202         logger << "Syncer::getSwitchesState() - failed to parse data:\n" << data << std::endl;
 203         return false;
 204     }
 205     Lines::const_iterator it;
 206     for (it = lines.begin(); it != lines.end(); ++it) {
 207         std::pair<std::map<std::string, Switch>::iterator, bool> res(
 208             switches.insert(
 209                 std::make_pair(
 210                     it->switchIP,
 211                     Switch(
 212                         _settingsParser.settings(),
 213                         _snmp,
 214                         it->switchIP,
 215                         it->readCommunity,
 216                         it->writeCommunity,
 217                         it->uplinkPort
 218                     )
 219                 )
 220             )
 221         );
 222         res.first->second.addSubscriber(
 223             Subscriber(
 224                 it->mac,
 225                 it->userPort,
 226                 it->upShape,
 227                 it->downShape,
 228                 it->upBurst,
 229                 it->downBurst
 230             )
 231         );
 232     }
 233     return true;
 234 }