XrdHttpMon.cc

Go to the documentation of this file.

#include "XrdHttpMon.hh"
#include "XrdSys/XrdSysError.hh"
#include "XrdXrootd/XrdXrootdGStream.hh"
 
#include <iostream>
#include <sstream>
#include <thread>
 
XrdSysError eDest(0, "HttpMon");
 
typedef std::array<std::array<XrdHttpMon::HttpInfo, XrdHttpMon::StatusCodes::sc_Count>, XrdHttpReq::ReqType::rtCount>
    StatsMatrix;
 
StatsMatrix XrdHttpMon::statsInfo{};
 
XrdXrootdGStream* XrdHttpMon::gStream = nullptr;
XrdMonRoll* XrdHttpMon::mrollP = nullptr;
std::chrono::seconds XrdHttpMon::flushPeriod{0};
 
bool XrdHttpMon::hasGStream = false;
bool XrdHttpMon::hasMonRoll = false;
bool XrdHttpMon::isInitialized = false;
 
RAtomic_uint64_t XrdHttpMon::verbCounters[XrdHttpReq::ReqType::rtCount] = {0};
RAtomic_uint64_t XrdHttpMon::statusCounters[XrdHttpMon::StatusCodes::sc_Count] = {0};
 
// Combined schema for HTTP statistics
// This creates a JSON structure: {"httpReqStats": {...}, "httpStatusCodeStats": {...}}
std::vector<XrdMonRoll::Item> XrdHttpMon::statsSchema = {
// NOTE: Keep this mapping aligned to the XrdHttpReq enum
    XrdMonRoll::Item("request", XrdMonRoll::Item::Schema::begObject),
        XrdMonRoll::Item("Unknown",   verbCounters[0]),
        XrdMonRoll::Item("Malformed", verbCounters[1]),
        XrdMonRoll::Item("GET",       verbCounters[2]),
        XrdMonRoll::Item("HEAD",      verbCounters[3]),
        XrdMonRoll::Item("PUT",       verbCounters[4]),
        XrdMonRoll::Item("OPTIONS",   verbCounters[5]),
        XrdMonRoll::Item("PATCH",     verbCounters[6]),
        XrdMonRoll::Item("DELETE",    verbCounters[7]),
        XrdMonRoll::Item("PROPFIND",  verbCounters[8]),
        XrdMonRoll::Item("MKCOL",     verbCounters[9]),
        XrdMonRoll::Item("MOVE",      verbCounters[10]),
        XrdMonRoll::Item("POST",      verbCounters[11]),
        XrdMonRoll::Item("COPY",      verbCounters[12]),
    XrdMonRoll::Item("request", XrdMonRoll::Item::Schema::endObject),
 
// NOTE: Keep this mapping strictly aligned with StatusCodes enum XrdHttpMon::StatusCode
// The order and number of entries MUST match.
    XrdMonRoll::Item("response", XrdMonRoll::Item::Schema::begObject),
        XrdMonRoll::Item("100", statusCounters[sc_100]),
        XrdMonRoll::Item("200", statusCounters[sc_200]),
        XrdMonRoll::Item("201", statusCounters[sc_201]),
        XrdMonRoll::Item("202", statusCounters[sc_202]),
        XrdMonRoll::Item("206", statusCounters[sc_206]),
        XrdMonRoll::Item("207", statusCounters[sc_207]),
        XrdMonRoll::Item("302", statusCounters[sc_302]),
        XrdMonRoll::Item("307", statusCounters[sc_307]),
        XrdMonRoll::Item("400", statusCounters[sc_400]),
        XrdMonRoll::Item("401", statusCounters[sc_401]),
        XrdMonRoll::Item("403", statusCounters[sc_403]),
        XrdMonRoll::Item("404", statusCounters[sc_404]),
        XrdMonRoll::Item("405", statusCounters[sc_405]),
        XrdMonRoll::Item("409", statusCounters[sc_409]),
        XrdMonRoll::Item("416", statusCounters[sc_416]),
        XrdMonRoll::Item("423", statusCounters[sc_423]),
        XrdMonRoll::Item("500", statusCounters[sc_500]),
        XrdMonRoll::Item("502", statusCounters[sc_502]),
        XrdMonRoll::Item("504", statusCounters[sc_504]),
        XrdMonRoll::Item("507", statusCounters[sc_507]),
        XrdMonRoll::Item("OTHERS", statusCounters[sc_UNKNOWN]),
    XrdMonRoll::Item("response", XrdMonRoll::Item::Schema::endObject)
};
 
