fn apply(&mut self, srv: polixy::Server, ns_authzs: &AuthzIndex) {
        let srv_name = srv.name();
        let port = srv.spec.port;
        let protocol = mk_protocol(srv.spec.proxy_protocol.as_ref());

        match self.index.entry(srv_name) {
            HashEntry::Vacant(entry) => {
                let labels = k8s::Labels::from(srv.metadata.labels);
                let authzs = ns_authzs
                    .filter_selected(entry.key(), labels.clone())
                    .map(|(n, a)| (n, a.clone()))
                    .collect::<BTreeMap<_, _>>();
                let meta = ServerMeta {
                    labels,
                    port,
                    pod_selector: srv.spec.pod_selector.into(),
                    protocol: protocol.clone(),
                };
                debug!(authzs = ?authzs.keys());
                let (tx, rx) = watch::channel(InboundServer {
                    protocol,
                    authorizations: authzs.clone(),
                });
                entry.insert(Server {
                    meta,
                    rx,
                    tx,
                    authorizations: authzs,
                });
            }

            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().meta.labels.as_ref() != &srv.metadata.labels {
                    Some(k8s::Labels::from(srv.metadata.labels))
                } else {
                    None
                };

                let new_protocol = if entry.get().meta.protocol == protocol {
                    Some(protocol)
                } else {
                    None
                };

                trace!(?new_labels, ?new_protocol);
                if new_labels.is_some() || new_protocol.is_some() {
                    // NB: Only a single task applies server updates, so it's
                    // okay to borrow a version, modify, and send it.  We don't
                    // need a lock because serialization is guaranteed.
                    let mut config = entry.get().rx.borrow().clone();

                    if let Some(labels) = new_labels {
                        let authzs = ns_authzs
                            .filter_selected(entry.key(), labels.clone())
                            .map(|(n, a)| (n, a.clone()))
                            .collect::<BTreeMap<_, _>>();
                        debug!(authzs = ?authzs.keys());
                        config.authorizations = authzs.clone();
                        entry.get_mut().meta.labels = labels;
                        entry.get_mut().authorizations = authzs;
                    }

                    if let Some(protocol) = new_protocol {
                        config.protocol = protocol.clone();
                        entry.get_mut().meta.protocol = protocol;
                    }
                    entry
                        .get()
                        .tx
                        .send(config)
                        .expect("server update must succeed");
                }

                // If the pod/port selector didn't change, we don't need to
                // refresh the index.
                if *entry.get().meta.pod_selector == srv.spec.pod_selector
                    && entry.get().meta.port == port
                {
                    return;
                }

                entry.get_mut().meta.pod_selector = srv.spec.pod_selector.into();
                entry.get_mut().meta.port = port;
            }
        }
    }
/// Update the index with a server instance.
Update the index with a server instance.
Update the index with a server instance .
fn apply(&mut self, srv: polixy::Server, ns_authzs: &AuthzIndex) {
        let srv_name = srv.name();
        let port = srv.spec.port;
        let protocol = mk_protocol(srv.spec.proxy_protocol.as_ref());
        match self.index.entry(srv_name) {
            HashEntry::Vacant(entry) => {
                let labels = k8s::Labels::from(srv.metadata.labels);
                let authzs = ns_authzs
                    .filter_selected(entry.key(), labels.clone())
                    .map(|(n, a)| (n, a.clone()))
                    .collect::<BTreeMap<_, _>>();
                let meta = ServerMeta {
                    labels,
                    port,
                    pod_selector: srv.spec.pod_selector.into(),
                    protocol: protocol.clone(),
                };
                debug!(authzs = ?authzs.keys());
                let (tx, rx) = watch::channel(InboundServer {
                    protocol,
                    authorizations: authzs.clone(),
                });
                entry.insert(Server {
                    meta,
                    rx,
                    tx,
                    authorizations: authzs,
                });
            }
            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().meta.labels.as_ref() != &srv.metadata.labels {
                    Some(k8s::Labels::from(srv.metadata.labels))
                } else {
                    None
                };
                let new_protocol = if entry.get().meta.protocol == protocol {
                    Some(protocol)
                } else {
                    None
                };
                trace!(?new_labels, ?new_protocol);
                if new_labels.is_some() || new_protocol.is_some() {
                    // NB: Only a single task applies server updates, so it's
                    // okay to borrow a version, modify, and send it.  We don't
                    // need a lock because serialization is guaranteed.
                    let mut config = entry.get().rx.borrow().clone();
                    if let Some(labels) = new_labels {
                        let authzs = ns_authzs
                            .filter_selected(entry.key(), labels.clone())
                            .map(|(n, a)| (n, a.clone()))
                            .collect::<BTreeMap<_, _>>();
                        debug!(authzs = ?authzs.keys());
                        config.authorizations = authzs.clone();
                        entry.get_mut().meta.labels = labels;
                        entry.get_mut().authorizations = authzs;
                    }
                    if let Some(protocol) = new_protocol {
                        config.protocol = protocol.clone();
                        entry.get_mut().meta.protocol = protocol;
                    }
                    entry
                        .get()
                        .tx
                        .send(config)
                        .expect("server update must succeed");
                }
                // If the pod/port selector didn't change, we don't need to
                // refresh the index.
                if *entry.get().meta.pod_selector == srv.spec.pod_selector
                    && entry.get().meta.port == port
                {
                    return;
                }
                entry.get_mut().meta.pod_selector = srv.spec.pod_selector.into();
                entry.get_mut().meta.port = port;
            }
        }
    }
