Commit 0383db48 authored by Jan Koester's avatar Jan Koester
Browse files

bugfix

parent 4f85ac5c
Loading
Loading
Loading
Loading
+5 −3
Changes for include/snmpdd/api.h: 5 added lines, 3 removed lines.
Original line number Diff line number Diff line
@@ -63,11 +63,13 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *   GetStatus -> enabled, interval, last_report (unix time, 0: never),
 *       next_report (0: off), sent (notifications since start), status
 *       (ok|warning|critical|unknown: the worst item of the last report),
 *       engine_id, engine_boots (SNMPv3),
 *       collecting (yes while a report is collected), engine_id,
 *       engine_boots (SNMPv3),
 *       problem? (repeated: what couldn't be collected); then per
 *       target: target (host:port), sent, error?
 *   SendReport -> sent (notifications) -- collects and sends now; an
 *       error if no target could be reached */
 *   SendReport -> started (no: one is being collected already) --
 *       collects and sends a report now, in the background; GetStatus
 *       (last_report, the targets' errors) tells how it went */

#pragma once

+34 −11
Changes for src/collect.cpp: 34 added lines, 11 removed lines.
Original line number Diff line number Diff line
@@ -429,6 +429,17 @@ std::vector<Mount> mountedFilesystems() {
    return out;
}

// decimal digits of v at buf[n] (async-signal-safe, for the child below)
void putNumber(char *buf, size_t &n, unsigned long long v) {
    char tmp[24];
    size_t k = 0;
    do {
        tmp[k++] = (char)('0' + v % 10);
        v /= 10;
    } while (v);
    while (k) buf[n++] = tmp[--k];
}

struct Usage {
    bool ok = false;
    int error = 0;  // statvfs() failed with it (it answered, though)
@@ -440,6 +451,8 @@ struct Usage {
std::vector<Usage> usages(const std::vector<std::string> &paths, int timeout) {
    std::vector<Usage> out(paths.size());
    if (paths.empty()) return out;
    std::vector<const char *> cpaths;
    for (auto &path : paths) cpaths.push_back(path.c_str());
    int p[2];
    if (pipe2(p, O_CLOEXEC) != 0) return out;
    pid_t pid = fork();
@@ -449,21 +462,31 @@ std::vector<Usage> usages(const std::vector<std::string> &paths, int timeout) {
        return out;
    }
    if (pid == 0) {
        // forked from a process with threads: only async-signal-safe calls
        // here (no stdio, no allocation -- c_str() of the paths was taken
        // before, see cpaths)
        close(p[0]);
        for (size_t i = 0; i < paths.size(); ++i) {
        for (size_t i = 0; i < cpaths.size(); ++i) {
            struct statvfs st;
            char line[256];
            if (statvfs(paths[i].c_str(), &st) != 0) {
                int n = snprintf(line, sizeof(line), "%zu -%d\n", i, errno);
                if (write(p[1], line, (size_t)n) != n) _exit(1);
                continue;
            }
            char line[160];
            size_t n = 0;
            putNumber(line, n, i);
            if (statvfs(cpaths[i], &st) != 0) {
                line[n++] = ' ';
                line[n++] = '-';
                putNumber(line, n, (unsigned long long)errno);
            } else {
                unsigned long long f = st.f_frsize ? st.f_frsize : st.f_bsize;
            int n = snprintf(line, sizeof(line), "%zu %llu %llu %llu %llu %llu\n", i,
                             (unsigned long long)st.f_blocks * f, (unsigned long long)(st.f_blocks - st.f_bfree) * f,
                for (unsigned long long v : {(unsigned long long)st.f_blocks * f,
                                             (unsigned long long)(st.f_blocks - st.f_bfree) * f,
                                             (unsigned long long)st.f_bavail * f, (unsigned long long)st.f_files,
                             (unsigned long long)st.f_ffree);
            if (write(p[1], line, (size_t)n) != n) _exit(1);
                                             (unsigned long long)st.f_ffree}) {
                    line[n++] = ' ';
                    putNumber(line, n, v);
                }
            }
            line[n++] = '\n';
            if (write(p[1], line, n) != (ssize_t)n) _exit(1);
        }
        _exit(0);
    }
+79 −22
Changes for src/main.cpp: 79 added lines, 22 removed lines.
Original line number Diff line number Diff line
@@ -25,11 +25,13 @@ ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*******************************************************************************/

/* snmpdd: see CMakeLists.txt and include/snmpdd/api.h. One thread,
 * one poll() loop over the sysbus service connection; a second sysbus
 * connection is subscribed to the signals (storaged's SmartWarning,
 * syslite's UnitStateChanged). Reports are collected and sent in this
 * loop. SIGHUP re-reads snmpdd.yaml and the settings, SIGTERM stops. */
/* snmpdd: see CMakeLists.txt and include/snmpdd/api.h. One poll() loop
 * over the sysbus service connection; a second sysbus connection is
 * subscribed to the signals (storaged's SmartWarning, syslite's
 * UnitStateChanged). A report is collected in a thread of its own -- it
 * asks storaged about every disk and may take a while, and sysbus calls
 * (webctl) are answered meanwhile -- and sent from the loop once it is
 * done. SIGHUP re-reads snmpdd.yaml and the settings, SIGTERM stops. */

#include <algorithm>
#include <cerrno>
@@ -38,7 +40,9 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include <ctime>
#include <iostream>
#include <map>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

#include <fcntl.h>
@@ -93,7 +97,9 @@ private:
    sysbus::Message sendReport(const sysbus::Message &call);

    void apply(const Settings &s);
    unsigned report();
    void startReport();
    void finishReport();
    unsigned report(const Snapshot &snap);
    void events();
    void onBusSignal(const sysbus::Message &m);
    bool wantService(const std::string &name) const;
@@ -118,8 +124,63 @@ private:
    // disk -> result of its last self-test (for SelfTestFailed)
    std::map<std::string, std::string> _selfTests;
    bool _stop = false;

    // the collecting thread: done writes a byte to _donePipe
    int _donePipe[2] = {-1, -1};
    std::thread _worker;
    bool _collecting = false;
    std::mutex _doneMutex;
    Snapshot _done;

public:
    bool init();
    ~Daemon() {
        // a collection that still runs (storaged doesn't answer) isn't waited for
        if (_worker.joinable()) _worker.detach();
    }
};

bool Daemon::init() {
    if (pipe2(_donePipe, O_CLOEXEC | O_NONBLOCK) != 0) {
        std::cerr << "snmpdd: pipe: " << strerror(errno) << std::endl;
        return false;
    }
    return load();
}

void Daemon::startReport() {
    if (_collecting) return;
    _collecting = true;
    if (_settings.enabled) _nextReport = time(nullptr) + _settings.interval;
    Settings s = _settings;
    Config c = _config;
    int fd = _donePipe[1];
    _worker = std::thread([this, s, c, fd] {
        Snapshot snap = collect(s, c);
        {
            std::lock_guard<std::mutex> lock(_doneMutex);
            _done = std::move(snap);
        }
        char x = 1;
        (void)!write(fd, &x, 1);
    });
}

void Daemon::finishReport() {
    char buf[16];
    while (read(_donePipe[0], buf, sizeof(buf)) > 0) {
    }
    if (!_collecting) return;
    _worker.join();
    _collecting = false;
    Snapshot snap;
    {
        std::lock_guard<std::mutex> lock(_doneMutex);
        snap = std::move(_done);
    }
    report(snap);
}

bool Daemon::load() {
    Config c;
    try {
@@ -173,8 +234,7 @@ bool Daemon::wantService(const std::string &name) const {
    return false;
}

unsigned Daemon::report() {
    Snapshot snap = collect(_settings, _config);
unsigned Daemon::report(const Snapshot &snap) {
    unsigned n = _sender.report(snap, hostName());
    _lastReport = snap.at;
    _lastStatus = snap.worst();
@@ -362,6 +422,7 @@ sysbus::Message Daemon::getStatus(const sysbus::Message &call) {
    r.add("next_report", (long long)_nextReport);
    r.add("sent", (long long)_sender.sent());
    r.add("status", _lastReport ? statusName(_lastStatus) : "unknown");
    r.add("collecting", yesNo(_collecting));
    r.add("engine_id", usm::toHex(_sender.engineId()));
    r.add("engine_boots", (long long)_engineBoots);
    for (auto &p : _lastProblems) r.add("problem", p);
@@ -376,15 +437,11 @@ sysbus::Message Daemon::getStatus(const sysbus::Message &call) {
sysbus::Message Daemon::sendReport(const sysbus::Message &call) {
    if (_settings.targets.empty())
        return sysbus::Message::error(call, sysbus::error::InvalidArgs, tr("no targets", "keine Ziele"));
    unsigned n = report();
    if (_settings.enabled) _nextReport = time(nullptr) + _settings.interval;
    if (_sender.allFailed()) {
        std::string why;
        for (auto &t : _sender.targets()) why += (why.empty() ? "" : "; ") + t.target + ": " + t.error;
        return sysbus::Message::error(call, sysbus::error::Failed, tr("no target reached: %1", "kein Ziel erreicht: %1", {why}));
    }
    // collected in the background; GetStatus tells how it went
    bool already = _collecting;
    startReport();
    sysbus::Message r = sysbus::Message::reply(call);
    r.add("sent", (long long)n);
    r.add("started", yesNo(!already));
    return r;
}

@@ -417,11 +474,12 @@ void Daemon::run() {

        std::vector<pollfd> fds;
        fds.push_back({g_signalPipe[0], POLLIN, 0});
        fds.push_back({_donePipe[0], POLLIN, 0});
        if (_bus.running()) fds.push_back({_bus.fd(), POLLIN, 0});
        // a second at most: the signal connection has no fd of its own to
        // wait on (sysbus::Client), it is looked at every round
        int timeout = 1000;
        if (_nextReport) timeout = (int)std::min<long long>(timeout, std::max<long long>(0, (_nextReport - time(nullptr)) * 1000ll));
        if (_nextReport && !_collecting) timeout = (int)std::min<long long>(timeout, std::max<long long>(0, (_nextReport - time(nullptr)) * 1000ll));
        int r = poll(fds.data(), fds.size(), timeout);
        if (r < 0 && errno != EINTR) {
            std::cerr << "snmpdd: poll: " << strerror(errno) << std::endl;
@@ -439,6 +497,8 @@ void Daemon::run() {
                        _stop = true;
                    }
                }
            } else if (p.fd == _donePipe[0]) {
                finishReport();
            } else if (_bus.running() && p.fd == _bus.fd()) {
                if (!_bus.process()) {
                    std::cerr << "snmpdd: sysbus: " << _bus.lastError() << std::endl;
@@ -448,10 +508,7 @@ void Daemon::run() {
        }
        if (_stop) break;
        events();
        if (_nextReport && time(nullptr) >= _nextReport) {
            report();
            _nextReport = time(nullptr) + _settings.interval;
        }
        if (_nextReport && !_collecting && time(nullptr) >= _nextReport) startReport();
        reapChildren();
    }
}
@@ -492,7 +549,7 @@ int main(int argc, char **argv) {
    ::signal(SIGPIPE, SIG_IGN);

    Daemon d(config);
    if (!d.load()) return 1;
    if (!d.init()) return 1;
    d.run();
    return 0;
}