void XrdHttpMon::Initialize(XrdSysLogger *logP, XrdXrootdGStream *gStream, XrdMonRoll *mrollP) {
    eDest.logger(logP);
    XrdHttpMon::gStream = gStream;
    XrdHttpMon::mrollP = mrollP;
 
    if (gStream != nullptr){
        hasGStream = true;
        flushPeriod = std::chrono::seconds(gStream->GetAutoFlush());
    }
 
    if (mrollP != nullptr) {
        hasMonRoll = true;
        mrollP->Register(XrdMonRoll::AddOn, "http_plugin", statsSchema);
    }
 
    isInitialized = true;
}
 
void XrdHttpMon::Report() {
    std::string json = GetMonitoringJson();
    if (!gStream->Insert(json.c_str(), json.size() + 1)) {
        eDest.Emsg("HttpMon", "Gstream Buffer Rejected");
    }
}
 
void* XrdHttpMon::Start(void*) {
    while (true) {
        std::this_thread::sleep_for(flushPeriod);
        Report();
    }
}
 
void XrdHttpMon::Record(XrdHttpReq &req, int code)
{
    // Early return if monitoring is not enabled.
    if (!isInitialized) return;
 
    // Only record once we have a "final" (>= 200) response code. 100-Continue is interim.
    if (code < 200) return;
 
    std::chrono::steady_clock::duration duration{};
    if (hasGStream) {
        const auto now = std::chrono::steady_clock::now();
        duration = now - req.startTime;
    }
 
    StatusCodes statusCode = ToStatusCode(code);
    XrdHttpMonState st = req.monState;
 
    if (req.request >= XrdHttpReq::ReqType::rtCount || req.request < 0) {
        eDest.Emsg("Record", "ERROR: Record called with invalid request type");
        return;
    }
 
    switch (st) {
        case XrdHttpMonState::NEW:
            RecordGStreamCount(req.request, statusCode);
            RecordMonRollVerb(req.request);
            req.monState = XrdHttpMonState::ACTIVE;
            return;
 
        case XrdHttpMonState::ACTIVE:
            RecordGStreamSuccess(req.request, statusCode, duration);
            RecordMonRollStatus(statusCode);
            req.monState = XrdHttpMonState::DONE;
            return;
 
        case XrdHttpMonState::ERR_NET:
            RecordGStreamErrNet(req.request, statusCode, duration);
            RecordMonRollStatus(statusCode);
            req.monState = XrdHttpMonState::DONE;
            return;
 
        case XrdHttpMonState::ERR_PROT:
            RecordGStreamErrProt(req.request, statusCode, duration);
            RecordMonRollStatus(statusCode);
            req.monState = XrdHttpMonState::DONE;
            return;
 
        case XrdHttpMonState::DONE:
            eDest.Emsg("Record", "ERROR: Record called after state was set to DONE");
            return;
    }
}
 
void XrdHttpMon::RecordCount(XrdHttpReq::ReqType op, StatusCodes sc) {
    auto& info = statsInfo[op][sc];
    info.count++;
}
 
void XrdHttpMon::RecordSuccess(XrdHttpReq::ReqType op, StatusCodes sc, std::chrono::steady_clock::duration duration) {
    auto& info = statsInfo[op][sc];
    info.success++;
    info.duration_us += std::chrono::duration_cast<std::chrono::microseconds>(duration).count();
}
 
void XrdHttpMon::RecordErrProt(XrdHttpReq::ReqType op, StatusCodes sc, std::chrono::steady_clock::duration duration) {
    auto& info = statsInfo[op][sc];
    info.error_xrootd++;
    info.duration_us += std::chrono::duration_cast<std::chrono::microseconds>(duration).count();
}
 
void XrdHttpMon::RecordErrNet(XrdHttpReq::ReqType op, StatusCodes sc, std::chrono::steady_clock::duration duration) {
    auto& info = statsInfo[op][sc];
    info.error_network++;
    info.duration_us += std::chrono::duration_cast<std::chrono::microseconds>(duration).count();
}
 
