Compare commits

...

2 Commits

26 changed files with 2191 additions and 387 deletions

3
.gitmodules vendored
View File

@ -4,3 +4,6 @@
[submodule "json11"] [submodule "json11"]
path = json11 path = json11
url = ../json11.git url = ../json11.git
[submodule "libnfs"]
path = libnfs
url = ../libnfs.git

1
libnfs Submodule

@ -0,0 +1 @@
Subproject commit 5a991e1fcbcdcca9a191c8e377b4cfb89142761e

View File

@ -152,9 +152,70 @@ target_link_libraries(vitastor-nbd
vitastor_client vitastor_client
) )
# vitastor-nfs
add_executable(vitastor-nfs
nfs_proxy.cpp
nfs_conn.cpp
nfs_portmap.cpp
sha256.c
../libnfs/lib/init.c
../libnfs/lib/pdu.c
../libnfs/lib/libnfs-zdr.c
../libnfs/lib/socket.c
../libnfs/portmap/libnfs-raw-portmap.c
../libnfs/nfs/libnfs-raw-nfs.c
../libnfs/mount/libnfs-raw-mount.c
)
set_source_files_properties(
../libnfs/nfs/libnfs-raw-nfs.c
PROPERTIES
COMPILE_FLAGS "-Wno-unused-but-set-variable"
)
# Simplified static configuration
# The other option is to build patched libnfs packages until all distros get my fixes
target_compile_options(vitastor-nfs
PRIVATE
-DHAVE_ARPA_INET_H
-DHAVE_INTTYPES_H
-DHAVE_MEMORY_H
-DHAVE_NETDB_H
-DHAVE_NETINET_IN_H
-DHAVE_NETINET_TCP_H
-DHAVE_NET_IF_H
-DHAVE_POLL_H
-DHAVE_STDINT_H
-DHAVE_STDLIB_H
-DHAVE_STRINGS_H
-DHAVE_STRING_H
-DHAVE_SYS_IOCTL_H
-DHAVE_SYS_SOCKET_H
-DHAVE_SYS_STATVFS_H
-DHAVE_SYS_STAT_H
-DHAVE_SYS_SYSMACROS_H
-DHAVE_SYS_TIME_H
-DHAVE_SYS_TYPES_H
-DHAVE_SYS_VFS_H
-DHAVE_UNISTD_H
-DHAVE_UTIME_H
-DHAVE_SOCKADDR_STORAGE
-DHAVE_STRUCT_STAT_ST_MTIM_TV_NSEC
-D_U_=
)
target_include_directories(vitastor-nfs
PRIVATE
../libnfs/include
../libnfs/include/nfsc
../libnfs/portmap
../libnfs/nfs
../libnfs/mount
)
target_link_libraries(vitastor-nfs
vitastor_client
)
# vitastor-cli # vitastor-cli
add_executable(vitastor-cli add_executable(vitastor-cli
cli.cpp cli_alloc_osd.cpp cli_simple_offsets.cpp cli_status.cpp cli_df.cpp cli.cpp cli_common.cpp cli_alloc_osd.cpp cli_simple_offsets.cpp cli_status.cpp cli_df.cpp
cli_ls.cpp cli_create.cpp cli_modify.cpp cli_flatten.cpp cli_merge.cpp cli_rm_data.cpp cli_rm.cpp cli_ls.cpp cli_create.cpp cli_modify.cpp cli_flatten.cpp cli_merge.cpp cli_rm_data.cpp cli_rm.cpp
) )
target_link_libraries(vitastor-cli target_link_libraries(vitastor-cli

View File

@ -2,8 +2,7 @@
// License: VNPL-1.1 (see README.md for details) // License: VNPL-1.1 (see README.md for details)
/** /**
* CLI tool * CLI tool and also a library for administrative tasks
* Currently can (a) remove inodes and (b) merge snapshot/clone layers
*/ */
#include <vector> #include <vector>
@ -17,7 +16,9 @@
static const char *exe_name = NULL; static const char *exe_name = NULL;
json11::Json::object cli_tool_t::parse_args(int narg, const char *args[]) static void help();
static json11::Json::object parse_args(int narg, const char *args[])
{ {
json11::Json::object cfg; json11::Json::object cfg;
json11::Json::array cmd; json11::Json::array cmd;
@ -79,7 +80,7 @@ json11::Json::object cli_tool_t::parse_args(int narg, const char *args[])
return cfg; return cfg;
} }
void cli_tool_t::help() static void help()
{ {
printf( printf(
"Vitastor command-line tool\n" "Vitastor command-line tool\n"
@ -164,224 +165,126 @@ void cli_tool_t::help()
exit(0); exit(0);
} }
void cli_tool_t::change_parent(inode_t cur, inode_t new_parent) static int run(cli_tool_t *p, json11::Json cfg)
{
auto cur_cfg_it = cli->st_cli.inode_config.find(cur);
if (cur_cfg_it == cli->st_cli.inode_config.end())
{
fprintf(stderr, "Inode 0x%lx disappeared\n", cur);
exit(1);
}
inode_config_t new_cfg = cur_cfg_it->second;
std::string cur_name = new_cfg.name;
std::string cur_cfg_key = base64_encode(cli->st_cli.etcd_prefix+
"/config/inode/"+std::to_string(INODE_POOL(cur))+
"/"+std::to_string(INODE_NO_POOL(cur)));
new_cfg.parent_id = new_parent;
json11::Json::object cur_cfg_json = cli->st_cli.serialize_inode_cfg(&new_cfg);
waiting++;
cli->st_cli.etcd_txn_slow(json11::Json::object {
{ "compare", json11::Json::array {
json11::Json::object {
{ "target", "MOD" },
{ "key", cur_cfg_key },
{ "result", "LESS" },
{ "mod_revision", new_cfg.mod_revision+1 },
},
} },
{ "success", json11::Json::array {
json11::Json::object {
{ "request_put", json11::Json::object {
{ "key", cur_cfg_key },
{ "value", base64_encode(json11::Json(cur_cfg_json).dump()) },
} }
},
} },
}, [this, new_parent, cur, cur_name](std::string err, json11::Json res)
{
if (err != "")
{
fprintf(stderr, "Error changing parent of %s: %s\n", cur_name.c_str(), err.c_str());
exit(1);
}
if (!res["succeeded"].bool_value())
{
fprintf(stderr, "Inode %s was modified during snapshot deletion\n", cur_name.c_str());
exit(1);
}
if (new_parent)
{
auto new_parent_it = cli->st_cli.inode_config.find(new_parent);
std::string new_parent_name = new_parent_it != cli->st_cli.inode_config.end()
? new_parent_it->second.name : "<unknown>";
printf(
"Parent of layer %s (inode %lu in pool %u) changed to %s (inode %lu in pool %u)\n",
cur_name.c_str(), INODE_NO_POOL(cur), INODE_POOL(cur),
new_parent_name.c_str(), INODE_NO_POOL(new_parent), INODE_POOL(new_parent)
);
}
else
{
printf(
"Parent of layer %s (inode %lu in pool %u) detached\n",
cur_name.c_str(), INODE_NO_POOL(cur), INODE_POOL(cur)
);
}
waiting--;
ringloop->wakeup();
});
}
void cli_tool_t::etcd_txn(json11::Json txn)
{
waiting++;
cli->st_cli.etcd_txn_slow(txn, [this](std::string err, json11::Json res)
{
waiting--;
if (err != "")
{
fprintf(stderr, "Error reading from etcd: %s\n", err.c_str());
exit(1);
}
etcd_result = res;
ringloop->wakeup();
});
}
inode_config_t* cli_tool_t::get_inode_cfg(const std::string & name)
{
for (auto & ic: cli->st_cli.inode_config)
{
if (ic.second.name == name)
{
return &ic.second;
}
}
fprintf(stderr, "Layer %s not found\n", name.c_str());
exit(1);
}
void cli_tool_t::run(json11::Json cfg)
{ {
p->parse_config(cfg);
json11::Json::array cmd = cfg["command"].array_items(); json11::Json::array cmd = cfg["command"].array_items();
std::function<bool(cli_result_t &)> action_cb;
if (!cmd.size()) if (!cmd.size())
{ {
fprintf(stderr, "command is missing\n"); fprintf(stderr, "command is missing\n");
exit(1); return EINVAL;
} }
else if (cmd[0] == "status") else if (cmd[0] == "status")
{ {
// Show cluster status // Show cluster status
action_cb = start_status(cfg); action_cb = p->start_status(cfg);
} }
else if (cmd[0] == "df") else if (cmd[0] == "df")
{ {
// Show pool space stats // Show pool space stats
action_cb = start_df(cfg); action_cb = p->start_df(cfg);
} }
else if (cmd[0] == "ls") else if (cmd[0] == "ls")
{ {
// List images // List images
action_cb = start_ls(cfg); action_cb = p->start_ls(cfg);
} }
else if (cmd[0] == "create" || cmd[0] == "snap-create") else if (cmd[0] == "create" || cmd[0] == "snap-create")
{ {
// Create image/snapshot // Create image/snapshot
action_cb = start_create(cfg); action_cb = p->start_create(cfg);
} }
else if (cmd[0] == "modify") else if (cmd[0] == "modify")
{ {
// Modify image // Modify image
action_cb = start_modify(cfg); action_cb = p->start_modify(cfg);
} }
else if (cmd[0] == "rm-data") else if (cmd[0] == "rm-data")
{ {
// Delete inode data // Delete inode data
action_cb = start_rm(cfg); action_cb = p->start_rm(cfg);
} }
else if (cmd[0] == "merge-data") else if (cmd[0] == "merge-data")
{ {
// Merge layer data without affecting metadata // Merge layer data without affecting metadata
action_cb = start_merge(cfg); action_cb = p->start_merge(cfg);
} }
else if (cmd[0] == "flatten") else if (cmd[0] == "flatten")
{ {
// Merge layer data without affecting metadata // Merge layer data without affecting metadata
action_cb = start_flatten(cfg); action_cb = p->start_flatten(cfg);
} }
else if (cmd[0] == "rm") else if (cmd[0] == "rm")
{ {
// Remove multiple snapshots and rebase their children // Remove multiple snapshots and rebase their children
action_cb = start_snap_rm(cfg); action_cb = p->start_snap_rm(cfg);
} }
else if (cmd[0] == "alloc-osd") else if (cmd[0] == "alloc-osd")
{ {
// Allocate a new OSD number // Allocate a new OSD number
action_cb = start_alloc_osd(cfg); action_cb = p->start_alloc_osd(cfg);
} }
else if (cmd[0] == "simple-offsets") else if (cmd[0] == "simple-offsets")
{ {
// Calculate offsets for simple & stupid OSD deployment without superblock // Calculate offsets for simple & stupid OSD deployment without superblock
action_cb = simple_offsets(cfg); action_cb = p->simple_offsets(cfg);
} }
else else
{ {
fprintf(stderr, "unknown command: %s\n", cmd[0].string_value().c_str()); fprintf(stderr, "unknown command: %s\n", cmd[0].string_value().c_str());
exit(1); return EINVAL;
} }
if (action_cb == NULL) if (action_cb == NULL)
{ {
return; return 0;
} }
color = !cfg["no-color"].bool_value();
json_output = cfg["json"].bool_value();
iodepth = cfg["iodepth"].uint64_value();
if (!iodepth)
iodepth = 32;
parallel_osds = cfg["parallel_osds"].uint64_value();
if (!parallel_osds)
parallel_osds = 4;
log_level = cfg["log_level"].int64_value();
progress = cfg["progress"].uint64_value() ? true : false;
list_first = cfg["wait-list"].uint64_value() ? true : false;
// Create client // Create client
ringloop = new ring_loop_t(512); p->ringloop = new ring_loop_t(512);
epmgr = new epoll_manager_t(ringloop); p->epmgr = new epoll_manager_t(p->ringloop);
cli = new cluster_client_t(ringloop, epmgr->tfd, cfg); p->cli = new cluster_client_t(p->ringloop, p->epmgr->tfd, cfg);
// Smaller timeout by default for more interactiveness // Smaller timeout by default for more interactiveness
cli->st_cli.etcd_slow_timeout = cli->st_cli.etcd_quick_timeout; p->cli->st_cli.etcd_slow_timeout = p->cli->st_cli.etcd_quick_timeout;
cli->on_ready([this]() ring_consumer_t consumer;
cli_result_t result;
p->cli->on_ready([&]()
{ {
// Initialize job // Initialize job
consumer.loop = [this]() consumer.loop = [&]()
{ {
if (action_cb != NULL) if (action_cb != NULL)
{ {
bool done = action_cb(); bool done = action_cb(result);
if (done) if (done)
{ {
action_cb = NULL; action_cb = NULL;
} }
} }
ringloop->submit(); p->ringloop->submit();
}; };
ringloop->register_consumer(&consumer); p->ringloop->register_consumer(&consumer);
consumer.loop(); consumer.loop();
}); });
// Loop until it completes // Loop until it completes
while (action_cb != NULL) while (action_cb != NULL)
{ {
ringloop->loop(); p->ringloop->loop();
if (action_cb != NULL) if (action_cb != NULL)
ringloop->wait(); p->ringloop->wait();
}
// Print result
if (result.text != "")
{
fprintf(stderr, "%s\n", result.text.c_str());
} }
// Destroy the client // Destroy the client
delete cli; delete p->cli;
delete epmgr; delete p->epmgr;
delete ringloop; delete p->ringloop;
cli = NULL; p->cli = NULL;
epmgr = NULL; p->epmgr = NULL;
ringloop = NULL; p->ringloop = NULL;
return result.err;
} }
int main(int narg, const char *args[]) int main(int narg, const char *args[])
@ -390,7 +293,7 @@ int main(int narg, const char *args[])
setvbuf(stderr, NULL, _IONBF, 0); setvbuf(stderr, NULL, _IONBF, 0);
exe_name = args[0]; exe_name = args[0];
cli_tool_t *p = new cli_tool_t(); cli_tool_t *p = new cli_tool_t();
p->run(cli_tool_t::parse_args(narg, args)); int r = run(p, parse_args(narg, args));
delete p; delete p;
return 0; return r;
} }

View File

@ -19,6 +19,13 @@ class epoll_manager_t;
class cluster_client_t; class cluster_client_t;
struct inode_config_t; struct inode_config_t;
struct cli_result_t
{
int err;
std::string text;
json11::Json data;
};
class cli_tool_t class cli_tool_t
{ {
public: public:
@ -35,33 +42,28 @@ public:
int waiting = 0; int waiting = 0;
json11::Json etcd_result; json11::Json etcd_result;
ring_consumer_t consumer;
std::function<bool(void)> action_cb;
void run(json11::Json cfg); void parse_config(json11::Json cfg);
void change_parent(inode_t cur, inode_t new_parent); void change_parent(inode_t cur, inode_t new_parent);
inode_config_t* get_inode_cfg(const std::string & name); inode_config_t* get_inode_cfg(const std::string & name);
static json11::Json::object parse_args(int narg, const char *args[]);
static void help();
friend struct rm_inode_t; friend struct rm_inode_t;
friend struct snap_merger_t; friend struct snap_merger_t;
friend struct snap_flattener_t; friend struct snap_flattener_t;
friend struct snap_remover_t; friend struct snap_remover_t;
std::function<bool(void)> start_status(json11::Json cfg); std::function<bool(cli_result_t &)> start_status(json11::Json);
std::function<bool(void)> start_df(json11::Json); std::function<bool(cli_result_t &)> start_df(json11::Json);
std::function<bool(void)> start_ls(json11::Json); std::function<bool(cli_result_t &)> start_ls(json11::Json);
std::function<bool(void)> start_create(json11::Json); std::function<bool(cli_result_t &)> start_create(json11::Json);
std::function<bool(void)> start_modify(json11::Json); std::function<bool(cli_result_t &)> start_modify(json11::Json);
std::function<bool(void)> start_rm(json11::Json); std::function<bool(cli_result_t &)> start_rm(json11::Json);
std::function<bool(void)> start_merge(json11::Json); std::function<bool(cli_result_t &)> start_merge(json11::Json);
std::function<bool(void)> start_flatten(json11::Json); std::function<bool(cli_result_t &)> start_flatten(json11::Json);
std::function<bool(void)> start_snap_rm(json11::Json); std::function<bool(cli_result_t &)> start_snap_rm(json11::Json);
std::function<bool(void)> start_alloc_osd(json11::Json cfg, uint64_t *out = NULL); std::function<bool(cli_result_t &)> start_alloc_osd(json11::Json cfg);
std::function<bool(void)> simple_offsets(json11::Json cfg); std::function<bool(cli_result_t &)> simple_offsets(json11::Json cfg);
void etcd_txn(json11::Json txn); void etcd_txn(json11::Json txn);
}; };

View File

@ -102,20 +102,20 @@ struct alloc_osd_t
} }
}; };
std::function<bool(void)> cli_tool_t::start_alloc_osd(json11::Json cfg, uint64_t *out) std::function<bool(cli_result_t &)> cli_tool_t::start_alloc_osd(json11::Json cfg)
{ {
json11::Json::array cmd = cfg["command"].array_items(); json11::Json::array cmd = cfg["command"].array_items();
auto alloc_osd = new alloc_osd_t(); auto alloc_osd = new alloc_osd_t();
alloc_osd->parent = this; alloc_osd->parent = this;
return [alloc_osd, out]() return [alloc_osd](cli_result_t & result)
{ {
alloc_osd->loop(); alloc_osd->loop();
if (alloc_osd->is_done()) if (alloc_osd->is_done())
{ {
if (out) result = (cli_result_t){
*out = alloc_osd->new_id; .text = std::to_string(alloc_osd->new_id),
else if (alloc_osd->new_id) .data = json11::Json(alloc_osd->new_id),
printf("%lu\n", alloc_osd->new_id); };
delete alloc_osd; delete alloc_osd;
return true; return true;
} }

118
src/cli_common.cpp Normal file
View File

@ -0,0 +1,118 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
#include "base64.h"
#include "cluster_client.h"
#include "cli.h"
void cli_tool_t::change_parent(inode_t cur, inode_t new_parent)
{
auto cur_cfg_it = cli->st_cli.inode_config.find(cur);
if (cur_cfg_it == cli->st_cli.inode_config.end())
{
fprintf(stderr, "Inode 0x%lx disappeared\n", cur);
exit(1);
}
inode_config_t new_cfg = cur_cfg_it->second;
std::string cur_name = new_cfg.name;
std::string cur_cfg_key = base64_encode(cli->st_cli.etcd_prefix+
"/config/inode/"+std::to_string(INODE_POOL(cur))+
"/"+std::to_string(INODE_NO_POOL(cur)));
new_cfg.parent_id = new_parent;
json11::Json::object cur_cfg_json = cli->st_cli.serialize_inode_cfg(&new_cfg);
waiting++;
cli->st_cli.etcd_txn_slow(json11::Json::object {
{ "compare", json11::Json::array {
json11::Json::object {
{ "target", "MOD" },
{ "key", cur_cfg_key },
{ "result", "LESS" },
{ "mod_revision", new_cfg.mod_revision+1 },
},
} },
{ "success", json11::Json::array {
json11::Json::object {
{ "request_put", json11::Json::object {
{ "key", cur_cfg_key },
{ "value", base64_encode(json11::Json(cur_cfg_json).dump()) },
} }
},
} },
}, [this, new_parent, cur, cur_name](std::string err, json11::Json res)
{
if (err != "")
{
fprintf(stderr, "Error changing parent of %s: %s\n", cur_name.c_str(), err.c_str());
exit(1);
}
if (!res["succeeded"].bool_value())
{
fprintf(stderr, "Inode %s was modified during snapshot deletion\n", cur_name.c_str());
exit(1);
}
if (new_parent)
{
auto new_parent_it = cli->st_cli.inode_config.find(new_parent);
std::string new_parent_name = new_parent_it != cli->st_cli.inode_config.end()
? new_parent_it->second.name : "<unknown>";
printf(
"Parent of layer %s (inode %lu in pool %u) changed to %s (inode %lu in pool %u)\n",
cur_name.c_str(), INODE_NO_POOL(cur), INODE_POOL(cur),
new_parent_name.c_str(), INODE_NO_POOL(new_parent), INODE_POOL(new_parent)
);
}
else
{
printf(
"Parent of layer %s (inode %lu in pool %u) detached\n",
cur_name.c_str(), INODE_NO_POOL(cur), INODE_POOL(cur)
);
}
waiting--;
ringloop->wakeup();
});
}
void cli_tool_t::etcd_txn(json11::Json txn)
{
waiting++;
cli->st_cli.etcd_txn_slow(txn, [this](std::string err, json11::Json res)
{
waiting--;
if (err != "")
{
fprintf(stderr, "Error reading from etcd: %s\n", err.c_str());
exit(1);
}
etcd_result = res;
ringloop->wakeup();
});
}
inode_config_t* cli_tool_t::get_inode_cfg(const std::string & name)
{
for (auto & ic: cli->st_cli.inode_config)
{
if (ic.second.name == name)
{
return &ic.second;
}
}
fprintf(stderr, "Layer %s not found\n", name.c_str());
exit(1);
}
void cli_tool_t::parse_config(json11::Json cfg)
{
color = !cfg["no-color"].bool_value();
json_output = cfg["json"].bool_value();
iodepth = cfg["iodepth"].uint64_value();
if (!iodepth)
iodepth = 32;
parallel_osds = cfg["parallel_osds"].uint64_value();
if (!parallel_osds)
parallel_osds = 4;
log_level = cfg["log_level"].int64_value();
progress = cfg["progress"].uint64_value() ? true : false;
list_first = cfg["wait-list"].uint64_value() ? true : false;
}

View File

@ -33,6 +33,7 @@ struct image_creator_t
uint64_t max_id_mod_rev = 0, cfg_mod_rev = 0, idx_mod_rev = 0; uint64_t max_id_mod_rev = 0, cfg_mod_rev = 0, idx_mod_rev = 0;
int state = 0; int state = 0;
cli_result_t result;
bool is_done() bool is_done()
{ {
@ -43,13 +44,27 @@ struct image_creator_t
{ {
if (state >= 1) if (state >= 1)
goto resume_1; goto resume_1;
if (image_name == "")
{
// FIXME: EINVAL -> specific codes for every error
result = (cli_result_t){ .err = EINVAL, .text = "Image name is missing" };
state = 100;
return;
}
if (image_name.find('@') != std::string::npos)
{
result = (cli_result_t){ .err = EINVAL, .text = "Image name can't contain @ character" };
state = 100;
return;
}
if (new_pool_id) if (new_pool_id)
{ {
auto & pools = parent->cli->st_cli.pool_config; auto & pools = parent->cli->st_cli.pool_config;
if (pools.find(new_pool_id) == pools.end()) if (pools.find(new_pool_id) == pools.end())
{ {
fprintf(stderr, "Pool %u does not exist\n", new_pool_id); result = (cli_result_t){ .err = ENOENT, .text = "Pool "+std::to_string(new_pool_id)+" does not exist" };
exit(1); state = 100;
return;
} }
} }
else if (new_pool_name != "") else if (new_pool_name != "")
@ -64,8 +79,9 @@ struct image_creator_t
} }
if (!new_pool_id) if (!new_pool_id)
{ {
fprintf(stderr, "Pool %s does not exist\n", new_pool_name.c_str()); result = (cli_result_t){ .err = ENOENT, .text = "Pool "+new_pool_name+" does not exist" };
exit(1); state = 100;
return;
} }
} }
else if (parent->cli->st_cli.pool_config.size() == 1) else if (parent->cli->st_cli.pool_config.size() == 1)
@ -91,8 +107,9 @@ struct image_creator_t
{ {
if (ic.second.name == image_name) if (ic.second.name == image_name)
{ {
fprintf(stderr, "Image %s already exists\n", image_name.c_str()); result = (cli_result_t){ .err = EEXIST, .text = "Image "+image_name+" already exists" };
exit(1); state = 100;
return;
} }
if (ic.second.name == new_parent) if (ic.second.name == new_parent)
{ {
@ -109,18 +126,21 @@ struct image_creator_t
} }
if (new_parent != "" && !new_parent_id) if (new_parent != "" && !new_parent_id)
{ {
fprintf(stderr, "Parent image not found\n"); result = (cli_result_t){ .err = ENOENT, .text = "Parent image "+new_parent+" not found" };
exit(1); state = 100;
return;
} }
if (!new_pool_id) if (!new_pool_id)
{ {
fprintf(stderr, "Pool name or ID is missing\n"); result = (cli_result_t){ .err = EINVAL, .text = "Pool name or ID is missing" };
exit(1); state = 100;
return;
} }
if (!size) if (!size)
{ {
fprintf(stderr, "Image size is missing\n"); result = (cli_result_t){ .err = EINVAL, .text = "Image size is missing" };
exit(1); state = 100;
return;
} }
do do
{ {
@ -140,14 +160,12 @@ resume_3:
if (!parent->etcd_result["succeeded"].bool_value() && if (!parent->etcd_result["succeeded"].bool_value() &&
parent->etcd_result["responses"][0]["response_range"]["kvs"].array_items().size() > 0) parent->etcd_result["responses"][0]["response_range"]["kvs"].array_items().size() > 0)
{ {
fprintf(stderr, "Image %s already exists\n", image_name.c_str()); result = (cli_result_t){ .err = EEXIST, .text = "Image "+image_name+" already exists" };
exit(1); state = 100;
return;
} }
} while (!parent->etcd_result["succeeded"].bool_value()); } while (!parent->etcd_result["succeeded"].bool_value());
if (parent->progress) result = (cli_result_t){ .err = 0, .text = "Image "+image_name+" created" };
{
printf("Image %s created\n", image_name.c_str());
}
state = 100; state = 100;
} }
@ -163,14 +181,16 @@ resume_3:
{ {
if (ic.second.name == image_name+"@"+new_snap) if (ic.second.name == image_name+"@"+new_snap)
{ {
fprintf(stderr, "Snapshot %s@%s already exists\n", image_name.c_str(), new_snap.c_str()); result = (cli_result_t){ .err = EEXIST, .text = "Snapshot "+image_name+"@"+new_snap+" already exists" };
exit(1); state = 100;
return;
} }
} }
if (new_parent != "") if (new_parent != "")
{ {
fprintf(stderr, "--parent can't be used with snapshots\n"); result = (cli_result_t){ .err = EINVAL, .text = "Parent can't be specified for snapshots" };
exit(1); state = 100;
return;
} }
do do
{ {
@ -182,8 +202,9 @@ resume_3:
return; return;
if (!old_id) if (!old_id)
{ {
fprintf(stderr, "Image %s does not exist\n", image_name.c_str()); result = (cli_result_t){ .err = ENOENT, .text = "Image "+image_name+" does not exist" };
exit(1); state = 100;
return;
} }
if (!new_pool_id) if (!new_pool_id)
{ {
@ -198,14 +219,12 @@ resume_4:
if (!parent->etcd_result["succeeded"].bool_value() && if (!parent->etcd_result["succeeded"].bool_value() &&
parent->etcd_result["responses"][0]["response_range"]["kvs"].array_items().size() > 0) parent->etcd_result["responses"][0]["response_range"]["kvs"].array_items().size() > 0)
{ {
fprintf(stderr, "Snapshot %s@%s already exists\n", image_name.c_str(), new_snap.c_str()); result = (cli_result_t){ .err = EEXIST, .text = "Snapshot "+image_name+"@"+new_snap+" already exists" };
exit(1); state = 100;
return;
} }
} while (!parent->etcd_result["succeeded"].bool_value()); } while (!parent->etcd_result["succeeded"].bool_value());
if (parent->progress) result = (cli_result_t){ .err = 0, .text = "Snapshot "+image_name+"@"+new_snap+" created" };
{
printf("Snapshot %s@%s created\n", image_name.c_str(), new_snap.c_str());
}
state = 100; state = 100;
} }
@ -288,8 +307,9 @@ resume_2:
idx_mod_rev = kv.mod_revision; idx_mod_rev = kv.mod_revision;
if (!old_id || !old_pool_id || old_pool_id >= POOL_ID_MAX) if (!old_id || !old_pool_id || old_pool_id >= POOL_ID_MAX)
{ {
fprintf(stderr, "Invalid pool or inode ID in etcd key %s\n", kv.key.c_str()); result = (cli_result_t){ .err = ENOENT, .text = "Invalid pool or inode ID in etcd key "+kv.key };
exit(1); state = 100;
return;
} }
} }
parent->etcd_txn(json11::Json::object { parent->etcd_txn(json11::Json::object {
@ -469,13 +489,12 @@ uint64_t parse_size(std::string size_str)
uint64_t size = json11::Json(size_str).uint64_value() * mul; uint64_t size = json11::Json(size_str).uint64_value() * mul;
if (size == 0 && size_str != "0" && (size_str != "" || mul != 1)) if (size == 0 && size_str != "0" && (size_str != "" || mul != 1))
{ {
fprintf(stderr, "Invalid syntax for size: %s\n", size_str.c_str()); return UINT64_MAX;
exit(1);
} }
return size; return size;
} }
std::function<bool(void)> cli_tool_t::start_create(json11::Json cfg) std::function<bool(cli_result_t &)> cli_tool_t::start_create(json11::Json cfg)
{ {
json11::Json::array cmd = cfg["command"].array_items(); json11::Json::array cmd = cfg["command"].array_items();
auto image_creator = new image_creator_t(); auto image_creator = new image_creator_t();
@ -492,8 +511,12 @@ std::function<bool(void)> cli_tool_t::start_create(json11::Json cfg)
int p = image_creator->image_name.find('@'); int p = image_creator->image_name.find('@');
if (p == std::string::npos || p == image_creator->image_name.length()-1) if (p == std::string::npos || p == image_creator->image_name.length()-1)
{ {
fprintf(stderr, "Please specify new snapshot name after @\n"); delete image_creator;
exit(1); return [](cli_result_t & result)
{
result = (cli_result_t){ .err = EINVAL, .text = "Please specify new snapshot name after @" };
return true;
};
} }
image_creator->new_snap = image_creator->image_name.substr(p + 1); image_creator->new_snap = image_creator->image_name.substr(p + 1);
image_creator->image_name = image_creator->image_name.substr(0, p); image_creator->image_name = image_creator->image_name.substr(0, p);
@ -502,32 +525,39 @@ std::function<bool(void)> cli_tool_t::start_create(json11::Json cfg)
if (cfg["size"].string_value() != "") if (cfg["size"].string_value() != "")
{ {
image_creator->size = parse_size(cfg["size"].string_value()); image_creator->size = parse_size(cfg["size"].string_value());
if (image_creator->size == UINT64_MAX)
{
return [size = cfg["size"].string_value()](cli_result_t & result)
{
result = (cli_result_t){ .err = EINVAL, .text = "Invalid syntax for size: "+size };
return true;
};
}
if (image_creator->size % 4096) if (image_creator->size % 4096)
{ {
fprintf(stderr, "Size should be a multiple of 4096\n"); delete image_creator;
exit(1); return [](cli_result_t & result)
{
result = (cli_result_t){ .err = EINVAL, .text = "Size should be a multiple of 4096" };
return true;
};
} }
if (image_creator->new_snap != "") if (image_creator->new_snap != "")
{ {
fprintf(stderr, "--size can't be specified for snapshots\n"); delete image_creator;
exit(1); return [](cli_result_t & result)
{
result = (cli_result_t){ .err = EINVAL, .text = "Size can't be specified for snapshots" };
return true;
};
} }
} }
if (image_creator->image_name == "") return [image_creator](cli_result_t & result)
{
fprintf(stderr, "Image name is missing\n");
exit(1);
}
if (image_creator->image_name.find('@') != std::string::npos)
{
fprintf(stderr, "Image name can't contain @ character\n");
exit(1);
}
return [image_creator]()
{ {
image_creator->loop(); image_creator->loop();
if (image_creator->is_done()) if (image_creator->is_done())
{ {
result = image_creator->result;
delete image_creator; delete image_creator;
return true; return true;
} }

View File

@ -12,6 +12,7 @@ struct pool_lister_t
int state = 0; int state = 0;
json11::Json space_info; json11::Json space_info;
cli_result_t result;
std::map<pool_id_t, json11::Json::object> pool_stats; std::map<pool_id_t, json11::Json::object> pool_stats;
bool is_done() bool is_done()
@ -150,10 +151,10 @@ resume_1:
get_stats(); get_stats();
if (parent->waiting > 0) if (parent->waiting > 0)
return; return;
result.data = to_list();
if (parent->json_output) if (parent->json_output)
{ {
// JSON output // JSON output
printf("%s\n", json11::Json(to_list()).dump().c_str());
state = 100; state = 100;
return; return;
} }
@ -206,21 +207,22 @@ resume_1:
: 100)+"%"; : 100)+"%";
kv.second["eff_fmt"] = format_q(kv.second["space_efficiency"].number_value()*100)+"%"; kv.second["eff_fmt"] = format_q(kv.second["space_efficiency"].number_value()*100)+"%";
} }
printf("%s", print_table(to_list(), cols, parent->color).c_str()); result.text = print_table(result.data, cols, parent->color);
state = 100; state = 100;
} }
}; };
std::function<bool(void)> cli_tool_t::start_df(json11::Json cfg) std::function<bool(cli_result_t &)> cli_tool_t::start_df(json11::Json cfg)
{ {
json11::Json::array cmd = cfg["command"].array_items(); json11::Json::array cmd = cfg["command"].array_items();
auto lister = new pool_lister_t(); auto lister = new pool_lister_t();
lister->parent = this; lister->parent = this;
return [lister]() return [lister](cli_result_t & result)
{ {
lister->loop(); lister->loop();
if (lister->is_done()) if (lister->is_done())
{ {
result = lister->result;
delete lister; delete lister;
return true; return true;
} }

View File

@ -22,7 +22,8 @@ struct snap_flattener_t
std::string top_parent_name; std::string top_parent_name;
inode_t target_id = 0; inode_t target_id = 0;
int state = 0; int state = 0;
std::function<bool(void)> merger_cb; std::function<bool(cli_result_t &)> merger_cb;
cli_result_t result;
void get_merge_parents() void get_merge_parents()
{ {
@ -37,23 +38,34 @@ struct snap_flattener_t
auto it = parent->cli->st_cli.inode_config.find(cur->parent_id); auto it = parent->cli->st_cli.inode_config.find(cur->parent_id);
if (it == parent->cli->st_cli.inode_config.end()) if (it == parent->cli->st_cli.inode_config.end())
{ {
fprintf(stderr, "Parent inode of layer %s (id %ld) not found\n", cur->name.c_str(), cur->parent_id); result = (cli_result_t){
exit(1); .err = ENOENT,
.text = "Parent inode of layer "+cur->name+" (id "+std::to_string(cur->parent_id)+") does not exist",
.data = json11::Json::object {
{ "error", "parent-not-found" },
{ "inode_id", cur->num },
{ "inode_name", cur->name },
{ "parent_id", cur->parent_id },
},
};
state = 100;
return;
} }
cur = &it->second; cur = &it->second;
chain_list.push_back(cur->num); chain_list.push_back(cur->num);
} }
if (cur->parent_id != 0) if (cur->parent_id != 0)
{ {
fprintf(stderr, "Layer %s has a loop in parents\n", target_name.c_str()); result = (cli_result_t){ .err = EBADF, .text = "Layer "+target_name+" has a loop in parents" };
exit(1); state = 100;
return;
} }
top_parent_name = cur->name; top_parent_name = cur->name;
} }
bool is_done() bool is_done()
{ {
return state == 5; return state == 100;
} }
void loop() void loop()
@ -64,8 +76,16 @@ struct snap_flattener_t
goto resume_2; goto resume_2;
else if (state == 3) else if (state == 3)
goto resume_3; goto resume_3;
if (target_name == "")
{
result = (cli_result_t){ .err = EINVAL, .text = "Layer to flatten not specified" };
state = 100;
return;
}
// Get parent layers // Get parent layers
get_merge_parents(); get_merge_parents();
if (state == 100)
return;
// Start merger // Start merger
merger_cb = parent->start_merge(json11::Json::object { merger_cb = parent->start_merge(json11::Json::object {
{ "command", json11::Json::array{ "merge-data", top_parent_name, target_name } }, { "command", json11::Json::array{ "merge-data", top_parent_name, target_name } },
@ -76,12 +96,17 @@ struct snap_flattener_t
}); });
// Wait for it // Wait for it
resume_1: resume_1:
while (!merger_cb()) while (!merger_cb(result))
{ {
state = 1; state = 1;
return; return;
} }
merger_cb = NULL; merger_cb = NULL;
if (result.err)
{
state = 100;
return;
}
// Change parent // Change parent
parent->change_parent(target_id, 0); parent->change_parent(target_id, 0);
// Wait for it to complete // Wait for it to complete
@ -92,31 +117,27 @@ resume_2:
state = 3; state = 3;
resume_3: resume_3:
// Done // Done
return; state = 100;
} }
}; };
std::function<bool(void)> cli_tool_t::start_flatten(json11::Json cfg) std::function<bool(cli_result_t &)> cli_tool_t::start_flatten(json11::Json cfg)
{ {
json11::Json::array cmd = cfg["command"].array_items(); json11::Json::array cmd = cfg["command"].array_items();
auto flattener = new snap_flattener_t(); auto flattener = new snap_flattener_t();
flattener->parent = this; flattener->parent = this;
flattener->target_name = cmd.size() > 1 ? cmd[1].string_value() : ""; flattener->target_name = cmd.size() > 1 ? cmd[1].string_value() : "";
if (flattener->target_name == "")
{
fprintf(stderr, "Layer to flatten argument is missing\n");
exit(1);
}
flattener->fsync_interval = cfg["fsync-interval"].uint64_value(); flattener->fsync_interval = cfg["fsync-interval"].uint64_value();
if (!flattener->fsync_interval) if (!flattener->fsync_interval)
flattener->fsync_interval = 128; flattener->fsync_interval = 128;
if (!cfg["cas"].is_null()) if (!cfg["cas"].is_null())
flattener->use_cas = cfg["cas"].uint64_value() ? 2 : 0; flattener->use_cas = cfg["cas"].uint64_value() ? 2 : 0;
return [flattener]() return [flattener](cli_result_t & result)
{ {
flattener->loop(); flattener->loop();
if (flattener->is_done()) if (flattener->is_done())
{ {
result = flattener->result;
delete flattener; delete flattener;
return true; return true;
} }

View File

@ -24,6 +24,7 @@ struct image_lister_t
int state = 0; int state = 0;
std::map<inode_t, json11::Json::object> stats; std::map<inode_t, json11::Json::object> stats;
json11::Json space_info; json11::Json space_info;
cli_result_t result;
bool is_done() bool is_done()
{ {
@ -44,8 +45,9 @@ struct image_lister_t
} }
if (!list_pool_id) if (!list_pool_id)
{ {
fprintf(stderr, "Pool %s does not exist\n", list_pool_name.c_str()); result = (cli_result_t){ .err = ENOENT, .text = "Pool "+list_pool_name+" does not exist" };
exit(1); state = 100;
return;
} }
} }
for (auto & ic: parent->cli->st_cli.inode_config) for (auto & ic: parent->cli->st_cli.inode_config)
@ -246,10 +248,10 @@ resume_1:
if (parent->waiting > 0) if (parent->waiting > 0)
return; return;
} }
result.data = to_list();
if (parent->json_output) if (parent->json_output)
{ {
// JSON output // JSON output
printf("%s\n", json11::Json(to_list()).dump().c_str());
state = 100; state = 100;
return; return;
} }
@ -359,7 +361,7 @@ resume_1:
kv.second["size_fmt"] = format_size(kv.second["size"].uint64_value()); kv.second["size_fmt"] = format_size(kv.second["size"].uint64_value());
kv.second["ro"] = kv.second["readonly"].bool_value() ? "RO" : "-"; kv.second["ro"] = kv.second["readonly"].bool_value() ? "RO" : "-";
} }
printf("%s", print_table(to_list(), cols, parent->color).c_str()); result.text = print_table(to_list(), cols, parent->color);
state = 100; state = 100;
} }
}; };
@ -546,7 +548,7 @@ back:
return true; return true;
} }
std::function<bool(void)> cli_tool_t::start_ls(json11::Json cfg) std::function<bool(cli_result_t &)> cli_tool_t::start_ls(json11::Json cfg)
{ {
json11::Json::array cmd = cfg["command"].array_items(); json11::Json::array cmd = cfg["command"].array_items();
auto lister = new image_lister_t(); auto lister = new image_lister_t();
@ -562,11 +564,12 @@ std::function<bool(void)> cli_tool_t::start_ls(json11::Json cfg)
{ {
lister->only_names.insert(cmd[i].string_value()); lister->only_names.insert(cmd[i].string_value());
} }
return [lister]() return [lister](cli_result_t & result)
{ {
lister->loop(); lister->loop();
if (lister->is_done()) if (lister->is_done())
{ {
result = lister->result;
delete lister; delete lister;
return true; return true;
} }

View File

@ -12,6 +12,9 @@ struct snap_rw_op_t
cluster_op_t op; cluster_op_t op;
int todo = 0; int todo = 0;
uint32_t start = 0, end = 0; uint32_t start = 0, end = 0;
int error_code = 0;
uint64_t error_offset = 0;
bool error_read = false;
}; };
// Layer merge is the base for multiple operations: // Layer merge is the base for multiple operations:
@ -54,17 +57,27 @@ struct snap_merger_t
uint64_t last_written_offset = 0; uint64_t last_written_offset = 0;
int deleted_unsynced = 0; int deleted_unsynced = 0;
uint64_t processed = 0, to_process = 0; uint64_t processed = 0, to_process = 0;
std::string rwo_error;
cli_result_t result;
void start_merge() void start_merge()
{ {
if (from_name == "" || to_name == "")
{
result = (cli_result_t){ .err = EINVAL, .text = "Beginning or end of the merge sequence is missing" };
state = 100;
return;
}
check_delete_source = delete_source || check_delete_source; check_delete_source = delete_source || check_delete_source;
inode_config_t *from_cfg = parent->get_inode_cfg(from_name); inode_config_t *from_cfg = parent->get_inode_cfg(from_name);
inode_config_t *to_cfg = parent->get_inode_cfg(to_name); inode_config_t *to_cfg = parent->get_inode_cfg(to_name);
inode_config_t *target_cfg = target_name == "" ? from_cfg : parent->get_inode_cfg(target_name); inode_config_t *target_cfg = target_name == "" ? from_cfg : parent->get_inode_cfg(target_name);
if (to_cfg->num == from_cfg->num) if (to_cfg->num == from_cfg->num)
{ {
fprintf(stderr, "Only one layer specified, nothing to merge\n"); result = (cli_result_t){ .err = EINVAL, .text = "Only one layer specified, nothing to merge" };
exit(1); state = 100;
return;
} }
// Check that to_cfg is actually a child of from_cfg and target_cfg is somewhere between them // Check that to_cfg is actually a child of from_cfg and target_cfg is somewhere between them
std::vector<inode_t> chain_list; std::vector<inode_t> chain_list;
@ -78,8 +91,18 @@ struct snap_merger_t
auto it = parent->cli->st_cli.inode_config.find(cur->parent_id); auto it = parent->cli->st_cli.inode_config.find(cur->parent_id);
if (it == parent->cli->st_cli.inode_config.end()) if (it == parent->cli->st_cli.inode_config.end())
{ {
fprintf(stderr, "Parent inode of layer %s (id %ld) not found\n", cur->name.c_str(), cur->parent_id); result = (cli_result_t){
exit(1); .err = ENOENT,
.text = "Parent inode of layer "+cur->name+" (id "+std::to_string(cur->parent_id)+") does not exist",
.data = json11::Json::object {
{ "error", "parent-not-found" },
{ "inode_id", cur->num },
{ "inode_name", cur->name },
{ "parent_id", cur->parent_id },
},
};
state = 100;
return;
} }
cur = &it->second; cur = &it->second;
chain_list.push_back(cur->num); chain_list.push_back(cur->num);
@ -87,8 +110,9 @@ struct snap_merger_t
} }
if (cur->parent_id != from_cfg->num) if (cur->parent_id != from_cfg->num)
{ {
fprintf(stderr, "Layer %s is not a child of %s\n", to_name.c_str(), from_name.c_str()); result = (cli_result_t){ .err = EINVAL, .text = "Layer "+to_name+" is not a child of "+from_name };
exit(1); state = 100;
return;
} }
chain_list.push_back(from_cfg->num); chain_list.push_back(from_cfg->num);
layer_block_size[from_cfg->num] = get_block_size(from_cfg->num); layer_block_size[from_cfg->num] = get_block_size(from_cfg->num);
@ -99,8 +123,9 @@ struct snap_merger_t
} }
if (sources.find(target_cfg->num) == sources.end()) if (sources.find(target_cfg->num) == sources.end())
{ {
fprintf(stderr, "Layer %s is not between %s and %s\n", target_name.c_str(), to_name.c_str(), from_name.c_str()); result = (cli_result_t){ .err = EINVAL, .text = "Layer "+target_name+" is not between "+to_name+" and "+from_name };
exit(1); state = 100;
return;
} }
target = target_cfg->num; target = target_cfg->num;
target_rank = sources.at(target); target_rank = sources.at(target);
@ -130,14 +155,15 @@ struct snap_merger_t
int parent_rank = it->second; int parent_rank = it->second;
if (parent_rank < to_rank && (parent_rank >= target_rank || check_delete_source)) if (parent_rank < to_rank && (parent_rank >= target_rank || check_delete_source))
{ {
fprintf( result = (cli_result_t){
stderr, "Layers at or above %s, but below %s are not allowed" .err = EINVAL,
" to have other children, but %s is a child of %s\n", .text = "Layers at or above "+(check_delete_source ? from_name : target_name)+
(check_delete_source ? from_name.c_str() : target_name.c_str()), ", but below "+to_name+" are not allowed to have other children, but "+
to_name.c_str(), ic.second.name.c_str(), ic.second.name+" is a child of "+
parent->cli->st_cli.inode_config.at(ic.second.parent_id).name.c_str() parent->cli->st_cli.inode_config.at(ic.second.parent_id).name,
); };
exit(1); state = 100;
return;
} }
if (parent_rank >= to_rank) if (parent_rank >= to_rank)
{ {
@ -152,11 +178,14 @@ struct snap_merger_t
use_cas = 0; use_cas = 0;
} }
sources.erase(target); sources.erase(target);
printf( if (parent->progress)
"Merging %ld layer(s) into target %s%s (inode %lu in pool %u)\n", {
sources.size(), target_cfg->name.c_str(), printf(
use_cas ? " online (with CAS)" : "", INODE_NO_POOL(target), INODE_POOL(target) "Merging %ld layer(s) into target %s%s (inode %lu in pool %u)\n",
); sources.size(), target_cfg->name.c_str(),
use_cas ? " online (with CAS)" : "", INODE_NO_POOL(target), INODE_POOL(target)
);
}
target_block_size = get_block_size(target); target_block_size = get_block_size(target);
} }
@ -253,7 +282,8 @@ struct snap_merger_t
oit = merge_offsets.begin(); oit = merge_offsets.begin();
resume_5: resume_5:
// Now read, overwrite and optionally delete offsets one by one // Now read, overwrite and optionally delete offsets one by one
while (in_flight < parent->iodepth*parent->parallel_osds && oit != merge_offsets.end()) while (in_flight < parent->iodepth*parent->parallel_osds &&
oit != merge_offsets.end() && !rwo_error.size())
{ {
in_flight++; in_flight++;
read_and_write(*oit); read_and_write(*oit);
@ -264,6 +294,15 @@ struct snap_merger_t
printf("\rOverwriting blocks: %lu/%lu", processed, to_process); printf("\rOverwriting blocks: %lu/%lu", processed, to_process);
} }
} }
if (in_flight == 0 && rwo_error.size())
{
result = (cli_result_t){
.err = EIO,
.text = rwo_error,
};
state = 100;
return;
}
if (in_flight > 0 || oit != merge_offsets.end()) if (in_flight > 0 || oit != merge_offsets.end())
{ {
// Wait until overwrites finish // Wait until overwrites finish
@ -396,8 +435,9 @@ struct snap_merger_t
{ {
if (op->retval != op->len) if (op->retval != op->len)
{ {
fprintf(stderr, "error reading target at offset %lx: %s\n", op->offset, strerror(-op->retval)); rwo->error_code = -op->retval;
exit(1); rwo->error_offset = op->offset;
rwo->error_read = true;
} }
next_write(rwo); next_write(rwo);
}; };
@ -410,7 +450,7 @@ struct snap_merger_t
// FIXME: Allow to use single write with "holes" (OSDs don't allow it yet) // FIXME: Allow to use single write with "holes" (OSDs don't allow it yet)
uint32_t gran = parent->cli->get_bs_bitmap_granularity(); uint32_t gran = parent->cli->get_bs_bitmap_granularity();
uint64_t bitmap_size = target_block_size / gran; uint64_t bitmap_size = target_block_size / gran;
while (rwo->end < bitmap_size) while (rwo->end < bitmap_size && !rwo->error_code)
{ {
auto bit = ((*((uint8_t*)rwo->op.bitmap_buf + (rwo->end >> 3))) & (1 << (rwo->end & 0x7))); auto bit = ((*((uint8_t*)rwo->op.bitmap_buf + (rwo->end >> 3))) & (1 << (rwo->end & 0x7)));
if (!bit) if (!bit)
@ -434,7 +474,7 @@ struct snap_merger_t
rwo->end++; rwo->end++;
} }
} }
if (rwo->end > rwo->start) if (rwo->end > rwo->start && !rwo->error_code)
{ {
// write start->end // write start->end
rwo->todo++; rwo->todo++;
@ -473,8 +513,9 @@ struct snap_merger_t
delete subop; delete subop;
return; return;
} }
fprintf(stderr, "error writing target at offset %lx: %s\n", subop->offset, strerror(-subop->retval)); rwo->error_code = -subop->retval;
exit(1); rwo->error_offset = subop->offset;
rwo->error_read = false;
} }
// Increment CAS version // Increment CAS version
rwo->op.version++; rwo->op.version++;
@ -510,11 +551,12 @@ struct snap_merger_t
{ {
if (!rwo->todo) if (!rwo->todo)
{ {
if (last_written_offset < rwo->op.offset+target_block_size) if (!rwo->error_code &&
last_written_offset < rwo->op.offset+target_block_size)
{ {
last_written_offset = rwo->op.offset+target_block_size; last_written_offset = rwo->op.offset+target_block_size;
} }
if (delete_source) if (!rwo->error_code && delete_source)
{ {
deleted_unsynced++; deleted_unsynced++;
if (deleted_unsynced >= fsync_interval) if (deleted_unsynced >= fsync_interval)
@ -545,13 +587,20 @@ struct snap_merger_t
} }
free(rwo->buf); free(rwo->buf);
delete rwo; delete rwo;
if (rwo->error_code)
{
char buf[1024];
snprintf(buf, 1024, "Error %s target at offset %lx: %s",
rwo->error_read ? "reading" : "writing", rwo->error_offset, strerror(rwo->error_code));
rwo_error = std::string(buf);
}
in_flight--; in_flight--;
continue_merge_reent(); continue_merge_reent();
} }
} }
}; };
std::function<bool(void)> cli_tool_t::start_merge(json11::Json cfg) std::function<bool(cli_result_t &)> cli_tool_t::start_merge(json11::Json cfg)
{ {
json11::Json::array cmd = cfg["command"].array_items(); json11::Json::array cmd = cfg["command"].array_items();
auto merger = new snap_merger_t(); auto merger = new snap_merger_t();
@ -559,22 +608,18 @@ std::function<bool(void)> cli_tool_t::start_merge(json11::Json cfg)
merger->from_name = cmd.size() > 1 ? cmd[1].string_value() : ""; merger->from_name = cmd.size() > 1 ? cmd[1].string_value() : "";
merger->to_name = cmd.size() > 2 ? cmd[2].string_value() : ""; merger->to_name = cmd.size() > 2 ? cmd[2].string_value() : "";
merger->target_name = cfg["target"].string_value(); merger->target_name = cfg["target"].string_value();
if (merger->from_name == "" || merger->to_name == "")
{
fprintf(stderr, "Beginning or end of the merge sequence is missing\n");
exit(1);
}
merger->delete_source = cfg["delete-source"].string_value() != ""; merger->delete_source = cfg["delete-source"].string_value() != "";
merger->fsync_interval = cfg["fsync-interval"].uint64_value(); merger->fsync_interval = cfg["fsync-interval"].uint64_value();
if (!merger->fsync_interval) if (!merger->fsync_interval)
merger->fsync_interval = 128; merger->fsync_interval = 128;
if (!cfg["cas"].is_null()) if (!cfg["cas"].is_null())
merger->use_cas = cfg["cas"].uint64_value() ? 2 : 0; merger->use_cas = cfg["cas"].uint64_value() ? 2 : 0;
return [merger]() return [merger](cli_result_t & result)
{ {
merger->continue_merge_reent(); merger->continue_merge_reent();
if (merger->is_done()) if (merger->is_done())
{ {
result = merger->result;
delete merger; delete merger;
return true; return true;
} }

View File

@ -23,7 +23,8 @@ struct image_changer_t
bool has_children = false; bool has_children = false;
int state = 0; int state = 0;
std::function<bool(void)> cb; std::function<bool(cli_result_t &)> cb;
cli_result_t result;
bool is_done() bool is_done()
{ {
@ -36,6 +37,18 @@ struct image_changer_t
goto resume_1; goto resume_1;
else if (state == 2) else if (state == 2)
goto resume_2; goto resume_2;
if (image_name == "")
{
result = (cli_result_t){ .err = EINVAL, .text = "Image name is missing" };
state = 100;
return;
}
if (new_size != 0 && (new_size % 4096))
{
result = (cli_result_t){ .err = EINVAL, .text = "Image size should be a multiple of 4096" };
state = 100;
return;
}
for (auto & ic: parent->cli->st_cli.inode_config) for (auto & ic: parent->cli->st_cli.inode_config)
{ {
if (ic.second.name == image_name) if (ic.second.name == image_name)
@ -46,14 +59,16 @@ struct image_changer_t
} }
if (new_name != "" && ic.second.name == new_name) if (new_name != "" && ic.second.name == new_name)
{ {
fprintf(stderr, "Image %s already exists\n", new_name.c_str()); result = (cli_result_t){ .err = EEXIST, .text = "Image "+new_name+" already exists" };
exit(1); state = 100;
return;
} }
} }
if (!inode_num) if (!inode_num)
{ {
fprintf(stderr, "Image %s does not exist\n", image_name.c_str()); result = (cli_result_t){ .err = ENOENT, .text = "Image "+image_name+" does not exist" };
exit(1); state = 100;
return;
} }
for (auto & ic: parent->cli->st_cli.inode_config) for (auto & ic: parent->cli->st_cli.inode_config)
{ {
@ -68,7 +83,7 @@ struct image_changer_t
(!new_size || cfg.size == new_size) && (!new_size || cfg.size == new_size) &&
(new_name == "" || new_name == image_name)) (new_name == "" || new_name == image_name))
{ {
printf("No change\n"); result = (cli_result_t){ .text = "No change" };
state = 100; state = 100;
return; return;
} }
@ -79,8 +94,9 @@ struct image_changer_t
// Check confirmation when trimming an image with children // Check confirmation when trimming an image with children
if (has_children && !force) if (has_children && !force)
{ {
fprintf(stderr, "Image %s has children. Refusing to shrink it without --force\n", image_name.c_str()); result = (cli_result_t){ .err = EINVAL, .text = "Image "+image_name+" has children. Refusing to shrink it without --force" };
exit(1); state = 100;
return;
} }
// Shrink the image first // Shrink the image first
cb = parent->start_rm(json11::Json::object { cb = parent->start_rm(json11::Json::object {
@ -90,12 +106,17 @@ struct image_changer_t
{ "min-offset", new_size }, { "min-offset", new_size },
}); });
resume_1: resume_1:
while (!cb()) while (!cb(result))
{ {
state = 1; state = 1;
return; return;
} }
cb = NULL; cb = NULL;
if (result.err)
{
state = 100;
return;
}
} }
cfg.size = new_size; cfg.size = new_size;
} }
@ -109,8 +130,9 @@ resume_1:
// Check confirmation when making an image with children read-write // Check confirmation when making an image with children read-write
if (has_children && !force) if (has_children && !force)
{ {
fprintf(stderr, "Image %s has children. Refusing to make it read-write without --force\n", image_name.c_str()); result = (cli_result_t){ .err = EINVAL, .text = "Image "+image_name+" has children. Refusing to make it read-write without --force" };
exit(1); state = 100;
return;
} }
} }
if (new_name != "") if (new_name != "")
@ -180,32 +202,23 @@ resume_2:
return; return;
if (!parent->etcd_result["succeeded"].bool_value()) if (!parent->etcd_result["succeeded"].bool_value())
{ {
fprintf(stderr, "Image %s was modified by someone else, please repeat your request\n", image_name.c_str()); result = (cli_result_t){ .err = EAGAIN, .text = "Image "+image_name+" was modified by someone else, please repeat your request" };
exit(1); state = 100;
return;
} }
printf("Image %s modified\n", image_name.c_str()); result = (cli_result_t){ .err = 0, .text = "Image "+image_name+" modified" };
state = 100; state = 100;
} }
}; };
std::function<bool(void)> cli_tool_t::start_modify(json11::Json cfg) std::function<bool(cli_result_t &)> cli_tool_t::start_modify(json11::Json cfg)
{ {
json11::Json::array cmd = cfg["command"].array_items(); json11::Json::array cmd = cfg["command"].array_items();
auto changer = new image_changer_t(); auto changer = new image_changer_t();
changer->parent = this; changer->parent = this;
changer->image_name = cmd.size() > 1 ? cmd[1].string_value() : ""; changer->image_name = cmd.size() > 1 ? cmd[1].string_value() : "";
if (changer->image_name == "")
{
fprintf(stderr, "Image name is missing\n");
exit(1);
}
changer->new_name = cfg["rename"].string_value(); changer->new_name = cfg["rename"].string_value();
changer->new_size = parse_size(cfg["resize"].string_value()); changer->new_size = parse_size(cfg["resize"].string_value());
if (changer->new_size != 0 && (changer->new_size % 4096))
{
fprintf(stderr, "Image size should be a multiple of 4096\n");
exit(1);
}
changer->force = cfg["force"].bool_value(); changer->force = cfg["force"].bool_value();
changer->set_readonly = cfg["readonly"].bool_value(); changer->set_readonly = cfg["readonly"].bool_value();
changer->set_readwrite = cfg["readwrite"].bool_value(); changer->set_readwrite = cfg["readwrite"].bool_value();
@ -213,11 +226,12 @@ std::function<bool(void)> cli_tool_t::start_modify(json11::Json cfg)
if (!changer->fsync_interval) if (!changer->fsync_interval)
changer->fsync_interval = 128; changer->fsync_interval = 128;
// FIXME Check that the image doesn't have children when shrinking // FIXME Check that the image doesn't have children when shrinking
return [changer]() return [changer](cli_result_t & result)
{ {
changer->loop(); changer->loop();
if (changer->is_done()) if (changer->is_done())
{ {
result = changer->result;
delete changer; delete changer;
return true; return true;
} }

View File

@ -63,7 +63,9 @@ struct snap_remover_t
inode_t new_parent = 0; inode_t new_parent = 0;
int state = 0; int state = 0;
int current_child = 0; int current_child = 0;
std::function<bool(void)> cb; std::function<bool(cli_result_t &)> cb;
cli_result_t result;
bool is_done() bool is_done()
{ {
@ -90,11 +92,25 @@ struct snap_remover_t
goto resume_8; goto resume_8;
else if (state == 9) else if (state == 9)
goto resume_9; goto resume_9;
if (from_name == "")
{
result = (cli_result_t){ .err = EINVAL, .text = "Layer to remove argument is missing" };
state = 100;
return;
}
if (to_name == "")
{
to_name = from_name;
}
// Get children to merge // Get children to merge
get_merge_children(); get_merge_children();
if (state == 100)
return;
// Try to select an inode for the "inverse" optimized scenario // Try to select an inode for the "inverse" optimized scenario
// Read statistics from etcd to do it // Read statistics from etcd to do it
read_stats(); read_stats();
if (state == 100)
return;
state = 1; state = 1;
resume_1: resume_1:
if (parent->waiting > 0) if (parent->waiting > 0)
@ -106,13 +122,20 @@ resume_1:
if (merge_children[current_child] == inverse_child) if (merge_children[current_child] == inverse_child)
continue; continue;
start_merge_child(merge_children[current_child], merge_children[current_child]); start_merge_child(merge_children[current_child], merge_children[current_child]);
if (state == 100)
return;
resume_2: resume_2:
while (!cb()) while (!cb(result))
{ {
state = 2; state = 2;
return; return;
} }
cb = NULL; cb = NULL;
if (result.err)
{
state = 100;
return;
}
parent->change_parent(merge_children[current_child], new_parent); parent->change_parent(merge_children[current_child], new_parent);
state = 3; state = 3;
resume_3: resume_3:
@ -123,25 +146,41 @@ resume_3:
if (inverse_child != 0) if (inverse_child != 0)
{ {
start_merge_child(inverse_child, inverse_parent); start_merge_child(inverse_child, inverse_parent);
if (state == 100)
return;
resume_4: resume_4:
while (!cb()) while (!cb(result))
{ {
state = 4; state = 4;
return; return;
} }
cb = NULL; cb = NULL;
if (result.err)
{
state = 100;
return;
}
// Delete "inverse" child data // Delete "inverse" child data
start_delete_source(inverse_child); start_delete_source(inverse_child);
if (state == 100)
return;
resume_5: resume_5:
while (!cb()) while (!cb(result))
{ {
state = 5; state = 5;
return; return;
} }
cb = NULL; cb = NULL;
if (result.err)
{
state = 100;
return;
}
// Delete "inverse" child metadata, rename parent over it, // Delete "inverse" child metadata, rename parent over it,
// and also change parent links of the previous "inverse" child // and also change parent links of the previous "inverse" child
rename_inverse_parent(); rename_inverse_parent();
if (state == 100)
return;
state = 6; state = 6;
resume_6: resume_6:
if (parent->waiting > 0) if (parent->waiting > 0)
@ -154,13 +193,20 @@ resume_6:
continue; continue;
start_delete_source(chain_list[current_child]); start_delete_source(chain_list[current_child]);
resume_7: resume_7:
while (!cb()) while (!cb(result))
{ {
state = 7; state = 7;
return; return;
} }
cb = NULL; cb = NULL;
if (result.err)
{
state = 100;
return;
}
delete_inode_config(chain_list[current_child]); delete_inode_config(chain_list[current_child]);
if (state == 100)
return;
state = 8; state = 8;
resume_8: resume_8:
if (parent->waiting > 0) if (parent->waiting > 0)
@ -186,16 +232,19 @@ resume_9:
auto it = parent->cli->st_cli.inode_config.find(cur->parent_id); auto it = parent->cli->st_cli.inode_config.find(cur->parent_id);
if (it == parent->cli->st_cli.inode_config.end()) if (it == parent->cli->st_cli.inode_config.end())
{ {
fprintf(stderr, "Parent inode of layer %s (id %ld) not found\n", cur->name.c_str(), cur->parent_id); char buf[1024];
exit(1); snprintf(buf, 1024, "Parent inode of layer %s (id 0x%lx) not found", cur->name.c_str(), cur->parent_id);
state = 100;
return;
} }
cur = &it->second; cur = &it->second;
chain_list.push_back(cur->num); chain_list.push_back(cur->num);
} }
if (cur->num != from_cfg->num) if (cur->num != from_cfg->num)
{ {
fprintf(stderr, "Layer %s is not a child of %s\n", to_name.c_str(), from_name.c_str()); result = (cli_result_t){ .err = EINVAL, .text = "Layer "+to_name+" is not a child of "+from_name };
exit(1); state = 100;
return;
} }
new_parent = from_cfg->parent_id; new_parent = from_cfg->parent_id;
// Calculate ranks // Calculate ranks
@ -263,8 +312,9 @@ resume_9:
parent->waiting--; parent->waiting--;
if (err != "") if (err != "")
{ {
fprintf(stderr, "Error reading layer statistics from etcd: %s\n", err.c_str()); result = (cli_result_t){ .err = EIO, .text = "Error reading layer statistics from etcd: "+err };
exit(1); state = 100;
return;
} }
for (auto inode_result: data["responses"].array_items()) for (auto inode_result: data["responses"].array_items())
{ {
@ -275,14 +325,16 @@ resume_9:
sscanf(kv.key.c_str() + parent->cli->st_cli.etcd_prefix.length()+13, "%u/%lu%c", &pool_id, &inode, &null_byte); sscanf(kv.key.c_str() + parent->cli->st_cli.etcd_prefix.length()+13, "%u/%lu%c", &pool_id, &inode, &null_byte);
if (!inode || null_byte != 0) if (!inode || null_byte != 0)
{ {
fprintf(stderr, "Bad key returned from etcd: %s\n", kv.key.c_str()); result = (cli_result_t){ .err = EIO, .text = "Bad key returned from etcd: "+kv.key };
exit(1); state = 100;
return;
} }
auto pool_cfg_it = parent->cli->st_cli.pool_config.find(pool_id); auto pool_cfg_it = parent->cli->st_cli.pool_config.find(pool_id);
if (pool_cfg_it == parent->cli->st_cli.pool_config.end()) if (pool_cfg_it == parent->cli->st_cli.pool_config.end())
{ {
fprintf(stderr, "Pool %u does not exist\n", pool_id); result = (cli_result_t){ .err = ENOENT, .text = "Pool "+std::to_string(pool_id)+" does not exist" };
exit(1); state = 100;
return;
} }
inode = INODE_WITH_POOL(pool_id, inode); inode = INODE_WITH_POOL(pool_id, inode);
auto & pool_cfg = pool_cfg_it->second; auto & pool_cfg = pool_cfg_it->second;
@ -324,14 +376,20 @@ resume_9:
auto child_it = parent->cli->st_cli.inode_config.find(inverse_child); auto child_it = parent->cli->st_cli.inode_config.find(inverse_child);
if (child_it == parent->cli->st_cli.inode_config.end()) if (child_it == parent->cli->st_cli.inode_config.end())
{ {
fprintf(stderr, "Inode %ld disappeared\n", inverse_child); char buf[1024];
exit(1); snprintf(buf, 1024, "Inode 0x%lx disappeared", inverse_child);
result = (cli_result_t){ .err = EIO, .text = std::string(buf) };
state = 100;
return;
} }
auto target_it = parent->cli->st_cli.inode_config.find(inverse_parent); auto target_it = parent->cli->st_cli.inode_config.find(inverse_parent);
if (target_it == parent->cli->st_cli.inode_config.end()) if (target_it == parent->cli->st_cli.inode_config.end())
{ {
fprintf(stderr, "Inode %ld disappeared\n", inverse_parent); char buf[1024];
exit(1); snprintf(buf, 1024, "Inode 0x%lx disappeared", inverse_parent);
result = (cli_result_t){ .err = EIO, .text = std::string(buf) };
state = 100;
return;
} }
inode_config_t *child_cfg = &child_it->second; inode_config_t *child_cfg = &child_it->second;
inode_config_t *target_cfg = &target_it->second; inode_config_t *target_cfg = &target_it->second;
@ -422,16 +480,19 @@ resume_9:
parent->waiting--; parent->waiting--;
if (err != "") if (err != "")
{ {
fprintf(stderr, "Error renaming %s to %s: %s\n", target_name.c_str(), child_name.c_str(), err.c_str()); result = (cli_result_t){ .err = EIO, .text = "Error renaming "+target_name+" to "+child_name+": "+err };
exit(1); state = 100;
return;
} }
if (!res["succeeded"].bool_value()) if (!res["succeeded"].bool_value())
{ {
fprintf( result = (cli_result_t){
stderr, "Parent (%s), child (%s), or one of its children" .err = EIO,
" configuration was modified during rename\n", target_name.c_str(), child_name.c_str() .text = "Parent ("+target_name+"), child ("+child_name+"), or one of its children"
); " configuration was modified during rename",
exit(1); };
state = 100;
return;
} }
printf("Layer %s renamed to %s\n", target_name.c_str(), child_name.c_str()); printf("Layer %s renamed to %s\n", target_name.c_str(), child_name.c_str());
parent->ringloop->wakeup(); parent->ringloop->wakeup();
@ -443,8 +504,11 @@ resume_9:
auto cur_cfg_it = parent->cli->st_cli.inode_config.find(cur); auto cur_cfg_it = parent->cli->st_cli.inode_config.find(cur);
if (cur_cfg_it == parent->cli->st_cli.inode_config.end()) if (cur_cfg_it == parent->cli->st_cli.inode_config.end())
{ {
fprintf(stderr, "Inode 0x%lx disappeared\n", cur); char buf[1024];
exit(1); snprintf(buf, 1024, "Inode 0x%lx disappeared", cur);
result = (cli_result_t){ .err = EIO, .text = std::string(buf) };
state = 100;
return;
} }
inode_config_t *cur_cfg = &cur_cfg_it->second; inode_config_t *cur_cfg = &cur_cfg_it->second;
std::string cur_name = cur_cfg->name; std::string cur_name = cur_cfg->name;
@ -480,13 +544,15 @@ resume_9:
parent->waiting--; parent->waiting--;
if (err != "") if (err != "")
{ {
fprintf(stderr, "Error deleting %s: %s\n", cur_name.c_str(), err.c_str()); result = (cli_result_t){ .err = EIO, .text = "Error deleting "+cur_name+": "+err };
exit(1); state = 100;
return;
} }
if (!res["succeeded"].bool_value()) if (!res["succeeded"].bool_value())
{ {
fprintf(stderr, "Layer %s configuration was modified during deletion\n", cur_name.c_str()); result = (cli_result_t){ .err = EIO, .text = "Layer "+cur_name+" was modified during deletion" };
exit(1); state = 100;
return;
} }
printf("Layer %s deleted\n", cur_name.c_str()); printf("Layer %s deleted\n", cur_name.c_str());
parent->ringloop->wakeup(); parent->ringloop->wakeup();
@ -498,14 +564,20 @@ resume_9:
auto child_it = parent->cli->st_cli.inode_config.find(child_inode); auto child_it = parent->cli->st_cli.inode_config.find(child_inode);
if (child_it == parent->cli->st_cli.inode_config.end()) if (child_it == parent->cli->st_cli.inode_config.end())
{ {
fprintf(stderr, "Inode %ld disappeared\n", child_inode); char buf[1024];
exit(1); snprintf(buf, 1024, "Inode 0x%lx disappeared", child_inode);
result = (cli_result_t){ .err = EIO, .text = std::string(buf) };
state = 100;
return;
} }
auto target_it = parent->cli->st_cli.inode_config.find(target_inode); auto target_it = parent->cli->st_cli.inode_config.find(target_inode);
if (target_it == parent->cli->st_cli.inode_config.end()) if (target_it == parent->cli->st_cli.inode_config.end())
{ {
fprintf(stderr, "Inode %ld disappeared\n", target_inode); char buf[1024];
exit(1); snprintf(buf, 1024, "Inode 0x%lx disappeared", target_inode);
result = (cli_result_t){ .err = EIO, .text = std::string(buf) };
state = 100;
return;
} }
cb = parent->start_merge(json11::Json::object { cb = parent->start_merge(json11::Json::object {
{ "command", json11::Json::array{ "merge-data", from_name, child_it->second.name } }, { "command", json11::Json::array{ "merge-data", from_name, child_it->second.name } },
@ -521,8 +593,11 @@ resume_9:
auto source = parent->cli->st_cli.inode_config.find(inode); auto source = parent->cli->st_cli.inode_config.find(inode);
if (source == parent->cli->st_cli.inode_config.end()) if (source == parent->cli->st_cli.inode_config.end())
{ {
fprintf(stderr, "Inode %ld disappeared\n", inode); char buf[1024];
exit(1); snprintf(buf, 1024, "Inode 0x%lx disappeared", inode);
result = (cli_result_t){ .err = EIO, .text = std::string(buf) };
state = 100;
return;
} }
cb = parent->start_rm(json11::Json::object { cb = parent->start_rm(json11::Json::object {
{ "inode", inode }, { "inode", inode },
@ -532,22 +607,13 @@ resume_9:
} }
}; };
std::function<bool(void)> cli_tool_t::start_snap_rm(json11::Json cfg) std::function<bool(cli_result_t &)> cli_tool_t::start_snap_rm(json11::Json cfg)
{ {
json11::Json::array cmd = cfg["command"].array_items(); json11::Json::array cmd = cfg["command"].array_items();
auto snap_remover = new snap_remover_t(); auto snap_remover = new snap_remover_t();
snap_remover->parent = this; snap_remover->parent = this;
snap_remover->from_name = cmd.size() > 1 ? cmd[1].string_value() : ""; snap_remover->from_name = cmd.size() > 1 ? cmd[1].string_value() : "";
snap_remover->to_name = cmd.size() > 2 ? cmd[2].string_value() : ""; snap_remover->to_name = cmd.size() > 2 ? cmd[2].string_value() : "";
if (snap_remover->from_name == "")
{
fprintf(stderr, "Layer to remove argument is missing\n");
exit(1);
}
if (snap_remover->to_name == "")
{
snap_remover->to_name = snap_remover->from_name;
}
snap_remover->fsync_interval = cfg["fsync-interval"].uint64_value(); snap_remover->fsync_interval = cfg["fsync-interval"].uint64_value();
if (!snap_remover->fsync_interval) if (!snap_remover->fsync_interval)
snap_remover->fsync_interval = 128; snap_remover->fsync_interval = 128;
@ -555,11 +621,12 @@ std::function<bool(void)> cli_tool_t::start_snap_rm(json11::Json cfg)
snap_remover->use_cas = cfg["cas"].uint64_value() ? 2 : 0; snap_remover->use_cas = cfg["cas"].uint64_value() ? 2 : 0;
if (!cfg["writers_stopped"].is_null()) if (!cfg["writers_stopped"].is_null())
snap_remover->writers_stopped = true; snap_remover->writers_stopped = true;
return [snap_remover]() return [snap_remover](cli_result_t & result)
{ {
snap_remover->loop(); snap_remover->loop();
if (snap_remover->is_done()) if (snap_remover->is_done())
{ {
result = snap_remover->result;
delete snap_remover; delete snap_remover;
return true; return true;
} }

View File

@ -32,6 +32,9 @@ struct rm_inode_t
uint64_t pgs_to_list = 0; uint64_t pgs_to_list = 0;
bool lists_done = false; bool lists_done = false;
int state = 0; int state = 0;
int error_count = 0;
cli_result_t result;
void start_delete() void start_delete()
{ {
@ -74,8 +77,13 @@ struct rm_inode_t
}); });
if (!lister) if (!lister)
{ {
fprintf(stderr, "Failed to list inode %lu from pool %u objects\n", INODE_NO_POOL(inode), INODE_POOL(inode)); result = (cli_result_t){
exit(1); .err = EIO,
.text = "Failed to list objects of inode "+std::to_string(INODE_NO_POOL(inode))+
" from pool "+std::to_string(INODE_POOL(inode)),
};
state = 100;
return;
} }
pgs_to_list = parent->cli->list_pg_count(lister); pgs_to_list = parent->cli->list_pg_count(lister);
parent->cli->list_inode_next(lister, parent->parallel_osds); parent->cli->list_inode_next(lister, parent->parallel_osds);
@ -118,6 +126,7 @@ struct rm_inode_t
fprintf(stderr, "Failed to remove object %lx:%lx from PG %u (OSD %lu) (retval=%ld)\n", fprintf(stderr, "Failed to remove object %lx:%lx from PG %u (OSD %lu) (retval=%ld)\n",
op->req.rw.inode, op->req.rw.offset, op->req.rw.inode, op->req.rw.offset,
cur_list->pg_num, cur_list->rm_osd_num, op->reply.hdr.retval); cur_list->pg_num, cur_list->rm_osd_num, op->reply.hdr.retval);
error_count++;
} }
delete op; delete op;
cur_list->obj_done++; cur_list->obj_done++;
@ -161,31 +170,41 @@ struct rm_inode_t
} }
if (lists_done && !lists.size()) if (lists_done && !lists.size())
{ {
printf("Done, inode %lu in pool %u data removed\n", INODE_NO_POOL(inode), pool_id); result = (cli_result_t){
state = 2; .err = error_count > 0 ? EIO : 0,
.text = error_count > 0 ? "Some blocks were not removed" : (
"Done, inode "+std::to_string(INODE_NO_POOL(inode))+" from pool "+
std::to_string(pool_id)+" removed"),
};
state = 100;
} }
} }
bool loop() bool is_done()
{ {
if (state == 0) return state == 100;
}
void loop()
{
if (state == 1)
goto resume_1;
if (!pool_id)
{ {
start_delete(); result = (cli_result_t){ .err = EINVAL, .text = "Pool is not specified" };
state = 1; state = 100;
return;
} }
else if (state == 1) start_delete();
{ if (state == 100)
continue_delete(); return;
} state = 1;
else if (state == 2) resume_1:
{ continue_delete();
return true;
}
return false;
} }
}; };
std::function<bool(void)> cli_tool_t::start_rm(json11::Json cfg) std::function<bool(cli_result_t &)> cli_tool_t::start_rm(json11::Json cfg)
{ {
auto remover = new rm_inode_t(); auto remover = new rm_inode_t();
remover->parent = this; remover->parent = this;
@ -196,16 +215,13 @@ std::function<bool(void)> cli_tool_t::start_rm(json11::Json cfg)
remover->inode = (remover->inode & (((uint64_t)1 << (64-POOL_ID_BITS)) - 1)) | (((uint64_t)remover->pool_id) << (64-POOL_ID_BITS)); remover->inode = (remover->inode & (((uint64_t)1 << (64-POOL_ID_BITS)) - 1)) | (((uint64_t)remover->pool_id) << (64-POOL_ID_BITS));
} }
remover->pool_id = INODE_POOL(remover->inode); remover->pool_id = INODE_POOL(remover->inode);
if (!remover->pool_id)
{
fprintf(stderr, "pool is missing\n");
exit(1);
}
remover->min_offset = cfg["min-offset"].uint64_value(); remover->min_offset = cfg["min-offset"].uint64_value();
return [remover]() return [remover](cli_result_t & result)
{ {
if (remover->loop()) remover->loop();
if (remover->is_done())
{ {
result = remover->result;
delete remover; delete remover;
return true; return true;
} }

View File

@ -11,7 +11,7 @@
#include <sys/stat.h> #include <sys/stat.h>
// Calculate offsets for a block device and print OSD command line parameters // Calculate offsets for a block device and print OSD command line parameters
std::function<bool(void)> cli_tool_t::simple_offsets(json11::Json cfg) std::function<bool(cli_result_t &)> cli_tool_t::simple_offsets(json11::Json cfg)
{ {
std::string device = cfg["command"][1].string_value(); std::string device = cfg["command"][1].string_value();
uint64_t object_size = parse_size(cfg["object_size"].string_value()); uint64_t object_size = parse_size(cfg["object_size"].string_value());

View File

@ -277,16 +277,17 @@ resume_2:
} }
}; };
std::function<bool(void)> cli_tool_t::start_status(json11::Json cfg) std::function<bool(cli_result_t &)> cli_tool_t::start_status(json11::Json cfg)
{ {
json11::Json::array cmd = cfg["command"].array_items(); json11::Json::array cmd = cfg["command"].array_items();
auto printer = new status_printer_t(); auto printer = new status_printer_t();
printer->parent = this; printer->parent = this;
return [printer]() return [printer](cli_result_t & result)
{ {
printer->loop(); printer->loop();
if (printer->is_done()) if (printer->is_done())
{ {
result = { .err = 0 };
delete printer; delete printer;
return true; return true;
} }

View File

@ -954,6 +954,10 @@ void etcd_state_client_t::parse_state(const etcd_kv_t & kv)
} }
if (!value.is_object()) if (!value.is_object())
{ {
if (on_inode_change_hook != NULL)
{
on_inode_change_hook(inode_num, true);
}
this->inode_config.erase(inode_num); this->inode_config.erase(inode_num);
} }
else else
@ -995,6 +999,10 @@ void etcd_state_client_t::parse_state(const etcd_kv_t & kv)
} }
} }
} }
if (on_inode_change_hook != NULL)
{
on_inode_change_hook(inode_num, false);
}
} }
} }
} }

