src/NVMeMetricStore.cpp

#include "NVMeMetricStore.hpp"

#include <phosphor-logging/lg2.hpp>
#include <xyz/openbmc_project/Common/File/error.hpp>
#include <xyz/openbmc_project/Common/error.hpp>

sdbusplus::message::unix_fd
    MetricStore::writeToFd(std::span<const uint8_t> header,
                           std::span<const uint8_t> data)
{
    std::array<int, 2> pipe{};
    if (::pipe(pipe.data()) < 0)
    {
        lg2::error("Metric Fetcher failed due to pipe creation: {ERROR}",
                   "ERROR", strerror(errno));
        throw sdbusplus::xyz::openbmc_project::Common::File::Error::Open{};
    }

    auto asyncPipe =
        std::make_shared<boost::asio::posix::stream_descriptor>(io, pipe[1]);
    std::weak_ptr<boost::asio::posix::stream_descriptor> weakPipe = asyncPipe;

    auto asyncBuffer = std::make_shared<std::vector<uint8_t>>(header.begin(),
                                                              header.end());
    asyncBuffer->insert(asyncBuffer->end(), data.begin(), data.end());

    boost::asio::async_write(
        *asyncPipe, boost::asio::buffer(*asyncBuffer),
        [asyncPipe, asyncBuffer](const boost::system::error_code& error,
                                 std::size_t /* transferredSize */) {
        if (error)
        {
            lg2::error("Metric Fetcher fails to write fd: {ERROR}", "ERROR",
                       error.message());
        }
    });

    // close the read end of the pipe after waiting for 5 seconds
    // It is expected that the client has duplicated the fd in this time
    auto timer = std::make_shared<boost::asio::steady_timer>(io);
    timer->expires_after(std::chrono::seconds(5));
    timer->async_wait([timer, fd{pipe[0]},
                       weakPipe](const boost::system::error_code& /* ec */) {
        auto pipe = weakPipe.lock();
        if (pipe)
        {
            pipe->cancel();
        }
        close(fd);
    });

    // Return the read end wrapped in a stream descriptor
    return sdbusplus::message::unix_fd{pipe[0]};
}

sdbusplus::message::unix_fd MetricStore::getMetric(std::string name)
{
    auto it = metricStore.find(name);
    if (it == metricStore.end() || !it->second)
    {
        lg2::error("Cannot find the target Metric", "PATH", this->path, "NAME",
                   name);
        throw sdbusplus::xyz::openbmc_project::Common::Error::InvalidArgument{};
    }

    if (!it->second->isCacheValid())
    {
        lg2::error("The Metric cache is invalid", "PATH", this->path, "NAME",
                   name);
        throw sdbusplus::xyz::openbmc_project::Common::Error::Unavailable{};
    }
    auto data = it->second->getCache();
    // Send the data and header to be written to pipe
    MetricHeader header{};
    header.lens = boost::endian::little_uint32_t(sizeof(header));
    header.version = 0;
    header.dataFormat = 0;
    header.startTime = boost::endian::little_uint64_t(
        std::get<0>(data).time_since_epoch().count());
    header.finishTime = boost::endian::little_uint64_t(
        std::get<1>(data).time_since_epoch().count());

    return writeToFd(
        // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast)
        std::span<uint8_t>(reinterpret_cast<std::uint8_t*>(&header),
                           sizeof(header)),
        std::get<2>(data));
}

static void ensureFresh(
    const std::shared_ptr<MetricBase<std::chrono::steady_clock>>& metric,
    std::chrono::steady_clock::time_point requestTime,
    std::function<void(std::error_code ec)>&& cb)
{
    metric->refresh(
        [metric, requestTime, cb = std::move(cb)](std::error_code ec) mutable {
        if (ec)
        {
            cb(ec);
            return;
        }
        if (std::get<0>(metric->getCache()) < requestTime)
        {
            ensureFresh(metric, requestTime, std::move(cb));
        }
        else
        {
            cb(ec);
        }
    });
}

