-
Notifications
You must be signed in to change notification settings - Fork 1.3k
/
main.rs
182 lines (153 loc) · 5.18 KB
/
main.rs
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
174
175
176
177
178
179
180
181
182
#![deny(warnings, rust_2018_idioms)]
#![forbid(unsafe_code)]
use anyhow::{bail, Result};
use clap::Parser;
use futures::prelude::*;
use kube::api::ListParams;
use linkerd_policy_controller::{admission, k8s};
use linkerd_policy_controller_core::IpNet;
use parking_lot::Mutex;
use std::{net::SocketAddr, sync::Arc};
use tokio::time;
use tracing::{info, info_span, instrument, Instrument};
#[cfg(all(target_os = "linux", target_arch = "x86_64", target_env = "gnu"))]
#[global_allocator]
static GLOBAL: jemallocator::Jemalloc = jemallocator::Jemalloc;
const DETECT_TIMEOUT: time::Duration = time::Duration::from_secs(10);
#[derive(Debug, Parser)]
#[clap(name = "policy", about = "A policy resource prototype")]
struct Args {
#[clap(
parse(try_from_str),
long,
default_value = "linkerd=info,warn",
env = "LINKERD_POLICY_CONTROLLER_LOG"
)]
log_level: kubert::LogFilter,
#[clap(long, default_value = "plain")]
log_format: kubert::LogFormat,
#[clap(flatten)]
client: kubert::ClientArgs,
#[clap(flatten)]
server: kubert::ServerArgs,
#[clap(flatten)]
admin: kubert::AdminArgs,
/// Disables the admission controller server.
#[clap(long)]
admission_controller_disabled: bool,
#[clap(long, default_value = "0.0.0.0:8090")]
grpc_addr: SocketAddr,
/// Network CIDRs of pod IPs.
///
/// The default includes all private networks.
#[clap(
long,
default_value = "10.0.0.0/8,100.64.0.0/10,172.16.0.0/12,192.168.0.0/16"
)]
cluster_networks: IpNets,
#[clap(long, default_value = "cluster.local")]
identity_domain: String,
#[clap(long, default_value = "all-unauthenticated")]
default_policy: k8s::DefaultPolicy,
#[clap(long, default_value = "linkerd")]
control_plane_namespace: String,
}
#[tokio::main]
async fn main() -> Result<()> {
let Args {
admin,
client,
log_level,
log_format,
server,
grpc_addr,
admission_controller_disabled,
identity_domain,
cluster_networks: IpNets(cluster_networks),
default_policy,
control_plane_namespace,
} = Args::parse();
let server = if admission_controller_disabled {
None
} else {
Some(server)
};
let mut runtime = kubert::Runtime::builder()
.with_log(log_level, log_format)
.with_admin(admin)
.with_client(client)
.with_optional_server(server)
.build()
.await?;
// Build the index data structure, which will be used to process events from all watches
// The lookup handle is used by the gRPC server.
let (lookup, index) = {
let cluster = k8s::ClusterInfo {
networks: cluster_networks.clone(),
identity_domain,
control_plane_ns: control_plane_namespace,
};
let (lookup, idx) = k8s::Index::new(cluster, default_policy, DETECT_TIMEOUT);
(lookup, Arc::new(Mutex::new(idx)))
};
// Spawn resource indexers that update the index and publish lookups for the gRPC server.
let pods = runtime.watch_all(ListParams::default().labels("linkerd.io/control-plane-ns"));
tokio::spawn(k8s::index_pods(index.clone(), pods).instrument(info_span!("pods")));
let servers = runtime.watch_all(ListParams::default());
tokio::spawn(k8s::index_servers(index.clone(), servers).instrument(info_span!("servers")));
let serverauthorizations = runtime.watch_all(ListParams::default());
tokio::spawn(
k8s::index_serverauthorizations(index, serverauthorizations)
.instrument(info_span!("serverauthorizations")),
);
// Run the gRPC server, serving results by looking up against the index handle.
tokio::spawn(grpc(
grpc_addr,
cluster_networks,
lookup,
runtime.shutdown_handle(),
));
// TODO use resource caches instead of a client
let client = runtime.client();
let runtime = runtime.spawn_server(|| admission::Service { client });
// Block the main thread on the shutdown signal. Once it fires, wait for the background tasks to
// complete before exiting.
if runtime.run().await.is_err() {
bail!("Aborted");
}
Ok(())
}
#[derive(Debug)]
struct IpNets(Vec<IpNet>);
impl std::str::FromStr for IpNets {
type Err = anyhow::Error;
fn from_str(s: &str) -> Result<Self> {
s.split(',')
.map(|n| n.parse().map_err(Into::into))
.collect::<Result<Vec<IpNet>>>()
.map(Self)
}
}
#[instrument(skip(handle, drain))]
async fn grpc(
addr: SocketAddr,
cluster_networks: Vec<IpNet>,
handle: linkerd_policy_controller_k8s_index::Reader,
drain: drain::Watch,
) -> Result<()> {
let server =
linkerd_policy_controller_grpc::Server::new(handle, cluster_networks, drain.clone());
let (close_tx, close_rx) = tokio::sync::oneshot::channel();
tokio::pin! {
let srv = server.serve(addr, close_rx.map(|_| {}));
}
info!(%addr, "gRPC server listening");
tokio::select! {
res = (&mut srv) => res?,
handle = drain.signaled() => {
let _ = close_tx.send(());
handle.release_after(srv).await?
}
}
Ok(())
}