View File

@ -109,6 +109,7 @@ public:
std::function<void(pool_id_t, pg_num_t)> on_change_pg_history_hook; std::function<void(pool_id_t, pg_num_t)> on_change_pg_history_hook;
std::function<void(osd_num_t)> on_change_osd_state_hook; std::function<void(osd_num_t)> on_change_osd_state_hook;
std::function<void()> on_reload_hook; std::function<void()> on_reload_hook;
std::function<void(inode_t, bool)> on_inode_change_hook;
json11::Json::object serialize_inode_cfg(inode_config_t *cfg); json11::Json::object serialize_inode_cfg(inode_config_t *cfg);
etcd_kv_t parse_etcd_kv(const json11::Json & kv_json); etcd_kv_t parse_etcd_kv(const json11::Json & kv_json);

748
src/nfs_conn.cpp Normal file
View File

@ -0,0 +1,748 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
//
// NFS connection handler for NFS proxy
#include <sys/time.h>
#include "libnfs-raw-mount.h"
#include "libnfs-raw-nfs.h"
#include "base64.h"
#include "nfs_proxy.h"
static unsigned len_pad4(unsigned len)
{
return len + (len&3 ? 4-(len&3) : 0);
}
static int nfs3_null_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
rpc_send_reply(rpc, call, NULL, (zdrproc_t)zdr_void, 0);
return 0;
}
static fattr3 get_dir_attributes(nfs_client_t *self, std::string dir)
{
return (fattr3){
.type = NF3DIR,
.mode = 0755,
.nlink = 1,
.uid = 0,
.gid = 0,
.size = 4096,
.used = 4096,
.rdev = (specdata3){ 0 },
.fsid = self->parent->fsid,
.fileid = dir == "" ? 1 : self->parent->dir_ids.at(dir),
//.atime = (nfstime3){ .seconds = now.tv_sec, .nseconds = now.tv_nsec },
//.mtime = (nfstime3){ .seconds = now.tv_sec, .nseconds = now.tv_nsec },
//.ctime = (nfstime3){ .seconds = now.tv_sec, .nseconds = now.tv_nsec },
};
}
static int nfs3_getattr_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
GETATTR3args *args = (GETATTR3args*)call->body.cbody.args;
GETATTR3res reply;
std::string dirhash = std::string(args->object.data.data_val, args->object.data.data_len);
bool is_dir = false;
std::string dir;
if (dirhash == "roothandle")
is_dir = true;
else
{
auto dir_it = self->parent->dir_by_hash.find(dirhash);
if (dir_it != self->parent->dir_by_hash.end())
{
is_dir = true;
dir = dir_it->second;
}
}
if (is_dir)
{
// Directory info
reply.status = NFS3_OK;
reply.GETATTR3res_u.resok.obj_attributes = get_dir_attributes(self, dir);
}
else
{
uint64_t inode_num;
auto inode_num_it = self->parent->inode_by_hash.find(dirhash);
if (inode_num_it != self->parent->inode_by_hash.end())
inode_num = inode_num_it->second;
auto inode_it = self->parent->cli->st_cli.inode_config.find(inode_num);
if (inode_it != self->parent->cli->st_cli.inode_config.end())
{
// File info
auto & inode_cfg = inode_it->second;
reply.status = NFS3_OK;
reply.GETATTR3res_u.resok.obj_attributes = {
.type = NF3REG,
.mode = 0644,
.nlink = 1,
.uid = 0,
.gid = 0,
.size = inode_cfg.size,
.used = inode_cfg.size,
.rdev = (specdata3){ 0 },
.fsid = self->parent->fsid,
.fileid = inode_it->first,
//.atime = (nfstime3){ .seconds = now.tv_sec, .nseconds = now.tv_nsec },
//.mtime = (nfstime3){ .seconds = now.tv_sec, .nseconds = now.tv_nsec },
//.ctime = (nfstime3){ .seconds = now.tv_sec, .nseconds = now.tv_nsec },
};
}
else
{
// File not exists
reply.status = NFS3ERR_NOENT;
}
}
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_GETATTR3res, sizeof(GETATTR3res));
return 0;
}
static int nfs3_setattr_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
SETATTR3args *args = (SETATTR3args*)call->body.cbody.args;
SETATTR3res reply;
// Not supported yet
reply.status = NFS3ERR_NOTSUPP;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_SETATTR3res, sizeof(SETATTR3res));
return 0;
}
static int nfs3_lookup_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
LOOKUP3args *args = (LOOKUP3args*)call->body.cbody.args;
LOOKUP3res reply;
std::string dirhash = std::string(args->what.dir.data.data_val, args->what.dir.data.data_len);
std::string dir;
if (dirhash != "roothandle")
{
auto dir_it = self->parent->dir_by_hash.find(dirhash);
if (dir_it != self->parent->dir_by_hash.end())
dir = dir_it->second;
}
std::string full_name = self->parent->name_prefix;
if (dir != "")
{
full_name += dir+"/";
}
full_name += std::string(args->what.name);
for (auto & ic: self->parent->cli->st_cli.inode_config)
{
if (ic.second.name == full_name)
{
std::string fh = "S"+base64_encode(sha256(full_name.substr(self->parent->name_prefix.size())));
reply.status = NFS3_OK;
reply.LOOKUP3res_u.resok.object.data.data_len = fh.size();
reply.LOOKUP3res_u.resok.object.data.data_val = (char*)fh.c_str();
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_LOOKUP3res, sizeof(LOOKUP3res));
return 0;
}
}
reply.status = NFS3ERR_NOENT;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_LOOKUP3res, sizeof(LOOKUP3res));
return 0;
}
static int nfs3_access_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
ACCESS3args *args = (ACCESS3args*)call->body.cbody.args;
ACCESS3res reply = {
.status = NFS3_OK,
.ACCESS3res_u = { .resok = {
.access = args->access,
} },
};
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_ACCESS3res, sizeof(ACCESS3res));
return 0;
}
static int nfs3_readlink_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
READLINK3args *args = (READLINK3args*)call->body.cbody.args;
READLINK3res reply = {};
// Not supported yet
reply.status = NFS3ERR_NOTSUPP;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_READLINK3res, sizeof(READLINK3res));
return 0;
}
#define MAX_REQUEST_SIZE 128*1024*1024
static int nfs3_read_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
READ3args *args = (READ3args*)call->body.cbody.args;
std::string handle = std::string(args->file.data.data_val, args->file.data.data_len);
auto ino_it = self->parent->inode_by_hash.find(handle);
if (ino_it == self->parent->inode_by_hash.end())
{
READ3res reply = { .status = NFS3ERR_NOENT };
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_READ3res, sizeof(READ3res));
return 0;
}
if (args->count > MAX_REQUEST_SIZE)
{
READ3res reply = { .status = NFS3ERR_INVAL };
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_READ3res, sizeof(READ3res));
return 0;
}
void *buf = malloc_or_die(args->count);
cluster_op_t *op = new cluster_op_t;
op->opcode = OSD_OP_READ;
op->inode = ino_it->second;
op->offset = args->offset;
op->len = args->count;
op->iov.push_back(buf, args->count);
op->callback = [rpc, call](cluster_op_t *op)
{
void *buf = op->iov.buf[0].iov_base;
READ3res reply = {};
if (op->retval != op->len)
{
if (op->retval == -EINVAL)
reply.status = NFS3ERR_INVAL;
else if (op->retval == -ENOSPC)
reply.status = NFS3ERR_NOSPC;
else
reply.status = NFS3ERR_IO;
}
else
{
reply.status = NFS3_OK;
auto & reply_ok = reply.READ3res_u.resok;
reply_ok.count = op->retval;
reply_ok.eof = FALSE;
reply_ok.data.data_len = reply_ok.count;
reply_ok.data.data_val = (char*)buf;
}
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_READ3res, sizeof(READ3res));
delete op;
free(buf);
};
self->parent->cli->execute(op);
return 0;
}
static int nfs3_write_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
WRITE3args *args = (WRITE3args*)call->body.cbody.args;
WRITE3res reply;
// Not supported yet
reply.status = NFS3ERR_NOTSUPP;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_WRITE3res, sizeof(WRITE3res));
return 0;
}
static int nfs3_create_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
CREATE3args *args = (CREATE3args*)call->body.cbody.args;
CREATE3res reply;
// Not supported yet
reply.status = NFS3ERR_NOTSUPP;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_CREATE3res, sizeof(CREATE3res));
return 0;
}
static int nfs3_mkdir_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
MKDIR3args *args = (MKDIR3args*)call->body.cbody.args;
MKDIR3res reply;
// Not supported yet
reply.status = NFS3ERR_NOTSUPP;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_MKDIR3res, sizeof(MKDIR3res));
return 0;
}
static int nfs3_symlink_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
SYMLINK3args *args = (SYMLINK3args*)call->body.cbody.args;
SYMLINK3res reply;
// Not supported yet
reply.status = NFS3ERR_NOTSUPP;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_SYMLINK3res, sizeof(SYMLINK3res));
return 0;
}
static int nfs3_mknod_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
MKNOD3args *args = (MKNOD3args*)call->body.cbody.args;
MKNOD3res reply;
// Not supported yet
reply.status = NFS3ERR_NOTSUPP;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_MKNOD3res, sizeof(MKNOD3res));
return 0;
}
static int nfs3_remove_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
REMOVE3args *args = (REMOVE3args*)call->body.cbody.args;
REMOVE3res reply;
// Not supported yet
reply.status = NFS3ERR_NOTSUPP;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_REMOVE3res, sizeof(REMOVE3res));
return 0;
}
static int nfs3_rmdir_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
RMDIR3args *args = (RMDIR3args*)call->body.cbody.args;
RMDIR3res reply;
// Not supported yet
reply.status = NFS3ERR_NOTSUPP;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_RMDIR3res, sizeof(RMDIR3res));
return 0;
}
static int nfs3_rename_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
RENAME3args *args = (RENAME3args*)call->body.cbody.args;
RENAME3res reply;
// Not supported yet
reply.status = NFS3ERR_NOTSUPP;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_RENAME3res, sizeof(RENAME3res));
return 0;
}
static int nfs3_link_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
LINK3args *args = (LINK3args*)call->body.cbody.args;
// We don't support hard links
LINK3res reply = { NFS3ERR_NOTSUPP };
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_LINK3res, sizeof(LINK3res));
return 0;
}
static void nfs3_readdir_common(struct rpc_context *rpc, struct rpc_msg *call, void *opaque, bool is_plus)
{
nfs_client_t *self = (nfs_client_t*)opaque;
READDIRPLUS3args plus_args;
READDIRPLUS3args *args = NULL;
if (is_plus)
args = ((READDIRPLUS3args*)call->body.cbody.args);
else
{
args = &plus_args;
READDIR3args *in_args = ((READDIR3args*)call->body.cbody.args);
args->dir = in_args->dir;
args->cookie = in_args->cookie;
*((uint64_t*)args->cookieverf) = *((uint64_t*)in_args->cookieverf);
args->dircount = 512;
args->maxcount = in_args->count;
}
std::string dirhash = std::string(args->dir.data.data_val, args->dir.data.data_len);
std::string dir;
if (dirhash != "roothandle")
{
auto dir_it = self->parent->dir_by_hash.find(dirhash);
if (dir_it != self->parent->dir_by_hash.end())
dir = dir_it->second;
}
std::string prefix = self->parent->name_prefix;
if (dir != "")
{
prefix += dir+"/";
}
//struct timespec now;
//clock_gettime(CLOCK_REALTIME, &now);
std::map<std::string, struct entryplus3> entries;
std::vector<std::string> handles;
for (auto & ic: self->parent->cli->st_cli.inode_config)
{
auto & inode_cfg = ic.second;
if (prefix != "" && inode_cfg.name.substr(0, prefix.size()) != prefix)
continue;
std::string subname = inode_cfg.name.substr(prefix.size());
int p = 0;
while (p < subname.size() && subname[p] == '/')
p++;
if (p > 0)
subname = subname.substr(p);
if (subname.size() == 0)
continue;
p = 0;
while (p < subname.size() && subname[p] != '/')
p++;
if (p >= subname.size())
{
entries[subname] = (struct entryplus3){
// fileid will change when the user creates snapshots
// however, we hope that clients tolerate it well
// Linux does, even though it complains about "fileid changed" in dmesg
.fileid = ic.first,
};
if (is_plus)
{
handles.push_back("S"+base64_encode(sha256(inode_cfg.name)));
entries[subname].name_attributes = {
.attributes_follow = TRUE,
.post_op_attr_u = { .attributes = {
.type = NF3REG,
.mode = 0644,
.nlink = 1,
.uid = 0,
.gid = 0,
.size = inode_cfg.size,
.used = inode_cfg.size, // FIXME take from statistics
.rdev = (specdata3){ 0 },
.fsid = self->parent->fsid,
.fileid = ic.first,
//.atime = (nfstime3){ .seconds = now.tv_sec, .nseconds = now.tv_nsec },
//.mtime = (nfstime3){ .seconds = now.tv_sec, .nseconds = now.tv_nsec },
//.ctime = (nfstime3){ .seconds = now.tv_sec, .nseconds = now.tv_nsec },
} },
};
entries[subname].name_handle = {
.handle_follows = TRUE,
.post_op_fh3_u = { .handle = {
.data = {
// FIXME: I really want ZDR with std::string
.data_len = handles[handles.size()-1].size(),
.data_val = (char*)handles[handles.size()-1].c_str(),
},
} },
};
}
}
else
{
auto subdir = dir == "" ? subname.substr(0, p) : dir+"/"+subname.substr(0, p);
entries[subdir] = (struct entryplus3){
// for directories, fileid will change when the user restarts proxy
.fileid = self->parent->dir_ids.at(subdir),
};
if (is_plus)
{
handles.push_back("S"+base64_encode(sha256(subdir)));
entries[subdir].name_attributes = {
.attributes_follow = TRUE,
.post_op_attr_u = { .attributes = get_dir_attributes(self, subdir) },
};
entries[subdir].name_handle = {
.handle_follows = TRUE,
.post_op_fh3_u = { .handle = {
.data = {
// FIXME: I really want ZDR with std::string
.data_len = (unsigned)handles[handles.size()-1].size(),
.data_val = (char*)handles[handles.size()-1].c_str(),
},
} },
};
}
}
}
// Offset results by the continuation cookie (equal to index in the listing)
uint64_t idx = 1;
void *prev = NULL;
for (auto it = entries.begin(); it != entries.end(); it++)
{
entryplus3 *entry = &it->second;
// First fields of entry3 and entryplus3 are the same: fileid, name, cookie
entry->name = (char*)it->first.c_str();
entry->cookie = idx++;
if (prev)
{
if (is_plus)
((entryplus3*)prev)->nextentry = entry;
else
((entry3*)prev)->nextentry = (entry3*)entry;
}
prev = entry;
if (args->cookie > 0 && entry->cookie == args->cookie+1)
{
entries.erase(entries.begin(), it);
}
}
// Now limit results based on maximum reply size
// Sadly we have to calculate reply size by hand
// reply without entries is 4+4+(dir_attributes ? sizeof(fattr3) : 0)+8+4 bytes
int reply_size = 20;
if (reply_size > args->maxcount)
{
// Error, too small max reply size
if (is_plus)
{
READDIRPLUS3res reply = { .status = NFS3ERR_TOOSMALL };
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_READDIRPLUS3res, sizeof(READDIRPLUS3res));
}
else
{
READDIR3res reply = { .status = NFS3ERR_TOOSMALL };
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_READDIR3res, sizeof(READDIR3res));
}
return;
}
// 1 entry3 is (8+4+(filename_len+3)/4*4+8) bytes
// 1 entryplus3 is (8+4+(filename_len+3)/4*4+8
// + 4+(name_attributes ? (sizeof(fattr3) = 84) : 0)
// + 4+(name_handle ? 4+(handle_len+3)/4*4 : 0)) bytes
bool eof = true;
for (auto it = entries.begin(); it != entries.end(); it++)
{
reply_size += 20+len_pad4(it->first.size())+(is_plus
? 8+84+len_pad4(it->second.name_handle.post_op_fh3_u.handle.data.data_len) : 0);
if (reply_size > args->maxcount)
{
// Stop
entries.erase(it, entries.end());
eof = false;
break;
}
}
if (entries.end() != entries.begin())
{
auto last_it = entries.end();
last_it--;
if (is_plus)
((entryplus3*)&last_it->second)->nextentry = NULL;
else
((entry3*)&last_it->second)->nextentry = NULL;
}
// Send reply
if (is_plus)
{
READDIRPLUS3res reply = { .status = NFS3_OK };
*(uint64_t*)(reply.READDIRPLUS3res_u.resok.cookieverf) = self->parent->dir_mod_rev.at(dir);
reply.READDIRPLUS3res_u.resok.reply.entries = &entries.begin()->second;
reply.READDIRPLUS3res_u.resok.reply.eof = eof;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_READDIRPLUS3res, sizeof(READDIRPLUS3res));
}
else
{
READDIR3res reply = { .status = NFS3_OK };
*(uint64_t*)(reply.READDIR3res_u.resok.cookieverf) = self->parent->dir_mod_rev.at(dir);
reply.READDIR3res_u.resok.reply.entries = (entry3*)&entries.begin()->second;
reply.READDIR3res_u.resok.reply.eof = eof;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_READDIR3res, sizeof(READDIR3res));
}
}
static int nfs3_readdir_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs3_readdir_common(rpc, call, opaque, false);
return 0;
}
static int nfs3_readdirplus_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs3_readdir_common(rpc, call, opaque, true);
return 0;
}
// Get file system statistics
static int nfs3_fsstat_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
FSSTAT3args *args = (FSSTAT3args*)call->body.cbody.args;
FSSTAT3res reply;
reply.status = NFS3_OK;
reply.FSSTAT3res_u.resok.obj_attributes.attributes_follow = TRUE;
reply.FSSTAT3res_u.resok.obj_attributes.post_op_attr_u.attributes = get_dir_attributes(self, "");
reply.FSSTAT3res_u.resok.tbytes = 4096; // total bytes
reply.FSSTAT3res_u.resok.fbytes = 4096; // free bytes
reply.FSSTAT3res_u.resok.abytes = 4096; // available bytes
reply.FSSTAT3res_u.resok.tfiles = 1 << 31; // total files
reply.FSSTAT3res_u.resok.ffiles = 1 << 31; // free files
reply.FSSTAT3res_u.resok.afiles = 1 << 31; // available files
reply.FSSTAT3res_u.resok.invarsec = 0;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_FSSTAT3res, sizeof(FSSTAT3res));
return 0;
}
static int nfs3_fsinfo_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
FSINFO3args *args = (FSINFO3args*)call->body.cbody.args;
FSINFO3res reply;
if (args->fsroot.data.data_len != 10)
{
// Example error
reply.status = NFS3ERR_INVAL;
}
else
{
// Fill info
reply.status = NFS3_OK;
reply.FSINFO3res_u.resok.obj_attributes.attributes_follow = TRUE;
reply.FSINFO3res_u.resok.obj_attributes.post_op_attr_u.attributes = get_dir_attributes(self, "");
reply.FSINFO3res_u.resok.rtmax = 128*1024*1024;
reply.FSINFO3res_u.resok.rtpref = 128*1024*1024;
reply.FSINFO3res_u.resok.rtmult = 4096;
reply.FSINFO3res_u.resok.wtmax = 128*1024*1024;
reply.FSINFO3res_u.resok.wtpref = 128*1024*1024;
reply.FSINFO3res_u.resok.wtmult = 4096;
reply.FSINFO3res_u.resok.dtpref = 128;
reply.FSINFO3res_u.resok.maxfilesize = 0x7fffffffffffffff;
reply.FSINFO3res_u.resok.time_delta.seconds = 1;
reply.FSINFO3res_u.resok.time_delta.nseconds = 0;
reply.FSINFO3res_u.resok.properties = FSF3_SYMLINK | FSF3_HOMOGENEOUS;
}
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_FSINFO3res, sizeof(FSINFO3res));
return 0;
}
static int nfs3_pathconf_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
PATHCONF3args *args = (PATHCONF3args*)call->body.cbody.args;
PATHCONF3res reply;
if (args->object.data.data_len != 10)
{
// Example error
reply.status = NFS3ERR_INVAL;
}
else
{
// Fill info
reply.status = NFS3_OK;
reply.PATHCONF3res_u.resok.obj_attributes.attributes_follow = FALSE;
reply.PATHCONF3res_u.resok.linkmax = 0;
reply.PATHCONF3res_u.resok.name_max = 255;
reply.PATHCONF3res_u.resok.no_trunc = TRUE;
reply.PATHCONF3res_u.resok.chown_restricted = FALSE;
reply.PATHCONF3res_u.resok.case_insensitive = FALSE;
reply.PATHCONF3res_u.resok.case_preserving = TRUE;
}
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_PATHCONF3res, sizeof(PATHCONF3res));
return 0;
}
static int nfs3_commit_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
COMMIT3args *args = (COMMIT3args*)call->body.cbody.args;
COMMIT3res reply = {};
// Just pretend we did fsync :-)
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_COMMIT3res, sizeof(COMMIT3res));
return 0;
}
static int mount3_mnt_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
dirpath *arg = (dirpath*)call->body.cbody.args;
int flavor = AUTH_NONE;
mountres3 reply;
reply.fhs_status = MNT3_OK;
reply.mountres3_u.mountinfo.fhandle.fhandle3_len = 10;
reply.mountres3_u.mountinfo.fhandle.fhandle3_val = "roothandle";
reply.mountres3_u.mountinfo.auth_flavors.auth_flavors_len = 1;
reply.mountres3_u.mountinfo.auth_flavors.auth_flavors_val = &flavor;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_mountres3, sizeof(mountres3));
return 0;
}
static int mount3_dump_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
mountlist reply;
reply = (struct mountbody*)malloc(sizeof(struct mountbody));
reply->ml_hostname = (dirpath)"127.0.0.1";
reply->ml_directory = (dirpath)"/test";
reply->ml_next = NULL;
rpc_send_reply(rpc, call, NULL, (zdrproc_t)zdr_mountlist, sizeof(mountlist));
free(reply);
return 0;
}
static int mount3_umnt_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
dirpath *arg = (dirpath*)call->body.cbody.args;
// do nothing
rpc_send_reply(rpc, call, NULL, (zdrproc_t)zdr_void, 0);
return 0;
}
static int mount3_umntall_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
// do nothing
rpc_send_reply(rpc, call, NULL, (zdrproc_t)zdr_void, 0);
return 0;
}
static int mount3_export_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
nfs_client_t *self = (nfs_client_t*)opaque;
exports reply;
reply = (struct exportnode*)malloc(sizeof(struct exportnode) + sizeof(struct groupnode));
reply->ex_dir = (dirpath)"/test";
reply->ex_groups = (struct groupnode*)(reply+1);
reply->ex_groups->gr_name = (dirpath)"127.0.0.1";
reply->ex_groups->gr_next = NULL;
reply->ex_next = NULL;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_exports, sizeof(exports));
free(reply);
return 0;
}
nfs_client_t::nfs_client_t()
{
struct service_proc nfs3_pt_a[22] = {
{NFS3_NULL, nfs3_null_proc, (zdrproc_t)zdr_void, 0, this},
{NFS3_GETATTR, nfs3_getattr_proc, (zdrproc_t)zdr_GETATTR3args, sizeof(GETATTR3args), this},
{NFS3_SETATTR, nfs3_setattr_proc, (zdrproc_t)zdr_SETATTR3args, sizeof(SETATTR3args), this},
{NFS3_LOOKUP, nfs3_lookup_proc, (zdrproc_t)zdr_LOOKUP3args, sizeof(LOOKUP3args), this},
{NFS3_ACCESS, nfs3_access_proc, (zdrproc_t)zdr_ACCESS3args, sizeof(ACCESS3args), this},
{NFS3_READLINK, nfs3_readlink_proc, (zdrproc_t)zdr_READLINK3args, sizeof(READLINK3args), this},
{NFS3_READ, nfs3_read_proc, (zdrproc_t)zdr_READ3args, sizeof(READ3args), this},
{NFS3_WRITE, nfs3_write_proc, (zdrproc_t)zdr_WRITE3args, sizeof(WRITE3args), this},
{NFS3_CREATE, nfs3_create_proc, (zdrproc_t)zdr_CREATE3args, sizeof(CREATE3args), this},
{NFS3_MKDIR, nfs3_mkdir_proc, (zdrproc_t)zdr_MKDIR3args, sizeof(MKDIR3args), this},
{NFS3_SYMLINK, nfs3_symlink_proc, (zdrproc_t)zdr_SYMLINK3args, sizeof(SYMLINK3args), this},
{NFS3_MKNOD, nfs3_mknod_proc, (zdrproc_t)zdr_MKNOD3args, sizeof(MKNOD3args), this},
{NFS3_REMOVE, nfs3_remove_proc, (zdrproc_t)zdr_REMOVE3args, sizeof(REMOVE3args), this},
{NFS3_RMDIR, nfs3_rmdir_proc, (zdrproc_t)zdr_RMDIR3args, sizeof(RMDIR3args), this},
{NFS3_RENAME, nfs3_rename_proc, (zdrproc_t)zdr_RENAME3args, sizeof(RENAME3args), this},
{NFS3_LINK, nfs3_link_proc, (zdrproc_t)zdr_LINK3args, sizeof(LINK3args), this},
{NFS3_READDIR, nfs3_readdir_proc, (zdrproc_t)zdr_READDIR3args, sizeof(READDIR3args), this},
{NFS3_READDIRPLUS, nfs3_readdirplus_proc, (zdrproc_t)zdr_READDIRPLUS3args, sizeof(READDIRPLUS3args), this},
{NFS3_FSSTAT, nfs3_fsstat_proc, (zdrproc_t)zdr_FSSTAT3args, sizeof(FSSTAT3args), this},
{NFS3_FSINFO, nfs3_fsinfo_proc, (zdrproc_t)zdr_FSINFO3args, sizeof(FSINFO3args), this},
{NFS3_PATHCONF, nfs3_pathconf_proc, (zdrproc_t)zdr_PATHCONF3args, sizeof(PATHCONF3args), this},
{NFS3_COMMIT, nfs3_commit_proc, (zdrproc_t)zdr_COMMIT3args, sizeof(COMMIT3args), this},
};
for (int i = 0; i < sizeof(nfs3_pt_a)/sizeof(service_proc); i++)
{
nfs3_pt.push_back(nfs3_pt_a[i]);
}
struct service_proc nfs3_mount_pt_a[6] = {
{MOUNT3_NULL, nfs3_null_proc, (zdrproc_t)zdr_void, 0, this},
{MOUNT3_MNT, mount3_mnt_proc, (zdrproc_t)zdr_dirpath, sizeof(dirpath), this},
{MOUNT3_DUMP, mount3_dump_proc, (zdrproc_t)zdr_void, 0, this},
{MOUNT3_UMNT, mount3_umnt_proc, (zdrproc_t)zdr_dirpath, sizeof(dirpath), this},
{MOUNT3_UMNTALL, mount3_umntall_proc, (zdrproc_t)zdr_void, 0, this},
{MOUNT3_EXPORT, mount3_export_proc, (zdrproc_t)zdr_void, 0, this},
};
for (int i = 0; i < sizeof(nfs3_mount_pt_a)/sizeof(service_proc); i++)
{
nfs3_mount_pt.push_back(nfs3_mount_pt_a[i]);
}
}
nfs_client_t::~nfs_client_t()
{
if (rpc)
{
rpc_disconnect(rpc, NULL);
rpc_destroy_context(rpc);
}
}

