#include "PersistentTmStore.h" #include #include #include #include #include #include #include #include "fsfw/ipc/CommandMessage.h" #include "fsfw/ipc/QueueFactory.h" #include "fsfw/tmstorage/TmStoreMessage.h" #include "mission/persistentTmStoreDefs.h" using namespace returnvalue; PersistentTmStore::PersistentTmStore(PersistentTmStoreArgs args) : SystemObject(args.objectId), tmStore(args.tmStore), baseDir(args.baseDir), baseName(std::move(args.baseName)), sdcMan(args.sdcMan) { tcQueue = QueueFactory::instance()->createMessageQueue(); calcDiffSeconds(args.intervalUnit, args.intervalCount); } ReturnValue_t PersistentTmStore::cancelDump() { state = State::IDLE; return returnvalue::OK; } ReturnValue_t PersistentTmStore::assignAndOrCreateMostRecentFile() { if (not activeFile.has_value()) { return createMostRecentFile(std::nullopt); } return returnvalue::OK; } ReturnValue_t PersistentTmStore::handleCommandQueue(StorageManagerIF& ipcStore, Command_t& execCmd) { CommandMessage cmdMessage; ReturnValue_t result = tcQueue->receiveMessage(&cmdMessage); if (result != returnvalue::OK) { return result; } if (cmdMessage.getMessageType() == messagetypes::TM_STORE) { Command_t cmd = cmdMessage.getCommand(); if (cmd == TmStoreMessage::DELETE_STORE_CONTENT_TIME) { Clock::getClock_timeval(¤tTv); store_address_t storeId = TmStoreMessage::getStoreId(&cmdMessage); auto accessor = ipcStore.getData(storeId); uint32_t deleteUpToUnixSeconds = 0; size_t size = accessor.second.size(); SerializeAdapter::deSerialize(&deleteUpToUnixSeconds, accessor.second.data(), &size, SerializeIF::Endianness::NETWORK); execCmd = cmd; deleteUpTo(deleteUpToUnixSeconds); } else if (cmd == TmStoreMessage::DOWNLINK_STORE_CONTENT_TIME) { Clock::getClock_timeval(¤tTv); store_address_t storeId = TmStoreMessage::getStoreId(&cmdMessage); auto accessor = ipcStore.getData(storeId); if (accessor.second.size() < 8) { return returnvalue::FAILED; } uint32_t dumpFromUnixSeconds = 0; uint32_t dumpUntilUnixSeconds = 0; size_t size = 8; SerializeAdapter::deSerialize(&dumpFromUnixSeconds, accessor.second.data(), &size, SerializeIF::Endianness::NETWORK); SerializeAdapter::deSerialize(&dumpUntilUnixSeconds, accessor.second.data() + 4, &size, SerializeIF::Endianness::NETWORK); result = startDumpFromUpTo(dumpFromUnixSeconds, dumpUntilUnixSeconds); if (result == BUSY_DUMPING) { triggerEvent(persTmStore::BUSY_DUMPING_EVENT); } else { execCmd = cmd; } } } return result; } ReturnValue_t PersistentTmStore::startDumpFrom(uint32_t fromUnixSeconds) { return startDumpFromUpTo(fromUnixSeconds, currentTv.tv_sec); } ReturnValue_t PersistentTmStore::storePacket(PusTmReader& reader) { using namespace std::filesystem; if (baseDirUninitialized) { updateBaseDir(); } Clock::getClock_timeval(¤tTv); // It is assumed here that the filesystem is usable. if (not activeFile.has_value()) { ReturnValue_t result = assignAndOrCreateMostRecentFile(); if (result != returnvalue::OK) { return result; } } bool createNewFile = false; std::optional suffix = std::nullopt; std::error_code e; size_t fileSize = file_size(activeFile.value(), e); if (e) { sif::error << "PersistentTmStore: Could not retrieve file size, " "error " << e.message() << std::endl; return returnvalue::FAILED; } if (currentTv.tv_sec > activeFileTv.tv_sec + static_cast(rolloverDiffSeconds)) { createNewFile = true; currentSameSecNumber = 0; } else if (fileSize + reader.getFullPacketLen() > fileBuf.size()) { createNewFile = true; if (currentSameSecNumber >= MAX_FILES_IN_ONE_SECOND) { currentSameSecNumber = 0; } if (currentTv.tv_sec == activeFileTv.tv_sec) { suffix = currentSameSecNumber++; } else { currentSameSecNumber = 0; } } if (createNewFile) { createMostRecentFile(suffix); } // Rollover conditions were handled, write to file now std::ofstream of(activeFile.value(), std::ios::app | std::ios::binary); of.write(reinterpret_cast(reader.getFullData()), static_cast(reader.getFullPacketLen())); return returnvalue::OK; } MessageQueueId_t PersistentTmStore::getCommandQueue() const { return tcQueue->getId(); } void PersistentTmStore::calcDiffSeconds(RolloverInterval intervalUnit, uint32_t intervalCount) { if (intervalUnit == RolloverInterval::MINUTELY) { rolloverDiffSeconds = 60 * intervalCount; } else if (intervalUnit == RolloverInterval::HOURLY) { rolloverDiffSeconds = 60 * 60 * intervalCount; } else if (intervalUnit == RolloverInterval::DAILY) { rolloverDiffSeconds = 60 * 60 * 24 * intervalCount; } } bool PersistentTmStore::updateBaseDir() { using namespace std::filesystem; const char* currentPrefix = sdcMan.getCurrentMountPrefix(); if (currentPrefix == nullptr) { return false; } basePath = path(currentPrefix) / baseDir / baseName; std::error_code e; if (not exists(basePath, e)) { create_directories(basePath); } baseDirUninitialized = false; return true; } void PersistentTmStore::deleteUpTo(uint32_t unixSeconds) { using namespace std::filesystem; for (auto const& file : directory_iterator(basePath)) { if (file.is_directory() or (activeFile.has_value() and (activeFile.value() == file.path()))) { continue; } // Convert file time to the UNIX epoch struct tm fileTime {}; if (pathToTime(file.path(), fileTime) != returnvalue::OK) { sif::error << "Time extraction for " << file << "failed" << std::endl; continue; } time_t fileEpoch = timegm(&fileTime); if (fileEpoch + rolloverDiffSeconds < unixSeconds) { std::error_code e; std::filesystem::remove(file.path(), e); } } } ReturnValue_t PersistentTmStore::startDumpFromUpTo(uint32_t fromUnixSeconds, uint32_t upToUnixSeconds) { using namespace std::filesystem; if (state == State::DUMPING) { return returnvalue::FAILED; } dumpParams.dirIter = directory_iterator(basePath); if (dumpParams.dirIter == directory_iterator()) { return returnvalue::FAILED; } dumpParams.fromUnixTime = fromUnixSeconds; dumpParams.untilUnixTime = upToUnixSeconds; state = State::DUMPING; if (loadNextDumpFile() == DUMP_DONE) { // State will be set inside the function loading the next file. return DUMP_DONE; } return returnvalue::OK; } ReturnValue_t PersistentTmStore::loadNextDumpFile() { using namespace std::filesystem; dumpParams.currentSize = 0; std::error_code e; for (; dumpParams.dirIter != directory_iterator(); dumpParams.dirIter++) { dumpParams.dirEntry = *dumpParams.dirIter; if (dumpParams.dirEntry.is_directory(e)) { continue; } dumpParams.fileSize = std::filesystem::file_size(dumpParams.dirEntry.path(), e); if (e) { sif::error << "PersistentTmStore: Could not retrieve file size: " << e.message() << std::endl; continue; } // sif::debug << "Path: " << dumpParams.dirEntry.path() << std::endl; // Can't even read CCSDS header. if (dumpParams.fileSize <= 6) { continue; } if (dumpParams.fileSize > fileBuf.size()) { sif::error << "PersistentTmStore: File too large, is deleted" << std::endl; triggerEvent(persTmStore::FILE_TOO_LARGE, dumpParams.fileSize, fileBuf.size()); std::filesystem::remove(dumpParams.dirEntry.path(), e); continue; } const path& file = dumpParams.dirEntry.path(); struct tm fileTime {}; if (pathToTime(file, fileTime) != returnvalue::OK) { sif::error << "Time extraction for file " << file << "failed" << std::endl; continue; } auto fileEpoch = static_cast(timegm(&fileTime)); if ((fileEpoch > dumpParams.fromUnixTime) and (fileEpoch + rolloverDiffSeconds <= dumpParams.untilUnixTime)) { dumpParams.currentFileUnixStamp = fileEpoch; std::ifstream ifile(file, std::ios::binary); if (ifile.bad()) { sif::error << "PersistentTmStore: File is bad" << std::endl; continue; } ifile.read(reinterpret_cast(fileBuf.data()), static_cast(dumpParams.fileSize)); // Increment iterator for next cycle. dumpParams.dirIter++; break; } } if (dumpParams.dirIter == directory_iterator()) { state = State::IDLE; return DUMP_DONE; } return returnvalue::OK; } ReturnValue_t PersistentTmStore::dumpNextPacket(DirectTmSinkIF& tmSink, size_t& dumpedLen, bool& fileHasSwapped) { if (state == State::IDLE) { return returnvalue::FAILED; } PusTmReader reader(&timeReader, fileBuf.data() + dumpParams.currentSize, fileBuf.size() - dumpParams.currentSize); // CRC check to fully ensure this is a valid TM ReturnValue_t result = reader.parseDataWithCrcCheck(); if (result == returnvalue::OK) { result = tmSink.write(fileBuf.data() + dumpParams.currentSize, reader.getFullPacketLen()); if (result == DirectTmSinkIF::IS_BUSY) { return result; } else if (result != returnvalue::OK) { // TODO: Event? sif::error << "PersistentTmStore: Writing to TM sink failed" << std::endl; return result; } dumpParams.currentSize += reader.getFullPacketLen(); dumpedLen = reader.getFullPacketLen(); if (dumpParams.currentSize >= dumpParams.fileSize) { fileHasSwapped = true; return loadNextDumpFile(); } } else { sif::error << "PersistentTmStore: Parsing of PUS TM failed with code " << result << std::endl; triggerEvent(persTmStore::POSSIBLE_FILE_CORRUPTION, result, dumpParams.currentFileUnixStamp); // Delete the file and load next. Could use better algorithm to partially // restore the file dump, but for now do not trust the file. dumpedLen = 0; std::error_code e; std::filesystem::remove(dumpParams.dirEntry.path().c_str(), e); fileHasSwapped = true; return loadNextDumpFile(); } return returnvalue::OK; } ReturnValue_t PersistentTmStore::pathToTime(const std::filesystem::path& path, struct tm& time) { auto pathStr = path.string(); size_t splitChar = pathStr.find('_'); auto timeOnlyStr = pathStr.substr(splitChar + 1); if (nullptr == strptime(timeOnlyStr.c_str(), FILE_DATE_FORMAT, &time)) { return returnvalue::FAILED; } return returnvalue::OK; } ReturnValue_t PersistentTmStore::createMostRecentFile(std::optional suffix) { using namespace std::filesystem; unsigned currentIdx = 0; path pathStart = basePath / baseName; memcpy(fileBuf.data() + currentIdx, pathStart.c_str(), pathStart.string().length()); currentIdx += pathStart.string().length(); fileBuf[currentIdx] = '_'; currentIdx += 1; time_t epoch = currentTv.tv_sec; struct tm* time = gmtime(&epoch); size_t writtenBytes = strftime(reinterpret_cast(fileBuf.data() + currentIdx), fileBuf.size(), FILE_DATE_FORMAT, time); if (writtenBytes == 0) { sif::error << "PersistentTmStore::createMostRecentFile: Could not create file timestamp" << std::endl; return returnvalue::FAILED; } currentIdx += writtenBytes; char* res = strcpy(reinterpret_cast(fileBuf.data() + currentIdx), ".bin"); if (res == nullptr) { return returnvalue::FAILED; } currentIdx += 4; if (suffix.has_value()) { std::string fullSuffix = "." + std::to_string(suffix.value()); res = strcpy(reinterpret_cast(fileBuf.data() + currentIdx), fullSuffix.c_str()); if (res == nullptr) { return returnvalue::FAILED; } currentIdx += fullSuffix.size(); } path newPath(std::string(reinterpret_cast(fileBuf.data()), currentIdx)); std::ofstream of(newPath, std::ios::binary); activeFile = newPath; activeFileTv = currentTv; return returnvalue::OK; } ReturnValue_t PersistentTmStore::initializeTmStore() { Clock::getClock_timeval(¤tTv); updateBaseDir(); return assignAndOrCreateMostRecentFile(); } PersistentTmStore::State PersistentTmStore::getState() const { return state; } void PersistentTmStore::getStartAndEndTimeCurrentOrLastDump(uint32_t& startTime, uint32_t& endTime) const { startTime = dumpParams.fromUnixTime; endTime = dumpParams.untilUnixTime; }