Struct TelemetryDriver

pub struct TelemetryDriver { /* private fields */ }

Triggers registered ProcessesTelemetryMessages to poll for messages.

Runs its own background thread. The thread stops once this struct is dropped.

A TelemetryDriver can be ‘mounted’ into the hierarchy. If done so, it will still poll its children on its own thread independently.

Optional Metrics

The driver can be configured to collect metrics on its own activities.

The metrics will be added to all snapshots under a field named _metrix which contains the following fields:

• collections_per_second: The number of observation collection runs done per second

• collection_times_us: A histogram of the time each observation collection took in microseconds.

• observations_processed_per_second: The number of observations processed per second.

• observations_processed_per_collection: A histogram of the number of observations processed during each run

• observations_dropped_per_second: The number of observations dropped per second. See also max_observation_age.

• observations_dropped_per_collection: A histogram of the number of observations dropped during each run. See also max_observation_age.

• snapshots_per_second: The number of snapshots taken per second.

• snapshots_times_us: A histogram of the times it took to take a snapshot in microseconds

• dropped_observations_alarm: Will be true if observations have been dropped. Will by default stay true for 60 seconds once triggered.

• inactivity_alarm: Will be true if no observations have been made for a certain amount of time. The default is 60 seconds.

Implementations

impl TelemetryDriver

pub fn new( name: Option<String>, title: Option<String>, description: Option<String>, processing_strategy: ProcessingStrategy, with_driver_metrics: bool, ) -> TelemetryDriver

Creates a new TelemetryDriver.

max_observation_age is the maximum age of an Observation to be taken into account. This is determined by the timestamp field of an Observation. Observations that are too old are simply dropped. The default is 60 seconds.

pub fn name(&self) -> Option<&str>

Gets the name of this driver

pub fn change_processing_stragtegy(&self, strategy: ProcessingStrategy)

Changes the ProcessingStrategy

pub fn pause(&self)

Pauses processing of observations.

pub fn resume(&self)

Resumes processing of observations

pub fn snapshot(&self, descriptive: bool) -> Result<Snapshot, GetSnapshotError>

Examples found in repository

examples/demo_run.rs (line 184)

fn main() {
    let builder = DriverBuilder::new("demo");
    let mut driver = builder.build();
    //driver.change_processing_stragtegy(ProcessingStrategy::DropAll);
    //driver.pause();

    let (foo_transmitter, foo_processor) = create_foo_metrics();
    let (bar_transmitter, bar_processor) = create_bar_metrics();

    driver.add_processor(foo_processor);
    driver.add_processor(bar_processor);

    let polled_counter = PolledCounter::new();
    let mut polled_instrument =
        PollingInstrument::new_with_defaults("polled_instrument_3", polled_counter);
    polled_instrument.set_title("The polled counter 3");
    polled_instrument.set_description("A counter that is increased when a snapshot is polled");

    driver.add_snapshooter(polled_instrument);

    let start = Instant::now();

    let handle1 = {
        let foo_transmitter = foo_transmitter.clone();
        let bar_transmitter = bar_transmitter.clone();

        thread::spawn(move || {
            for n in 0..5_000_000 {
                foo_transmitter.observed_one_value(FooLabel::A, n, Instant::now());
                bar_transmitter.measure_time(BarLabel::C, start);
            }
        })
    };

    // Poll a snapshot for the counter
    let _ = driver.snapshot(true).unwrap();

    let handle2 = {
        let foo_transmitter = foo_transmitter;
        let bar_transmitter = bar_transmitter.clone();

        thread::spawn(move || {
            for n in 0..5_000_000u64 {
                foo_transmitter.observed_one_value(FooLabel::B, n, Instant::now());
                bar_transmitter.observed_one_value(BarLabel::B, n * n, Instant::now());
            }
        })
    };

    // Poll a snapshot for the counter
    let _ = driver.snapshot(true).unwrap();

    let handle3 = {
        let bar_transmitter = bar_transmitter;

        thread::spawn(move || {
            for i in 0..5_000_000 {
                bar_transmitter.observed_one_value(BarLabel::A, i, Instant::now());
            }
        })
    };

    handle1.join().unwrap();
    handle2.join().unwrap();
    handle3.join().unwrap();

    //driver.resume();

    println!(
        "Sending observations took {:?}. Sleeping 1 secs to collect remaining data. \
         Depending on your machine you might see that not all metrics have a count \
         of 5 million observations.",
        start.elapsed()
    );

    thread::sleep(Duration::from_secs(1));

    println!("\n\n\n=======================\n\n");

    println!(
        "Get snapshot. If it still blocks here there are still many messages to be processed..."
    );

    println!("\n\n\n=======================\n\n");

    let snapshot = driver.snapshot(true).unwrap();

    let mut config = JsonConfig::default();
    config.pretty = Some(4);

    println!("{:?}", snapshot);
    println!("\n\n\n=======================\n\n");
    println!("{}", snapshot.to_json(&config));
}

pub fn snapshot_async( &self, descriptive: bool, ) -> impl Future<Output = Result<Snapshot, GetSnapshotError>> + Send + 'static

Trait Implementations

impl AggregatesProcessors for TelemetryDriver

fn add_processor<P: ProcessesTelemetryMessages>(&mut self, processor: P)

Add a processor.

fn add_snapshooter<S: PutsSnapshot>(&mut self, snapshooter: S)

Add a snapshooter.

fn attached_mount(&mut self, mount: ProcessorMount) -> AttachedMount

impl Clone for TelemetryDriver

fn clone(&self) -> TelemetryDriver

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Default for TelemetryDriver

fn default() -> TelemetryDriver

Returns the “default value” for a type.

impl Descriptive for TelemetryDriver

fn title(&self) -> Option<&str>

fn description(&self) -> Option<&str>

impl ProcessesTelemetryMessages for TelemetryDriver

fn process( &mut self, _max: usize, _strategy: ProcessingStrategy, ) -> ProcessingOutcome

Receive and handle pending operations

impl PutsSnapshot for TelemetryDriver

fn put_snapshot(&self, into: &mut Snapshot, descriptive: bool)

Puts the current snapshot values into the given Snapshot thereby following the guidelines of PutsSnapshot.