loggingserver.cpp

Go to the documentation of this file.

#include <QAtomicInt>
#include <QMutex>
#include <QMutexLocker>
#include <QWaitCondition>
#include <QList>
#include <QQueue>
#include <QHash>
#include <QCoreApplication>
#include <QFileInfo>
#include <QStringList>
#include <QMap>
#include <QRegExp>
#include <QSocketNotifier>
#include <iostream>

using namespace std;

#include "mythlogging.h"
#include "logging.h"
#include "loggingserver.h"
#include "mythdb.h"
#include "mythcorecontext.h"
#include "mythsignalingtimer.h"
#include "dbutil.h"
#include "exitcodes.h"
#include "compat.h"

#include <cstdlib>
#ifndef _WIN32
#include <mythsyslog.h>
#if CONFIG_SYSTEMD_JOURNAL
#define SD_JOURNAL_SUPPRESS_LOCATION 1
#include <systemd/sd-journal.h>
#endif
#endif
#include <cstdarg>
#include <cstring>
#include <sys/stat.h>
#include <sys/types.h>
#include <fcntl.h>
#include <cstdio>
#include <unistd.h>
#if HAVE_GETTIMEOFDAY
#include <sys/time.h>
#endif
#include <csignal>

// Various ways to get to thread's tid
#if defined(linux)
#include <sys/syscall.h>
#elif defined(__FreeBSD__)
extern "C" {
#include <sys/ucontext.h>
#include <sys/thr.h>
}
#elif CONFIG_DARWIN
#include <mach/mach.h>
#endif

static QMutex                      loggerMapMutex;
static QMap<QString, LoggerBase *> loggerMap;

LogForwardThread                   *logForwardThread = nullptr;

using LoggerList = QList<LoggerBase *>;

struct LoggerListItem {
    LoggerList *m_itemList;
    qlonglong   m_itemEpoch;
};
using  ClientMap = QMap<QString, LoggerListItem *>;

using  ClientList = QList<QString>;
using  RevClientMap = QMap<LoggerBase *, ClientList *>;

static QMutex                       logClientMapMutex;
static ClientMap                    logClientMap;
static QAtomicInt                   logClientCount;

static QMutex                       logRevClientMapMutex;
static RevClientMap                 logRevClientMap;

static QMutex                       logMsgListMutex;
static LogMessageList               logMsgList;
static QWaitCondition               logMsgListNotEmpty;

#define TIMESTAMP_MAX 30
#define MAX_STRING_LENGTH (LOGLINE_MAX+120)

LoggerBase::LoggerBase(const char *string)
{
    QMutexLocker locker(&loggerMapMutex);
    if (string)
    {
        m_handle = strdup(string);
        loggerMap.insert(QString(m_handle), this);
    }
    else
    {
        m_handle = nullptr;
        loggerMap.insert(QString(""), this);
    }
}


LoggerBase::~LoggerBase()
{
    QMutexLocker locker(&loggerMapMutex);
    loggerMap.remove(QString(m_handle));

    if (m_handle)
        free(m_handle);
}


FileLogger::FileLogger(const char *filename) :
        LoggerBase(filename)
{
    m_fd = open(filename, O_WRONLY|O_CREAT|O_APPEND, 0664);
    m_opened = (m_fd != -1);
    LOG(VB_GENERAL, LOG_INFO, QString("Added logging to %1")
             .arg(filename));
}


FileLogger::~FileLogger()
{
    if( m_opened )
    {
        LOG(VB_GENERAL, LOG_INFO, QString("Removed logging to %1")
            .arg(m_handle));
        close(m_fd);
        m_fd = -1;
        m_opened = false;
    }
}

FileLogger *FileLogger::create(const QString& filename, QMutex *mutex)
{
    QByteArray ba = filename.toLocal8Bit();
    const char *file = ba.constData();
    FileLogger *logger =
        dynamic_cast<FileLogger *>(loggerMap.value(filename, nullptr));

    if (logger)
        return logger;

    // Need to add a new FileLogger
    mutex->unlock();
    // inserts into loggerMap
    logger = new FileLogger(file);
    mutex->lock();

    auto *clients = new ClientList;
    logRevClientMap.insert(logger, clients);
    return logger;
}

