diff --git a/console-api/proto/instrument.proto b/console-api/proto/instrument.proto index b2b96c47a..7308a9e1d 100644 --- a/console-api/proto/instrument.proto +++ b/console-api/proto/instrument.proto @@ -70,6 +70,17 @@ message Update { // Any new span metadata that was registered since the last update. common.RegisterMetadata new_metadata = 5; + + // The time "state" of the aggregator, such as paused or live. + Temporality temporality = 6; +} + +// The time "state" of the aggregator. +enum Temporality { + // The aggregator is currently live. + LIVE = 0; + // The aggregator is currently paused. + PAUSED = 1; } // `PauseResponse` is the value returned after a pause request. diff --git a/console-api/src/generated/rs.tokio.console.instrument.rs b/console-api/src/generated/rs.tokio.console.instrument.rs index 0fc2f9723..a2c773794 100644 --- a/console-api/src/generated/rs.tokio.console.instrument.rs +++ b/console-api/src/generated/rs.tokio.console.instrument.rs @@ -54,6 +54,9 @@ pub struct Update { /// Any new span metadata that was registered since the last update. #[prost(message, optional, tag = "5")] pub new_metadata: ::core::option::Option, + /// The time "state" of the aggregator, such as paused or live. + #[prost(enumeration = "Temporality", tag = "6")] + pub temporality: i32, } /// `PauseResponse` is the value returned after a pause request. #[allow(clippy::derive_partial_eq_without_eq)] @@ -63,6 +66,35 @@ pub struct PauseResponse {} #[allow(clippy::derive_partial_eq_without_eq)] #[derive(Clone, Copy, PartialEq, ::prost::Message)] pub struct ResumeResponse {} +/// The time "state" of the aggregator. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] +#[repr(i32)] +pub enum Temporality { + /// The aggregator is currently live. + Live = 0, + /// The aggregator is currently paused. + Paused = 1, +} +impl Temporality { + /// String value of the enum field names used in the ProtoBuf definition. + /// + /// The values are not transformed in any way and thus are considered stable + /// (if the ProtoBuf definition does not change) and safe for programmatic use. + pub fn as_str_name(&self) -> &'static str { + match self { + Temporality::Live => "LIVE", + Temporality::Paused => "PAUSED", + } + } + /// Creates an enum from field names used in the ProtoBuf definition. + pub fn from_str_name(value: &str) -> ::core::option::Option { + match value { + "LIVE" => Some(Self::Live), + "PAUSED" => Some(Self::Paused), + _ => None, + } + } +} /// Generated client implementations. pub mod instrument_client { #![allow(unused_variables, dead_code, missing_docs, clippy::let_unit_value)] diff --git a/console-subscriber/examples/grpc_web/app/src/gen/instrument_pb.ts b/console-subscriber/examples/grpc_web/app/src/gen/instrument_pb.ts index 09ec3b34d..1e47b774d 100644 --- a/console-subscriber/examples/grpc_web/app/src/gen/instrument_pb.ts +++ b/console-subscriber/examples/grpc_web/app/src/gen/instrument_pb.ts @@ -10,6 +10,32 @@ import { TaskUpdate } from "./tasks_pb.js"; import { ResourceUpdate } from "./resources_pb.js"; import { AsyncOpUpdate } from "./async_ops_pb.js"; +/** + * The time "state" of the aggregator. + * + * @generated from enum rs.tokio.console.instrument.Temporality + */ +export enum Temporality { + /** + * The aggregator is currently live. + * + * @generated from enum value: LIVE = 0; + */ + LIVE = 0, + + /** + * The aggregator is currently paused. + * + * @generated from enum value: PAUSED = 1; + */ + PAUSED = 1, +} +// Retrieve enum metadata with: proto3.getEnumType(Temporality) +proto3.util.setEnumType(Temporality, "rs.tokio.console.instrument.Temporality", [ + { no: 0, name: "LIVE" }, + { no: 1, name: "PAUSED" }, +]); + /** * InstrumentRequest requests the stream of updates * to observe the async runtime state over time. @@ -207,6 +233,13 @@ export class Update extends Message { */ newMetadata?: RegisterMetadata; + /** + * The time "state" of the aggregator, such as paused or live. + * + * @generated from field: rs.tokio.console.instrument.Temporality temporality = 6; + */ + temporality = Temporality.LIVE; + constructor(data?: PartialMessage) { super(); proto3.util.initPartial(data, this); @@ -220,6 +253,7 @@ export class Update extends Message { { no: 3, name: "resource_update", kind: "message", T: ResourceUpdate }, { no: 4, name: "async_op_update", kind: "message", T: AsyncOpUpdate }, { no: 5, name: "new_metadata", kind: "message", T: RegisterMetadata }, + { no: 6, name: "temporality", kind: "enum", T: proto3.getEnumType(Temporality) }, ]); static fromBinary(bytes: Uint8Array, options?: Partial): Update { diff --git a/console-subscriber/src/aggregator/mod.rs b/console-subscriber/src/aggregator/mod.rs index b6846d62e..24a816884 100644 --- a/console-subscriber/src/aggregator/mod.rs +++ b/console-subscriber/src/aggregator/mod.rs @@ -7,6 +7,7 @@ use std::{ }; use console_api as proto; +use console_api::instrument::Temporality; use prost::Message; use proto::resources::resource; use tokio::sync::{mpsc, Notify}; @@ -102,11 +103,6 @@ pub(crate) struct Flush { triggered: AtomicBool, } -#[derive(Debug)] -enum Temporality { - Live, - Paused, -} // Represent static data for resources struct Resource { id: Id, @@ -293,6 +289,7 @@ impl Aggregator { new_metadata: Some(proto::RegisterMetadata { metadata: (*self.all_metadata).clone(), }), + temporality: self.temporality.into(), }; let message_size = update.encoded_len(); if message_size < MAX_MESSAGE_SIZE { @@ -417,6 +414,7 @@ impl Aggregator { task_update, resource_update, async_op_update, + temporality: self.temporality.into(), }; self.watchers diff --git a/tokio-console/src/state/mod.rs b/tokio-console/src/state/mod.rs index 4b9928a8f..819ea9e7d 100644 --- a/tokio-console/src/state/mod.rs +++ b/tokio-console/src/state/mod.rs @@ -5,6 +5,7 @@ use crate::{ warnings::Linter, }; use console_api as proto; +use console_api::instrument::Temporality; use ratatui::{ style::{Color, Modifier}, text::Span, @@ -71,12 +72,6 @@ pub(crate) enum FieldValue { Debug(String), } -#[derive(Debug)] -enum Temporality { - Live, - Paused, -} - #[derive(Debug, Eq, PartialEq)] pub(crate) struct Attribute { field: Field, @@ -107,6 +102,14 @@ impl State { current_view: &view::ViewState, update: proto::instrument::Update, ) { + match Temporality::try_from(update.temporality) { + Ok(temporality) => { + self.temporality = temporality; + } + Err(..) => { + tracing::warn!(?update.temporality, "invalid temporality"); + } + } if let Some(now) = update.now.map(|v| v.try_into().unwrap()) { self.last_updated_at = Some(now); } @@ -250,12 +253,6 @@ impl State { } } -impl Default for Temporality { - fn default() -> Self { - Self::Live - } -} - impl Metadata { fn from_proto(pb: proto::Metadata, id: u64, strings: &mut intern::Strings) -> Self { Self {