common/flight_recorder.hpp

#pragma once

#include <common/utils.hpp>

#include <chrono>
#include <deque>
#include <fstream>
#include <iomanip>
#include <iostream>
#include <unordered_map>
#include <vector>

namespace pldm
{
namespace flightrecorder
{

using TxOrRx = bool;
using FlightRecorderData = std::vector<uint8_t>;
using FlightRecorderTimeStamp =
    std::chrono::time_point<std::chrono::system_clock>;
using FlightRecorderRecord =
    std::tuple<FlightRecorderTimeStamp, TxOrRx, FlightRecorderData>;

static constexpr auto flightRecorderDumpPath = "/tmp/pldm_flight_recorder";

/** @class FlightRecorder
 *
 *  The class for implementing the PLDM flight recorder logic. This class
 *  handles the insertion of the data into the recorder and also provides
 *  API's to dump the flight recorder into a file.
 */

class FlightRecorder
{
  private:
    FlightRecorder()
    {
        flightRecorderPolicy = (FLIGHT_RECORDER_MAX_ENTRIES > 0) ? true : false;
    }

    enum LatencyRangeEnum
    {
        MS1_10 = 0,
        MS10_100,
        MS100_1000,
        S1_5,
        S5_MAX,
    };

    struct Metrics
    {
        uint64_t txPackets;
        uint64_t txBytes;
        uint64_t txRetryPackets;

        uint64_t rxPackets;
        uint64_t rxBytes;
        uint64_t latencyRange[S5_MAX + 1];

        uint64_t retries;
        // In-kernel MCTP socket acts like UDP. Tx error is uncommon.
        // Any Tx error will be refelcted as Rx timeout.
        uint64_t errors;
    };

    LatencyRangeEnum mapLatencyRange(int ms)
    {
        if (ms <= 10)
            return MS1_10;
        if (ms <= 100)
            return MS10_100;
        if (ms <= 1000)
            return MS100_1000;
        if (ms <= 5000)
            return S1_5;
        return S5_MAX;
    }

    void saveMetric(uint8_t eid, size_t size, TxOrRx isTx, bool isRetry)
    {
        if (METRICS_DUMP_PERIOD == 0)
            return;
        auto emplaceResult = metricsMap.try_emplace(eid, Metrics{});
        Metrics& metrics = emplaceResult.first->second;
        if (isTx)
        {
            metrics.txBytes += size;
            metrics.txPackets++;
            lastTxTimestamp = std::chrono::steady_clock::now();
            isLastTxTimestampValid = true;

            if (isRetry)
            {
                metrics.retries++;
            }
        }
        else
        {
            metrics.rxBytes += size;
            metrics.rxPackets++;

            if (isLastTxTimestampValid)
            {
                int elapsedMs = static_cast<int>(
                    std::chrono::duration_cast<std::chrono::milliseconds>(
                        std::chrono::steady_clock::now() - lastTxTimestamp)
                        .count());
                metrics.latencyRange[mapLatencyRange(elapsedMs)]++;
                isLastTxTimestampValid = false;
            }
        }
    }

  protected:
    bool flightRecorderPolicy;
    std::unordered_map<uint8_t, std::deque<FlightRecorderRecord>> recordMap;
    bool isLastTxTimestampValid;
    std::chrono::time_point<std::chrono::steady_clock> lastTxTimestamp;
    std::chrono::time_point<std::chrono::steady_clock> lastMetricsDumpTime;
    std::unordered_map<uint8_t, Metrics> metricsMap;
    std::unordered_map<uint8_t, Metrics> lasMetricsSnapshot;

  public:
    FlightRecorder(const FlightRecorder&) = delete;
    FlightRecorder(FlightRecorder&&) = delete;
    FlightRecorder& operator=(const FlightRecorder&) = delete;
    FlightRecorder& operator=(FlightRecorder&&) = delete;
    ~FlightRecorder() = default;

    static FlightRecorder& GetInstance()
    {
        static FlightRecorder flightRecorder;
        return flightRecorder;
    }

    void increaseError(uint8_t eid)
    {
        if (METRICS_DUMP_PERIOD == 0)
            return;
        auto iter = metricsMap.find(eid);
        if (iter != metricsMap.end())
        {
            iter->second.errors++;
        }
    }