//  This creates a json with the following format:
//  http_GET_200 = {count: <>, bytes: <>, duration: <>}
std::string XrdHttpMon::GetMonitoringJson() {
    std::ostringstream oss;
    oss << "{";
 
    bool first = true;
    for (size_t op = 0; op < XrdHttpReq::ReqType::rtCount; ++op) {
        std::string opName = GetOperationString(static_cast<XrdHttpReq::ReqType>(op));
        for (size_t sc = 0; sc < StatusCodes::sc_Count; ++sc) {
            auto& info = statsInfo[op][sc];
 
            uint64_t total_count = info.count;
            if (total_count == 0) continue;
 
            uint64_t error_count_network = info.error_network;
            uint64_t error_count_xrootd = info.error_xrootd;
            uint64_t success_count = info.success;
            double duration_us = std::chrono::duration<double>(std::chrono::microseconds(static_cast<uint64_t>(info.duration_us))).count();
 
            std::string key = "HTTP_" + opName + "_" + GetStatusCodeString(static_cast<StatusCodes>(sc));
 
            if (!first) oss << ",";
            first = false;
 
            oss << "\"" << key << "\":{";
            oss << "\"count\":" << total_count << ",";
            oss << "\"errors_network\":" << error_count_network << ",";
            oss << "\"errors_xrootd\":" << error_count_xrootd << ",";
            oss << "\"success\":" << success_count << ",";
            oss << "\"duration_seconds\":" << duration_us;
            oss << "}";
        }
    }
 
    oss << "}";
    return oss.str();
}
 
std::string XrdHttpMon::GetOperationString(XrdHttpReq::ReqType op) {
    switch (op) {
        case XrdHttpReq::ReqType::rtDELETE:
            return "DELETE";
        case XrdHttpReq::ReqType::rtHEAD:
            return "HEAD";
        case XrdHttpReq::ReqType::rtGET:
            return "GET";
        case XrdHttpReq::ReqType::rtMKCOL:
            return "MKCOL";
        case XrdHttpReq::ReqType::rtMOVE:
            return "MOVE";
        case XrdHttpReq::ReqType::rtOPTIONS:
            return "OPTIONS";
        case XrdHttpReq::ReqType::rtPROPFIND:
            return "PROPFIND";
        case XrdHttpReq::ReqType::rtPUT:
            return "PUT";
        case XrdHttpReq::ReqType::rtCOPY:
            return "COPY";
        case XrdHttpReq::ReqType::rtMalformed:
            return "Malformed";
        default:
            return "UNKNOWN";
    }
}
 
std::string XrdHttpMon::GetStatusCodeString(StatusCodes sc) {
    switch (sc) {
        case sc_100:
            return "100";
        case sc_200:
            return "200";
        case sc_201:
            return "201";
        case sc_202:
            return "202";
        case sc_206:
            return "206";
        case sc_207:
            return "207";
        case sc_302:
            return "302";
        case sc_307:
            return "307";
        case sc_400:
            return "400";
        case sc_401:
            return "401";
        case sc_403:
            return "403";
        case sc_404:
            return "404";
        case sc_405:
            return "405";
        case sc_409:
            return "409";
        case sc_416:
            return "416";
        case sc_423:
            return "423";
        case sc_500:
            return "500";
        case sc_502:
            return "502";
        case sc_504:
            return "504";
        case sc_507:
            return "507";
        default:
            return "UNKNOWN";
    }
}
 
XrdHttpMon::StatusCodes XrdHttpMon::ToStatusCode(int code) {
    switch (code) {
        case 100:
            return sc_100;
        case 200:
            return sc_200;
        case 201:
            return sc_201;
        case 202:
            return sc_202;
        case 206:
            return sc_206;
        case 207:
            return sc_207;
        case 302:
            return sc_302;
        case 307:
            return sc_307;
        case 400:
            return sc_400;
        case 401:
            return sc_401;
        case 403:
            return sc_403;
        case 404:
            return sc_404;
        case 405:
            return sc_405;
        case 409:
            return sc_409;
        case 416:
            return sc_416;
        case 423:
            return sc_423;
        case 500:
            return sc_500;
        case 502:
            return sc_502;
        case 504:
            return sc_504;
        case 507:
            return sc_507;
        default:
            return sc_UNKNOWN;
    }
}