workspace.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_dpm
 * @copyright (c) Copyright ESO 2022
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 *
 * @brief daq::dpm::Workspace interface and implementation declaration
 */
#include <daq/dpm/workspace.hpp>
 
#include <fstream>
#include <stdexcept>
#include <system_error>
 
#include <fmt/format.h>
#include <nlohmann/json.hpp>
 
#include <daq/json.hpp>
 
namespace fs = std::filesystem;
 
namespace daq::dpm {
namespace {
 
constexpr const std::string_view PATH_IN_PROGRESS = "in-progress";
constexpr const std::string_view PATH_ARCHIVE = "archive";
constexpr const std::string_view PATH_RESULT = "result";
constexpr const std::string_view PATH_SOURCES = "sources";
constexpr const std::string_view FILE_DAQ_QUEUE = "queue.json";
constexpr const std::string_view FILE_DP_SPEC = "specification.json";
constexpr const std::string_view FILE_DAQ_STATUS = "status.json";
constexpr const std::string_view FILE_DAQ_SOURCES = "sources.json";
constexpr const std::string_view FILE_TEMP_EXT = ".swap";
constexpr const std::string_view FILE_SYMLINK = "result";
 
template <class T>
void SafeWrite(T const& content, fs::path const& destination) {
    try {
        auto temp = destination;
        auto temp_file = destination.filename().native();
        temp_file += FILE_TEMP_EXT;
        temp.replace_filename(temp_file);
        // Write to temporary file
        {
            std::fstream temp_file(temp, std::ios::out | std::ios::trunc);
            temp_file << std::setw(4) << content;
            if (!temp_file) {
                throw std::runtime_error(
                    fmt::format("File contains error after write: '{}'", temp.native()));
            }
            temp_file.close();
        }
        // Move (atomically) into location
        fs::rename(temp, destination);
    } catch (...) {
        std::throw_with_nested(
            std::runtime_error(fmt::format("Failed to write file '{}'", destination.native())));
    }
}
 
/**
 * Initialize workspace
 *
 * @note Is safe to run for existing and valid workspace.
 */
void InitRootWorkspace(fs::path const& root) {
    // If there are non-managed files refuse to initialize new worksace as we are misconfigured
    auto status = fs::status(root);
    if (fs::exists(status)) {
        if (!fs::is_directory(status)) {
            throw std::invalid_argument(
                fmt::format("Workspace root exist but is not a directory: {}", root.native()));
        }
 
        for (auto const& entry : fs::directory_iterator(root)) {
            if (entry.path().extension() == FILE_TEMP_EXT) {
                // Old temporary files -> delete
                fs::remove(entry);
                continue;
            }
            if (entry.path().filename() == FILE_DAQ_QUEUE) {
                continue;
            }
            if (entry.path().filename() == PATH_IN_PROGRESS) {
                continue;
            }
            if (entry.path().filename() == PATH_ARCHIVE) {
                continue;
            }
            if (entry.path().filename() == PATH_RESULT) {
                continue;
            }
            throw std::invalid_argument(
                fmt::format("Workspace {} has unexpected contents - aborting to prevent filesystem "
                            "modifications due to misconfiguration",
                            root.native()));
        }
    }
 
    fs::create_directories(root);
    fs::create_directory(root / PATH_IN_PROGRESS);
    fs::create_directory(root / PATH_ARCHIVE);
    fs::create_directory(root / PATH_RESULT);
}
 
/**
 * Init new data acquisition directory inside root.
 * It assumes root already exits.
 *
 * @param root Workspace root (not DAQ workspace)
 */
auto MakeInProgressDaqPath(fs::path const& root, std::string const& daq_id) -> fs::path {
    return root / PATH_IN_PROGRESS / daq_id;
}
 
auto InitDaqWorkspace(fs::path const& root, std::string const& daq_id) -> fs::path {
    auto daq_root = MakeInProgressDaqPath(root, daq_id);
    if (fs::exists(daq_root)) {
        throw std::invalid_argument(
            fmt::format("DAQ workspace directory '{}' already exists", daq_root.string()).c_str());
    }
    fs::create_directory(daq_root);
    fs::create_directory(daq_root / PATH_SOURCES);
    return daq_root;
}
 
}  // namespace
 
WorkspaceImpl::WorkspaceImpl(fs::path root) : m_root(std::move(root)) {
    auto status = fs::status(m_root);
    if (status.type() != fs::file_type::directory && status.type() != fs::file_type::not_found) {
        throw std::invalid_argument(
            fmt::format("Workspace root directory '{}' is not a directory", m_root.string())
                .c_str());
    }
 
    InitRootWorkspace(m_root);
    // Initialize empty DAQ queue if necessary
    if (!fs::exists(m_root / FILE_DAQ_QUEUE)) {
        WorkspaceImpl::StoreQueue({});
    }
}
 
auto WorkspaceImpl::EnumerateDaqs() const -> std::vector<std::string> {
    return {};
}
 
fs::space_info WorkspaceImpl::QueryStorageStatus() const {
    return fs::space(m_root);
}
 
auto WorkspaceImpl::InitializeDaq(std::string const& daq_id) -> std::unique_ptr<DaqWorkspace> {
    auto root = InitDaqWorkspace(m_root, daq_id);
    return std::make_unique<DaqWorkspaceImpl>(root, m_root / PATH_RESULT);
}
 
auto WorkspaceImpl::LoadDaq(std::string const& daq_id) -> std::unique_ptr<DaqWorkspace> {
    auto root = MakeInProgressDaqPath(m_root, daq_id);
    return std::make_unique<DaqWorkspaceImpl>(root, m_root / PATH_RESULT);
}
 
void WorkspaceImpl::RemoveDaq(std::string const& daq_id) {
    fs::remove_all(m_root / PATH_IN_PROGRESS / daq_id);
}
 
auto WorkspaceImpl::ArchiveDaq(std::string const& daq_id) -> std::filesystem::path {
    auto target = m_root / PATH_ARCHIVE / daq_id;
    if (fs::exists(target)) {
        target += ".";
        for (auto idx = 1u; idx < 0xffff; ++idx) {
            auto tmp = target;
            tmp += std::to_string(idx);
            if (!fs::exists(tmp)) {
                target.swap(tmp);
                break;
            }
        }
    }
    fs::rename(m_root / PATH_IN_PROGRESS / daq_id, target);
    return target;
}
 
auto WorkspaceImpl::LoadQueue() const -> std::vector<std::string> {
    auto file = m_root / FILE_DAQ_QUEUE;
    try {
        std::fstream fs(file, std::ios::in);
        auto json = nlohmann::json::parse(fs);
        return json.get<std::vector<std::string>>();
    } catch (...) {
        std::throw_with_nested(
            std::runtime_error(fmt::format("Failed to load DAQ queue '{}'", file.native())));
    }
}
 
void WorkspaceImpl::StoreQueue(std::vector<std::string> const& queue) const {
    try {
        nlohmann::json json;
        json = queue;
        SafeWrite(json, m_root / FILE_DAQ_QUEUE);
    } catch (...) {
        std::throw_with_nested(std::runtime_error("Failed to store DAQ queue"));
    }
}
 
DaqWorkspaceImpl::DaqWorkspaceImpl(fs::path root, fs::path result)
    : m_root(std::move(root)), m_result(std::move(result)) {
    if (!m_root.is_absolute()) {
        throw std::invalid_argument(
            fmt::format("DAQ workspace root path must be absolute: {}", m_root.native()));
    }
    if (!m_result.is_absolute()) {
        throw std::invalid_argument(
            fmt::format("DAQ workspace result path must be absolute: {}", m_result.native()));
    }
}
 
auto DaqWorkspaceImpl::GetSourcesPath() const -> std::filesystem::path {
    return PATH_SOURCES;
}
 
auto DaqWorkspaceImpl::LoadStatus() const -> Status {
    try {
        std::fstream fs(m_root / FILE_DAQ_STATUS, std::ios::in);
        auto json = nlohmann::json::parse(fs);
        return json.get<Status>();
    } catch (...) {
        std::throw_with_nested(std::runtime_error("Failed to load DAQ status"));
    }
}
 
void DaqWorkspaceImpl::StoreStatus(Status const& status) const {
    try {
        nlohmann::json json;
        json = status;
        SafeWrite(json, m_root / FILE_DAQ_STATUS);
    } catch (...) {
        std::throw_with_nested(std::runtime_error("Failed to store DAQ status"));
    }
}
 
auto DaqWorkspaceImpl::GetSourceLookupPath() const -> std::filesystem::path {
    return FILE_DAQ_SOURCES;
}
 
void DaqWorkspaceImpl::MakeResultSymlink(std::filesystem::path const& result) const {
    auto relative = result.lexically_relative(m_root);
    fs::create_symlink(relative, m_root / FILE_SYMLINK);
}
 
auto DaqWorkspaceImpl::LoadSourceLookup() const -> SourceResolver::Mapping {
    try {
        std::fstream fs(m_root / FILE_DAQ_SOURCES, std::ios::in);
        auto json = nlohmann::json::parse(fs);
        return json.get<SourceResolver::Mapping>();
    } catch (...) {
        std::throw_with_nested(std::runtime_error("Failed to load DAQ source lookup"));
    }
}
 
void DaqWorkspaceImpl::StoreSourceLookup(SourceResolver::Mapping const& lookup) const {
    try {
        nlohmann::json json;
        json = lookup;
        SafeWrite(json, m_root / FILE_DAQ_SOURCES);
    } catch (...) {
        std::throw_with_nested(std::runtime_error("Failed to store DAQ source lookup"));
    }
}
 
auto DaqWorkspaceImpl::GetSpecificationPath() const -> std::filesystem::path {
    return FILE_DP_SPEC;
}
 
auto DaqWorkspaceImpl::LoadSpecification() const -> json::DpSpec {
    auto file = m_root / FILE_DP_SPEC;
    try {
        std::fstream fs(file, std::ios::in);
        auto json = nlohmann::json::parse(fs);
        return json::ParseDpSpec(json);
    } catch (...) {
        std::throw_with_nested(std::runtime_error(
            fmt::format("Failed to load Data Product Specification '{}'", file.native())));
    }
}
 
void DaqWorkspaceImpl::StoreSpecification(std::string const& specification) const {
    try {
        SafeWrite(specification, m_root / FILE_DP_SPEC);
    } catch (...) {
        std::throw_with_nested(std::runtime_error("Failed to store DAQ status"));
    }
}
 
}  // namespace daq::dpm