Trait TransmitsTelemetryData

pub trait TransmitsTelemetryData<L> {
    // Required methods
    fn transmit(&self, observation: Observation<L>) -> &Self;
    fn add_handler<H: HandlesObservations<Label = L>>(
        &self,
        handler: H,
    ) -> &Self
       where L: Send + 'static;
    fn add_cockpit(&self, cockpit: Cockpit<L>) -> &Self;
    fn remove_cockpit<T: Into<String>>(&self, name: T) -> &Self;
    fn add_panel_to_cockpit<T: Into<String>>(
        &self,
        cockpit_name: T,
        panel: Panel<L>,
    ) -> &Self;
    fn remove_panel_from_cockpit<U: Into<String>, V: Into<String>>(
        &self,
        cockpit_name: U,
        panel_name: V,
    ) -> &Self;

    // Provided methods
    fn observed(&self, label: L, count: u64, timestamp: Instant) -> &Self { ... }
    fn observed_one(&self, label: L, timestamp: Instant) -> &Self { ... }
    fn observed_one_value<V: Into<ObservedValue>>(
        &self,
        label: L,
        value: V,
        timestamp: Instant,
    ) -> &Self { ... }
    fn observed_duration(
        &self,
        label: L,
        duration: Duration,
        timestamp: Instant,
    ) -> &Self { ... }
    fn observed_now(&self, label: L, count: u64) -> &Self { ... }
    fn observed_one_now(&self, label: L) -> &Self { ... }
    fn observed_one_value_now<V: Into<ObservedValue>>(
        &self,
        label: L,
        value: V,
    ) -> &Self { ... }
    fn observed_one_duration_now(&self, label: L, duration: Duration) -> &Self { ... }
    fn measure_time(&self, label: L, from: Instant) -> &Self { ... }
}

Transmits telemetry data to the backend.

Implementors should transfer Observations to a backend and manipulate the instruments there to not to interfere to much with the actual task being measured/observed

Required Methods

fn transmit(&self, observation: Observation<L>) -> &Self

Transit an observation to the backend.

fn add_handler<H: HandlesObservations<Label = L>>(&self, handler: H) -> &Self

where L: Send + 'static,

Add a handler.

fn add_cockpit(&self, cockpit: Cockpit<L>) -> &Self

Add a Copckpit

If the cockpit has a name and another cockpit with the same name is already present the cockpit will not be added.

fn remove_cockpit<T: Into<String>>(&self, name: T) -> &Self

fn add_panel_to_cockpit<T: Into<String>>( &self, cockpit_name: T, panel: Panel<L>, ) -> &Self

Add a Panel to a Cockpit if that Cockpit has the given name.

fn remove_panel_from_cockpit<U: Into<String>, V: Into<String>>( &self, cockpit_name: U, panel_name: V, ) -> &Self

Removes the panel with the given name from a cockpit with the given name.

This means the cockpit and the panel must have a name set.

Provided Methods

fn observed(&self, label: L, count: u64, timestamp: Instant) -> &Self

Observed count occurrences at time timestamp

Convenience method. Simply calls transmit

fn observed_one(&self, label: L, timestamp: Instant) -> &Self

Observed one occurrence at time timestamp

Convenience method. Simply calls transmit

fn observed_one_value<V: Into<ObservedValue>>( &self, label: L, value: V, timestamp: Instant, ) -> &Self

Observed one occurrence with value value at time timestamp

Convenience method. Simply calls transmit

Examples found in repository

examples/demo_run.rs (line 177)

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));
}

fn observed_duration( &self, label: L, duration: Duration, timestamp: Instant, ) -> &Self

Sends a Duration as an observed value observed at timestamp. The Duration is converted to nanoseconds.

fn observed_now(&self, label: L, count: u64) -> &Self

Observed count occurrences at now.

Convenience method. Simply calls observed with the current timestamp.

fn observed_one_now(&self, label: L) -> &Self

Observed one occurrence now

Convenience method. Simply calls observed_one with the current timestamp.

fn observed_one_value_now<V: Into<ObservedValue>>( &self, label: L, value: V, ) -> &Self

Observed one occurrence with value value now

Convenience method. Simply calls observed_one_value with the current timestamp.

fn observed_one_duration_now(&self, label: L, duration: Duration) -> &Self

Sends a Duration as an observed value observed with the current timestamp.

The Duration is converted to nanoseconds internally.

fn measure_time(&self, label: L, from: Instant) -> &Self

Measures the time from from until now.

The resulting duration is an observed value with the measured duration in nanoseconds.

Examples found in repository

examples/demo_run.rs (line 178)

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));
}

Dyn Compatibility

This trait is not dyn compatible.

In older versions of Rust, dyn compatibility was called "object safety", so this trait is not object safe.

Implementors

impl<L: Send + 'static> TransmitsTelemetryData<L> for TelemetryTransmitter<L>