void FileLogger::reopen(void)
{
    close(m_fd);

    m_fd = open(m_handle, O_WRONLY|O_CREAT|O_APPEND, 0664);
    m_opened = (m_fd != -1);
    LOG(VB_GENERAL, LOG_INFO, QString("Rolled logging on %1") .arg(m_handle));
}

bool FileLogger::logmsg(LoggingItem *item)
{
    char                line[MAX_STRING_LENGTH];
    char                usPart[9];
    char                timestamp[TIMESTAMP_MAX];

    if (!m_opened)
        return false;

    time_t epoch = item->epoch();
    struct tm tm {};
    localtime_r(&epoch, &tm);

    strftime(timestamp, TIMESTAMP_MAX-8, "%Y-%m-%d %H:%M:%S",
             (const struct tm *)&tm);
    snprintf( usPart, 9, ".%06d", (int)(item->usec()) );
    strcat( timestamp, usPart );

    char shortname = '-';
    {
        QMutexLocker locker(&loglevelMapMutex);
        LoglevelMap::iterator it = loglevelMap.find(item->level());
        if (it != loglevelMap.end())
            shortname = (*it)->shortname;
    }

    if( item->tid() )
    {
        snprintf( line, MAX_STRING_LENGTH,
                  "%s %c [%d/%" PREFIX64 "d] %s %s:%d (%s) - %s\n",
                  timestamp, shortname, item->pid(), item->tid(),
                  item->rawThreadName(), item->rawFile(), item->line(),
                  item->rawFunction(), item->rawMessage() );
    }
    else
    {
        snprintf( line, MAX_STRING_LENGTH,
                  "%s %c [%d] %s %s:%d (%s) - %s\n",
                  timestamp, shortname, item->pid(), item->rawThreadName(),
                  item->rawFile(), item->line(), item->rawFunction(),
                  item->rawMessage() );
    }

    int result = write(m_fd, line, strlen(line));

    if( result == -1 )
    {
        LOG(VB_GENERAL, LOG_ERR,
                 QString("Closed Log output on fd %1 due to errors").arg(m_fd));
        m_opened = false;
        close( m_fd );
        return false;
    }
    return true;
}

#ifndef _WIN32
SyslogLogger::SyslogLogger(bool open) :
    LoggerBase(nullptr)
{
    if (open)
    {
        openlog(nullptr, LOG_NDELAY, 0 );
        m_opened = true;
    }

    LOG(VB_GENERAL, LOG_INFO, "Added syslogging");
}

SyslogLogger::~SyslogLogger()
{
    LOG(VB_GENERAL, LOG_INFO, "Removing syslogging");
    if (m_opened)
        closelog();
}

SyslogLogger *SyslogLogger::create(QMutex *mutex, bool open)
{
    SyslogLogger *logger =
        dynamic_cast<SyslogLogger *>(loggerMap.value("", nullptr));

    if (logger)
        return logger;

    // Need to add a new FileLogger
    mutex->unlock();
    // inserts into loggerMap
    logger = new SyslogLogger(open);
    mutex->lock();

    auto *clients = new ClientList;
    logRevClientMap.insert(logger, clients);
    return logger;
}


bool SyslogLogger::logmsg(LoggingItem *item)
{
    if (!m_opened || item->facility() <= 0)
        return false;

    char shortname = '-';

    {
        QMutexLocker locker(&loglevelMapMutex);
        LoglevelDef *lev = loglevelMap.value(item->level(), nullptr);
        if (lev != nullptr)
            shortname = lev->shortname;
    }
    syslog(item->level() | item->facility(), "%s[%d]: %c %s %s:%d (%s) %s",
           item->rawAppName(), item->pid(), shortname, item->rawThreadName(),
           item->rawFile(), item->line(), item->rawFunction(),
           item->rawMessage());

    return true;
}

#if CONFIG_SYSTEMD_JOURNAL
JournalLogger::JournalLogger() :
    LoggerBase(nullptr)
{
    LOG(VB_GENERAL, LOG_INFO, "Added journal logging");
}

