Skip to content
Permalink
Browse files

Revert remote execution from tower to grpcio (pantsbuild#7256)

We're seeing weird broken connection errors with tower.

We'll probably just chuck some retries in and be happy, but for now, let's get back to a more stable time.

* Revert "Remove unused operation wrapper (pantsbuild#7194)"

This reverts commit 9400024.

* Revert "Switch operation getting to tower (pantsbuild#7108)"

This reverts commit 0375b30.

* Revert "Remote execution uses tower-grpc to start executions (pantsbuild#7049)"

This reverts commit 28683c7.
  • Loading branch information...
illicitonion committed Feb 18, 2019
1 parent 3da2165 commit 222bc11bb3cf6703f118484605178c8d54982b5a

Some generated files are not rendered by default. Learn more.

Oops, something went wrong.
@@ -13,28 +13,18 @@ bytes = "0.4.5"
digest = "0.8"
fs = { path = "../fs" }
futures = "^0.1.16"
# TODO: Switch to a release once https://github.com/alexcrichton/futures-timer/pull/11 and https://github.com/alexcrichton/futures-timer/pull/12 merge
futures-timer = { git = "https://github.com/pantsbuild/futures-timer", rev = "0b747e565309a58537807ab43c674d8951f9e5a0" }
h2 = "0.1.13"
grpcio = { git = "https://github.com/pantsbuild/grpc-rs.git", rev = "4dfafe9355dc996d7d0702e7386a6fedcd9734c0", default_features = false, features = ["protobuf-codec"] }
hashing = { path = "../hashing" }
http = "0.1"
log = "0.4"
parking_lot = "0.6"
prost = "0.4"
prost-types = "0.4"
protobuf = { version = "2.0.4", features = ["with-bytes"] }
resettable = { path = "../resettable" }
sha2 = "0.8"
tempfile = "3"
# TODO: Switch to a release once https://github.com/alexcrichton/futures-timer/pull/11 and https://github.com/alexcrichton/futures-timer/pull/12 merge
futures-timer = { git = "https://github.com/pantsbuild/futures-timer", rev = "0b747e565309a58537807ab43c674d8951f9e5a0" }
time = "0.1.40"
tokio = "0.1.14"
tokio-codec = "0.1"
tokio-connect = { git = "https://github.com/pantsbuild/tokio-connect.git", rev = "f7ad1ca437973d6e24037ac6f7d5ef1013833c0b" }
tokio-process = "0.2.1"
tower-grpc = { git = "https://github.com/pantsbuild/tower-grpc.git", rev = "ef19f2e1715f415ecb699e8f17f5845ad2b45daf" }
tower-h2 = { git = "https://github.com/pantsbuild/tower-h2.git", rev = "44b0efb4983b769283efd5b2a3bc3decbf7c33de" }
tower-http = { git = "https://github.com/pantsbuild/tower-http.git", rev = "56049ee7f31d4f6c549f5d1d5fbbfd7937df3d00" }
tower-util = { git = "https://github.com/pantsbuild/tower.git", rev = "7b61c1fc1992c1df684fd3f179644ef0ca9bfa4c" }

[dev-dependencies]
mock = { path = "../testutil/mock" }
@@ -10,7 +10,6 @@ bytes = "0.4.5"
futures = "^0.1.16"
grpcio = { git = "https://github.com/pantsbuild/grpc-rs.git", rev = "4dfafe9355dc996d7d0702e7386a6fedcd9734c0", default_features = false, features = ["protobuf-codec"] }
hashing = { path = "../../hashing" }
log = "0.4"
prost = "0.4"
prost-derive = "0.4"
prost-types = "0.4"
@@ -176,11 +176,9 @@ fn generate_for_tower(thirdpartyprotobuf: &Path, out_dir: PathBuf) {
.enable_server(true)
.enable_client(true)
.build(
&[
PathBuf::from("build/bazel/remote/execution/v2/remote_execution.proto"),
PathBuf::from("google/rpc/code.proto"),
PathBuf::from("google/rpc/error_details.proto"),
],
&[PathBuf::from(
"build/bazel/remote/execution/v2/remote_execution.proto",
)],
&std::fs::read_dir(&thirdpartyprotobuf)
.unwrap()
.into_iter()
@@ -1,7 +1,4 @@
use bytes::BytesMut;
use hashing;
use log::error;
use prost::Message;

impl<'a> From<&'a hashing::Digest> for crate::remote_execution::Digest {
fn from(d: &hashing::Digest) -> Self {
@@ -21,31 +18,19 @@ impl<'a> From<&'a hashing::Digest> for crate::build::bazel::remote::execution::v
}
}

impl<'a> From<&'a crate::remote_execution::Digest> for Result<hashing::Digest, String> {
fn from(d: &crate::remote_execution::Digest) -> Self {
impl<'a> From<&'a super::remote_execution::Digest> for Result<hashing::Digest, String> {
fn from(d: &super::remote_execution::Digest) -> Self {
hashing::Fingerprint::from_hex_string(d.get_hash())
.map_err(|err| format!("Bad fingerprint in Digest {:?}: {:?}", d.get_hash(), err))
.map(|fingerprint| hashing::Digest(fingerprint, d.get_size_bytes() as usize))
}
}

impl<'a> From<&'a crate::build::bazel::remote::execution::v2::Digest>
for Result<hashing::Digest, String>
{
fn from(d: &crate::build::bazel::remote::execution::v2::Digest) -> Self {
hashing::Fingerprint::from_hex_string(&d.hash)
.map_err(|err| format!("Bad fingerprint in Digest {:?}: {:?}", d.hash, err))
.map(|fingerprint| hashing::Digest(fingerprint, d.size_bytes as usize))
}
}

impl From<crate::google::longrunning::Operation> for crate::operations::Operation {
fn from(op: crate::google::longrunning::Operation) -> Self {
let mut dst = Self::new();
dst.set_name(op.name);
if let Some(metadata) = op.metadata {
dst.set_metadata(prost_any_to_gcprio_any(metadata));
}
dst.set_metadata(prost_any_to_gcprio_any(op.metadata.unwrap()));
dst.set_done(op.done);
match op.result {
Some(crate::google::longrunning::operation::Result::Response(response)) => {
@@ -60,87 +45,6 @@ impl From<crate::google::longrunning::Operation> for crate::operations::Operatio
}
}

// This should only be used in test contexts. It should be deleted when the mock systems use tower.
impl From<crate::remote_execution::ExecuteRequest>
for crate::build::bazel::remote::execution::v2::ExecuteRequest
{
fn from(req: crate::remote_execution::ExecuteRequest) -> Self {
if req.has_execution_policy() || req.has_results_cache_policy() {
panic!("Can't convert ExecuteRequest protos with execution policy or results cache policy");
}
let digest: Result<hashing::Digest, String> = req.get_action_digest().into();
Self {
action_digest: Some((&digest.expect("Bad digest converting ExecuteRequest proto")).into()),
instance_name: req.instance_name,
execution_policy: None,
results_cache_policy: None,
skip_cache_lookup: req.skip_cache_lookup,
}
}
}

// This should only be used in test contexts. It should be deleted when the mock systems use tower.
impl From<crate::build::bazel::remote::execution::v2::ExecuteRequest>
for crate::remote_execution::ExecuteRequest
{
fn from(req: crate::build::bazel::remote::execution::v2::ExecuteRequest) -> Self {
if req.execution_policy.is_some() || req.results_cache_policy.is_some() {
panic!("Can't convert ExecuteRequest protos with execution policy or results cache policy");
}
let digest: Result<hashing::Digest, String> = (&req
.action_digest
.expect("Missing digest converting ExecuteRequest proto"))
.into();

let mut ret = Self::new();
ret.set_action_digest((&digest.expect("Bad digest converting ExecuteRequest proto")).into());
ret.set_instance_name(req.instance_name);
ret.set_skip_cache_lookup(req.skip_cache_lookup);
ret
}
}

// This should only be used in test contexts. It should be deleted when the mock systems use tower.
impl Into<grpcio::RpcStatus> for crate::google::rpc::Status {
fn into(self) -> grpcio::RpcStatus {
let mut buf = BytesMut::with_capacity(self.encoded_len());
self.encode(&mut buf).unwrap();
grpcio::RpcStatus {
status: self.code.into(),
details: None,
status_proto_bytes: Some(buf.to_vec()),
}
}
}

// TODO: Use num_enum or similar here when TryInto is stable.
pub fn code_from_i32(i: i32) -> crate::google::rpc::Code {
use crate::google::rpc::Code::*;
match i {
0 => Ok,
1 => Cancelled,
2 => Unknown,
3 => InvalidArgument,
4 => DeadlineExceeded,
5 => NotFound,
6 => AlreadyExists,
7 => PermissionDenied,
8 => ResourceExhausted,
9 => FailedPrecondition,
10 => Aborted,
11 => OutOfRange,
12 => Unimplemented,
13 => Internal,
14 => Unavailable,
15 => DataLoss,
16 => Unauthenticated,
_ => {
error!("Unknown grpc error code: {}, default to Unknown", i);
Unknown
}
}
}

pub fn prost_any_to_gcprio_any(any: prost_types::Any) -> protobuf::well_known_types::Any {
let prost_types::Any { type_url, value } = any;
let mut dst = protobuf::well_known_types::Any::new();
@@ -13,6 +13,5 @@ mod gen_for_tower;
pub use crate::gen_for_tower::*;

mod conversions;
pub use crate::conversions::code_from_i32;
mod verification;
pub use crate::verification::verify_directory_canonical;
Oops, something went wrong.

0 comments on commit 222bc11

Please sign in to comment.
You can’t perform that action at this time.