github-mirrors/netris-cdc-file-transfer
1
0
Fork 0
mirror of https://github.com/nestriness/cdc-file-transfer.git synced 2026-01-30 12:25:35 +02:00
Code Issues Packages Projects Releases Wiki Activity

[cdc_rsync] Detect remote architecture (#86)

Browse Source 
Improves ServerArch so that it can detect the remote architecture by
running uname and checking %PROCESSOR_ARCHITECTURE%. So far, only
x64 Linux and x64 Windows are supported, but in the future it is easy
to add support for others, e.g. aarch64, as well.

Before the detection is run, the remote architecture is guessed first
based on the destination. For instance, if the destination directory
starts with "C:\", it pretty much means Windows. If cdc_rsync_server
exists and runs fine, there's no need for detection.

Since also PortManager depends on the remote architecture, it has to
be adjusted as well. So far, PortManager assumeed that "local" means
Windows and "remote" means Linux. This is no longer the case for
syncing to Windows devices, so this CL adds the necessary abstractions
to PortManager.

Also refactors ArchType into a separate class in common, since it is
used now from several places. It is also expanded to handle future
changes that add support for different processor architectures, e.g.
aarch64.
This commit is contained in:
Lutz Justen
2023-02-01 11:51:20 +01:00
committed by GitHub
parent 3194678007
commit ee4118c6bf
19 changed files with 696 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
common/BUILD
View File

@@ -18,6 +18,24 @@ cc_test(
],
)
cc_library(
name = "arch_type",
srcs = ["arch_type.cc"],
hdrs = ["arch_type.h"],
deps = [":platform"],
)
cc_test(
name = "arch_type_test",
srcs = ["arch_type_test.cc"],
deps = [
":arch_type",
"@com_google_absl//absl/strings",
"@com_google_googletest//:gtest",
"@com_google_googletest//:gtest_main",
],
)
cc_library(
name = "buffer",
srcs = ["buffer.cc"],
@@ -254,6 +272,7 @@ cc_library(
hdrs = ["port_manager.h"],
target_compatible_with = ["@platforms//os:windows"],
deps = [
":arch_type",
":remote_util",
":status",
":stopwatch",
@@ -360,6 +379,7 @@ cc_library(
srcs = ["remote_util.cc"],
hdrs = ["remote_util.h"],
deps = [
":arch_type",
":platform",
":process",
":sdk_util",

70
common/arch_type.cc Normal file
View File

@@ -0,0 +1,70 @@
// Copyright 2023 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 "common/arch_type.h"
#include <cassert>
#include "common/platform.h"
namespace cdc_ft {
static constexpr char kUnhandledArchType[] = "Unhandled arch type";
ArchType GetLocalArchType() {
// TODO(ljusten): Take CPU architecture into account.
#if PLATFORM_WINDOWS
return ArchType::kWindows_x86_64;
#elif PLATFORM_LINUX
return ArchType::kLinux_x86_64;
#endif
}
bool IsWindowsArchType(ArchType arch_type) {
switch (arch_type) {
case ArchType::kWindows_x86_64:
return true;
case ArchType::kLinux_x86_64:
return false;
default:
assert(!kUnhandledArchType);
return false;
}
}
bool IsLinuxArchType(ArchType arch_type) {
switch (arch_type) {
case ArchType::kWindows_x86_64:
return false;
case ArchType::kLinux_x86_64:
return true;
default:
assert(!kUnhandledArchType);
return false;
}
}
const char* GetArchTypeStr(ArchType arch_type) {
switch (arch_type) {
case ArchType::kWindows_x86_64:
return "Windows_x86_64";
case ArchType::kLinux_x86_64:
return "Linux_x86_64";
default:
assert(!kUnhandledArchType);
return "Unknown";
}
}
} // namespace cdc_ft

41
common/arch_type.h Normal file
View File

@@ -0,0 +1,41 @@
/*
* Copyright 2023 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 COMMON_ARCH_TYPE_H_
#define COMMON_ARCH_TYPE_H_
namespace cdc_ft {
enum class ArchType {
kWindows_x86_64 = 0,
kLinux_x86_64 = 1,
};
// Returns the arch type of the current process.
ArchType GetLocalArchType();
// Returns true if |arch_type| is a Windows operating system.
bool IsWindowsArchType(ArchType arch_type);
// Returns true if |arch_type| is a Linux operating system.
bool IsLinuxArchType(ArchType arch_type);
// Returns a human readable string for |arch_type|.
const char* GetArchTypeStr(ArchType arch_type);
} // namespace cdc_ft
#endif // COMMON_ARCH_TYPE_H_

49
common/arch_type_test.cc Normal file
View File

@@ -0,0 +1,49 @@
// Copyright 2023 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 "common/arch_type.h"
#include "absl/strings/match.h"
#include "gtest/gtest.h"
namespace cdc_ft {
namespace {
TEST(ArchTypeTest, GetLocalArchType) {
#if PLATFORM_WINDOWS
EXPECT_TRUE(IsPlatformWindows(GetLocalArchType()));
#elif PLATFORM_LINUX
EXPECT_TRUE(IsPlatformLinux(GetLocalArchType()));
#endif
}
TEST(ArchTypeTest, IsWindowsArchType) {
EXPECT_TRUE(IsWindowsArchType(ArchType::kWindows_x86_64));
EXPECT_FALSE(IsWindowsArchType(ArchType::kLinux_x86_64));
}
TEST(ArchTypeTest, IsLinuxArchType) {
EXPECT_FALSE(IsLinuxArchType(ArchType::kWindows_x86_64));
EXPECT_TRUE(IsLinuxArchType(ArchType::kLinux_x86_64));
}
TEST(ArchTypeTest, GetArchTypeStr) {
EXPECT_TRUE(
absl::StrContains(GetArchTypeStr(ArchType::kWindows_x86_64), "Windows"));
EXPECT_TRUE(
absl::StrContains(GetArchTypeStr(ArchType::kLinux_x86_64), "Linux"));
}
} // namespace
} // namespace cdc_ft

27
common/port_manager.h
View File

@@ -17,12 +17,12 @@
#ifndef COMMON_PORT_MANAGER_H_
#define COMMON_PORT_MANAGER_H_
#include <absl/status/statusor.h>
#include <memory>
#include <string>
#include <unordered_set>
#include "absl/status/statusor.h"
#include "common/arch_type.h"
#include "common/clock.h"
namespace cdc_ft {
@@ -53,9 +53,12 @@ class PortManager {
// explicitly.
// |remote_timeout_sec| is the timeout for finding available ports on the
// remote instance.
// Returns a DeadlineExceeded error if the timeout is exceeded.
// Returns a ResourceExhausted error if no ports are available.
absl::StatusOr<int> ReservePort(int remote_timeout_sec);
// |remote_arch_type| is the architecture of the remote device.
// Both |remote_timeout_sec| and |remote_arch_type| are ignored if
// |remote_util| is nullptr. Returns a DeadlineExceeded error if the timeout
// is exceeded. Returns a ResourceExhausted error if no ports are available.
absl::StatusOr<int> ReservePort(int remote_timeout_sec,
ArchType remote_arch_type);
// Releases a reserved port.
absl::Status ReleasePort(int port);
@@ -66,34 +69,34 @@ class PortManager {
// Finds available ports in the range [first_port, last_port] for port
// forwarding on the local workstation.
// |ip| is the IP address to filter by.
// |arch_type| is the architecture of the local device.
// |process_factory| is used to create a netstat process.
// Returns ResourceExhaustedError if no port is available.
static absl::StatusOr<std::unordered_set<int>> FindAvailableLocalPorts(
int first_port, int last_port, const char* ip,
int first_port, int last_port, ArchType arch_type,
ProcessFactory* process_factory);
// Finds available ports in the range [first_port, last_port] for port
// forwarding on the instance.
// |ip| is the IP address to filter by.
// |arch_type| is the architecture of the remote device.
// |process_factory| is used to create a netstat process.
// |remote_util| is used to connect to the instance.
// |timeout_sec| is the connection timeout in seconds.
// Returns a DeadlineExceeded error if the timeout is exceeded.
// Returns ResourceExhaustedError if no port is available.
static absl::StatusOr<std::unordered_set<int>> FindAvailableRemotePorts(
int first_port, int last_port, const char* ip,
int first_port, int last_port, ArchType arch_type,
ProcessFactory* process_factory, RemoteUtil* remote_util, int timeout_sec,
SteadyClock* steady_clock = DefaultSteadyClock::GetInstance());
private:
// Returns a list of available ports in the range [|first_port|, |last_port|]
// from the given |netstat_output|. |ip| is the IP address to look for, e.g.
// "127.0.0.1".
// from the given |netstat_output|.
// |arch_type| is the architecture of the device where netstat was called.
// Returns ResourceExhaustedError if no port is available.
static absl::StatusOr<std::unordered_set<int>> FindAvailablePorts(
int first_port, int last_port, const std::string& netstat_output,
const char* ip);
ArchType arch_type);
int first_port_;
int last_port_;

180
common/port_manager_test.cc
View File

@@ -25,7 +25,6 @@
namespace cdc_ft {
namespace {
constexpr int kSshPort = 12345;
constexpr char kUserHost[] = "user@1.2.3.4";
constexpr char kGuid[] = "f77bcdfe-368c-4c45-9f01-230c5e7e2132";
@@ -35,12 +34,12 @@ constexpr int kNumPorts = kLastPort - kFirstPort + 1;
constexpr int kTimeoutSec = 1;
constexpr char kLocalNetstat[] = "netstat -a -n -p tcp";
constexpr char kRemoteNetstat[] = "netstat --numeric --listening --tcp";
constexpr char kWindowsNetstat[] = "netstat -a -n -p tcp";
constexpr char kLinuxNetstat[] = "netstat --numeric --listening --tcp";
constexpr char kLocalNetstatOutFmt[] =
constexpr char kWindowsNetstatOutFmt[] =
"TCP 127.0.0.1:50000 127.0.0.1:%i ESTABLISHED";
constexpr char kRemoteNetstatOutFmt[] =
constexpr char kLinuxNetstatOutFmt[] =
"tcp 0 0 0.0.0.0:%i 0.0.0.0:* LISTEN";
class PortManagerTest : public ::testing::Test {
@@ -67,10 +66,11 @@ class PortManagerTest : public ::testing::Test {
};
TEST_F(PortManagerTest, ReservePortSuccess) {
process_factory_.SetProcessOutput(kLocalNetstat, "", "", 0);
process_factory_.SetProcessOutput(kRemoteNetstat, "", "", 0);
process_factory_.SetProcessOutput(kWindowsNetstat, "", "", 0);
process_factory_.SetProcessOutput(kLinuxNetstat, "", "", 0);
absl::StatusOr<int> port = port_manager_.ReservePort(kTimeoutSec);
absl::StatusOr<int> port =
port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64);
ASSERT_OK(port);
EXPECT_EQ(*port, kFirstPort);
}
@@ -78,12 +78,13 @@ TEST_F(PortManagerTest, ReservePortSuccess) {
TEST_F(PortManagerTest, ReservePortAllLocalPortsTaken) {
std::string local_netstat_out = "";
for (int port = kFirstPort; port <= kLastPort; ++port) {
local_netstat_out += absl::StrFormat(kLocalNetstatOutFmt, port);
local_netstat_out += absl::StrFormat(kWindowsNetstatOutFmt, port);
}
process_factory_.SetProcessOutput(kLocalNetstat, local_netstat_out, "", 0);
process_factory_.SetProcessOutput(kRemoteNetstat, "", "", 0);
process_factory_.SetProcessOutput(kWindowsNetstat, local_netstat_out, "", 0);
process_factory_.SetProcessOutput(kLinuxNetstat, "", "", 0);
absl::StatusOr<int> port = port_manager_.ReservePort(kTimeoutSec);
absl::StatusOr<int> port =
port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64);
EXPECT_TRUE(absl::IsResourceExhausted(port.status()));
EXPECT_TRUE(
absl::StrContains(port.status().message(), "No port available in range"));
@@ -92,22 +93,24 @@ TEST_F(PortManagerTest, ReservePortAllLocalPortsTaken) {
TEST_F(PortManagerTest, ReservePortAllRemotePortsTaken) {
std::string remote_netstat_out = "";
for (int port = kFirstPort; port <= kLastPort; ++port) {
remote_netstat_out += absl::StrFormat(kRemoteNetstatOutFmt, port);
remote_netstat_out += absl::StrFormat(kLinuxNetstatOutFmt, port);
}
process_factory_.SetProcessOutput(kLocalNetstat, "", "", 0);
process_factory_.SetProcessOutput(kRemoteNetstat, remote_netstat_out, "", 0);
process_factory_.SetProcessOutput(kWindowsNetstat, "", "", 0);
process_factory_.SetProcessOutput(kLinuxNetstat, remote_netstat_out, "", 0);
absl::StatusOr<int> port = port_manager_.ReservePort(kTimeoutSec);
absl::StatusOr<int> port =
port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64);
EXPECT_TRUE(absl::IsResourceExhausted(port.status()));
EXPECT_TRUE(
absl::StrContains(port.status().message(), "No port available in range"));
}
TEST_F(PortManagerTest, ReservePortLocalNetstatFails) {
process_factory_.SetProcessOutput(kLocalNetstat, "", "", 1);
process_factory_.SetProcessOutput(kRemoteNetstat, "", "", 0);
process_factory_.SetProcessOutput(kWindowsNetstat, "", "", 1);
process_factory_.SetProcessOutput(kLinuxNetstat, "", "", 0);
absl::StatusOr<int> port = port_manager_.ReservePort(kTimeoutSec);
absl::StatusOr<int> port =
port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64);
EXPECT_NOT_OK(port);
EXPECT_TRUE(
absl::StrContains(port.status().message(),
@@ -115,21 +118,23 @@ TEST_F(PortManagerTest, ReservePortLocalNetstatFails) {
}
TEST_F(PortManagerTest, ReservePortRemoteNetstatFails) {
process_factory_.SetProcessOutput(kLocalNetstat, "", "", 0);
process_factory_.SetProcessOutput(kRemoteNetstat, "", "", 1);
process_factory_.SetProcessOutput(kWindowsNetstat, "", "", 0);
process_factory_.SetProcessOutput(kLinuxNetstat, "", "", 1);
absl::StatusOr<int> port = port_manager_.ReservePort(kTimeoutSec);
absl::StatusOr<int> port =
port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64);
EXPECT_NOT_OK(port);
EXPECT_TRUE(absl::StrContains(port.status().message(),
"Failed to find available ports on instance"));
}
TEST_F(PortManagerTest, ReservePortRemoteNetstatTimesOut) {
process_factory_.SetProcessOutput(kLocalNetstat, "", "", 0);
process_factory_.SetProcessNeverExits(kRemoteNetstat);
process_factory_.SetProcessOutput(kWindowsNetstat, "", "", 0);
process_factory_.SetProcessNeverExits(kLinuxNetstat);
steady_clock_.AutoAdvance(kTimeoutSec * 2 * 1000);
absl::StatusOr<int> port = port_manager_.ReservePort(kTimeoutSec);
absl::StatusOr<int> port =
port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64);
EXPECT_NOT_OK(port);
EXPECT_TRUE(absl::IsDeadlineExceeded(port.status()));
EXPECT_TRUE(absl::StrContains(port.status().message(),
@@ -137,8 +142,8 @@ TEST_F(PortManagerTest, ReservePortRemoteNetstatTimesOut) {
}
TEST_F(PortManagerTest, ReservePortMultipleInstances) {
process_factory_.SetProcessOutput(kLocalNetstat, "", "", 0);
process_factory_.SetProcessOutput(kRemoteNetstat, "", "", 0);
process_factory_.SetProcessOutput(kWindowsNetstat, "", "", 0);
process_factory_.SetProcessOutput(kLinuxNetstat, "", "", 0);
PortManager port_manager2(kGuid, kFirstPort, kLastPort, &process_factory_,
&remote_util_);
@@ -146,55 +151,79 @@ TEST_F(PortManagerTest, ReservePortMultipleInstances) {
// Port managers use shared memory, so different instances know about each
// other. This would even work if |port_manager_| and |port_manager2| belonged
// to different processes, but we don't test that here.
EXPECT_EQ(*port_manager_.ReservePort(kTimeoutSec), kFirstPort + 0);
EXPECT_EQ(*port_manager2.ReservePort(kTimeoutSec), kFirstPort + 1);
EXPECT_EQ(*port_manager_.ReservePort(kTimeoutSec), kFirstPort + 2);
EXPECT_EQ(*port_manager2.ReservePort(kTimeoutSec), kFirstPort + 3);
EXPECT_EQ(*port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64),
kFirstPort + 0);
EXPECT_EQ(*port_manager2.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64),
kFirstPort + 1);
EXPECT_EQ(*port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64),
kFirstPort + 2);
EXPECT_EQ(*port_manager2.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64),
kFirstPort + 3);
}
TEST_F(PortManagerTest, ReservePortReusesPortsInLRUOrder) {
process_factory_.SetProcessOutput(kLocalNetstat, "", "", 0);
process_factory_.SetProcessOutput(kRemoteNetstat, "", "", 0);
process_factory_.SetProcessOutput(kWindowsNetstat, "", "", 0);
process_factory_.SetProcessOutput(kLinuxNetstat, "", "", 0);
for (int n = 0; n < kNumPorts * 2; ++n) {
EXPECT_EQ(*port_manager_.ReservePort(kTimeoutSec),
EXPECT_EQ(*port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64),
kFirstPort + n % kNumPorts);
system_clock_.Advance(1000);
}
}
TEST_F(PortManagerTest, ReleasePort) {
process_factory_.SetProcessOutput(kLocalNetstat, "", "", 0);
process_factory_.SetProcessOutput(kRemoteNetstat, "", "", 0);
process_factory_.SetProcessOutput(kWindowsNetstat, "", "", 0);
process_factory_.SetProcessOutput(kLinuxNetstat, "", "", 0);
absl::StatusOr<int> port = port_manager_.ReservePort(kTimeoutSec);
absl::StatusOr<int> port =
port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64);
EXPECT_EQ(*port, kFirstPort);
EXPECT_OK(port_manager_.ReleasePort(*port));
port = port_manager_.ReservePort(kTimeoutSec);
port = port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64);
EXPECT_EQ(*port, kFirstPort);
}
TEST_F(PortManagerTest, ReleasePortOnDestruction) {
process_factory_.SetProcessOutput(kLocalNetstat, "", "", 0);
process_factory_.SetProcessOutput(kRemoteNetstat, "", "", 0);
process_factory_.SetProcessOutput(kWindowsNetstat, "", "", 0);
process_factory_.SetProcessOutput(kLinuxNetstat, "", "", 0);
auto port_manager2 = std::make_unique<PortManager>(
kGuid, kFirstPort, kLastPort, &process_factory_, &remote_util_);
EXPECT_EQ(*port_manager2->ReservePort(kTimeoutSec), kFirstPort + 0);
EXPECT_EQ(*port_manager_.ReservePort(kTimeoutSec), kFirstPort + 1);
EXPECT_EQ(*port_manager2->ReservePort(kTimeoutSec, ArchType::kLinux_x86_64),
kFirstPort + 0);
EXPECT_EQ(*port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64),
kFirstPort + 1);
port_manager2.reset();
EXPECT_EQ(*port_manager_.ReservePort(kTimeoutSec), kFirstPort + 0);
EXPECT_EQ(*port_manager_.ReservePort(kTimeoutSec, ArchType::kLinux_x86_64),
kFirstPort + 0);
}
TEST_F(PortManagerTest, FindAvailableLocalPortsSuccess) {
// First port is taken
TEST_F(PortManagerTest, FindAvailableLocalPortsSuccessWindows) {
// First port is in use.
std::string local_netstat_out =
absl::StrFormat(kLocalNetstatOutFmt, kFirstPort);
process_factory_.SetProcessOutput(kLocalNetstat, local_netstat_out, "", 0);
absl::StrFormat(kWindowsNetstatOutFmt, kFirstPort);
process_factory_.SetProcessOutput(kWindowsNetstat, local_netstat_out, "", 0);
absl::StatusOr<std::unordered_set<int>> ports =
PortManager::FindAvailableLocalPorts(kFirstPort, kLastPort, "127.0.0.1",
&process_factory_);
PortManager::FindAvailableLocalPorts(
kFirstPort, kLastPort, ArchType::kWindows_x86_64, &process_factory_);
ASSERT_OK(ports);
EXPECT_EQ(ports->size(), kNumPorts - 1);
for (int port = kFirstPort + 1; port <= kLastPort; ++port) {
EXPECT_TRUE(ports->find(port) != ports->end());
}
}
TEST_F(PortManagerTest, FindAvailableLocalPortsSuccessLinux) {
// First port is in use.
std::string local_netstat_out =
absl::StrFormat(kLinuxNetstatOutFmt, kFirstPort);
process_factory_.SetProcessOutput(kLinuxNetstat, local_netstat_out, "", 0);
absl::StatusOr<std::unordered_set<int>> ports =
PortManager::FindAvailableLocalPorts(
kFirstPort, kLastPort, ArchType::kLinux_x86_64, &process_factory_);
ASSERT_OK(ports);
EXPECT_EQ(ports->size(), kNumPorts - 1);
for (int port = kFirstPort + 1; port <= kLastPort; ++port) {
@@ -203,31 +232,48 @@ TEST_F(PortManagerTest, FindAvailableLocalPortsSuccess) {
}
TEST_F(PortManagerTest, FindAvailableLocalPortsFailsNoPorts) {
// All ports taken
// All ports are in use.
std::string local_netstat_out = "";
for (int port = kFirstPort; port <= kLastPort; ++port) {
local_netstat_out += absl::StrFormat(kLocalNetstatOutFmt, port);
local_netstat_out += absl::StrFormat(kWindowsNetstatOutFmt, port);
}
process_factory_.SetProcessOutput(kLocalNetstat, local_netstat_out, "", 0);
process_factory_.SetProcessOutput(kWindowsNetstat, local_netstat_out, "", 0);
absl::StatusOr<std::unordered_set<int>> ports =
PortManager::FindAvailableLocalPorts(kFirstPort, kLastPort, "127.0.0.1",
&process_factory_);
PortManager::FindAvailableLocalPorts(
kFirstPort, kLastPort, ArchType::kWindows_x86_64, &process_factory_);
EXPECT_TRUE(absl::IsResourceExhausted(ports.status()));
EXPECT_TRUE(absl::StrContains(ports.status().message(),
"No port available in range"));
}
TEST_F(PortManagerTest, FindAvailableRemotePortsSuccess) {
// First port is taken
TEST_F(PortManagerTest, FindAvailableRemotePortsSuccessLinux) {
// First port is in use.
std::string remote_netstat_out =
absl::StrFormat(kRemoteNetstatOutFmt, kFirstPort);
process_factory_.SetProcessOutput(kRemoteNetstat, remote_netstat_out, "", 0);
absl::StrFormat(kLinuxNetstatOutFmt, kFirstPort);
process_factory_.SetProcessOutput(kLinuxNetstat, remote_netstat_out, "", 0);
absl::StatusOr<std::unordered_set<int>> ports =
PortManager::FindAvailableRemotePorts(kFirstPort, kLastPort, "0.0.0.0",
&process_factory_, &remote_util_,
kTimeoutSec);
PortManager::FindAvailableRemotePorts(
kFirstPort, kLastPort, ArchType::kLinux_x86_64, &process_factory_,
&remote_util_, kTimeoutSec);
ASSERT_OK(ports);
EXPECT_EQ(ports->size(), kNumPorts - 1);
for (int port = kFirstPort + 1; port <= kLastPort; ++port) {
EXPECT_TRUE(ports->find(port) != ports->end());
}
}
TEST_F(PortManagerTest, FindAvailableRemotePortsSuccessWindows) {
// First port is in use.
std::string remote_netstat_out =
absl::StrFormat(kWindowsNetstatOutFmt, kFirstPort);
process_factory_.SetProcessOutput(kWindowsNetstat, remote_netstat_out, "", 0);
absl::StatusOr<std::unordered_set<int>> ports =
PortManager::FindAvailableRemotePorts(
kFirstPort, kLastPort, ArchType::kWindows_x86_64, &process_factory_,
&remote_util_, kTimeoutSec);
ASSERT_OK(ports);
EXPECT_EQ(ports->size(), kNumPorts - 1);
for (int port = kFirstPort + 1; port <= kLastPort; ++port) {
@@ -236,17 +282,17 @@ TEST_F(PortManagerTest, FindAvailableRemotePortsSuccess) {
}
TEST_F(PortManagerTest, FindAvailableRemotePortsFailsNoPorts) {
// All ports taken
// All ports are in use.
std::string remote_netstat_out = "";
for (int port = kFirstPort; port <= kLastPort; ++port) {
remote_netstat_out += absl::StrFormat(kRemoteNetstatOutFmt, port);
remote_netstat_out += absl::StrFormat(kLinuxNetstatOutFmt, port);
}
process_factory_.SetProcessOutput(kRemoteNetstat, remote_netstat_out, "", 0);
process_factory_.SetProcessOutput(kLinuxNetstat, remote_netstat_out, "", 0);
absl::StatusOr<std::unordered_set<int>> ports =
PortManager::FindAvailableRemotePorts(kFirstPort, kLastPort, "0.0.0.0",
&process_factory_, &remote_util_,
kTimeoutSec);
PortManager::FindAvailableRemotePorts(
kFirstPort, kLastPort, ArchType::kLinux_x86_64, &process_factory_,
&remote_util_, kTimeoutSec);
EXPECT_TRUE(absl::IsResourceExhausted(ports.status()));
EXPECT_TRUE(absl::StrContains(ports.status().message(),
"No port available in range"));

87
common/port_manager_win.cc
View File

@@ -20,6 +20,7 @@
#include <map>
#include "absl/strings/str_split.h"
#include "common/arch_type.h"
#include "common/log.h"
#include "common/process.h"
#include "common/remote_util.h"
@@ -30,6 +31,42 @@
namespace cdc_ft {
constexpr char kErrorArchTypeUnhandled[] = "arch_type_unhandled";
// Returns the arch-specific netstat command.
const char* GetNetstatCommand(ArchType arch_type) {
if (IsWindowsArchType(arch_type)) {
// -a to get the connection and ports the computer is listening on.
// -n to get numerical addresses to avoid the overhead of getting names.
// -p tcp to limit the output to TCPv4 connections.
return "netstat -a -n -p tcp";
}
if (IsLinuxArchType(arch_type)) {
// --numeric to get numerical addresses.
// --listening to get only listening sockets.
// --tcp to get only TCP connections.
return "netstat --numeric --listening --tcp";
}
assert(!kErrorArchTypeUnhandled);
return "";
}
// Returns the arch-specific IP address to filter netstat results by.
const char* GetNetstatFilterIp(ArchType arch_type) {
if (IsWindowsArchType(arch_type)) {
return "127.0.0.1";
}
if (IsLinuxArchType(arch_type)) {
return "0.0.0.0";
}
assert(!kErrorArchTypeUnhandled);
return "";
}
class SharedMemory {
public:
// Creates a new shared memory instance with given |name| and |size| in bytes.
@@ -121,22 +158,25 @@ PortManager::~PortManager() {
}
}
absl::StatusOr<int> PortManager::ReservePort(int remote_timeout_sec) {
absl::StatusOr<int> PortManager::ReservePort(int remote_timeout_sec,
ArchType remote_arch_type) {
// Find available port on workstation.
std::unordered_set<int> local_ports;
ASSIGN_OR_RETURN(local_ports,
FindAvailableLocalPorts(first_port_, last_port_, "127.0.0.1",
process_factory_),
"Failed to find available ports on workstation");
ASSIGN_OR_RETURN(
local_ports,
FindAvailableLocalPorts(first_port_, last_port_,
ArchType::kWindows_x86_64, process_factory_),
"Failed to find available ports on workstation");
// Find available port on remote instance.
std::unordered_set<int> remote_ports = local_ports;
if (remote_util_ != nullptr) {
ASSIGN_OR_RETURN(remote_ports,
FindAvailableRemotePorts(
first_port_, last_port_, "0.0.0.0", process_factory_,
remote_util_, remote_timeout_sec, steady_clock_),
"Failed to find available ports on instance");
ASSIGN_OR_RETURN(
remote_ports,
FindAvailableRemotePorts(first_port_, last_port_, remote_arch_type,
process_factory_, remote_util_,
remote_timeout_sec, steady_clock_),
"Failed to find available ports on instance");
}
// Fetch shared memory.
@@ -203,14 +243,11 @@ absl::Status PortManager::ReleasePort(int port) {
// static
absl::StatusOr<std::unordered_set<int>> PortManager::FindAvailableLocalPorts(
int first_port, int last_port, const char* ip,
int first_port, int last_port, ArchType arch_type,
ProcessFactory* process_factory) {
// -a to get the connection and ports the computer is listening on.
// -n to get numerical addresses to avoid the overhead of determining names.
// -p tcp to limit the output to TCPv4 connections.
// TODO: Use Windows API instead of netstat.
// TODO: Use local APIs instead of netstat.
ProcessStartInfo start_info;
start_info.command = "netstat -a -n -p tcp";
start_info.command = GetNetstatCommand(arch_type);
start_info.name = "netstat";
start_info.flags = ProcessFlags::kNoWindow;
@@ -231,18 +268,15 @@ absl::StatusOr<std::unordered_set<int>> PortManager::FindAvailableLocalPorts(
}
LOG_DEBUG("netstat (workstation) output:\n%s", output);
return FindAvailablePorts(first_port, last_port, output, ip);
return FindAvailablePorts(first_port, last_port, output, arch_type);
}
// static
absl::StatusOr<std::unordered_set<int>> PortManager::FindAvailableRemotePorts(
int first_port, int last_port, const char* ip,
int first_port, int last_port, ArchType arch_type,
ProcessFactory* process_factory, RemoteUtil* remote_util, int timeout_sec,
SteadyClock* steady_clock) {
// --numeric to get numerical addresses.
// --listening to get only listening sockets.
// --tcp to get only TCP connections.
std::string remote_command = "netstat --numeric --listening --tcp";
std::string remote_command = GetNetstatCommand(arch_type);
ProcessStartInfo start_info =
remote_util->BuildProcessStartInfoForSsh(remote_command);
start_info.name = "netstat";
@@ -281,24 +315,25 @@ absl::StatusOr<std::unordered_set<int>> PortManager::FindAvailableRemotePorts(
}
LOG_DEBUG("netstat (instance) output:\n%s", output);
return FindAvailablePorts(first_port, last_port, output, ip);
return FindAvailablePorts(first_port, last_port, output, arch_type);
}
// static
absl::StatusOr<std::unordered_set<int>> PortManager::FindAvailablePorts(
int first_port, int last_port, const std::string& netstat_output,
const char* ip) {
ArchType arch_type) {
std::unordered_set<int> available_ports;
std::vector<std::string> lines;
const char* filter_ip = GetNetstatFilterIp(arch_type);
for (const auto& line : absl::StrSplit(netstat_output, '\n')) {
if (absl::StrContains(line, ip)) {
if (absl::StrContains(line, filter_ip)) {
lines.push_back(std::string(line));
}
}
for (int port = first_port; port <= last_port; ++port) {
bool port_occupied = false;
std::string portToken = absl::StrFormat("%s:%i", ip, port);
std::string portToken = absl::StrFormat("%s:%i", filter_ip, port);
for (const std::string& line : lines) {
// Ports in the TIME_WAIT state can be reused. It is common that ports
// stay in this state for O(minutes).

25
common/remote_util.cc
View File

@@ -148,6 +148,31 @@ absl::Status RemoteUtil::Run(std::string remote_command, std::string name) {
return process_factory_->Run(start_info);
}
absl::Status RemoteUtil::RunWithCapture(std::string remote_command,
std::string name, std::string* std_out,
std::string* std_err) {
ProcessStartInfo start_info =
BuildProcessStartInfoForSsh(std::move(remote_command));
start_info.name = std::move(name);
start_info.forward_output_to_log = forward_output_to_log_;
if (std_out) {
start_info.stdout_handler = [std_out](const char* data, size_t size) {
std_out->append(data, size);
return absl::OkStatus();
};
}
if (std_err) {
start_info.stderr_handler = [std_err](const char* data, size_t size) {
std_err->append(data, size);
return absl::OkStatus();
};
}
return process_factory_->Run(start_info);
}
ProcessStartInfo RemoteUtil::BuildProcessStartInfoForSsh(
std::string remote_command) {
return BuildProcessStartInfoForSshInternal("", "-- " + remote_command);

31
common/remote_util.h
View File

@@ -25,7 +25,7 @@
namespace cdc_ft {
// Utilities for executing remote commands on a gamelet through SSH.
// Utilities for executing remote commands on a remote device through SSH.
// Windows-only.
class RemoteUtil {
public:
@@ -63,8 +63,8 @@ class RemoteUtil {
// Returns bad results for tricky strings like "C:\scp.path\scp.exe".
static std::string ScpToSftpCommand(std::string scp_command);
// Copies |source_filepaths| to the remote folder |dest| on the gamelet using
// scp. If |compress| is true, compressed upload is used.
// Copies |source_filepaths| to the remote folder |dest| on the remove device
// using scp. If |compress| is true, compressed upload is used.
absl::Status Scp(std::vector<std::string> source_filepaths,
const std::string& dest, bool compress);
@@ -86,27 +86,32 @@ class RemoteUtil {
absl::Status Sftp(const std::string& commands,
const std::string& initial_local_dir, bool compress);
// Calls 'chmod |mode| |remote_path|' on the gamelet.
// Calls 'chmod |mode| |remote_path|' on the remote device.
absl::Status Chmod(const std::string& mode, const std::string& remote_path,
bool quiet = false);
// Runs |remote_command| on the gamelet. The command must be properly escaped.
// |name| is the name of the command displayed in the logs.
// Runs |remote_command| on the remote device. The command must be properly
// escaped. |name| is the name of the command displayed in the logs.
absl::Status Run(std::string remote_command, std::string name);
// Builds an SSH command that executes |remote_command| on the gamelet.
// Same as Run(), but captures both stdout and stderr.
// If |std_out| or |std_err| are nullptr, the output is not captured.
absl::Status RunWithCapture(std::string remote_command, std::string name,
std::string* std_out, std::string* std_err);
// Builds an SSH command that executes |remote_command| on the remote device.
ProcessStartInfo BuildProcessStartInfoForSsh(std::string remote_command);
// Builds an SSH command that runs SSH port forwarding to the gamelet, using
// the given |local_port| and |remote_port|.
// If |reverse| is true, sets up reverse port forwarding.
// Builds an SSH command that runs SSH port forwarding to the remote device,
// using the given |local_port| and |remote_port|. If |reverse| is true, sets
// up reverse port forwarding.
ProcessStartInfo BuildProcessStartInfoForSshPortForward(int local_port,
int remote_port,
bool reverse);
// Builds an SSH command that executes |remote_command| on the gamelet, using
// port forwarding with given |local_port| and |remote_port|.
// If |reverse| is true, sets up reverse port forwarding.
// Builds an SSH command that executes |remote_command| on the remote device,
// using port forwarding with given |local_port| and |remote_port|. If
// |reverse| is true, sets up reverse port forwarding.
ProcessStartInfo BuildProcessStartInfoForSshPortForwardAndCommand(
int local_port, int remote_port, bool reverse,
std::string remote_command);