policy-controller: Update kube, k8s-openapi dependencies (#6661)

kube v0.59 depends on k8s-openapi v0.13, which includes breaking
changes.

This change updates these dependencies and modifies our code to account
for these changes.

Furthermore, we now use the k8s-openapi feature `v1_16` so that we use
an API version that is compatible with Linkerd's minimum support
kubernetes version.

Closes #6657 #6658 #6659

policy-controller/Cargo.toml

@@ -10,7 +10,7 @@ anyhow = "1"
 drain = "0.1"
 futures = "0.3"
 hyper = { version = "0.14", features = ["http1", "http2", "runtime", "server"] }
-kube = { version = "0.58", default-features = false, features = ["client", "derive", "native-tls"] }
+kube = { version = "0.59", default-features = false, features = ["client", "derive", "native-tls"] }
 linkerd-policy-controller-core = { path = "./core" }
 linkerd-policy-controller-grpc = { path = "./grpc" }
 linkerd-policy-controller-k8s-index = { path = "./k8s/index" }

policy-controller/k8s/api/Cargo.toml

@@ -7,9 +7,9 @@ publish = false
 
 [dependencies]
 futures = { version = "0.3", default-features = false }
-k8s-openapi = { version = "0.12.0", default-features = false, features = ["v1_20"] }
-kube = { version = "0.58.1", default-features = false, features = ["client", "derive", "native-tls"] }
-kube-runtime = { version = "0.58.1", default-features = false }
+k8s-openapi = { version = "0.13", default-features = false, features = ["v1_16"] }
+kube = { version = "0.59", default-features = false, features = ["client", "derive", "native-tls"] }
+kube-runtime = { version = "0.59", default-features = false }
 schemars = "0.8"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

policy-controller/k8s/api/src/labels.rs

@@ -5,7 +5,7 @@ use std::{
     sync::Arc,
 };
 
-#[derive(Clone, Debug, Eq, Default)]
+#[derive(Clone, Debug, Default)]
 pub struct Labels(Arc<Map>);
 
 pub type Map = BTreeMap<String, String>;
@@ -95,6 +95,13 @@ impl std::iter::FromIterator<Expression> for Selector {
 
 // === Labels ===
 
+impl From<Option<Map>> for Labels {
+    #[inline]
+    fn from(labels: Option<Map>) -> Self {
+        labels.unwrap_or_default().into()
+    }
+}
+
 impl From<Map> for Labels {
     #[inline]
     fn from(labels: Map) -> Self {
@@ -109,13 +116,23 @@ impl AsRef<Map> for Labels {
     }
 }
 
-impl<T: AsRef<Map>> std::cmp::PartialEq<T> for Labels {
+impl std::cmp::PartialEq<Self> for Labels {
     #[inline]
-    fn eq(&self, t: &T) -> bool {
+    fn eq(&self, t: &Self) -> bool {
         self.0.as_ref().eq(t.as_ref())
     }
 }
 
+impl std::cmp::PartialEq<Option<Map>> for Labels {
+    #[inline]
+    fn eq(&self, t: &Option<Map>) -> bool {
+        match t {
+            None => self.0.is_empty(),
+            Some(t) => t.eq(self.0.as_ref()),
+        }
+    }
+}
+
 impl std::iter::FromIterator<(String, String)> for Labels {
     fn from_iter<T: IntoIterator<Item = (String, String)>>(iter: T) -> Self {
         Self(Arc::new(iter.into_iter().collect()))

policy-controller/k8s/index/src/default_allow.rs

@@ -33,12 +33,14 @@ impl DefaultAllow {
     pub const ANNOTATION: &'static str = "policy.linkerd.io/default-allow";
 
     pub fn from_annotation(meta: &k8s::ObjectMeta) -> Result<Option<Self>> {
-        if let Some(v) = meta.annotations.get(Self::ANNOTATION) {
-            let mode = v.parse()?;
-            Ok(Some(mode))
-        } else {
-            Ok(None)
+        if let Some(ann) = meta.annotations.as_ref() {
+            if let Some(v) = ann.get(Self::ANNOTATION) {
+                let mode = v.parse()?;
+                return Ok(Some(mode));
+            }
         }
+
+        Ok(None)
     }
 }
 

policy-controller/k8s/index/src/node.rs

@@ -206,17 +206,17 @@ impl KubeletIps {
     fn try_from_node(node: k8s::Node) -> Result<Self> {
         let spec = node.spec.ok_or_else(|| anyhow!("node missing spec"))?;
 
-        let addrs = if spec.pod_cidrs.is_empty() {
+        let addrs = if let Some(cidrs) = spec.pod_cidrs {
+            cidrs
+                .into_iter()
+                .map(Self::try_from_cidr)
+                .collect::<Result<Vec<_>>>()?
+        } else {
             let cidr = spec
                 .pod_cidr
                 .ok_or_else(|| anyhow!("node missing pod_cidr"))?;
             let ip = Self::try_from_cidr(cidr)?;
             vec![ip]
-        } else {
-            spec.pod_cidrs
-                .into_iter()
-                .map(Self::try_from_cidr)
-                .collect::<Result<Vec<_>>>()?
         };
 
         Ok(Self(addrs.into()))

policy-controller/k8s/index/src/pod.rs

@@ -260,7 +260,7 @@ impl PodIndex {
                 // labels have changed, then we relink servers to pods in case label selections have
                 // changed.
                 let p = entry.get_mut();
-                if p.labels.as_ref() != &pod.metadata.labels {
+                if p.labels != pod.metadata.labels {
                     p.labels = pod.metadata.labels.into();
                     p.link_servers(servers);
                 }
@@ -281,7 +281,7 @@ impl PodIndex {
         let mut lookups = HashMap::new();
 
         for container in spec.containers.into_iter() {
-            for p in container.ports.into_iter() {
+            for p in container.ports.into_iter().flatten() {
                 if p.protocol.map(|p| p == "TCP").unwrap_or(true) {
                     let port = p.container_port as u16;
                     if ports.by_port.contains_key(&port) {

policy-controller/k8s/index/src/server.rs

@@ -210,7 +210,7 @@ impl SrvIndex {
             HashEntry::Occupied(mut entry) => {
                 // If something about the server changed, we need to update the config to reflect
                 // the change.
-                let new_labels = if entry.get().labels.as_ref() != &srv.metadata.labels {
+                let new_labels = if entry.get().labels != srv.metadata.labels {
                     Some(k8s::Labels::from(srv.metadata.labels))
                 } else {
                     None

policy-controller/k8s/index/src/tests.rs

@@ -462,7 +462,7 @@ fn mk_node(name: impl Into<String>, pod_net: IpNet) -> k8s::Node {
         },
         spec: Some(k8s::api::core::v1::NodeSpec {
             pod_cidr: Some(pod_net.to_string()),
-            pod_cidrs: vec![pod_net.to_string()],
+            pod_cidrs: Some(vec![pod_net.to_string()]),
             ..Default::default()
         }),
         status: Some(k8s::api::core::v1::NodeStatus::default()),
@@ -488,22 +488,24 @@ fn mk_pod(
                 .into_iter()
                 .map(|(name, ports)| k8s::api::core::v1::Container {
                     name: name.into(),
-                    ports: ports
-                        .into_iter()
-                        .map(|p| k8s::api::core::v1::ContainerPort {
-                            container_port: p as i32,
-                            ..Default::default()
-                        })
-                        .collect(),
+                    ports: Some(
+                        ports
+                            .into_iter()
+                            .map(|p| k8s::api::core::v1::ContainerPort {
+                                container_port: p as i32,
+                                ..Default::default()
+                            })
+                            .collect(),
+                    ),
                     ..Default::default()
                 })
                 .collect(),
             ..Default::default()
         }),
         status: Some(k8s::api::core::v1::PodStatus {
-            pod_ips: vec![k8s::api::core::v1::PodIP {
+            pod_ips: Some(vec![k8s::api::core::v1::PodIP {
                 ip: Some(pod_ip.to_string()),
-            }],
+            }]),
             ..Default::default()
         }),
     }
@@ -522,10 +524,12 @@ fn mk_server(
         metadata: k8s::ObjectMeta {
             namespace: Some(ns.into()),
             name: Some(name.into()),
-            labels: srv_labels
-                .into_iter()
-                .map(|(k, v)| (k.to_string(), v.to_string()))
-                .collect(),
+            labels: Some(
+                srv_labels
+                    .into_iter()
+                    .map(|(k, v)| (k.to_string(), v.to_string()))
+                    .collect(),
+            ),
             ..Default::default()
         },
         spec: k8s::policy::ServerSpec {