JournalLogger::~JournalLogger()
{
    LOG(VB_GENERAL, LOG_INFO, "Removing journal logging");
}

JournalLogger *JournalLogger::create(QMutex *mutex)
{
    JournalLogger *logger =
        dynamic_cast<JournalLogger *>(loggerMap.value("", nullptr));

    if (logger)
        return logger;

    // Need to add a new FileLogger
    mutex->unlock();
    // inserts into loggerMap
    logger = new JournalLogger();
    mutex->lock();

    auto *clients = new ClientList;
    logRevClientMap.insert(logger, clients);
    return logger;
}


bool JournalLogger::logmsg(LoggingItem *item)
{
    sd_journal_send(
        "MESSAGE=%s", item->rawMessage(),
        "PRIORITY=%d", item->level(),
        "CODE_FILE=%s", item->rawFile(),
        "CODE_LINE=%d", item->line(),
        "CODE_FUNC=%s", item->rawFunction(),
        "SYSLOG_IDENTIFIER=%s", item->rawAppName(),
        "SYSLOG_PID=%d", item->pid(),
        "MYTH_THREAD=%s", item->rawThreadName(),
        NULL
        );
    return true;
}
#endif
#endif

const int DatabaseLogger::kMinDisabledTime = 1000;

DatabaseLogger::DatabaseLogger(const char *table) :
    LoggerBase(table)
{
    m_query = QString(
        "INSERT INTO %1 "
        "    (host, application, pid, tid, thread, filename, "
        "     line, function, msgtime, level, message) "
        "VALUES (:HOST, :APP, :PID, :TID, :THREAD, :FILENAME, "
        "        :LINE, :FUNCTION, :MSGTIME, :LEVEL, :MESSAGE)")
        .arg(m_handle);

    LOG(VB_GENERAL, LOG_INFO, QString("Added database logging to table %1")
        .arg(m_handle));

    m_thread = new DBLoggerThread(this);
    m_thread->start();
}

DatabaseLogger::~DatabaseLogger()
{
    LOG(VB_GENERAL, LOG_INFO, "Removing database logging");

    DatabaseLogger::stopDatabaseAccess();
}

DatabaseLogger *DatabaseLogger::create(const QString& table, QMutex *mutex)
{
    QByteArray ba = table.toLocal8Bit();
    const char *tble = ba.constData();
    DatabaseLogger *logger =
        dynamic_cast<DatabaseLogger *>(loggerMap.value(table, nullptr));

    if (logger)
        return logger;

    // Need to add a new FileLogger
    mutex->unlock();
    // inserts into loggerMap
    logger = new DatabaseLogger(tble);
    mutex->lock();

    auto *clients = new ClientList;
    logRevClientMap.insert(logger, clients);
    return logger;
}

void DatabaseLogger::stopDatabaseAccess(void)
{
    if( m_thread )
    {
        m_thread->stop();
        m_thread->wait();
        delete m_thread;
        m_thread = nullptr;
    }
}

bool DatabaseLogger::logmsg(LoggingItem *item)
{
    if (!m_thread)
        return false;

    if (!m_thread->isRunning())
    {
        m_disabledTime.start();
    }

    if (!m_disabledTime.isValid() && m_thread->queueFull())
    {
        m_disabledTime.start();
        LOG(VB_GENERAL, LOG_CRIT,
            "Disabling DB Logging: too many messages queued");
        return false;
    }

    if (m_disabledTime.isValid() && m_disabledTime.hasExpired(kMinDisabledTime))
    {
        if (isDatabaseReady() && !m_thread->queueFull())
        {
            m_disabledTime.invalidate();
            LOG(VB_GENERAL, LOG_CRIT, "Reenabling DB Logging");
        }
    }

    if (m_disabledTime.isValid())
        return false;

    m_thread->enqueue(item);
    return true;
}


