pub(crate) async fn service_discovery<F>(
    config: Arc<config::Config>,
    registry_client: Arc<impl RegistryClient>,
    log: Logger,
    metrics: Arc<metrics::Metrics>,
    graceful: F,
) where
    F: Future<Output = ()>,
{
    let mut graceful = Box::pin(graceful);
    let mut interval = time::interval(config.discover_every);

    let mut extra_labels: Labels = Default::default();
    extra_labels.insert(IC_NAME.to_string(), config.ic_name.clone());

    loop {
        let mut event_timer = ecs::Timer::start();

        let registry_version = registry_client.get_latest_version();

        let result = get_ic_topology(&registry_client, registry_version, extra_labels.clone())
            .map_err(ServiceDiscoveryError::from)
            .and_then(|node_records| {
                write_file(
                    &node_records,
                    &config.service_discovery_file,
                    config.service_discovery_file_mode,
                )
            });

        let mut canonical_event = ecs::Event::new(ecs::Kind::Event);
        canonical_event.event.set(ecs::Category::Database);
        canonical_event.event.set(ecs::Type::Access);

        let metric_status_value = match &result {
            Ok(_) => {
                metrics
                    .ic_topology_registry_version
                    .set(registry_version.get() as i64);
                canonical_event.event.set(ecs::Outcome::Success);

                "success"
            }
            Err(error) => {
                canonical_event.event.set(ecs::Outcome::Failure);
                let ecs_error = ecs::error_fields::Error {
                    message: Some(error.to_string()),
                    ..Default::default()
                };
                canonical_event.error = Some(ecs_error);

                error.into()
            }
        };

        let mut ic_registry_map = BTreeMap::new();
        ic_registry_map.insert("version".into(), registry_version.get().into());
        let mut ic_map = BTreeMap::new();
        ic_map.insert("registry".into(), ic_registry_map.into());
        if !ic_map.is_empty() {
            canonical_event.extra_values.insert("IC", ic_map.into());
        }

        info!(log, ""; &canonical_event);

        event_timer.finish();
        canonical_event.event.set(&event_timer);

        metrics
            .ic_service_discovery_duration_seconds
            .with_label_values(&[metric_status_value])
            .observe(event_timer.elapsed_secs());

        loop {
            // Wait for the next interval, or a graceful shutdown signal
            tokio::select! {
                _ = graceful.as_mut() => {
                    info!(&log, "Graceful shutdown");
                    return;
                },
                _ = interval.tick() => {}
            }

            // Wait if the registry version hasn't changed and the last run was a
            // successful.
            if registry_client.get_latest_version() == registry_version && result.is_ok() {
                metrics.ic_service_discovery_skipped_total.inc();
                continue;
            }

            // Back to the outer loop to update the topology
            break;
        }
    }
}
// Periodically monitor the registry for changes. Translate and write the
// registry content changes to prometheus service discovery file.
Periodically monitor the registry for changes. Translate and write the
registry content changes to prometheus service discovery file.
Periodically monitor the registry for changes . Translate and write the registry content changes to prometheus service discovery file .
pub(crate) async fn service_discovery<F>(
    config: Arc<config::Config>,
    registry_client: Arc<impl RegistryClient>,
    log: Logger,
    metrics: Arc<metrics::Metrics>,
    graceful: F,
) where
    F: Future<Output = ()>,
{
    let mut graceful = Box::pin(graceful);
    let mut interval = time::interval(config.discover_every);
    let mut extra_labels: Labels = Default::default();
    extra_labels.insert(IC_NAME.to_string(), config.ic_name.clone());
    loop {
        let mut event_timer = ecs::Timer::start();
        let registry_version = registry_client.get_latest_version();
        let result = get_ic_topology(&registry_client, registry_version, extra_labels.clone())
            .map_err(ServiceDiscoveryError::from)
            .and_then(|node_records| {
                write_file(
                    &node_records,
                    &config.service_discovery_file,
                    config.service_discovery_file_mode,
                )
            });
        let mut canonical_event = ecs::Event::new(ecs::Kind::Event);
        canonical_event.event.set(ecs::Category::Database);
        canonical_event.event.set(ecs::Type::Access);
        let metric_status_value = match &result {
            Ok(_) => {
                metrics
                    .ic_topology_registry_version
                    .set(registry_version.get() as i64);
                canonical_event.event.set(ecs::Outcome::Success);
                "success"
            }
            Err(error) => {
                canonical_event.event.set(ecs::Outcome::Failure);
                let ecs_error = ecs::error_fields::Error {
                    message: Some(error.to_string()),
                    ..Default::default()
                };
                canonical_event.error = Some(ecs_error);
                error.into()
            }
        };
        let mut ic_registry_map = BTreeMap::new();
        ic_registry_map.insert("version".into(), registry_version.get().into());
        let mut ic_map = BTreeMap::new();
        ic_map.insert("registry".into(), ic_registry_map.into());
        if !ic_map.is_empty() {
            canonical_event.extra_values.insert("IC", ic_map.into());
        }
        info!(log, ""; &canonical_event);
        event_timer.finish();
        canonical_event.event.set(&event_timer);
        metrics
            .ic_service_discovery_duration_seconds
            .with_label_values(&[metric_status_value])
            .observe(event_timer.elapsed_secs());
        loop {
            // Wait for the next interval, or a graceful shutdown signal
            tokio::select! {
                _ = graceful.as_mut() => {
                    info!(&log, "Graceful shutdown");
                    return;
                },
                _ = interval.tick() => {}
            }
            // Wait if the registry version hasn't changed and the last run was a
            // successful.
            if registry_client.get_latest_version() == registry_version && result.is_ok() {
                metrics.ic_service_discovery_skipped_total.inc();
                continue;
            }
            // Back to the outer loop to update the topology
            break;
        }
    }
}
pub ( crate ) async fn service_discovery < F > ( config : Arc < config :: Config > , registry_client : Arc < impl RegistryClient > , log : Logger , metrics : Arc < metrics :: Metrics > , graceful : F , ) where F : Future < Output = ( ) > , { let mut graceful = Box :: pin ( graceful ) ; let mut interval = time :: interval ( config . discover_every ) ; let mut extra_labels : Labels = Default :: default ( ) ; extra_labels . insert ( IC_NAME . to_string ( ) , config . ic_name . clone ( ) ) ; loop { let mut event_timer = ecs :: Timer :: start ( ) ; let registry_version = registry_client . get_latest_version ( ) ; let result = get_ic_topology ( & registry_client , registry_version , extra_labels . clone ( ) ) . map_err ( ServiceDiscoveryError :: from ) . and_then ( | node_records | { write_file ( & node_records , & config . service_discovery_file , config . service_discovery_file_mode , ) } ) ; let mut canonical_event = ecs :: Event :: new ( ecs :: Kind :: Event ) ; canonical_event . event . set ( ecs :: Category :: Database ) ; canonical_event . event . set ( ecs :: Type :: Access ) ; let metric_status_value = match & result { Ok ( _ ) => { metrics . ic_topology_registry_version . set ( registry_version . get ( ) as i64 ) ; canonical_event . event . set ( ecs :: Outcome :: Success ) ; " " } Err ( error ) => { canonical_event . event . set ( ecs :: Outcome :: Failure ) ; let ecs_error = ecs :: error_fields :: Error { message : Some ( error . to_string ( ) ) , .. Default :: default ( ) } ; canonical_event . error = Some ( ecs_error ) ; error . into ( ) } } ; let mut ic_registry_map = BTreeMap :: new ( ) ; ic_registry_map . insert ( " " . into ( ) , registry_version . get ( ) . into ( ) ) ; let mut ic_map = BTreeMap :: new ( ) ; ic_map . insert ( " " . into ( ) , ic_registry_map . into ( ) ) ; if ! ic_map . is_empty ( ) { canonical_event . extra_values . insert ( " " , ic_map . into ( ) ) ; } info ! ( log " " canonical_event ) ; event_timer . finish ( ) ; canonical_event . event . set ( & event_timer ) ; metrics . ic_service_discovery_duration_seconds . with_label_values ( & [ metric_status_value ] ) . observe ( event_timer . elapsed_secs ( ) ) ; loop { // Wait for the next interval, or a graceful shutdown signal tokio :: select ! { graceful as_mut ( ) { info ( log " " ) return } interval tick ( ) { } } // Wait if the registry version hasn't changed and the last run was a // successful. if registry_client . get_latest_version ( ) == registry_version && result . is_ok ( ) { metrics . ic_service_discovery_skipped_total . inc ( ) ; continue ; } // Back to the outer loop to update the topology break ; } } }
Periodically monitor the registry for changes.
Periodically monitor the registry for changes .
config
Arc<config::Config>
registry_client
Arc<impl RegistryClient>
log
Logger
metrics
Arc<metrics::Metrics>
graceful
F
config
Arc<config::Config>
registry_client
Arc<impl RegistryClient>
log
Logger
metrics
Arc<metrics::Metrics>
graceful
F