Kinesin/daggy
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Adding environment interpolation for noop, forking, and slurm executors

Browse Source
This commit is contained in:
Ian Roddis
2021-11-13 12:09:51 -04:00
parent c0315b4f0b
commit 14d0ef4a3f
6 changed files with 119 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
libdaggy/src/executors/task/ForkingTaskExecutor.cpp
View File

@@ -135,8 +135,7 @@ daggy::AttemptRecord ForkingTaskExecutor::runTask(const Task &task,
close(stdoutPipe[0]);
close(stderrPipe[0]);
char **env = (envp.empty() ? nullptr : envp.data());
auto res = execvpe(argv[0], argv.data(), env);
std::cout << res << std::endl;
execvpe(argv[0], argv.data(), env);
exit(errno);
}
@@ -217,11 +216,23 @@ std::vector<daggy::ConfigValues> ForkingTaskExecutor::expandTaskParameters(
{
std::vector<ConfigValues> newValues;
const auto command = std::get<Command>(job.at("command"));
for (const auto &expandedCommand :
interpolateValues(command, expansionValues)) {
auto command =
(job.count("command") == 0 ? Command{}
: std::get<Command>(job.at("command")));
auto environment = (job.count("environment") == 0
? Command{}
: std::get<Command>(job.at("environment")));
Command both(command);
std::copy(environment.begin(), environment.end(), std::back_inserter(both));
for (const auto &parts : interpolateValues(both, expansionValues)) {
ConfigValues newCommand{job};
newCommand.at("command") = expandedCommand;
newCommand["command"] =
Command(parts.begin(), parts.begin() + command.size());
newCommand["environment"] =
Command(parts.begin() + command.size(), parts.end());
newValues.emplace_back(newCommand);
}

19
libdaggy/src/executors/task/NoopTaskExecutor.cpp
View File

@@ -31,12 +31,23 @@ namespace daggy::executors::task {
const ConfigValues &job, const ConfigValues &expansionValues)
{
std::vector<ConfigValues> newValues;
auto command =
(job.count("command") == 0 ? Command{}
: std::get<Command>(job.at("command")));
const auto command = std::get<Command>(job.at("command"));
for (const auto &expandedCommand :
interpolateValues(command, expansionValues)) {
auto environment = (job.count("environment") == 0
? Command{}
: std::get<Command>(job.at("environment")));
Command both(command);
std::copy(environment.begin(), environment.end(), std::back_inserter(both));
for (const auto &parts : interpolateValues(both, expansionValues)) {
ConfigValues newCommand{job};
newCommand.at("command") = expandedCommand;
newCommand["command"] =
Command(parts.begin(), parts.begin() + command.size());
newCommand["environment"] =
Command(parts.begin() + command.size(), parts.end());
newValues.emplace_back(newCommand);
}

20
libdaggy/src/executors/task/SlurmTaskExecutor.cpp
View File

@@ -125,11 +125,23 @@ namespace daggy::executors::task {
{
std::vector<ConfigValues> newValues;
const auto command = std::get<Command>(job.at("command"));
for (const auto &expandedCommand :
interpolateValues(command, expansionValues)) {
auto command =
(job.count("command") == 0 ? Command{}
: std::get<Command>(job.at("command")));
auto environment = (job.count("environment") == 0
? Command{}
: std::get<Command>(job.at("environment")));
Command both(command);
std::copy(environment.begin(), environment.end(), std::back_inserter(both));
for (const auto &parts : interpolateValues(both, expansionValues)) {
ConfigValues newCommand{job};
newCommand.at("command") = expandedCommand;
newCommand["command"] =
Command(parts.begin(), parts.begin() + command.size());
newCommand["environment"] =
Command(parts.begin() + command.size(), parts.end());
newValues.emplace_back(newCommand);
}

1
libdaggy/tests/CMakeLists.txt
View File

@@ -7,6 +7,7 @@ add_executable(${PROJECT_NAME} main.cpp
unit_dagrun_loggers.cpp
unit_executor_forkingexecutor.cpp
unit_executor_slurmexecutor.cpp
unit_executor_noopexecutor.cpp
unit_serialization.cpp
unit_threadpool.cpp
unit_utilities.cpp

43
libdaggy/tests/unit_executor_noopexecutor.cpp Normal file
View File

@@ -0,0 +1,43 @@
#include <catch2/catch.hpp>
#include <filesystem>
#include <fstream>
#include <iostream>
#include <thread>
#include "daggy/Serialization.hpp"
#include "daggy/Utilities.hpp"
#include "daggy/executors/task/NoopTaskExecutor.hpp"
namespace fs = std::filesystem;
TEST_CASE("noop_executor", "[noop_executor]")
{
daggy::executors::task::NoopTaskExecutor ex;
SECTION("Simple Run")
{
daggy::Task task{
.job{{"command", daggy::executors::task::NoopTaskExecutor::Command{
"/usr/bin/echo", "abc", "123"}}}};
REQUIRE(ex.validateTaskParameters(task.job));
auto recFuture = ex.execute(0, "command", task);
auto rec = recFuture.get();
REQUIRE(rec.rc == 0);
}
SECTION("Expansion with environment")
{
std::string testParams{R"({"DATE": ["2021-05-06", "2021-05-07" ]})"};
auto params = daggy::configFromJSON(testParams);
std::string taskJSON =
R"({"B": {"job": {"command": ["/usr/bin/echo", "{{DATE}}"], "environment": [ "TEST={{DATE}}"] }, "children": ["C"]}})";
auto tasks = daggy::tasksFromJSON(taskJSON);
auto result = daggy::expandTaskSet(tasks, ex, params);
REQUIRE(result.size() == 2);
}
}

40
libdaggy/tests/unit_executor_slurmexecutor.cpp
View File

@@ -13,7 +13,7 @@ namespace fs = std::filesystem;
#ifdef DAGGY_ENABLE_SLURM
TEST_CASE("slurm environment", "[slurm_env]")
TEST_CASE("slurm environment", "[slurm][slurm_env]")
{
daggy::executors::task::SlurmTaskExecutor ex;
@@ -27,7 +27,7 @@ TEST_CASE("slurm environment", "[slurm_env]")
{"tmpDir", fs::current_path().string()}};
}
TEST_CASE("slurm_execution", "[slurm_executor]")
TEST_CASE("slurm_execution", "[slurm][slurm_executor]")
{
daggy::executors::task::SlurmTaskExecutor ex;
@@ -65,23 +65,36 @@ TEST_CASE("slurm_execution", "[slurm_executor]")
if (fs::exists(scriptFile))
fs::remove_all(scriptFile);
std::string valOne = "funky_times";
std::string valTwo = "bleep_bloop";
std::string valThree = "SOME-DATE";
std::string testParams{R"({"DATE": ")" + valThree + R"("})"};
auto params = daggy::configFromJSON(testParams);
std::ofstream ofh(scriptFile);
ofh << "#!/bin/bash\necho \"${DAGGY_TEST_VAR}\"\necho "
"\"${DAGGY_TEST_VAR2}\"\n";
ofh << R"(#!/bin/bash
echo "${DAGGY_TEST_VAR}"
echo "${DAGGY_TEST_VAR2}"
echo "${DATE}"
)";
ofh.close();
fs::permissions(scriptFile, fs::perms::owner_all,
fs::perm_options::replace);
std::string valOne = "funky_times";
std::string valTwo = "bleep_bloop";
daggy::Task rawTask{.job{
{"command",
daggy::executors::task::SlurmTaskExecutor::Command{
scriptFile.string()}},
{"environment", std::vector<std::string>{"DAGGY_TEST_VAR=" + valOne,
"DAGGY_TEST_VAR2=" + valTwo,
"DATE={{DATE}}"}}}};
rawTask.job.merge(defaultJobValues);
daggy::Task task{.job{{"command",
daggy::executors::task::SlurmTaskExecutor::Command{
scriptFile.string()}},
{"environment", std::vector<std::string>{
"DAGGY_TEST_VAR=" + valOne,
"DAGGY_TEST_VAR2=" + valTwo}}}};
task.job.merge(defaultJobValues);
auto tasks =
daggy::expandTaskSet(daggy::TaskSet{{"cmd", rawTask}}, ex, params);
REQUIRE(tasks.size() == 1);
auto task = tasks.at("cmd_0");
REQUIRE(ex.validateTaskParameters(task.job));
@@ -92,6 +105,7 @@ TEST_CASE("slurm_execution", "[slurm_executor]")
REQUIRE(rec.outputLog.size() >= 6);
REQUIRE(rec.outputLog.find(valOne) != std::string::npos);
REQUIRE(rec.outputLog.find(valTwo) != std::string::npos);
REQUIRE(rec.outputLog.find(valThree) != std::string::npos);
REQUIRE(rec.errorLog.empty());
if (fs::exists(scriptFile))