bool DatabaseLogger::logqmsg(MSqlQuery &query, LoggingItem *item)
{
    char        timestamp[TIMESTAMP_MAX];

    time_t epoch = item->epoch();
    struct tm tm {};
    localtime_r(&epoch, &tm);

    strftime(timestamp, TIMESTAMP_MAX-8, "%Y-%m-%d %H:%M:%S",
             (const struct tm *)&tm);

    query.bindValue(":TID",         item->tid());
    query.bindValue(":THREAD",      item->threadName());
    query.bindValue(":FILENAME",    item->file());
    query.bindValue(":LINE",        item->line());
    query.bindValue(":FUNCTION",    item->function());
    query.bindValue(":MSGTIME",     timestamp);
    query.bindValue(":LEVEL",       item->level());
    query.bindValue(":MESSAGE",     item->message());
    query.bindValue(":APP",         item->appName());
    query.bindValue(":PID",         item->pid());

    if (!query.exec())
    {
        // Suppress Driver not loaded errors that occur at startup.
        // and suppress additional errors for one second after the
        // previous error (to avoid spamming the log).
        QSqlError err = query.lastError();
        if ((err.type() != 1
             || !err.nativeErrorCode().isEmpty()
                ) &&
            (!m_errorLoggingTime.isValid() ||
             (m_errorLoggingTime.hasExpired(1000))))
        {
            MythDB::DBError("DBLogging", query);
            m_errorLoggingTime.start();
        }
        return false;
    }

    return true;
}

void DatabaseLogger::prepare(MSqlQuery &query)
{
    query.prepare(m_query);
    query.bindValue(":HOST", gCoreContext->GetHostName());
}

bool DatabaseLogger::isDatabaseReady(void)
{
    bool ready = false;
    MythDB *db = GetMythDB();

    if ((db) && db->HaveValidDatabase())
    {
        if ( !m_loggingTableExists )
            m_loggingTableExists = tableExists(m_handle);

        if ( m_loggingTableExists )
            ready = true;
    }

    return ready;
}

bool DatabaseLogger::tableExists(const QString &table)
{
    bool result = false;
    MSqlQuery query(MSqlQuery::InitCon());
    if (query.isConnected())
    {
        QString sql = "SELECT COLUMN_NAME "
                      "  FROM INFORMATION_SCHEMA.COLUMNS "
                      " WHERE TABLE_SCHEMA = DATABASE() "
                      "   AND TABLE_NAME = :TABLENAME "
                      "   AND COLUMN_NAME = :COLUMNNAME;";
        if (query.prepare(sql))
        {
            query.bindValue(":TABLENAME", table);
            query.bindValue(":COLUMNNAME", "function");
            if (query.exec() && query.next())
                result = true;
        }
    }
    return result;
}


DBLoggerThread::DBLoggerThread(DatabaseLogger *logger) :
    MThread("DBLogger"), m_logger(logger),
    m_queue(new QQueue<LoggingItem *>),
    m_wait(new QWaitCondition())
{
}

DBLoggerThread::~DBLoggerThread()
{
    stop();
    wait();

    QMutexLocker qLock(&m_queueMutex);
    while (!m_queue->empty())
        m_queue->dequeue()->DecrRef();
    delete m_queue;
    delete m_wait;
    m_queue = nullptr;
    m_wait = nullptr;
}

void DBLoggerThread::run(void)
{
    RunProlog();

    // Wait a bit before we start logging to the DB..  If we wait too long,
    // then short-running tasks (like mythpreviewgen) will not log to the db
    // at all, and that's undesirable.
    while (true)
    {
        if ((m_aborted || (gCoreContext && m_logger->isDatabaseReady())))
            break;

        QMutexLocker locker(&m_queueMutex);
        m_wait->wait(locker.mutex(), 100);
    }

    if (!m_aborted)
    {
        // We want the query to be out of scope before the RunEpilog() so
        // shutdown occurs correctly as otherwise the connection appears still
        // in use, and we get a qWarning on shutdown.
        auto *query = new MSqlQuery(MSqlQuery::InitCon());
        m_logger->prepare(*query);

        QMutexLocker qLock(&m_queueMutex);
        while (!m_aborted || !m_queue->isEmpty())
        {
            if (m_queue->isEmpty())
            {
                m_wait->wait(qLock.mutex(), 100);
                continue;
            }

            LoggingItem *item = m_queue->dequeue();
            if (!item)
                continue;

            if (item->message()[0] != QChar('\0'))
            {
                qLock.unlock();
                bool logged = m_logger->logqmsg(*query, item);
                qLock.relock();

                if (!logged)
                {
                    m_queue->prepend(item);
                    m_wait->wait(qLock.mutex(), 100);
                    delete query;
                    query = new MSqlQuery(MSqlQuery::InitCon());
                    m_logger->prepare(*query);
                    continue;
                }
            }

            item->DecrRef();
        }

        delete query;
    }

    RunEpilog();
}

