-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
telemetry - introduce telemetry aggFile
- Loading branch information
1 parent
a806de4
commit 791a70c
Showing
8 changed files
with
289 additions
and
4 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,69 @@ | ||
/** | ||
* @file | ||
* @author Pavel Siska <siska@cesnet.cz> | ||
* @brief Aggregated telemetry file | ||
* | ||
* SPDX-License-Identifier: BSD-3-Clause | ||
*/ | ||
|
||
#pragma once | ||
|
||
#include "aggMethod.hpp" | ||
#include "content.hpp" | ||
#include "file.hpp" | ||
#include "node.hpp" | ||
|
||
#include <memory> | ||
#include <string> | ||
#include <string_view> | ||
#include <vector> | ||
|
||
namespace telemetry { | ||
|
||
/** | ||
* @brief Class representing an aggregated file | ||
* | ||
* AggregatedFile is a subclass of File and is responsible for aggregating telemetry data from | ||
* multiple files. It reads data from matched files based on a regex pattern and applies aggregation | ||
* operations defined by aggregation methods. | ||
*/ | ||
class AggregatedFile : public File { | ||
public: | ||
~AggregatedFile() override = default; | ||
|
||
// Object cannot be copied or moved as it would break references from directories. | ||
AggregatedFile(const AggregatedFile& other) = delete; | ||
AggregatedFile& operator=(const AggregatedFile& other) = delete; | ||
AggregatedFile(AggregatedFile&& other) = delete; | ||
AggregatedFile& operator=(AggregatedFile&& other) = delete; | ||
|
||
/** | ||
* @brief Execute read operation over all matched files and aggregate them. | ||
* | ||
* This function reads data from all matched files based on the regex pattern and aggregates it | ||
* using aggregation methods specified during object creation. | ||
* | ||
* @return Aggregated content. | ||
* @throw NodeException if an error occurs during aggregation. | ||
*/ | ||
Content read(); | ||
|
||
private: | ||
// Allow directory to call AggregatedFile constructor | ||
friend class Directory; | ||
// Can be created only from a directory. Must be always created as a shared_ptr. | ||
AggregatedFile( | ||
const std::shared_ptr<Node>& parent, | ||
std::string_view name, | ||
std::string aggFilesPattern, | ||
const std::vector<AggOperation>& ops); | ||
|
||
FileOps getOps(); | ||
|
||
const std::string m_filesRegexPattern; | ||
|
||
std::vector<std::string> m_paths; | ||
std::vector<std::unique_ptr<AggMethod>> m_aggMethods; | ||
}; | ||
|
||
} // namespace telemetry |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,161 @@ | ||
/** | ||
* @file | ||
* @author Pavel Siska <siska@cesnet.cz> | ||
* @brief Telemetry file aggregator | ||
* | ||
* SPDX-License-Identifier: BSD-3-Clause | ||
*/ | ||
|
||
#include "aggregator/aggMethodFactory.hpp" | ||
|
||
#include <telemetry/aggFile.hpp> | ||
#include <telemetry/directory.hpp> | ||
#include <telemetry/file.hpp> | ||
#include <telemetry/utility.hpp> | ||
|
||
#include <iomanip> | ||
#include <regex> | ||
|
||
namespace telemetry { | ||
|
||
template <typename T> | ||
static std::vector<std::shared_ptr<T>> | ||
getMatchesInDirectory(const std::regex& regex, const std::shared_ptr<Directory>& directory) | ||
{ | ||
std::vector<std::shared_ptr<T>> matches; | ||
|
||
const auto& entries = directory->listEntries(); | ||
for (const auto& entry : entries) { | ||
if (!std::regex_match(entry, regex)) { | ||
continue; | ||
} | ||
auto node = directory->getEntry(entry); | ||
if (const auto& derivedNode = std::dynamic_pointer_cast<T>(node)) { | ||
matches.push_back(derivedNode); | ||
} | ||
} | ||
|
||
return matches; | ||
} | ||
|
||
static std::vector<std::shared_ptr<File>> | ||
getFilesMatchingPattern(const std::string& regexPath, std::shared_ptr<Directory> parentDir) | ||
{ | ||
std::vector<std::shared_ptr<File>> matchingFiles; | ||
|
||
auto pathSegments = utils::parsePath(regexPath); | ||
if (pathSegments.empty()) { | ||
return matchingFiles; | ||
} | ||
|
||
const std::string topLevelName = pathSegments.back(); | ||
pathSegments.pop_back(); | ||
|
||
std::vector<std::shared_ptr<Directory>> matchedDirs = {std::move(parentDir)}; | ||
|
||
for (const auto& subDir : pathSegments) { | ||
std::vector<std::shared_ptr<Directory>> matchesInCurrentDir; | ||
const std::regex dirRegex(subDir); | ||
for (const auto& dir : matchedDirs) { | ||
const auto matchedSubDirs = getMatchesInDirectory<Directory>(dirRegex, dir); | ||
matchesInCurrentDir.insert( | ||
matchesInCurrentDir.end(), | ||
matchedSubDirs.begin(), | ||
matchedSubDirs.end()); | ||
} | ||
matchedDirs = matchesInCurrentDir; | ||
} | ||
|
||
const std::regex fileRegex(topLevelName); | ||
for (const auto& dir : matchedDirs) { | ||
const auto filesInDir = getMatchesInDirectory<File>(fileRegex, dir); | ||
matchingFiles.insert(matchingFiles.end(), filesInDir.begin(), filesInDir.end()); | ||
} | ||
|
||
return matchingFiles; | ||
} | ||
|
||
static void mergeContent(Content& content, const Content& newContent) | ||
{ | ||
if (std::holds_alternative<Dict>(content) && std::holds_alternative<Dict>(newContent)) { | ||
auto& dict = std::get<Dict>(content); | ||
const auto& newDict = std::get<Dict>(newContent); | ||
dict.insert(newDict.begin(), newDict.end()); | ||
return; | ||
} | ||
content = newContent; | ||
} | ||
|
||
static void validateAggOperations(const std::vector<AggOperation>& ops) | ||
{ | ||
const bool hasDictFieldName | ||
= std::any_of(ops.begin(), ops.end(), [](const AggOperation& aggOp) { | ||
return !aggOp.dictFieldName.empty(); | ||
}); | ||
|
||
const bool hasNoDictFieldName | ||
= std::any_of(ops.begin(), ops.end(), [](const AggOperation& aggOp) { | ||
return aggOp.dictFieldName.empty(); | ||
}); | ||
|
||
if (hasDictFieldName && hasNoDictFieldName) { | ||
throw NodeException( | ||
"Inconsistent AggOperation configurations: Some operations have 'dictFieldName' " | ||
"specified while others don't."); | ||
} | ||
|
||
if (hasDictFieldName && ops.size() > 1) { | ||
throw NodeException( | ||
"Invalid AggOperation configuration: When 'dictFieldName' is specified, only one " | ||
"operation is allowed."); | ||
} | ||
} | ||
|
||
Content AggregatedFile::read() | ||
{ | ||
Content content; | ||
|
||
const auto files = getFilesMatchingPattern( | ||
m_filesRegexPattern, | ||
std::dynamic_pointer_cast<Directory>(m_parent)); | ||
|
||
std::vector<Content> fileContents; | ||
fileContents.reserve(files.size()); | ||
for (const auto& file : files) { | ||
fileContents.emplace_back(file->read()); | ||
} | ||
|
||
for (const auto& aggMethod : m_aggMethods) { | ||
const Content methodResult = aggMethod->aggregate(fileContents); | ||
mergeContent(content, methodResult); | ||
} | ||
|
||
return content; | ||
} | ||
|
||
FileOps AggregatedFile::getOps() | ||
{ | ||
FileOps ops = {}; | ||
ops.read = [this]() { return read(); }; | ||
return ops; | ||
} | ||
|
||
AggregatedFile::AggregatedFile( | ||
const std::shared_ptr<Node>& parent, | ||
std::string_view name, | ||
std::string aggFilesPattern, | ||
const std::vector<AggOperation>& ops) | ||
: File(parent, name, getOps()) | ||
, m_filesRegexPattern(std::move(aggFilesPattern)) | ||
{ | ||
validateAggOperations(ops); | ||
|
||
for (const auto& aggOp : ops) { | ||
m_aggMethods.push_back(AggMethodFactory::createAggMethod( | ||
aggOp.method, | ||
aggOp.dictFieldName, | ||
aggOp.dictResultName)); | ||
} | ||
} | ||
|
||
} // namespace telemetry |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters