Merge cdc_stream into asset_stream_manager (#26)

Adds start and stop commands to asset_stream_manager.
asset_stream_manager will be renamed to cdc_stream next.

asset_stream_manager/BUILD

@@ -9,27 +9,71 @@ cc_binary(
     srcs = ["main.cc"],
     data = [":roots_pem"],
     deps = [
-        ":commands",
-        "//cdc_stream",
+        ":start_command",
+        ":start_service_command",
+        ":stop_command",
         "//common:log",
         "//common:path",
     ],
 )
 
 cc_library(
-    name = "commands",
-    srcs = [
-        "base_command.cc",
-        "start_service_command.cc",
-    ],
-    hdrs = [
-        "base_command.h",
-        "start_service_command.h",
+    name = "base_command",
+    srcs = ["base_command.cc"],
+    hdrs = ["base_command.h"],
+    deps = [
+        "//absl_helper:jedec_size_flag",
+        "@com_github_lyra//:lyra",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings:str_format",
     ],
+)
+
+cc_library(
+    name = "start_service_command",
+    srcs = ["start_service_command.cc"],
+    hdrs = ["start_service_command.h"],
     deps = [
         ":asset_stream_config",
+        ":base_command",
         ":session_management_server",
-        "@com_github_lyra//:lyra",
+    ],
+)
+
+cc_library(
+    name = "start_command",
+    srcs = ["start_command.cc"],
+    hdrs = ["start_command.h"],
+    deps = [
+        ":base_command",
+        ":local_assets_stream_manager_client",
+        ":session_management_server",
+        "//common:path",
+        "//common:status_macros",
+    ],
+)
+
+cc_library(
+    name = "stop_command",
+    srcs = ["stop_command.cc"],
+    hdrs = ["stop_command.h"],
+    deps = [
+        ":base_command",
+        ":local_assets_stream_manager_client",
+        ":session_management_server",
+        "//common:path",
+        "//common:remote_util",
+        "//common:status_macros",
+    ],
+)
+
+cc_library(
+    name = "local_assets_stream_manager_client",
+    srcs = ["local_assets_stream_manager_client.cc"],
+    hdrs = ["local_assets_stream_manager_client.h"],
+    deps = [
+        "//common:grpc_status",
+        "//proto:local_assets_stream_manager_grpc_proto",
         "@com_google_absl//absl/status",
     ],
 )
@@ -66,8 +110,8 @@ cc_library(
     srcs = ["asset_stream_config.cc"],
     hdrs = ["asset_stream_config.h"],
     deps = [
+        ":base_command",
         ":multi_session",
-        "//absl_helper:jedec_size_flag",
         "//common:log",
         "//common:path",
         "//common:status_macros",

asset_stream_manager/asset_stream_config.cc

@@ -19,6 +19,7 @@
 #include "absl/strings/str_format.h"
 #include "absl/strings/str_join.h"
 #include "absl_helper/jedec_size_flag.h"
+#include "asset_stream_manager/base_command.h"
 #include "common/buffer.h"
 #include "common/path.h"
 #include "common/status_macros.h"
@@ -38,12 +39,14 @@ AssetStreamConfig::AssetStreamConfig() = default;
 
 AssetStreamConfig::~AssetStreamConfig() = default;
 
-void AssetStreamConfig::RegisterCommandLineFlags(lyra::command& cmd) {
+void AssetStreamConfig::RegisterCommandLineFlags(lyra::command& cmd,
+                                                 BaseCommand& base_command) {
   session_cfg_.verbosity = kDefaultVerbosity;
   cmd.add_argument(lyra::opt(session_cfg_.verbosity, "num")
                        .name("--verbosity")
                        .help("Verbosity of the log output, default: " +
-                             std::to_string(kDefaultVerbosity)));
+                             std::to_string(kDefaultVerbosity) +
+                             ". Increase to make logs more verbose."));
 
   cmd.add_argument(
       lyra::opt(session_cfg_.stats)
@@ -84,13 +87,14 @@ void AssetStreamConfig::RegisterCommandLineFlags(lyra::command& cmd) {
                        .help("Check FUSE consistency and log check results"));
 
   session_cfg_.fuse_cache_capacity = DiskDataStore::kDefaultCapacity;
-  cmd.add_argument(lyra::opt(JedecParser("--cache-capacity",
+  cmd.add_argument(
+      lyra::opt(base_command.JedecParser("--cache-capacity",
                                          &session_cfg_.fuse_cache_capacity),
-                             "bytes")
-                       .name("--cache-capacity")
-                       .help("FUSE cache capacity, default: " +
-                             std::to_string(DiskDataStore::kDefaultCapacity) +
-                             ". Supports common unit suffixes K, M, G."));
+                "bytes")
+          .name("--cache-capacity")
+          .help("FUSE cache capacity, default: " +
+                std::to_string(DiskDataStore::kDefaultCapacity) +
+                ". Supports common unit suffixes K, M, G."));
 
   session_cfg_.fuse_cleanup_timeout_sec = DataProvider::kCleanupTimeoutSec;
   cmd.add_argument(
@@ -251,18 +255,5 @@ std::string AssetStreamConfig::GetFlagReadErrors() {
                                  error_str.empty() ? "" : "\n", flag, error);
   return error_str;
 }
-std::function<void(const std::string&)> AssetStreamConfig::JedecParser(
-    const char* flag_name, uint64_t* bytes) {
-  return [flag_name, bytes,
-          error = &jedec_parse_error_](const std::string& value) {
-    JedecSize size;
-    if (AbslParseFlag(value, &size, error)) {
-      *bytes = size.Size();
-    } else {
-      *error = absl::StrFormat("Failed to parse --%s=%s: %s", flag_name, value,
-                               *error);
-    }
-  };
-}
 
 }  // namespace cdc_ft

asset_stream_manager/asset_stream_config.h

@@ -31,6 +31,8 @@ class command;
 
 namespace cdc_ft {
 
+class BaseCommand;
+
 // Class containing all configuration settings for asset streaming.
 // Reads flags from the command line and optionally applies overrides from
 // a json file.
@@ -41,7 +43,7 @@ class AssetStreamConfig {
   ~AssetStreamConfig();
 
   // Registers arguments with Lyra.
-  void RegisterCommandLineFlags(lyra::command& cmd);
+  void RegisterCommandLineFlags(lyra::command& cmd, BaseCommand& base_command);
 
   // Loads a configuration from the JSON file at |path| and overrides any config
   // values that are set in this file. Sample json file:
@@ -88,16 +90,7 @@ class AssetStreamConfig {
   // Whether to log to a file or to stdout.
   bool log_to_stdout() const { return log_to_stdout_; }
 
-  // Workaround for Lyra not accepting errors from parsers.
-  const std::string& jedec_parse_error() const { return jedec_parse_error_; }
-
  private:
-  // Jedec parser for Lyra options. Usage:
-  //   lyra::opt(JedecParser("size-flag", &size_bytes), "bytes"))
-  // Sets jedec_parse_error_ on error, Lyra doesn't support errors from lambdas.
-  std::function<void(const std::string&)> JedecParser(const char* flag_name,
-                                                      uint64_t* bytes);
-
   SessionConfig session_cfg_;
   bool log_to_stdout_ = false;
 
@@ -111,9 +104,6 @@ class AssetStreamConfig {
 
   // Maps flags to errors occurred while reading this flag.
   std::map<std::string, std::string> flag_read_errors_;
-
-  // Errors from parsing JEDEC sizes.
-  std::string jedec_parse_error_;
 };
 
 };  // namespace cdc_ft

asset_stream_manager/asset_stream_manager.vcxproj

@@ -66,7 +66,7 @@
   </ItemDefinitionGroup>
   <!-- Bazel setup -->
   <PropertyGroup>
-    <BazelTargets>//asset_stream_manager //cdc_stream</BazelTargets>
+    <BazelTargets>//asset_stream_manager</BazelTargets>
     <BazelOutputFile>asset_stream_manager.exe</BazelOutputFile>
     <BazelIncludePaths>..\;..\third_party\absl;..\bazel-cdc-file-transfer\external\com_github_jsoncpp\include;..\bazel-cdc-file-transfer\external\com_github_blake3\c;..\third_party\googletest\googletest\include;..\bazel-cdc-file-transfer\external\com_google_protobuf\src;..\bazel-cdc-file-transfer\external\com_github_grpc_grpc\include;..\bazel-out\x64_windows-dbg\bin;..\bazel-cdc-file-transfer\external\com_github_lyra\include;$(VC_IncludePath);$(WindowsSDK_IncludePath)</BazelIncludePaths>
   </PropertyGroup>

asset_stream_manager/base_command.cc

@@ -14,6 +14,8 @@
 
 #include "asset_stream_manager/base_command.h"
 
+#include "absl/strings/str_format.h"
+#include "absl_helper/jedec_size_flag.h"
 #include "lyra/lyra.hpp"
 
 namespace cdc_ft {
@@ -33,14 +35,38 @@ void BaseCommand::Register(lyra::cli& cli) {
 
   RegisterCommandLineFlags(cmd);
 
-  // Workaround for Lyra treating --unknown_flags as positional argument.
-  // If this argument is not empty, it's an unknown arg.
-  cmd.add_argument(lyra::arg(invalid_arg_, ""));
+  // Detect extra positional args.
+  cmd.add_argument(lyra::arg(PosArgValidator(&extra_positional_arg_), ""));
 
   // Register command with CLI.
   cli.add_argument(std::move(cmd));
 }
 
+std::function<void(const std::string&)> BaseCommand::JedecParser(
+    const char* flag_name, uint64_t* bytes) {
+  return [flag_name, bytes,
+          error = &jedec_parse_error_](const std::string& value) {
+    JedecSize size;
+    if (AbslParseFlag(value, &size, error)) {
+      *bytes = size.Size();
+    } else {
+      *error = absl::StrFormat("Failed to parse %s=%s: %s", flag_name, value,
+                               *error);
+    }
+  };
+}
+
+std::function<void(const std::string&)> BaseCommand::PosArgValidator(
+    std::string* str) {
+  return [str, invalid_arg = &invalid_arg_](const std::string& value) {
+    if (!value.empty() && value[0] == '-') {
+      *invalid_arg = value;
+    } else {
+      *str = value;
+    }
+  };
+}
+
 void BaseCommand::CommandHandler(const lyra::group& g) {
   // Handle -h, --help.
   if (show_help_) {
@@ -57,6 +83,19 @@ void BaseCommand::CommandHandler(const lyra::group& g) {
     return;
   }
 
+  if (!jedec_parse_error_.empty()) {
+    std::cerr << "Error: " << jedec_parse_error_ << std::endl;
+    *exit_code_ = 1;
+    return;
+  }
+
+  if (!extra_positional_arg_.empty()) {
+    std::cerr << "Error: Extraneous positional argument '"
+              << extra_positional_arg_ << "'. Try -h for help." << std::endl;
+    *exit_code_ = 1;
+    return;
+  }
+
   // Run and print error.
   absl::Status status = Run();
   if (!status.ok()) {

asset_stream_manager/base_command.h

@@ -42,6 +42,18 @@ class BaseCommand {
   // Registers the command with Lyra. Must be called before parsing args.
   void Register(lyra::cli& cli);
 
+  // Jedec parser for Lyra options. Usage:
+  //   lyra::opt(JedecParser("size-flag", &size_bytes), "bytes"))
+  // Automatically reports a parse failure on error.
+  std::function<void(const std::string&)> JedecParser(const char* flag_name,
+                                                      uint64_t* bytes);
+
+  // Validator that should be used for all positional arguments. Lyra interprets
+  // -u, --unknown_flag as positional argument. This validator makes sure that
+  // a positional argument starting with - is reported as an error. Otherwise,
+  // writes the value to |str|.
+  std::function<void(const std::string&)> PosArgValidator(std::string* str);
+
  protected:
   // Adds all optional and required arguments used by the command.
   // Called by Register().
@@ -66,6 +78,13 @@ class BaseCommand {
   // Workaround for invalid args. Lyra doesn't interpret --invalid as invalid
   // argument, but as positional argument "--invalid".
   std::string invalid_arg_;
+
+  // Extraneous positional args. Gets reported as error if present.
+  std::string extra_positional_arg_;
+
+  // Errors from parsing JEDEC sizes.
+  // Works around Lyra not accepting errors from parsers.
+  std::string jedec_parse_error_;
 };
 
 }  // namespace cdc_ft

asset_stream_manager/local_assets_stream_manager_client.cc

@@ -0,0 +1,95 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "asset_stream_manager/local_assets_stream_manager_client.h"
+
+#include <vector>
+
+#include "absl/status/status.h"
+#include "absl/strings/str_format.h"
+#include "absl/strings/str_split.h"
+#include "common/grpc_status.h"
+
+namespace cdc_ft {
+
+using StartSessionRequest = localassetsstreammanager::StartSessionRequest;
+using StartSessionResponse = localassetsstreammanager::StartSessionResponse;
+using StopSessionRequest = localassetsstreammanager::StopSessionRequest;
+using StopSessionResponse = localassetsstreammanager::StopSessionResponse;
+
+LocalAssetsStreamManagerClient::LocalAssetsStreamManagerClient(
+    uint16_t service_port) {
+  std::string client_address = absl::StrFormat("localhost:%u", service_port);
+  std::shared_ptr<grpc::Channel> channel = grpc::CreateCustomChannel(
+      client_address, grpc::InsecureChannelCredentials(),
+      grpc::ChannelArguments());
+  stub_ = LocalAssetsStreamManager::NewStub(std::move(channel));
+}
+
+LocalAssetsStreamManagerClient::LocalAssetsStreamManagerClient(
+    std::shared_ptr<grpc::Channel> channel) {
+  stub_ = LocalAssetsStreamManager::NewStub(std::move(channel));
+}
+
+LocalAssetsStreamManagerClient::~LocalAssetsStreamManagerClient() = default;
+
+absl::Status LocalAssetsStreamManagerClient::StartSession(
+    const std::string& src_dir, const std::string& user_host, uint16_t ssh_port,
+    const std::string& mount_dir, const std::string& ssh_command,
+    const std::string& scp_command) {
+  StartSessionRequest request;
+  request.set_workstation_directory(src_dir);
+  request.set_user_host(user_host);
+  request.set_port(ssh_port);
+  request.set_mount_dir(mount_dir);
+  request.set_ssh_command(ssh_command);
+  request.set_scp_command(scp_command);
+
+  grpc::ClientContext context;
+  StartSessionResponse response;
+  return ToAbslStatus(stub_->StartSession(&context, request, &response));
+}
+
+absl::Status LocalAssetsStreamManagerClient::StopSession(
+    const std::string& user_host, const std::string& mount_dir) {
+  StopSessionRequest request;
+  request.set_user_host(user_host);
+  request.set_mount_dir(mount_dir);
+
+  grpc::ClientContext context;
+  StopSessionResponse response;
+  return ToAbslStatus(stub_->StopSession(&context, request, &response));
+}
+
+// static
+absl::Status LocalAssetsStreamManagerClient::ParseUserHostDir(
+    const std::string& user_host_dir, std::string* user_host,
+    std::string* dir) {
+  std::vector<std::string> parts =
+      absl::StrSplit(user_host_dir, absl::MaxSplits(':', 1));
+  if (parts.size() < 2 ||
+      (parts[0].size() == 1 && toupper(parts[0][0]) >= 'A' &&
+       toupper(parts[0][0]) <= 'Z')) {
+    return absl::InvalidArgumentError(
+        absl::StrFormat("Failed to parse '%s'. Make sure it is of the form "
+                        "[user@]host:linux_dir.",
+                        user_host_dir));
+  }
+
+  *user_host = parts[0];
+  *dir = parts[1];
+  return absl::OkStatus();
+}
+
+}  // namespace cdc_ft

asset_stream_manager/local_assets_stream_manager_client.h

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef ASSET_STREAM_MANAGER_LOCAL_ASSETS_STREAM_MANAGER_CLIENT_H_
+#define ASSET_STREAM_MANAGER_LOCAL_ASSETS_STREAM_MANAGER_CLIENT_H_
+
+#include <memory>
+
+#include "absl/status/status.h"
+#include "grpcpp/channel.h"
+#include "proto/local_assets_stream_manager.grpc.pb.h"
+
+namespace grpc_impl {
+class Channel;
+}
+
+namespace cdc_ft {
+
+// gRpc client for starting/stopping asset streaming sessions.
+class LocalAssetsStreamManagerClient {
+ public:
+  explicit LocalAssetsStreamManagerClient(uint16_t service_port);
+
+  // |channel| is a grpc channel to use.
+  explicit LocalAssetsStreamManagerClient(
+      std::shared_ptr<grpc::Channel> channel);
+
+  ~LocalAssetsStreamManagerClient();
+
+  // Starts streaming |src_dir| to |user_host|:|mount_dir|.
+  // Starting a second session to the same target will stop the first one.
+  // |src_dir| is the Windows source directory to stream.
+  // |user_host| is the Linux host, formatted as [user@:host].
+  // |ssh_port| is the SSH port to use while connecting to the host.
+  // |mount_dir| is the Linux target directory to stream to.
+  // |ssh_command| is the ssh command and extra arguments to use.
+  // |scp_command| is the scp command and extra arguments to use.
+  absl::Status StartSession(const std::string& src_dir,
+                            const std::string& user_host, uint16_t ssh_port,
+                            const std::string& mount_dir,
+                            const std::string& ssh_command,
+                            const std::string& scp_command);
+
+  // Stops the streaming session to the Linux target |user_host|:|mount_dir|.
+  // |user_host| is the Linux host, formatted as [user@:host].
+  // |mount_dir| is the Linux target directory.
+  absl::Status StopSession(const std::string& user_host,
+                           const std::string& mount_dir);
+
+  // Helper function that splits "user@host:dir" into "user@host" and "dir".
+  // Does not think that C: is a host.
+  static absl::Status ParseUserHostDir(const std::string& user_host_dir,
+                                       std::string* user_host,
+                                       std::string* dir);
+
+ private:
+  using LocalAssetsStreamManager =
+      localassetsstreammanager::LocalAssetsStreamManager;
+  std::unique_ptr<LocalAssetsStreamManager::Stub> stub_;
+};
+
+}  // namespace cdc_ft
+
+#endif  // ASSET_STREAM_MANAGER_LOCAL_ASSETS_STREAM_MANAGER_CLIENT_H_

asset_stream_manager/main.cc

@@ -12,7 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include "asset_stream_manager/start_command.h"
 #include "asset_stream_manager/start_service_command.h"
+#include "asset_stream_manager/stop_command.h"
 #include "lyra/lyra.hpp"
 
 int main(int argc, char* argv[]) {
@@ -22,8 +24,14 @@ int main(int argc, char* argv[]) {
   int exit_code = -1;
   cli.add_argument(lyra::help(show_help));
 
-  cdc_ft::StartServiceCommand start_service(&exit_code);
-  start_service.Register(cli);
+  cdc_ft::StartCommand start_cmd(&exit_code);
+  start_cmd.Register(cli);
+
+  cdc_ft::StopCommand stop_cmd(&exit_code);
+  stop_cmd.Register(cli);
+
+  cdc_ft::StartServiceCommand start_service_cmd(&exit_code);
+  start_service_cmd.Register(cli);
 
   // Parse args and run. Note that parse actually runs the commands.
   // exit_code is -1 if no command was run.

asset_stream_manager/session_management_server.h

@@ -36,6 +36,8 @@ class ProcessFactory;
 // - Background
 class SessionManagementServer {
  public:
+  static constexpr int kDefaultServicePort = 44432;
+
   SessionManagementServer(grpc::Service* session_service,
                           grpc::Service* background_service,
                           SessionManager* session_manager);

asset_stream_manager/start_command.cc

@@ -0,0 +1,110 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "asset_stream_manager/start_command.h"
+
+#include <memory>
+
+#include "asset_stream_manager/local_assets_stream_manager_client.h"
+#include "asset_stream_manager/session_management_server.h"
+#include "common/log.h"
+#include "common/path.h"
+#include "common/remote_util.h"
+#include "common/status_macros.h"
+#include "lyra/lyra.hpp"
+
+namespace cdc_ft {
+namespace {
+constexpr int kDefaultVerbosity = 2;
+}  // namespace
+
+StartCommand::StartCommand(int* exit_code)
+    : BaseCommand("start",
+                  "Start streaming files from a Windows to a Linux device",
+                  exit_code) {}
+
+StartCommand::~StartCommand() = default;
+
+void StartCommand::RegisterCommandLineFlags(lyra::command& cmd) {
+  verbosity_ = kDefaultVerbosity;
+  cmd.add_argument(lyra::opt(verbosity_, "num")
+                       .name("--verbosity")
+                       .help("Verbosity of the log output, default: " +
+                             std::to_string(kDefaultVerbosity) +
+                             ". Increase to make logs more verbose."));
+
+  service_port_ = SessionManagementServer::kDefaultServicePort;
+  cmd.add_argument(
+      lyra::opt(service_port_, "port")
+          .name("--service-port")
+          .help("Local port to use while connecting to the local "
+                "asset stream service, default: " +
+                std::to_string(SessionManagementServer::kDefaultServicePort)));
+
+  ssh_port_ = RemoteUtil::kDefaultSshPort;
+  cmd.add_argument(
+      lyra::opt(ssh_port_, "port")
+          .name("--ssh-port")
+          .help("Port to use while connecting to the remote instance being "
+                "streamed to, default: " +
+                std::to_string(RemoteUtil::kDefaultSshPort)));
+
+  path::GetEnv("CDC_SSH_COMMAND", &ssh_command_).IgnoreError();
+  cmd.add_argument(
+      lyra::opt(ssh_command_, "ssh_command")
+          .name("--ssh-command")
+          .help("Path and arguments of ssh command to use, e.g. "
+                "\"C:\\path\\to\\ssh.exe -F config_file\". Can also be "
+                "specified by the CDC_SSH_COMMAND environment variable."));
+
+  path::GetEnv("CDC_SCP_COMMAND", &scp_command_).IgnoreError();
+  cmd.add_argument(
+      lyra::opt(scp_command_, "scp_command")
+          .name("--scp-command")
+          .help("Path and arguments of scp command to use, e.g. "
+                "\"C:\\path\\to\\scp.exe -F config_file\". Can also be "
+                "specified by the CDC_SCP_COMMAND environment variable."));
+
+  cmd.add_argument(lyra::arg(PosArgValidator(&src_dir_), "dir")
+                       .required()
+                       .help("Windows directory to stream"));
+
+  cmd.add_argument(
+      lyra::arg(PosArgValidator(&user_host_dir_), "[user@]host:src-dir")
+          .required()
+          .help("Linux host and directory to stream to"));
+}
+
+absl::Status StartCommand::Run() {
+  LogLevel level = Log::VerbosityToLogLevel(verbosity_);
+  ScopedLog scoped_log(std::make_unique<ConsoleLog>(level));
+  LocalAssetsStreamManagerClient client(service_port_);
+
+  std::string full_src_dir = path::GetFullPath(src_dir_);
+  std::string user_host, mount_dir;
+  RETURN_IF_ERROR(LocalAssetsStreamManagerClient::ParseUserHostDir(
+      user_host_dir_, &user_host, &mount_dir));
+
+  absl::Status status =
+      client.StartSession(full_src_dir, user_host, ssh_port_, mount_dir,
+                          ssh_command_, scp_command_);
+  if (status.ok()) {
+    LOG_INFO("Started streaming directory '%s' to '%s:%s'", src_dir_, user_host,
+             mount_dir);
+  }
+
+  return status;
+}
+
+}  // namespace cdc_ft

asset_stream_manager/start_command.h

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef ASSET_STREAM_MANAGER_START_COMMAND_H_
+#define ASSET_STREAM_MANAGER_START_COMMAND_H_
+
+#include "absl/status/status.h"
+#include "asset_stream_manager/base_command.h"
+
+namespace cdc_ft {
+
+// Handler for the start command. Sends an RPC call to the service to starts a
+// new asset streaming session.
+class StartCommand : public BaseCommand {
+ public:
+  explicit StartCommand(int* exit_code);
+  ~StartCommand();
+
+  // BaseCommand:
+  void RegisterCommandLineFlags(lyra::command& cmd) override;
+  absl::Status Run() override;
+
+ private:
+  int verbosity_ = 0;
+  uint16_t service_port_ = 0;
+  uint16_t ssh_port_ = 0;
+  std::string ssh_command_;
+  std::string scp_command_;
+  std::string src_dir_;
+  std::string user_host_dir_;
+};
+
+}  // namespace cdc_ft
+
+#endif  // ASSET_STREAM_MANAGER_START_COMMAND_H_

asset_stream_manager/start_service_command.cc

@@ -30,8 +30,6 @@
 namespace cdc_ft {
 namespace {
 
-constexpr int kSessionManagementPort = 44432;
-
 std::string GetLogPath(const char* log_dir, const char* log_base_name) {
   DefaultSystemClock* clock = DefaultSystemClock::GetInstance();
   std::string timestamp_ext = clock->FormatNow(".%Y%m%d-%H%M%S.log", false);
@@ -57,19 +55,17 @@ void StartServiceCommand::RegisterCommandLineFlags(lyra::command& cmd) {
           .name("--log-dir")
           .help("Directory to store log files, default: " + log_dir_));
 
-  cfg_.RegisterCommandLineFlags(cmd);
+  cfg_.RegisterCommandLineFlags(cmd, *this);
 }
 
 absl::Status StartServiceCommand::Run() {
-  if (!cfg_.jedec_parse_error().empty()) {
-    return absl::InvalidArgumentError(cfg_.jedec_parse_error());
-  }
-
   // Set up config. Allow overriding this config with |config_file|.
   absl::Status cfg_load_status = path::ExpandPathVariables(&config_file_);
   cfg_load_status.Update(cfg_.LoadFromFile(config_file_));
 
-  RETURN_IF_ERROR(InitLogging());
+  std::unique_ptr<Log> logger;
+  ASSIGN_OR_RETURN(logger, GetLogger());
+  cdc_ft::ScopedLog scoped_log(std::move(logger));
 
   // Log status of loaded configuration. Errors are not critical.
   if (cfg_load_status.ok()) {
@@ -102,16 +98,14 @@ absl::Status StartServiceCommand::Run() {
     LOG_INFO("Asset stream manager shut down successfully.");
   }
 
-  Log::Shutdown();
   return status;
 }
 
-absl::Status StartServiceCommand::InitLogging() {
+absl::StatusOr<std::unique_ptr<Log>> StartServiceCommand::GetLogger() {
   LogLevel level = Log::VerbosityToLogLevel(cfg_.session_cfg().verbosity);
   if (cfg_.log_to_stdout()) {
     // Log to stdout.
-    Log::Initialize(std::make_unique<ConsoleLog>(level));
-    return absl::OkStatus();
+    return std::make_unique<ConsoleLog>(level);
   }
 
   // Log to file.
@@ -121,9 +115,8 @@ absl::Status StartServiceCommand::InitLogging() {
         absl::StrFormat("Failed to create log directory '%s'", log_dir_));
   }
 
-  Log::Initialize(std::make_unique<FileLog>(
-      level, GetLogPath(log_dir_.c_str(), "assets_stream_manager").c_str()));
-  return absl::OkStatus();
+  return std::make_unique<FileLog>(
+      level, GetLogPath(log_dir_.c_str(), "assets_stream_manager").c_str());
 }
 
 // Runs the session management service and returns when it finishes.
@@ -154,7 +147,8 @@ absl::Status StartServiceCommand::RunService() {
     RETURN_ABSL_IF_ERROR(
         session_service.StartSession(nullptr, &request, &response));
   }
-  RETURN_IF_ERROR(sm_server.Start(kSessionManagementPort));
+  RETURN_IF_ERROR(
+      sm_server.Start(SessionManagementServer::kDefaultServicePort));
   sm_server.RunUntilShutdown();
   return absl::OkStatus();
 }

asset_stream_manager/start_service_command.h

@@ -17,6 +17,10 @@
 #ifndef ASSET_STREAM_MANAGER_START_SERVICE_COMMAND_H_
 #define ASSET_STREAM_MANAGER_START_SERVICE_COMMAND_H_
 
+#include <memory>
+
+#include "absl/status/status.h"
+#include "absl/status/statusor.h"
 #include "asset_stream_manager/asset_stream_config.h"
 #include "asset_stream_manager/base_command.h"
 
@@ -34,9 +38,8 @@ class StartServiceCommand : public BaseCommand {
   absl::Status Run() override;
 
  private:
-  // Initializes LOG* logging.
-  // Depending on the flags, might log to console or to a file.
-  absl::Status InitLogging();
+  // Depending on the flags, returns a console or file logger.
+  absl::StatusOr<std::unique_ptr<Log>> GetLogger();
 
   // Runs the asset streaming service.
   absl::Status RunService();

asset_stream_manager/stop_command.cc

@@ -0,0 +1,75 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include "asset_stream_manager/stop_command.h"
+
+#include <memory>
+
+#include "asset_stream_manager/local_assets_stream_manager_client.h"
+#include "asset_stream_manager/session_management_server.h"
+#include "common/log.h"
+#include "common/path.h"
+#include "common/status_macros.h"
+#include "lyra/lyra.hpp"
+
+namespace cdc_ft {
+namespace {
+constexpr int kDefaultVerbosity = 2;
+}  // namespace
+
+StopCommand::StopCommand(int* exit_code)
+    : BaseCommand("stop", "Stops a streaming session", exit_code) {}
+
+StopCommand::~StopCommand() = default;
+
+void StopCommand::RegisterCommandLineFlags(lyra::command& cmd) {
+  verbosity_ = kDefaultVerbosity;
+  cmd.add_argument(lyra::opt(verbosity_, "num")
+                       .name("--verbosity")
+                       .help("Verbosity of the log output, default: " +
+                             std::to_string(kDefaultVerbosity) +
+                             ". Increase to make logs more verbose."));
+
+  service_port_ = SessionManagementServer::kDefaultServicePort;
+  cmd.add_argument(
+      lyra::opt(service_port_, "port")
+          .name("--service-port")
+          .help("Local port to use while connecting to the local "
+                "asset stream service, default: " +
+                std::to_string(SessionManagementServer::kDefaultServicePort)));
+
+  cmd.add_argument(
+      lyra::arg(PosArgValidator(&user_host_dir_), "[user@]host:src-dir")
+          .required()
+          .help("Linux host and directory to stream to"));
+}
+
+absl::Status StopCommand::Run() {
+  LogLevel level = Log::VerbosityToLogLevel(verbosity_);
+  ScopedLog scoped_log(std::make_unique<ConsoleLog>(level));
+  LocalAssetsStreamManagerClient client(service_port_);
+
+  std::string user_host, mount_dir;
+  RETURN_IF_ERROR(LocalAssetsStreamManagerClient::ParseUserHostDir(
+      user_host_dir_, &user_host, &mount_dir));
+
+  absl::Status status = client.StopSession(user_host, mount_dir);
+  if (status.ok()) {
+    LOG_INFO("Stopped streaming session to '%s:%s'", user_host, mount_dir);
+  }
+
+  return status;
+}
+
+}  // namespace cdc_ft

asset_stream_manager/stop_command.h

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef ASSET_STREAM_MANAGER_STOP_COMMAND_H_
+#define ASSET_STREAM_MANAGER_STOP_COMMAND_H_
+
+#include "absl/status/status.h"
+#include "asset_stream_manager/base_command.h"
+
+namespace cdc_ft {
+
+// Handler for the stop command. Sends an RPC call to the service to stop an
+// asset streaming session.
+class StopCommand : public BaseCommand {
+ public:
+  explicit StopCommand(int* exit_code);
+  ~StopCommand();
+
+  // BaseCommand:
+  void RegisterCommandLineFlags(lyra::command& cmd) override;
+  absl::Status Run() override;
+
+ private:
+  int verbosity_ = 0;
+  uint16_t service_port_ = 0;
+  std::string user_host_dir_;
+};
+
+}  // namespace cdc_ft
+
+#endif  // ASSET_STREAM_MANAGER_STOP_COMMAND_H_