172
src/nfs_portmap.cpp Normal file
View File

@ -0,0 +1,172 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
//
// Portmap service for NFS proxy
#include <netinet/in.h>
#include <string.h>
#include "nfs_portmap.h"
#include "libnfs-raw-portmap.h"
#include "sha256.h"
#include "base64.h"
/*
* The NULL procedure. All protocols/versions must provide a NULL procedure
* as index 0.
* It is used by clients, and rpcinfo, to "ping" a service and verify that
* the service is available and that it does support the indicated version.
*/
static int pmap2_null_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
rpc_send_reply(rpc, call, NULL, (zdrproc_t)zdr_void, 0);
return 0;
}
/*
* v2 GETPORT.
* This is the lookup function for portmapper version 2.
* A client provides program, version and protocol (tcp or udp)
* and portmapper returns which port that service is available on,
* (or 0 if no such program is registered.)
*/
static int pmap2_getport_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
portmap_service_t *self = (portmap_service_t *)opaque;
PMAP2GETPORTargs *args = (PMAP2GETPORTargs *)call->body.cbody.args;
uint32_t port = 0;
auto it = self->reg_ports.lower_bound((portmap_id_t){
.prog = args->prog,
.vers = args->vers,
.udp = args->prot == IPPROTO_UDP,
.ipv6 = false,
});
if (it != self->reg_ports.end() &&
it->prog == args->prog && it->vers == args->vers &&
it->udp == (args->prot == IPPROTO_UDP))
{
port = it->port;
}
rpc_send_reply(rpc, call, &port, (zdrproc_t)zdr_uint32_t, sizeof(uint32_t));
return 0;
}
/*
* v2 DUMP.
* This RPC returns a list of all endpoints that are registered with
* portmapper.
*/
static int pmap2_dump_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
portmap_service_t *self = (portmap_service_t *)opaque;
pmap2_mapping_list *list = new pmap2_mapping_list[self->reg_ports.size()];
int i = 0;
for (auto it = self->reg_ports.begin(); it != self->reg_ports.end(); it++)
{
if (it->ipv6)
continue;
list[i] = {
.map = {
.prog = it->prog,
.vers = it->vers,
.prot = it->udp ? IPPROTO_UDP : IPPROTO_TCP,
.port = it->port,
},
.next = list+i+1,
};
i++;
}
list[i-1].next = NULL;
// Send reply
PMAP2DUMPres reply;
reply.list = list;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_PMAP2DUMPres, sizeof(PMAP2DUMPres));
reply.list = NULL;
delete list;
return 0;
}
/*
* v3 GETADDR.
* This is the lookup function for portmapper version 3.
*/
static int pmap3_getaddr_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
portmap_service_t *self = (portmap_service_t *)opaque;
PMAP3GETADDRargs *args = (PMAP3GETADDRargs *)call->body.cbody.args;
portmap_id_t ref = (portmap_id_t){
.prog = args->prog,
.vers = args->vers,
.udp = !strcmp(args->netid, "udp") || !strcmp(args->netid, "udp6"),
.ipv6 = !strcmp(args->netid, "tcp6") || !strcmp(args->netid, "udp6"),
};
auto it = self->reg_ports.lower_bound(ref);
PMAP3GETADDRres reply;
if (it != self->reg_ports.end() &&
it->prog == ref.prog && it->vers == ref.vers &&
it->udp == ref.udp && it->ipv6 == ref.ipv6)
{
reply.addr = (char*)it->addr.c_str();
}
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_PMAP3GETADDRres, sizeof(PMAP3GETADDRres));
return 0;
}
/*
* v3 DUMP.
* This RPC returns a list of all endpoints that are registered with
* portmapper.
*/
static int pmap3_dump_proc(struct rpc_context *rpc, struct rpc_msg *call, void *opaque)
{
portmap_service_t *self = (portmap_service_t *)opaque;
pmap3_mapping_list *list = new pmap3_mapping_list[self->reg_ports.size()];
int i = 0;
for (auto it = self->reg_ports.begin(); it != self->reg_ports.end(); it++)
{
list[i] = (pmap3_mapping_list){
.map = {
.prog = it->prog,
.vers = it->vers,
.netid = (char*)(it->ipv6
? (it->udp ? "udp6" : "tcp6")
: (it->udp ? "udp" : "tcp")),
.addr = (char*)it->addr.c_str(), // 0.0.0.0.port
.owner = (char*)it->owner.c_str(),
},
.next = list+i+1,
};
i++;
}
list[i-1].next = NULL;
// Send reply
PMAP3DUMPres reply;
reply.list = list;
rpc_send_reply(rpc, call, &reply, (zdrproc_t)zdr_PMAP3DUMPres, sizeof(PMAP3DUMPres));
reply.list = NULL;
delete list;
return 0;
}
portmap_service_t::portmap_service_t()
{
pmap2_pt.push_back((service_proc){PMAP2_NULL, pmap2_null_proc, (zdrproc_t)zdr_void, 0, this});
pmap2_pt.push_back((service_proc){PMAP2_GETPORT, pmap2_getport_proc, (zdrproc_t)zdr_PMAP2GETPORTargs, sizeof(PMAP2GETPORTargs), this});
pmap2_pt.push_back((service_proc){PMAP2_DUMP, pmap2_dump_proc, (zdrproc_t)zdr_void, 0, this});
pmap3_pt.push_back((service_proc){PMAP3_NULL, pmap2_null_proc, (zdrproc_t)zdr_void, 0, this});
pmap3_pt.push_back((service_proc){PMAP3_GETADDR, pmap3_getaddr_proc, (zdrproc_t)zdr_PMAP3GETADDRargs, sizeof(PMAP3GETADDRargs), this});
pmap3_pt.push_back((service_proc){PMAP3_DUMP, pmap3_dump_proc, (zdrproc_t)zdr_void, 0, this});
}
std::string sha256(const std::string & str)
{
std::string hash;
hash.resize(32);
SHA256_CTX ctx;
sha256_init(&ctx);
sha256_update(&ctx, (uint8_t*)str.data(), str.size());
sha256_final(&ctx, (uint8_t*)hash.data());
return hash;
}