    /** @brief Add records to the flightRecorder
     *  @param[in] eid  - The eid of the message
     *  @param[in] buffer  - The request/respose byte buffer
     *  @param[in] isTx - Whether the message comes from TX or RX
     *
     *  @return void
     */
    void saveRecord(uint8_t eid, const FlightRecorderData& buffer, TxOrRx isTx,
                    bool isRetry = false)
    {
        saveMetric(eid, buffer.size(), isTx, isRetry);

        // if the flight recorder policy is enabled, then only insert the
        // messages into the flight recorder, if not this function will be just
        // a no-op
        if (flightRecorderPolicy)
        {
            auto emplaceResult =
                recordMap.try_emplace(eid, std::deque<FlightRecorderRecord>());
            std::deque<FlightRecorderRecord>& records =
                emplaceResult.first->second;
            auto entry = std::make_tuple(std::chrono::system_clock::now(), isTx,
                                         buffer);
            records.push_back(std::move(entry));
            if (records.size() > FLIGHT_RECORDER_MAX_ENTRIES)
            {
                records.pop_front();
            }
        }
    }

    /** @brief play flight recorder
     *
     *  @return void
     */

    void playRecorder()
    {
        if (flightRecorderPolicy)
        {
            std::ofstream recorderOutputFile(flightRecorderDumpPath);
            std::cerr << "Dumping the flight recorder into "
                      << flightRecorderDumpPath;
            for (const auto& [eid, records] : recordMap)
            {
                recorderOutputFile << "EID: " << (unsigned)eid << "\n";
                for (const auto& [timestamp, isTx, message] : records)
                {
                    recorderOutputFile << timestamp << " : ";
                    recorderOutputFile << (isTx ? "Tx : " : "Rx : ");
                    for (int byte : message)
                    {
                        recorderOutputFile << std::format("{:02x} ", byte);
                    }
                    recorderOutputFile << std::endl;
                }
            }
            recorderOutputFile.close();
        }
        else
        {
            std::cerr << "Fight recorder policy is disabled";
        }
    }

    void printEidHistory(uint8_t eid)
    {
        if (!flightRecorderPolicy)
            return;
        if (!recordMap.contains(eid))
            return;

        std::cerr << "EID: " << (unsigned)eid << " command history:\n";
        for (const auto& [timestamp, isTx, message] : recordMap[eid])
        {
            std::cerr << timestamp << " : ";
            std::cerr << (isTx ? "Tx : " : "Rx : ");
            for (int byte : message)
            {
                std::cerr << std::format("{:02x} ", byte);
            }
            std::cerr << std::endl;
        }
    }

    void dumpIncrementalMetrics()
    {
        if (METRICS_DUMP_PERIOD == 0)
            return;
        const auto now = std::chrono::steady_clock::now();
        auto elapsedMs = std::chrono::duration_cast<std::chrono::milliseconds>(
                             now - lastMetricsDumpTime)
                             .count();

        if (elapsedMs == 0)
        {
            // Unlikely, just avoid devide by zero
            elapsedMs = 1;
        }

        for (const auto& [eid, current] : metricsMap)
        {
            auto emplaceResult = lasMetricsSnapshot.try_emplace(eid, Metrics{});
            const Metrics& last = emplaceResult.first->second;
            std::string output = std::format(
                "Incremental metrics: EID: {:2}, TxPackets:{}, TxBytes:{},"
                " RxPackets:{}, RxBytes:{}, Retries:{}, Errors:{}, TxRate:{} B/s,"
                " RxRate:{} B/s",
                eid, current.txPackets - last.txPackets,
                current.txBytes - last.txBytes,
                current.rxPackets - last.rxPackets,
                current.rxBytes - last.rxBytes, current.retries - last.retries,
                current.errors - last.errors,
                (current.txBytes - last.txBytes) * 1000 / elapsedMs,
                (current.rxBytes - last.rxBytes) * 1000 / elapsedMs);
            std::cerr << output << std::endl;
        }

        for (const auto& [eid, current] : metricsMap)
        {
            auto iter = lasMetricsSnapshot.find(eid);
            if (iter == lasMetricsSnapshot.end())
            {
                continue;
            }

            const Metrics& last = iter->second;
            std::string output = std::format(
                "Latency histogram:   EID: {:2}, 1~10ms:{}, 10~100ms:{},"
                " 100~1000ms:{}, 1~5s:{}, 5~s:{}",
                eid, current.latencyRange[MS1_10] - last.latencyRange[MS1_10],
                current.latencyRange[MS10_100] - last.latencyRange[MS10_100],
                current.latencyRange[MS100_1000] -
                    last.latencyRange[MS100_1000],
                current.latencyRange[S1_5] - last.latencyRange[S1_5],
                current.latencyRange[S5_MAX] - last.latencyRange[S5_MAX]);
            std::cerr << output << std::endl;
        }

        lastMetricsDumpTime = now;
        lasMetricsSnapshot = metricsMap;
    }
};

} // namespace flightrecorder
} // namespace pldm