fn apply ( & mut self , srv : polixy :: Server , ns_authzs : & AuthzIndex ) { let srv_name = srv . name ( ) ; let port = srv . spec . port ; let protocol = mk_protocol ( srv . spec . proxy_protocol . as_ref ( ) ) ; match self . index . entry ( srv_name ) { HashEntry :: Vacant ( entry ) => { let labels = k8s :: Labels :: from ( srv . metadata . labels ) ; let authzs = ns_authzs . filter_selected ( entry . key ( ) , labels . clone ( ) ) . map ( | ( n , a ) | ( n , a . clone ( ) ) ) . collect :: < BTreeMap < _ , _ > > ( ) ; let meta = ServerMeta { labels , port , pod_selector : srv . spec . pod_selector . into ( ) , protocol : protocol . clone ( ) , } ; debug ! ( authzs authzs keys ( ) ) ; let ( tx , rx ) = watch :: channel ( InboundServer { protocol , authorizations : authzs . clone ( ) , } ) ; entry . insert ( Server { meta , rx , tx , authorizations : authzs , } ) ; } 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 ( ) . meta . labels . as_ref ( ) != & srv . metadata . labels { Some ( k8s :: Labels :: from ( srv . metadata . labels ) ) } else { None } ; let new_protocol = if entry . get ( ) . meta . protocol == protocol { Some ( protocol ) } else { None } ; trace ! ( new_labels new_protocol ) ; if new_labels . is_some ( ) || new_protocol . is_some ( ) { // NB: Only a single task applies server updates, so it's // okay to borrow a version, modify, and send it.  We don't // need a lock because serialization is guaranteed. let mut config = entry . get ( ) . rx . borrow ( ) . clone ( ) ; if let Some ( labels ) = new_labels { let authzs = ns_authzs . filter_selected ( entry . key ( ) , labels . clone ( ) ) . map ( | ( n , a ) | ( n , a . clone ( ) ) ) . collect :: < BTreeMap < _ , _ > > ( ) ; debug ! ( authzs authzs keys ( ) ) ; config . authorizations = authzs . clone ( ) ; entry . get_mut ( ) . meta . labels = labels ; entry . get_mut ( ) . authorizations = authzs ; } if let Some ( protocol ) = new_protocol { config . protocol = protocol . clone ( ) ; entry . get_mut ( ) . meta . protocol = protocol ; } entry . get ( ) . tx . send ( config ) . expect ( " " ) ; } // If the pod/port selector didn't change, we don't need to // refresh the index. if * entry . get ( ) . meta . pod_selector == srv . spec . pod_selector && entry . get ( ) . meta . port == port { return ; } entry . get_mut ( ) . meta . pod_selector = srv . spec . pod_selector . into ( ) ; entry . get_mut ( ) . meta . port = port ; } } }
Update the index with a server instance.
Update the index with a server instance .
&mut self
srv
polixy::Server
ns_authzs
&AuthzIndex
&mut self
srv
polixy::Server
ns_authzs
&AuthzIndex