41
src/nfs_portmap.h Normal file
View File

@ -0,0 +1,41 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
//
// Portmap service for NFS proxy
#pragma once
#include <string>
#include <set>
#include <vector>
#include "nfsc/libnfs.h"
#include "nfsc/libnfs-raw.h"
struct portmap_id_t
{
unsigned prog, vers;
bool udp;
bool ipv6;
unsigned port;
std::string owner;
std::string addr;
};
class portmap_service_t
{
public:
std::set<portmap_id_t> reg_ports;
std::vector<service_proc> pmap2_pt;
std::vector<service_proc> pmap3_pt;
portmap_service_t();
};
inline bool operator < (const portmap_id_t &a, const portmap_id_t &b)
{
return a.prog < b.prog || a.prog == b.prog && a.vers < b.vers ||
a.prog == b.prog && a.vers == b.vers && a.udp < b.udp ||
a.prog == b.prog && a.vers == b.vers && a.udp == b.udp && a.ipv6 < b.ipv6;
}
std::string sha256(const std::string & str);

301
src/nfs_proxy.cpp Normal file
View File

@ -0,0 +1,301 @@
// Copyright (c) Vitaliy Filippov, 2019+
// License: VNPL-1.1 (see README.md for details)
//
// Simplified NFS proxy
// Presents all images as files, stores small files directly in etcd
// Keeps image list in memory and thus is unsuitable for a lot of files
#include <netinet/tcp.h>
#include <sys/epoll.h>
#include <sys/poll.h>
#include <unistd.h>
#include <fcntl.h>
//#include <signal.h>
#include "libnfs-raw-mount.h"
#include "libnfs-raw-nfs.h"
#include "libnfs-raw-portmap.h"
#include "addr_util.h"
#include "base64.h"
#include "nfs_proxy.h"
const char *exe_name = NULL;
nfs_proxy_t::~nfs_proxy_t()
{
if (cli)
delete cli;
if (epmgr)
delete epmgr;
if (ringloop)
delete ringloop;
}
json11::Json::object nfs_proxy_t::parse_args(int narg, const char *args[])
{
json11::Json::object cfg;
for (int i = 1; i < narg; i++)
{
if (!strcmp(args[i], "-h") || !strcmp(args[i], "--help"))
{
printf(
"Vitastor NFS 3.0 proxy\n"
"(c) Vitaliy Filippov, 2021-2022 (VNPL-1.1)\n\n"
"USAGE:\n"
" %s [--etcd_address ADDR] [OTHER OPTIONS]\n",
exe_name
);
exit(0);
}
else if (args[i][0] == '-' && args[i][1] == '-')
{
const char *opt = args[i]+2;
cfg[opt] = !strcmp(opt, "json") || i == narg-1 ? "1" : args[++i];
}
}
return cfg;
}
void nfs_proxy_t::run(json11::Json cfg)
{
bind_address = cfg["bind_address"].string_value();
if (bind_address == "")
bind_address = "0.0.0.0";
// Create client
ringloop = new ring_loop_t(512);
epmgr = new epoll_manager_t(ringloop);
cli = new cluster_client_t(ringloop, epmgr->tfd, cfg);
// We need inode name hashes for NFS handles to remain stateless and <= 64 bytes long
dir_mod_rev[""] = 0;
dir_ids[""] = 1;
assert(cli->st_cli.on_inode_change_hook == NULL);
cli->st_cli.on_inode_change_hook = [this](inode_t changed_inode, bool removed)
{
if (removed)
{
auto ino_it = hash_by_inode.find(changed_inode);
if (ino_it != hash_by_inode.end())
{
inode_by_hash.erase(ino_it->second);
hash_by_inode.erase(ino_it);
}
// FIXME also calculate dir_mod_rev
}
else
{
auto & inode_cfg = cli->st_cli.inode_config.at(changed_inode);
std::string name = inode_cfg.name;
if (name_prefix != "")
{
if (name.substr(0, name_prefix.size()) != name_prefix)
return;
name = name.substr(name_prefix.size());
}
dir_mod_rev[""] = dir_mod_rev[""] < inode_cfg.mod_revision ? inode_cfg.mod_revision : dir_mod_rev[""];
std::string hash = "S"+base64_encode(sha256(name));
int pos = name.find('/');
while (pos >= 0)
{
std::string dir = name.substr(0, pos);
if (dir_ids.find(dir) == dir_ids.end())
dir_ids[dir] = next_dir_id++;
dir_mod_rev[dir] = dir_mod_rev[dir] < inode_cfg.mod_revision ? inode_cfg.mod_revision : dir_mod_rev[dir];
dir_by_hash["S"+base64_encode(sha256(dir))] = dir;
int next = name.substr(pos+1).find('/');
pos = next < 0 ? -1 : pos+1+next;
}
auto hbi_it = hash_by_inode.find(changed_inode);
if (hbi_it != hash_by_inode.end() && hbi_it->second != hash)
{
// inode had a different name, remove old hash=>inode pointer
inode_by_hash.erase(hbi_it->second);
}
inode_by_hash[hash] = changed_inode;
hash_by_inode[changed_inode] = hash;
}
};
// Load image metadata
while (!cli->is_ready())
{
ringloop->loop();
if (cli->is_ready())
break;
ringloop->wait();
}
// Create portmap socket
int portmap_socket = create_and_bind_socket(bind_address, 111, 128, NULL);
fcntl(portmap_socket, F_SETFL, fcntl(portmap_socket, F_GETFL, 0) | O_NONBLOCK);
// Create NFS socket
int nfs_socket = create_and_bind_socket(bind_address, 2049, 128, NULL);
fcntl(nfs_socket, F_SETFL, fcntl(nfs_socket, F_GETFL, 0) | O_NONBLOCK);
// Self-register portmap and NFS
pmap.reg_ports.insert((portmap_id_t){
.prog = PMAP_PROGRAM,
.vers = PMAP_V2,
.port = 111,
.owner = "portmapper-service",
.addr = "0.0.0.0.0.111",
});
pmap.reg_ports.insert((portmap_id_t){
.prog = PMAP_PROGRAM,
.vers = PMAP_V3,
.port = 111,
.owner = "portmapper-service",
.addr = "0.0.0.0.0.111",
});
pmap.reg_ports.insert((portmap_id_t){
.prog = NFS_PROGRAM,
.vers = NFS_V3,
.port = 2049,
.owner = "nfs-server",
.addr = "0.0.0.0.0.2049",
});
pmap.reg_ports.insert((portmap_id_t){
.prog = MOUNT_PROGRAM,
.vers = MOUNT_V3,
.port = 2049,
.owner = "rpc.mountd",
.addr = "0.0.0.0.0.2049",
});
// Add FDs to epoll
epmgr->tfd->set_fd_handler(portmap_socket, false, [this](int portmap_socket, int epoll_events)
{
if (epoll_events & EPOLLRDHUP)
{
fprintf(stderr, "Listening portmap socket disconnected, exiting\n");
exit(1);
}
else
{
do_accept(portmap_socket);
}
});
epmgr->tfd->set_fd_handler(nfs_socket, false, [this](int nfs_socket, int epoll_events)
{
if (epoll_events & EPOLLRDHUP)
{
fprintf(stderr, "Listening portmap socket disconnected, exiting\n");
exit(1);
}
else
{
do_accept(nfs_socket);
}
});
if (cfg["foreground"].is_null())
{
daemonize();
}
while (true)
{
ringloop->loop();
ringloop->wait();
}
/*// Sync at the end
cluster_op_t *close_sync = new cluster_op_t;
close_sync->opcode = OSD_OP_SYNC;
close_sync->callback = [&stop](cluster_op_t *op)
{
stop = true;
delete op;
};
cli->execute(close_sync);*/
// Destroy the client
delete cli;
delete epmgr;
delete ringloop;
cli = NULL;
epmgr = NULL;
ringloop = NULL;
}
void nfs_proxy_t::do_accept(int listen_fd)
{
struct sockaddr_storage addr;
socklen_t addr_size = sizeof(addr);
int nfs_fd = 0;
while ((nfs_fd = accept(listen_fd, (struct sockaddr *)&addr, &addr_size)) >= 0)
{
fprintf(stderr, "New client %d: connection from %s\n", nfs_fd, addr_to_string(addr).c_str());
fcntl(nfs_fd, F_SETFL, fcntl(nfs_fd, F_GETFL, 0) | O_NONBLOCK);
int one = 1;
setsockopt(nfs_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
auto cli = new nfs_client_t();
cli->parent = this;
cli->nfs_fd = nfs_fd;
cli->rpc = rpc_init_server_context(nfs_fd);
if (!cli->rpc)
{
delete cli;
close(nfs_fd);
fprintf(stderr, "Failed to init libnfs server context\n");
exit(1);
}
// Use both portmap and NFS everywhere
rpc_register_service(cli->rpc, PMAP_PROGRAM, PMAP_V2, pmap.pmap2_pt.data(), pmap.pmap2_pt.size());
rpc_register_service(cli->rpc, PMAP_PROGRAM, PMAP_V3, pmap.pmap3_pt.data(), pmap.pmap3_pt.size());
rpc_register_service(cli->rpc, NFS_PROGRAM, NFS_V3, cli->nfs3_pt.data(), cli->nfs3_pt.size());
rpc_register_service(cli->rpc, MOUNT_PROGRAM, MOUNT_V3, cli->nfs3_mount_pt.data(), cli->nfs3_mount_pt.size());
epmgr->tfd->set_fd_handler(nfs_fd, true, [this, cli](int nfs_fd, int epoll_events)
{
// Handle incoming event
if (epoll_events & EPOLLRDHUP)
{
fprintf(stderr, "Client %d disconnected\n", nfs_fd);
epmgr->tfd->set_fd_handler(cli->nfs_fd, true, NULL);
delete cli;
close(nfs_fd);
return;
}
int revents = 0;
if (epoll_events & EPOLLIN)
revents |= POLLIN;
if (epoll_events & EPOLLOUT)
revents |= POLLOUT;
// Let libnfs process the event
if (rpc_service(cli->rpc, revents) < 0)
{
fprintf(stderr, "libnfs error: %s, disconnecting client %d\n", rpc_get_error(cli->rpc), nfs_fd);
epmgr->tfd->set_fd_handler(cli->nfs_fd, true, NULL);
delete cli;
close(nfs_fd);
return;
}
// FIXME Add/remove events based on rpc_which_events(rpc) ?
});
}
if (nfs_fd < 0 && errno != EAGAIN)
{
fprintf(stderr, "Failed to accept connection: %s\n", strerror(errno));
exit(1);
}
}
void nfs_proxy_t::daemonize()
{
if (fork())
exit(0);
setsid();
if (fork())
exit(0);
if (chdir("/") != 0)
fprintf(stderr, "Warning: Failed to chdir into /\n");
close(0);
close(1);
close(2);
open("/dev/null", O_RDONLY);
open("/dev/null", O_WRONLY);
open("/dev/null", O_WRONLY);
}
int main(int narg, const char *args[])
{
setvbuf(stdout, NULL, _IONBF, 0);
setvbuf(stderr, NULL, _IONBF, 0);
exe_name = args[0];
nfs_proxy_t *p = new nfs_proxy_t();
p->run(nfs_proxy_t::parse_args(narg, args));
delete p;
return 0;
}

47
src/nfs_proxy.h Normal file
View File

@ -0,0 +1,47 @@
#pragma once
#include "cluster_client.h"
#include "epoll_manager.h"
#include "nfs_portmap.h"
#include "nfsc/libnfs-raw.h"
class nfs_proxy_t
{
public:
std::string bind_address;
std::string name_prefix;
int fsid = 1;
portmap_service_t pmap;
ring_loop_t *ringloop = NULL;
epoll_manager_t *epmgr = NULL;
cluster_client_t *cli = NULL;
uint64_t next_dir_id = 2;
std::map<std::string, std::string> dir_by_hash;
std::map<std::string, uint64_t> dir_ids;
std::map<std::string, uint64_t> dir_mod_rev;
std::map<inode_t, std::string> hash_by_inode;
std::map<std::string, inode_t> inode_by_hash;
~nfs_proxy_t();
static json11::Json::object parse_args(int narg, const char *args[]);
void run(json11::Json cfg);
void do_accept(int listen_fd);
void daemonize();
};
class nfs_client_t
{
public:
nfs_proxy_t *parent = NULL;
int nfs_fd;
struct rpc_context *rpc = NULL;
std::vector<service_proc> nfs3_pt;
std::vector<service_proc> nfs3_mount_pt;
nfs_client_t();
~nfs_client_t();
};

158
src/sha256.c Normal file
View File

@ -0,0 +1,158 @@
/*********************************************************************
* Filename: sha256.c
* Author: Brad Conte (brad AT bradconte.com)
* Copyright:
* Disclaimer: This code is presented "as is" without any guarantees.
* Details: Implementation of the SHA-256 hashing algorithm.
SHA-256 is one of the three algorithms in the SHA2
specification. The others, SHA-384 and SHA-512, are not
offered in this implementation.
Algorithm specification can be found here:
* http://csrc.nist.gov/publications/fips/fips180-2/fips180-2withchangenotice.pdf
This implementation uses little endian byte order.
*********************************************************************/
/*************************** HEADER FILES ***************************/
#include <stdlib.h>
#include <memory.h>
#include "sha256.h"
/****************************** MACROS ******************************/
#define ROTLEFT(a,b) (((a) << (b)) | ((a) >> (32-(b))))
#define ROTRIGHT(a,b) (((a) >> (b)) | ((a) << (32-(b))))
#define CH(x,y,z) (((x) & (y)) ^ (~(x) & (z)))
#define MAJ(x,y,z) (((x) & (y)) ^ ((x) & (z)) ^ ((y) & (z)))
#define EP0(x) (ROTRIGHT(x,2) ^ ROTRIGHT(x,13) ^ ROTRIGHT(x,22))
#define EP1(x) (ROTRIGHT(x,6) ^ ROTRIGHT(x,11) ^ ROTRIGHT(x,25))
#define SIG0(x) (ROTRIGHT(x,7) ^ ROTRIGHT(x,18) ^ ((x) >> 3))
#define SIG1(x) (ROTRIGHT(x,17) ^ ROTRIGHT(x,19) ^ ((x) >> 10))
/**************************** VARIABLES *****************************/
static const WORD k[64] = {
0x428a2f98,0x71374491,0xb5c0fbcf,0xe9b5dba5,0x3956c25b,0x59f111f1,0x923f82a4,0xab1c5ed5,
0xd807aa98,0x12835b01,0x243185be,0x550c7dc3,0x72be5d74,0x80deb1fe,0x9bdc06a7,0xc19bf174,
0xe49b69c1,0xefbe4786,0x0fc19dc6,0x240ca1cc,0x2de92c6f,0x4a7484aa,0x5cb0a9dc,0x76f988da,
0x983e5152,0xa831c66d,0xb00327c8,0xbf597fc7,0xc6e00bf3,0xd5a79147,0x06ca6351,0x14292967,
0x27b70a85,0x2e1b2138,0x4d2c6dfc,0x53380d13,0x650a7354,0x766a0abb,0x81c2c92e,0x92722c85,
0xa2bfe8a1,0xa81a664b,0xc24b8b70,0xc76c51a3,0xd192e819,0xd6990624,0xf40e3585,0x106aa070,
0x19a4c116,0x1e376c08,0x2748774c,0x34b0bcb5,0x391c0cb3,0x4ed8aa4a,0x5b9cca4f,0x682e6ff3,
0x748f82ee,0x78a5636f,0x84c87814,0x8cc70208,0x90befffa,0xa4506ceb,0xbef9a3f7,0xc67178f2
};
/*********************** FUNCTION DEFINITIONS ***********************/
void sha256_transform(SHA256_CTX *ctx, const BYTE data[])
{
WORD a, b, c, d, e, f, g, h, i, j, t1, t2, m[64];
for (i = 0, j = 0; i < 16; ++i, j += 4)
m[i] = (data[j] << 24) | (data[j + 1] << 16) | (data[j + 2] << 8) | (data[j + 3]);
for ( ; i < 64; ++i)
m[i] = SIG1(m[i - 2]) + m[i - 7] + SIG0(m[i - 15]) + m[i - 16];
a = ctx->state[0];
b = ctx->state[1];
c = ctx->state[2];
d = ctx->state[3];
e = ctx->state[4];
f = ctx->state[5];
g = ctx->state[6];
h = ctx->state[7];
for (i = 0; i < 64; ++i) {
t1 = h + EP1(e) + CH(e,f,g) + k[i] + m[i];
t2 = EP0(a) + MAJ(a,b,c);
h = g;
g = f;
f = e;
e = d + t1;
d = c;
c = b;
b = a;
a = t1 + t2;
}
ctx->state[0] += a;
ctx->state[1] += b;
ctx->state[2] += c;
ctx->state[3] += d;
ctx->state[4] += e;
ctx->state[5] += f;
ctx->state[6] += g;
ctx->state[7] += h;
}
void sha256_init(SHA256_CTX *ctx)
{
ctx->datalen = 0;
ctx->bitlen = 0;
ctx->state[0] = 0x6a09e667;
ctx->state[1] = 0xbb67ae85;
ctx->state[2] = 0x3c6ef372;
ctx->state[3] = 0xa54ff53a;
ctx->state[4] = 0x510e527f;
ctx->state[5] = 0x9b05688c;
ctx->state[6] = 0x1f83d9ab;
ctx->state[7] = 0x5be0cd19;
}
void sha256_update(SHA256_CTX *ctx, const BYTE data[], size_t len)
{
WORD i;
for (i = 0; i < len; ++i) {
ctx->data[ctx->datalen] = data[i];
ctx->datalen++;
if (ctx->datalen == 64) {
sha256_transform(ctx, ctx->data);
ctx->bitlen += 512;
ctx->datalen = 0;
}
}
}
void sha256_final(SHA256_CTX *ctx, BYTE hash[])
{
WORD i;
i = ctx->datalen;
// Pad whatever data is left in the buffer.
if (ctx->datalen < 56) {
ctx->data[i++] = 0x80;
while (i < 56)
ctx->data[i++] = 0x00;
}
else {
ctx->data[i++] = 0x80;
while (i < 64)
ctx->data[i++] = 0x00;
sha256_transform(ctx, ctx->data);
memset(ctx->data, 0, 56);
}
// Append to the padding the total message's length in bits and transform.
ctx->bitlen += ctx->datalen * 8;
ctx->data[63] = ctx->bitlen;
ctx->data[62] = ctx->bitlen >> 8;
ctx->data[61] = ctx->bitlen >> 16;
ctx->data[60] = ctx->bitlen >> 24;
ctx->data[59] = ctx->bitlen >> 32;
ctx->data[58] = ctx->bitlen >> 40;
ctx->data[57] = ctx->bitlen >> 48;
ctx->data[56] = ctx->bitlen >> 56;
sha256_transform(ctx, ctx->data);
// Since this implementation uses little endian byte ordering and SHA uses big endian,
// reverse all the bytes when copying the final state to the output hash.
for (i = 0; i < 4; ++i) {
hash[i] = (ctx->state[0] >> (24 - i * 8)) & 0x000000ff;
hash[i + 4] = (ctx->state[1] >> (24 - i * 8)) & 0x000000ff;
hash[i + 8] = (ctx->state[2] >> (24 - i * 8)) & 0x000000ff;
hash[i + 12] = (ctx->state[3] >> (24 - i * 8)) & 0x000000ff;
hash[i + 16] = (ctx->state[4] >> (24 - i * 8)) & 0x000000ff;
hash[i + 20] = (ctx->state[5] >> (24 - i * 8)) & 0x000000ff;
hash[i + 24] = (ctx->state[6] >> (24 - i * 8)) & 0x000000ff;
hash[i + 28] = (ctx->state[7] >> (24 - i * 8)) & 0x000000ff;
}
}

41
src/sha256.h Normal file
View File

@ -0,0 +1,41 @@
/*********************************************************************
* Filename: sha256.h
* Author: Brad Conte (brad AT bradconte.com)
* Copyright:
* Disclaimer: This code is presented "as is" without any guarantees.
* Details: Defines the API for the corresponding SHA1 implementation.
*********************************************************************/
#ifndef SHA256_H
#define SHA256_H
/*************************** HEADER FILES ***************************/
#include <stddef.h>
/****************************** MACROS ******************************/
#define SHA256_BLOCK_SIZE 32 // SHA256 outputs a 32 byte digest
#ifdef __cplusplus
extern "C" {
#endif
/**************************** DATA TYPES ****************************/
typedef unsigned char BYTE; // 8-bit byte
typedef unsigned int WORD; // 32-bit word, change to "long" for 16-bit machines
typedef struct {
BYTE data[64];
WORD datalen;
unsigned long long bitlen;
WORD state[8];
} SHA256_CTX;
/*********************** FUNCTION DECLARATIONS **********************/
void sha256_init(SHA256_CTX *ctx);
void sha256_update(SHA256_CTX *ctx, const BYTE data[], size_t len);
void sha256_final(SHA256_CTX *ctx, BYTE hash[]);
#ifdef __cplusplus
};
#endif
#endif // SHA256_H