Repository navigation
Expand file tree
/
Copy pathgrpc.rs
More file actions
173 lines (154 loc) · 5.04 KB
/
Copy pathgrpc.rs
File metadata and controls
173 lines (154 loc) · 5.04 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
//! A gRPC client for the inbound policy API.
//!
//! This client currently discovers a destination controller pod via the k8s API and uses port
//! forwarding to connect to a running instance.
use anyhow::Result;
use linkerd2_proxy_api::inbound::inbound_server_policies_client::InboundServerPoliciesClient;
pub use linkerd2_proxy_api::*;
use linkerd_policy_controller_k8s_api::{self as k8s, ResourceExt};
use tokio::io;
#[macro_export]
macro_rules! assert_is_default_all_unauthenticated {
($config:expr) => {
assert_eq!(
$config.labels,
Some((
"name".to_string(),
"default:all-unauthenticated".to_string()
))
.into_iter()
.collect()
);
assert_eq!($config.authorizations.len(), 1);
};
}
#[macro_export]
macro_rules! assert_protocol_detect {
($config:expr) => {{
use linkerd2_proxy_api::inbound;
assert_eq!(
$config.protocol,
Some(inbound::ProxyProtocol {
kind: Some(inbound::proxy_protocol::Kind::Detect(
inbound::proxy_protocol::Detect {
timeout: Some(time::Duration::from_secs(10).into()),
}
)),
}),
);
}};
}
#[derive(Debug)]
pub struct PolicyClient {
client: InboundServerPoliciesClient<GrpcHttp>,
}
#[derive(Debug)]
struct GrpcHttp {
tx: hyper::client::conn::SendRequest<tonic::body::BoxBody>,
}
// === impl PolicyClient ===
impl PolicyClient {
pub async fn port_forwarded(client: &kube::Client) -> Self {
let pod = Self::get_policy_controller_pod(client)
.await
.expect("failed to find a policy controller pod");
let io = Self::connect_port_forward(client, &pod)
.await
.expect("failed to establish a port forward");
let http = GrpcHttp::handshake(io)
.await
.expect("failed to connect to the gRPC server");
PolicyClient {
client: InboundServerPoliciesClient::new(http),
}
}
pub async fn get_port(
&mut self,
ns: &str,
pod: &str,
port: u16,
) -> Result<inbound::Server, tonic::Status> {
let rsp = self
.client
.get_port(tonic::Request::new(inbound::PortSpec {
workload: format!("{}:{}", ns, pod),
port: port as u32,
}))
.await?;
Ok(rsp.into_inner())
}
pub async fn watch_port(
&mut self,
ns: &str,
pod: &str,
port: u16,
) -> Result<tonic::Streaming<inbound::Server>, tonic::Status> {
let rsp = self
.client
.watch_port(tonic::Request::new(inbound::PortSpec {
workload: format!("{}:{}", ns, pod),
port: port as u32,
}))
.await?;
Ok(rsp.into_inner())
}
async fn get_policy_controller_pod(client: &kube::Client) -> Result<String> {
let params = kube::api::ListParams::default()
.labels("linkerd.io/control-plane-component=destination");
let mut pods = kube::Api::<k8s::Pod>::namespaced(client.clone(), "linkerd")
.list(¶ms)
.await?;
let pod = pods
.items
.pop()
.ok_or_else(|| anyhow::anyhow!("no destination controller pods found"))?;
Ok(pod.name())
}
async fn connect_port_forward(
client: &kube::Client,
pod: &str,
) -> Result<impl io::AsyncRead + io::AsyncWrite + Unpin> {
let mut pf = kube::Api::<k8s::Pod>::namespaced(client.clone(), "linkerd")
.portforward(pod, &[8090])
.await?;
let io = pf.ports()[0].stream().expect("must have a stream");
Ok(io)
}
}
// === impl GrpcHttp ===
impl GrpcHttp {
async fn handshake<I>(io: I) -> Result<Self>
where
I: io::AsyncRead + io::AsyncWrite + Unpin + Send + 'static,
{
let (tx, conn) = hyper::client::conn::Builder::new()
.http2_only(true)
.handshake(io)
.await?;
tokio::spawn(conn);
Ok(Self { tx })
}
}
impl hyper::service::Service<hyper::Request<tonic::body::BoxBody>> for GrpcHttp {
type Response = hyper::Response<hyper::Body>;
type Error = hyper::Error;
type Future = hyper::client::conn::ResponseFuture;
fn poll_ready(
&mut self,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Result<(), Self::Error>> {
self.tx.poll_ready(cx)
}
fn call(&mut self, req: hyper::Request<tonic::body::BoxBody>) -> Self::Future {
let (mut parts, body) = req.into_parts();
let mut uri = parts.uri.into_parts();
uri.scheme = Some(hyper::http::uri::Scheme::HTTP);
uri.authority = Some(
"linkerd-destination.linkerd.svc.cluster.local:8090"
.parse()
.unwrap(),
);
parts.uri = hyper::Uri::from_parts(uri).unwrap();
self.tx.call(hyper::Request::from_parts(parts, body))
}
}