use std::collections::{BTreeMap, HashMap};
use super::types::{
AuthorizationPolicy, AuthorizationPolicySpec, AuthorizationRule, ConnectionPoolSettings,
Destination, DestinationRule, DestinationRuleSpec, Gateway, GatewaySpec, HttpRoute,
HttpRouteDestination, LoadBalancerSettings, MutualTls, Operation, OutlierDetection,
PeerAuthentication, PeerAuthenticationSpec, Port, PortSelector, Server, Source, Subset,
TcpSettings, TrafficPolicy, VirtualService, VirtualServiceSpec, WorkloadSelector,
};
use kube::Api;
use tracing::info;
pub struct IstioConfigManager {
client: kube::Client,
config: IstioConfig,
}
impl IstioConfigManager {
pub fn new(client: kube::Client, config: IstioConfig) -> Self {
Self { client, config }
}
pub async fn configure(&self) -> Result<(), Error> {
info!("Configuring Istio for Amber services");
self.configure_namespace().await?;
self.create_destination_rules().await?;
self.create_virtual_services().await?;
self.create_gateway().await?;
self.configure_peer_authentication().await?;
self.configure_authorization_policy().await?;
info!("Istio configuration completed successfully");
Ok(())
}
async fn configure_namespace(&self) -> Result<(), Error> {
let namespaces: Api<k8s_openapi::api::core::v1::Namespace> = Api::all(self.client.clone());
match namespaces.get(&self.config.namespace).await {
Ok(_) => {
let patch: _ = serde_json::json!({
"metadata": {
"labels": {
"istio-injection": "enabled"
}
}
});
let params: _ = kube::api::PatchParams::default();
namespaces
.patch(
&self.config.namespace,
¶ms,
&kube::api::Patch::Merge(&patch),
)
.await?;
}
Err(kube::Error::Api(ref err)) if err.code == 404 => {
let namespace: _ = k8s_openapi::api::core::v1::Namespace {
metadata: k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta {
name: Some(self.config.namespace.clone()),
labels: Some(std::collections::BTreeMap::from([(
"istio-injection".to_string(),
"enabled".to_string(),
)])),
annotations: Some(std::collections::BTreeMap::from([(
"istio.io/rev".to_string(),
"default".to_string(),
)])),
..Default::default()
},
spec: None,
status: None,
};
namespaces
.create(&kube::api::PostParams::default(), &namespace)
.await?;
info!("Created namespace: {}", self.config.namespace);
}
Err(e) => return Err(Error::Kube(e)),
}
Ok(())
}
async fn create_destination_rules(&self) -> Result<(), Error> {
let destination_rules: Api<DestinationRule> =
Api::namespaced(self.client.clone(), &self.config.namespace);
for service in &self.config.services {
let dr_spec: _ = DestinationRuleSpec {
host: service.name.clone(),
traffic_policy: Some(TrafficPolicy {
load_balancer: Some(LoadBalancerSettings {
simple: Some(match self.config.traffic_policy.load_balancer {
LoadBalancerAlgorithm::RoundRobin => "ROUND_ROBIN".to_string(),
LoadBalancerAlgorithm::LeastRequest => "LEAST_REQUEST".to_string(),
LoadBalancerAlgorithm::Random => "RANDOM".to_string(),
LoadBalancerAlgorithm::ConsistentHash => "PASSTHROUGH".to_string(),
}),
}),
connection_pool: Some(ConnectionPoolSettings {
tcp: Some(TcpSettings {
max_connections: Some(
self.config.traffic_policy.connection_pool.max_connections as i32,
),
connect_timeout: Some("10s".to_string()),
}),
http: None,
}),
outlier_detection: Some(OutlierDetection {
consecutive_errors: Some(
self.config
.traffic_policy
.outlier_detection
.consecutive_errors as i32,
),
interval: Some(format!(
"{}s",
self.config
.traffic_policy
.outlier_detection
.interval
.as_secs()
)),
base_ejection_time: Some(format!(
"{}s",
self.config
.traffic_policy
.outlier_detection
.base_ejection_time
.as_secs()
)),
max_ejection_percent: Some(50),
}),
}),
subsets: Some(vec![
Subset {
name: "v1".to_string(),
labels: Some(std::collections::HashMap::from([(
"version".to_string(),
"v1".to_string(),
)])),
},
Subset {
name: "v2".to_string(),
labels: Some(std::collections::HashMap::from([(
"version".to_string(),
"v2".to_string(),
)])),
},
]),
};
let dr: _ = DestinationRule::new(&format!("{}-dr", service.name), dr_spec);
let params: _ = kube::api::PostParams::default();
destination_rules.create(¶ms, &dr).await?;
info!("Created DestinationRule: {}-dr", service.name);
}
Ok(())
}
async fn create_virtual_services(&self) -> Result<(), Error> {
let virtual_services: Api<VirtualService> =
Api::namespaced(self.client.clone(), &self.config.namespace);
for service in &self.config.services {
let vs_spec: _ = VirtualServiceSpec {
hosts: vec![service.name.clone()],
gateways: Some(vec![format!("{}-gateway", service.name)]),
http: Some(vec![HttpRoute {
r#match: None,
route: Some(vec![HttpRouteDestination {
destination: Destination {
host: service.name.clone(),
subset: Some("v1".to_string()),
port: Some(PortSelector {
number: Some(service.port),
}),
},
weight: Some(100),
}]),
fault: None,
timeout: None,
retries: None,
}]),
};
let vs: _ = VirtualService::new(&format!("{}-vs", service.name), vs_spec);
let params: _ = kube::api::PostParams::default();
virtual_services.create(¶ms, &vs).await?;
info!("Created VirtualService: {}-vs", service.name);
}
Ok(())
}
async fn create_gateway(&self) -> Result<(), Error> {
let gateways: Api<Gateway> = Api::namespaced(self.client.clone(), &self.config.namespace);
for service in &self.config.services {
let gw_spec: _ = GatewaySpec {
selector: std::collections::HashMap::from([(
"istio".to_string(),
"ingressgateway".to_string(),
)]),
servers: vec![Server {
port: Port {
number: service.port,
name: service.name.clone(),
protocol: "HTTP".to_string(),
},
hosts: vec!["*".to_string()],
tls: None,
}],
};
let gw: _ = Gateway::new(&format!("{}-gateway", service.name), gw_spec);
let params: _ = kube::api::PostParams::default();
gateways.create(¶ms, &gw).await?;
info!("Created Gateway: {}-gateway", service.name);
}
Ok(())
}
async fn configure_peer_authentication(&self) -> Result<(), Error> {
if !self.config.mtls_enabled {
return Ok(());
}
let peer_authentications: Api<PeerAuthentication> =
Api::namespaced(self.client.clone(), &self.config.namespace);
let pa_spec: _ = PeerAuthenticationSpec {
selector: None,
mtls: Some(MutualTls {
mode: Some("STRICT".to_string()),
}),
};
let pa: _ = PeerAuthentication::new("default", pa_spec);
let params: _ = kube::api::PostParams::default();
peer_authentications.create(¶ms, &pa).await?;
info!("Configured PeerAuthentication with STRICT mTLS");
Ok(())
}
async fn configure_authorization_policy(&self) -> Result<(), Error> {
let authorization_policies: Api<AuthorizationPolicy> =
Api::namespaced(self.client.clone(), &self.config.namespace);
for service in &self.config.services {
let ap_spec: _ = AuthorizationPolicySpec {
selector: Some(WorkloadSelector {
match_labels: Some(std::collections::HashMap::from([(
"app".to_string(),
service.name.clone(),
)])),
}),
action: Some("ALLOW".to_string()),
rules: Some(vec![AuthorizationRule {
from: Some(vec![Source {
principals: None,
namespaces: Some(vec![self.config.namespace.clone()]),
}]),
to: Some(vec![Operation {
paths: None,
methods: Some(vec!["GET".to_string(), "POST".to_string()]),
}]),
}]),
};
let ap: _ = AuthorizationPolicy::new(&format!("{}-authz", service.name), ap_spec);
let params: _ = kube::api::PostParams::default();
authorization_policies.create(¶ms, &ap).await?;
info!("Created AuthorizationPolicy: {}-authz", service.name);
}
Ok(())
}
}
#[derive(Debug, Clone)]
pub struct IstioConfig {
pub namespace: String,
pub mtls_enabled: bool,
pub services: Vec<IstioService>,
pub traffic_policy: TrafficPolicyConfig,
}
#[derive(Debug, Clone)]
pub struct IstioService {
pub name: String,
pub port: u32,
pub protocol: String,
}
#[derive(Debug, Clone)]
pub struct TrafficPolicyConfig {
pub load_balancer: LoadBalancerAlgorithm,
pub connection_pool: ConnectionPoolConfig,
pub outlier_detection: OutlierDetectionConfig,
}
#[derive(Debug, Clone)]
pub enum LoadBalancerAlgorithm {
RoundRobin,
LeastRequest,
Random,
ConsistentHash,
}
#[derive(Debug, Clone)]
pub struct ConnectionPoolConfig {
pub max_connections: u32,
pub max_pending_requests: u32,
}
#[derive(Debug, Clone)]
pub struct OutlierDetectionConfig {
pub consecutive_errors: u32,
pub interval: std::time::Duration,
pub base_ejection_time: std::time::Duration,
}
#[derive(Debug, thiserror::Error)]
pub enum Error {
#[error("Kubernetes error: {0}")]
Kube(#[from] kube::Error),
#[error("Serde error: {0}")]
Serde(#[from] serde_json::Error),
#[error("Other error: {0}")]
Other(String),
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
#[test]
fn test_istio_config_creation() {
let config: _ = IstioConfig {
namespace: "amberjs-system".to_string(),
mtls_enabled: true,
services: vec![IstioService {
name: "amberjs-api".to_string(),
port: 8080,
protocol: "HTTP".to_string(),
}],
traffic_policy: TrafficPolicyConfig {
load_balancer: LoadBalancerAlgorithm::LeastRequest,
connection_pool: ConnectionPoolConfig {
max_connections: 100,
max_pending_requests: 10,
},
outlier_detection: OutlierDetectionConfig {
consecutive_errors: 5,
interval: std::time::Duration::from_secs(10),
base_ejection_time: std::time::Duration::from_secs(30),
},
},
};
assert_eq!(config.namespace, "amberjs-system");
assert!(config.mtls_enabled);
assert_eq!(config.services.len(), 1);
assert_eq!(config.services[0].name, "amberjs-api");
assert_eq!(config.services[0].port, 8080);
}
}