void DBLoggerThread::stop(void)
{
    QMutexLocker qLock(&m_queueMutex);
    m_aborted = true;
    m_wait->wakeAll();
}

bool DBLoggerThread::enqueue(LoggingItem *item)
{
    QMutexLocker qLock(&m_queueMutex);
    if (!m_aborted)
    {
        if (item)
        {
            item->IncrRef();
        }
        m_queue->enqueue(item);
    }
    return true;
}


#ifndef _WIN32

void logSigHup(void)
{
    if (!logForwardThread)
        return;

    // This will be running in the thread that's used by SignalHandler
    // Emit the signal which is connected to a slot that runs in the actual
    // handling thread.
    emit logForwardThread->incomingSigHup();
}
#endif


LogForwardThread::LogForwardThread() :
    MThread("LogForward")
{
    moveToThread(qthread());
}

LogForwardThread::~LogForwardThread()
{
    stop();
    wait();
}

void LogForwardThread::run(void)
{
    RunProlog();

    connect(this, SIGNAL(incomingSigHup(void)), this, SLOT(handleSigHup(void)),
            Qt::QueuedConnection);

    qRegisterMetaType<QList<QByteArray> >("QList<QByteArray>");

    while (!m_aborted)
    {
        qApp->processEvents(QEventLoop::AllEvents, 10);
        qApp->sendPostedEvents(nullptr, QEvent::DeferredDelete);

        {
            QMutexLocker lock(&logMsgListMutex);
            if (logMsgList.isEmpty() &&
                !logMsgListNotEmpty.wait(lock.mutex(), 90))
            {
                continue;
            }

            int processed = 0;
            while (!logMsgList.isEmpty())
            {
                processed++;
                LogMessage *msg = logMsgList.takeFirst();
                lock.unlock();
                forwardMessage(msg);
                delete msg;

                // Force a processEvents every 128 messages so a busy queue
                // doesn't preclude timer notifications, etc.
                if ((processed & 127) == 0)
                {
                    qApp->processEvents(QEventLoop::AllEvents, 10);
                    qApp->sendPostedEvents(nullptr, QEvent::DeferredDelete);
                }

                lock.relock();
            }
        }
    }

    LoggerList loggers;

    {
        QMutexLocker lock(&loggerMapMutex);
        loggers = loggerMap.values();
    }

    while (!loggers.isEmpty())
    {
        LoggerBase *logger = loggers.takeFirst();
        delete logger;
    }

    RunEpilog();
}


void LogForwardThread::handleSigHup(void)
{
#ifndef _WIN32
    LOG(VB_GENERAL, LOG_INFO, "SIGHUP received, rolling log files.");

    /* SIGHUP was sent.  Close and reopen debug logfiles */
    QMutexLocker locker(&loggerMapMutex);
    QMap<QString, LoggerBase *>::iterator it;
    for (it = loggerMap.begin(); it != loggerMap.end(); ++it)
    {
        it.value()->reopen();
    }
#endif
}