sdbusplus::message::unix_fd MetricStore::refreshMetric(std::string name)
{
    if (name.empty())
    {
        lg2::error("Metric name is empty");
        throw sdbusplus::xyz::openbmc_project::Common::Error::InvalidArgument{};
    }

    auto it = metricStore.find(name);
    if (it == metricStore.end() || !it->second)
    {
        lg2::error("Cannot find the target Metric", "PATH", this->path, "NAME",
                   name);
        throw sdbusplus::xyz::openbmc_project::Common::Error::InvalidArgument{};
    }

    std::array<int, 2> pipe{};
    if (::pipe(pipe.data()) < 0)
    {
        lg2::error("Metric Fetcher failed due to pipe creation: {ERROR}",
                   "ERROR", strerror(errno));
        throw sdbusplus::xyz::openbmc_project::Common::File::Error::Open{};
    }

    auto asyncPipe =
        std::make_shared<boost::asio::posix::stream_descriptor>(io, pipe[1]);
    std::weak_ptr<boost::asio::posix::stream_descriptor> weakPipe = asyncPipe;

    auto requestTime = std::chrono::steady_clock::now();
    ensureFresh(it->second, requestTime,
                [asyncPipe, metric = it->second](std::error_code ec) {
        if (ec)
        {
            lg2::error("Metric Refresh fails: {ERROR}", "ERROR", ec.message());
            return;
        }

        auto data = metric->getCache();
        // Send the data and header to be written to pipe
        MetricHeader header{};
        header.lens = boost::endian::little_uint32_t(sizeof(header));
        header.version = 0;
        header.dataFormat = 0;
        header.startTime = boost::endian::little_uint64_t(
            std::get<0>(data).time_since_epoch().count());
        header.finishTime = boost::endian::little_uint64_t(
            std::get<1>(data).time_since_epoch().count());

        auto asyncBuffer =
            std::make_shared<std::vector<uint8_t>>(sizeof(header));
        std::memcpy(asyncBuffer->data(), &header, sizeof(header));
        asyncBuffer->insert(asyncBuffer->end(), std::get<2>(data).begin(),
                            std::get<2>(data).end());

        boost::asio::async_write(
            *asyncPipe, boost::asio::buffer(*asyncBuffer),
            [asyncPipe, asyncBuffer](const boost::system::error_code& error,
                                     std::size_t /* transferredSize */) {
            if (error)
            {
                lg2::error("Metric Fetcher fails to write fd: {ERROR}", "ERROR",
                           error.message());
            }
        });
    });

    // close the read end of the pipe after waiting for 5 seconds
    // It is expected that the client has duplicated the fd in this time
    auto timer = std::make_shared<boost::asio::steady_timer>(io);
    timer->expires_after(std::chrono::seconds(5));
    timer->async_wait([timer, fd{pipe[0]},
                       weakPipe](const boost::system::error_code& /* ec */) {
        auto pipe = weakPipe.lock();
        if (pipe)
        {
            pipe->cancel();
        }
        close(fd);
    });

    return sdbusplus::message::unix_fd{pipe[0]};
}

void MetricStore::signalHelper(
    std::string_view name,
    const std::shared_ptr<MetricBase<std::chrono::steady_clock>>&
        metric) noexcept
{
    if (!metric || !metric->isCacheValid())
    {
        // broadcast invalid metric signal
        metricUpdated(std::string{name}, {});
        return;
    }

    auto data = metric->getCache();
    std::vector<uint8_t> headerBuf;
    headerBuf.resize(sizeof(MetricHeader));
    // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast)
    MetricHeader& header = *reinterpret_cast<MetricHeader*>(headerBuf.data());
    header.lens = boost::endian::little_uint32_t(sizeof(header));
    header.version = 0;
    header.dataFormat = 0;
    header.startTime = boost::endian::little_uint64_t(
        std::get<0>(data).time_since_epoch().count());
    header.finishTime = boost::endian::little_uint64_t(
        std::get<1>(data).time_since_epoch().count());
    metricUpdated(std::string{name}, headerBuf);
};