/// Spawn a background task to accept and handle metrics connections.  This
/// task does not terminate and if/when we support clean shutdown this
/// task will need to be joined.
Spawn a background task to accept and handle metrics connections.  This
task does not terminate and if/when we support clean shutdown this
task will need to be joined.
Spawn a background task to accept and handle metrics connections . This task does not terminate and if / when we support clean shutdown this task will need to be joined .
fn start_http(&self, address: SocketAddr) {
        let metrics_registry = self.metrics_registry.clone();
        let log = self.log.clone();
        let aservice = service_fn(move |_req| {
            // Clone again to ensure that `metrics_registry` outlives this closure.
            let metrics_registry = metrics_registry.clone();
            let encoder = TextEncoder::new();
            async move {
                let metric_families = metrics_registry.prometheus_registry().gather();
                let mut buffer = vec![];
                encoder.encode(&metric_families, &mut buffer).unwrap();
                Ok::<_, hyper::Error>(Response::new(Body::from(buffer)))
            }
        });
        let crypto_tls = self.crypto_tls.clone();
        // Temporarily listen on [::] so that we accept both IPv4 and IPv6 connections.
        // This requires net.ipv6.bindv6only = 0.  TODO: revert this once we have rolled
        // out IPv6 in prometheus and ic_p8s_service_discovery.
        let mut addr = "[::]:9090".parse::<SocketAddr>().unwrap();
        addr.set_port(address.port());
        self.rt_handle.spawn(async move {
            let listener = match TcpListener::bind(&addr).await {
                Err(e) => {
                    error!(log, "HTTP exporter server error: {}", e);
                    return;
                }
                Ok(listener) => listener,
            };
            let http = Http::new();
            loop {
                let log = log.clone();
                let http = http.clone();
                let aservice = aservice.clone();
                let crypto_tls = crypto_tls.clone();
                if let Ok((stream, _)) = listener.accept().await {
                    tokio::spawn(async move {
                        let mut b = [0_u8; 1];
                        if stream.peek(&mut b).await.is_ok() && b[0] == 22 {
                            if let Some((registry_client, crypto)) = crypto_tls {
                                // Note: the unwrap() can't fail since we tested Some(crypto)
                                // above.
                                let registry_version = registry_client.get_latest_version();
                                match crypto
                                    .perform_tls_server_handshake_without_client_auth(
                                        stream,
                                        registry_version,
                                    )
                                    .await
                                {
                                    Err(e) => warn!(log, "TLS error: {}", e),
                                    Ok(stream) => {
                                        if let Err(e) =
                                            http.serve_connection(stream, aservice).await
                                        {
                                            trace!(log, "Connection error: {}", e);
                                        }
                                    }
                                };
                            }
                        } else {
                            // Fallback to Http.
                            if let Err(e) = http.serve_connection(stream, aservice).await {
                                trace!(log, "Connection error: {}", e);
                            }
                        }
                    });
                }
            }
        });
    }
fn start_http ( & self , address : SocketAddr ) { let metrics_registry = self . metrics_registry . clone ( ) ; let log = self . log . clone ( ) ; let aservice = service_fn ( move | _req | { // Clone again to ensure that `metrics_registry` outlives this closure. let metrics_registry = metrics_registry . clone ( ) ; let encoder = TextEncoder :: new ( ) ; async move { let metric_families = metrics_registry . prometheus_registry ( ) . gather ( ) ; let mut buffer = vec ! [ ] ; encoder . encode ( & metric_families , & mut buffer ) . unwrap ( ) ; Ok :: < _ , hyper :: Error > ( Response :: new ( Body :: from ( buffer ) ) ) } } ) ; let crypto_tls = self . crypto_tls . clone ( ) ; // Temporarily listen on [::] so that we accept both IPv4 and IPv6 connections. // This requires net.ipv6.bindv6only = 0.  TODO: revert this once we have rolled // out IPv6 in prometheus and ic_p8s_service_discovery. let mut addr = " " . parse :: < SocketAddr > ( ) . unwrap ( ) ; addr . set_port ( address . port ( ) ) ; self . rt_handle . spawn ( async move { let listener = match TcpListener :: bind ( & addr ) . await { Err ( e ) => { error ! ( log " " e ) ; return ; } Ok ( listener ) => listener , } ; let http = Http :: new ( ) ; loop { let log = log . clone ( ) ; let http = http . clone ( ) ; let aservice = aservice . clone ( ) ; let crypto_tls = crypto_tls . clone ( ) ; if let Ok ( ( stream , _ ) ) = listener . accept ( ) . await { tokio :: spawn ( async move { let mut b = [ 0_u8 ; 1 ] ; if stream . peek ( & mut b ) . await . is_ok ( ) && b [ 0 ] == 22 { if let Some ( ( registry_client , crypto ) ) = crypto_tls { // Note: the unwrap() can't fail since we tested Some(crypto) // above. let registry_version = registry_client . get_latest_version ( ) ; match crypto . perform_tls_server_handshake_without_client_auth ( stream , registry_version , ) . await { Err ( e ) => warn ! ( log " " e ) , Ok ( stream ) => { if let Err ( e ) = http . serve_connection ( stream , aservice ) . await { trace ! ( log " " e ) ; } } } ; } } else { // Fallback to Http. if let Err ( e ) = http . serve_connection ( stream , aservice ) . await { trace ! ( log " " e ) ; } } } ) ; } } } ) ; }
Spawn a background task to accept and handle metrics connections.
Spawn a background task to accept and handle metrics connections .
&self
address
SocketAddr