void LogForwardThread::forwardMessage(LogMessage *msg)
{
#ifdef DUMP_PACKET
    QList<QByteArray>::const_iterator it = msg->begin();
    int i = 0;
    for (; it != msg->end(); ++it, i++)
    {
        QByteArray buf = *it;
        cout << i << ":\t" << buf.size() << endl << "\t"
             << buf.toHex().constData() << endl << "\t"
             << buf.constData() << endl;
    }
#endif

    // First section is the client id
    QByteArray clientBa = msg->first();
    QString clientId = QString(clientBa.toHex());

    QByteArray json     = msg->at(1);

    if (json.size() == 0)
    {
        // cout << "invalid msg, no json data " << qPrintable(clientId) << endl;
        return;
    }

    QMutexLocker lock(&logClientMapMutex);
    LoggerListItem *logItem = logClientMap.value(clientId, nullptr);

    // cout << "msg  " << clientId.toLocal8Bit().constData() << endl;
    if (logItem)
    {
        loggingGetTimeStamp(&logItem->m_itemEpoch, nullptr);
    }
    else
    {
        LoggingItem *item = LoggingItem::create(json);

        logClientCount.ref();
        LOG(VB_FILE, LOG_DEBUG, QString("New Logging Client: ID: %1 (#%2)")
            .arg(clientId).arg(logClientCount.fetchAndAddOrdered(0)));

        QMutexLocker lock2(&loggerMapMutex);
        QMutexLocker lock3(&logRevClientMapMutex);

        // Need to find or create the loggers
        auto *loggers = new LoggerList;

        // FileLogger from logFile
        QString logfile = item->logFile();
        if (!logfile.isEmpty())
        {
            LoggerBase *logger = FileLogger::create(logfile, lock2.mutex());

            ClientList *clients = logRevClientMap.value(logger);

            if (clients)
                clients->insert(0, clientId);

            if (logger && loggers)
                loggers->insert(0, logger);
        }

#ifndef _WIN32
        // SyslogLogger from facility
        int facility = item->facility();
        if (facility > 0)
        {
            LoggerBase *logger = SyslogLogger::create(lock2.mutex());

            ClientList *clients = logRevClientMap.value(logger);

            if (clients)
                clients->insert(0, clientId);

            if (logger && loggers)
                loggers->insert(0, logger);
        }

#if CONFIG_SYSTEMD_JOURNAL
        // Journal Logger
        if (facility == SYSTEMD_JOURNAL_FACILITY)
        {
            LoggerBase *logger = JournalLogger::create(lock2.mutex());

            ClientList *clients = logRevClientMap.value(logger);

            if (clients)
                clients->insert(0, clientId);

            if (logger && loggers)
                loggers->insert(0, logger);
        }
#endif
#endif

        // DatabaseLogger from table
        QString table = item->table();
        if (!table.isEmpty())
        {
            LoggerBase *logger = DatabaseLogger::create(table, lock2.mutex());

            ClientList *clients = logRevClientMap.value(logger);

            if (clients)
                clients->insert(0, clientId);

            if (logger && loggers)
                loggers->insert(0, logger);
        }

        logItem = new LoggerListItem;
        loggingGetTimeStamp(&logItem->m_itemEpoch, nullptr);
        logItem->m_itemList = loggers;
        logClientMap.insert(clientId, logItem);

        item->DecrRef();
    }

    if (logItem && logItem->m_itemList && !logItem->m_itemList->isEmpty())
    {
        LoggingItem *item = LoggingItem::create(json);
        if (!item)
            return;
        foreach (auto & it, *logItem->m_itemList)
            it->logmsg(item);
        item->DecrRef();
    }
}

void LogForwardThread::stop(void)
{
    m_aborted = true;
}

bool logForwardStart(void)
{
    logForwardThread = new LogForwardThread();
    logForwardThread->start();

    usleep(10000);
    return (logForwardThread && logForwardThread->isRunning());
}

void logForwardStop(void)
{
    if (!logForwardThread)
        return;

    logForwardThread->stop();
    delete logForwardThread;
    logForwardThread = nullptr;

    QMutexLocker locker(&loggerMapMutex);
    for (auto it = loggerMap.begin(); it != loggerMap.end(); ++it)
    {
        it.value()->stopDatabaseAccess();
    }
}

void logForwardMessage(const QList<QByteArray> &msg)
{
    auto *message = new LogMessage(msg);
    QMutexLocker lock(&logMsgListMutex);

    bool wasEmpty = logMsgList.isEmpty();
    logMsgList.append(message);

    if (wasEmpty)
        logMsgListNotEmpty.wakeAll();
}

/*
 * vim:ts=4:sw=4:ai:et:si:sts=4
 */