path: root/src/stats_task.rs

use bollard::container::{BlkioStatsEntry, MemoryStatsStats, Stats, StatsOptions};
use bollard::models::ContainerSummary;
use bollard::Docker;
use opentelemetry::metrics::Meter;
use opentelemetry::KeyValue;
use std::mem::MaybeUninit;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::task::JoinHandle;
use tokio_stream::StreamExt;
use crate::s_log::*;

// I do not enjoy taking a bunch of Rcs but tokio needs ownership so fine.
pub fn launch_stats_task(
	container: ContainerSummary,
	docker: Arc<Docker>,
	meter: Arc<Meter>
) -> JoinHandle<()> {
	tokio::spawn(async move {
		// extract some container info
		let container_id = container.id.unwrap();
		let container_name = container
			.names
			.iter()
			.flatten()
			.next()
			.map(|n| n.trim_start_matches("/").to_owned());

		let mut stats_stream = docker.stats(
			container_id.as_str(),
			Some(StatsOptions {
				stream: true,
				one_shot: false,
			}),
		);

		// use the first read only for stats diffing for blkio - don't need for cpu thanks to precpu.
		#[allow(unused_assignments)]
		let mut first_read = MaybeUninit::uninit();

		loop {
			match stats_stream.next().await {
				None => return,
				Some(Ok(st)) => {
					first_read = MaybeUninit::new(st);
					break;
				}
				Some(Err(err)) => {
					error(format_args!("Failed to get stats for container {container_id}!: {err:?}"), [("container_id", &*container_id)]);
				}
			}
		}

		// I'm going to rust jail!
		let first_read = unsafe { first_read.assume_init() };
		let Stats {
			blkio_stats,
			networks: mut last_net_stats,
			memory_stats: mut last_mem_stats,
			..
		} = first_read;

		let mut last_io_stats = blkio_stats.io_service_bytes_recursive;

		// container labels shared for all metrics
		let mut shared_labels = vec![
			KeyValue::new("id", container_id.to_owned()),
			KeyValue::new(
				"image",
				container.image.unwrap_or(container.image_id.unwrap()),
			),
		];

		if let Some(name) = container_name {
			shared_labels.push(KeyValue::new("name", name));
		}

		if let Some(docker_labels) = &container.labels {
			for (key, value) in docker_labels {
				shared_labels.push(KeyValue::new(
					"container_label_".to_string() + key,
					value.clone(),
				))
			}
		}

		// other label sets that are static per container
		let mut labels_mem_container_min_c = shared_labels.clone();
		labels_mem_container_min_c.push(KeyValue::new("failure_type", "pgfault"));

		let mut labels_mem_container_maj_c = shared_labels.clone();
		labels_mem_container_maj_c.push(KeyValue::new("failure_type", "pgmajfault"));

		let mut labels_mem_container_min_h = labels_mem_container_min_c.clone();
		labels_mem_container_min_h.push(KeyValue::new("scope", "hierarchy"));
		labels_mem_container_min_c.push(KeyValue::new("scope", "container"));

		let mut labels_mem_container_maj_h = labels_mem_container_maj_c.clone();
		labels_mem_container_maj_h.push(KeyValue::new("scope", "hierarchy"));
		labels_mem_container_maj_c.push(KeyValue::new("scope", "container"));

		// free space and make immutable
		shared_labels.shrink_to_fit();
		let shared_labels = &shared_labels[..];

		labels_mem_container_min_c.shrink_to_fit();
		labels_mem_container_min_h.shrink_to_fit();
		labels_mem_container_maj_c.shrink_to_fit();
		labels_mem_container_maj_h.shrink_to_fit();
		let labels_mem_container_min_c = &labels_mem_container_min_c[..];
		let labels_mem_container_min_h = &labels_mem_container_min_h[..];
		let labels_mem_container_maj_c = &labels_mem_container_maj_c[..];
		let labels_mem_container_maj_h = &labels_mem_container_maj_h[..];

		//println!("Starting reporting for container: {shared_labels:?}");

		// create meters
		let meter_container_cpu_usage_seconds_total = meter
			.f64_counter("container_cpu_usage_seconds_total")
			.with_unit("s")
			.with_description("Cumulative cpu time consumed")
			.build();
		let meter_container_cpu_user_seconds_total = meter
			.f64_counter("container_cpu_user_seconds_total")
			.with_unit("s")
			.with_description("Cumulative userland cpu time consumed")
			.build();
		let meter_container_cpu_system_seconds_total = meter
			.f64_counter("container_cpu_system_seconds_total")
			.with_unit("s")
			.with_description("Cumulative kernel cpu time consumed")
			.build();

		let meter_container_cpu_cfs_periods_total = meter
			.u64_counter("container_cpu_cfs_periods_total")
			.with_description("Number of elapsed enforcement period intervals")
			.build();
		let meter_container_cpu_cfs_throttled_periods_total = meter
			.u64_counter("container_cpu_cfs_throttled_periods_total")
			.with_description("Number of throttled period intervals")
			.build();
		let meter_container_cpu_cfs_throttled_seconds_total = meter
			.f64_counter("container_cpu_cfs_throttled_seconds_total")
			.with_unit("s")
			.with_description("Total time duration the container has been throttled")
			.build();

		let meter_container_fs_reads_bytes_total = meter
			.u64_counter("container_fs_reads_bytes_total")
			.with_unit("By")
			.with_description("Cumulative bytes read")
			.build();
		let meter_container_fs_writes_bytes_total = meter
			.u64_counter("container_fs_writes_bytes_total")
			.with_unit("By")
			.with_description("Cumulative bytes written")
			.build();

		let meter_container_last_seen = meter
			.u64_gauge("container_last_seen")
			.with_description("Last time this container was seen by ContainerSpy")
			.build();

		// annoyingly a lot of the meter names cadvisor went with don't have units attached even though they have known units
		let meter_container_memory_cache = meter
			.u64_gauge("container_memory_cache")
			//.with_unit("By")
			.with_description("Total page cache memory")
			.build();
		let meter_container_memory_failures_total = meter
			.u64_counter("container_memory_failures_total")
			.with_description("Cumulative count of memory allocation failures")
			.build();
		let meter_container_memory_mapped_file = meter
			.u64_gauge("container_memory_mapped_file")
			//.with_unit("By")
			.with_description("Size of memory mapped files")
			.build();
		let meter_container_memory_rss = meter
			.u64_gauge("container_memory_rss")
			//.with_unit("By")
			.with_description("Size of RSS")
			.build();
		let meter_container_memory_usage_bytes = meter
			.u64_gauge("container_memory_usage_bytes")
			.with_unit("By")
			.with_description(
				"Current memory usage, including all memory regardless of when it was accessed",
			)
			.build();
		let meter_container_memory_working_set_bytes = meter
			.u64_gauge("container_memory_working_set_bytes")
			.with_unit("By")
			.with_description("Current working set")
			.build();

		let meter_container_network_receive_bytes_total = meter
			.u64_counter("container_network_receive_bytes_total")
			.with_unit("By")
			.with_description("Cumulative count of bytes received")
			.build();
		#[cfg(not(windows))]
		let meter_container_network_receive_errors_total = meter
			.u64_counter("container_network_receive_errors_total")
			.with_description("Cumulative count of errors encountered while receiving")
			.build();
		let meter_container_network_receive_packets_dropped_total = meter
			.u64_counter("container_network_receive_packets_dropped_total")
			.with_description("Cumulative count of packets dropped while receiving")
			.build();
		let meter_container_network_receive_packets_total = meter
			.u64_counter("container_network_receive_packets_total")
			.with_description("Cumulative count of packets received")
			.build();

		let meter_container_network_transmit_bytes_total = meter
			.u64_counter("container_network_transmit_bytes_total")
			.with_unit("By")
			.with_description("Cumulative count of bytes transmitted")
			.build();
		#[cfg(not(windows))]
		let meter_container_network_transmit_errors_total = meter
			.u64_counter("container_network_transmit_errors_total")
			.with_description("Cumulative count of errors encountered while transmitting")
			.build();
		let meter_container_network_transmit_packets_dropped_total = meter
			.u64_counter("container_network_transmit_packets_dropped_total")
			.with_description("Cumulative count of packets dropped while transmitting")
			.build();
		let meter_container_network_transmit_packets_total = meter
			.u64_counter(