[2/6] [oxql] move core types to a new oxql-types crate (#6364)

Means that the Nexus external API crate doesn't have to pull in
oximeter-db.

I did have to expose a few mutators, but my general reasoning is that
because these types derive `Deserialize`, it's always possible to make
these changes by serializing into JSON and working with them there.

Cargo.toml

@@ -80,6 +80,7 @@ members = [
  "oximeter/instruments",
  "oximeter/oximeter-macro-impl",
  "oximeter/oximeter",
+ "oximeter/oxql-types",
  "oximeter/producer",
  "oximeter/schema",
  "oximeter/test-utils",
@@ -196,6 +197,7 @@ default-members = [
  "oximeter/instruments",
  "oximeter/oximeter-macro-impl",
  "oximeter/oximeter",
+ "oximeter/oxql-types",
  "oximeter/producer",
  "oximeter/schema",
  "oximeter/test-utils",
@@ -470,6 +472,7 @@ oximeter-schema = { path = "oximeter/schema" }
 oximeter-test-utils = { path = "oximeter/test-utils" }
 oximeter-timeseries-macro = { path = "oximeter/timeseries-macro" }
 oximeter-types = { path = "oximeter/types" }
+oxql-types = { path = "oximeter/oxql-types" }
 p256 = "0.13"
 parse-display = "0.10.0"
 partial-io = { version = "0.5.4", features = ["proptest1", "tokio1"] }

nexus/Cargo.toml

@@ -57,6 +57,7 @@ openssl.workspace = true
 oximeter-client.workspace = true
 oximeter-db = { workspace = true, default-features = false, features = [ "oxql" ] }
 oxnet.workspace = true
+oxql-types.workspace = true
 parse-display.workspace = true
 paste.workspace = true
 # See omicron-rpaths for more about the "pq-sys" dependency.

nexus/src/app/metrics.rs

@@ -14,7 +14,7 @@ use nexus_db_queries::{
 };
 use omicron_common::api::external::{Error, InternalContext};
 use oximeter_db::{
- oxql, Measurement, TimeseriesSchema, TimeseriesSchemaPaginationParams,
+ Measurement, TimeseriesSchema, TimeseriesSchemaPaginationParams,
 };
 use std::num::NonZeroU32;
 
@@ -138,7 +138,7 @@ impl super::Nexus {
  &self,
  opctx: &OpContext,
  query: impl AsRef<str>,
- ) -> Result<Vec<oxql::Table>, Error> {
+ ) -> Result<Vec<oxql_types::Table>, Error> {
  // Must be a fleet user to list timeseries schema.
  //
  // TODO-security: We need to figure out how to implement proper security

nexus/src/external_api/http_entrypoints.rs

@@ -6386,7 +6386,7 @@ async fn timeseries_schema_list(
 async fn timeseries_query(
  rqctx: RequestContext<ApiContext>,
  body: TypedBody<params::TimeseriesQuery>,
-) -> Result<HttpResponseOk<Vec<oximeter_db::oxql::Table>>, HttpError> {
+) -> Result<HttpResponseOk<Vec<oxql_types::Table>>, HttpError> {
  let apictx = rqctx.context();
  let handler = async {
  let nexus = &apictx.context.nexus;

nexus/tests/integration_tests/metrics.rs

@@ -284,7 +284,7 @@ async fn test_timeseries_schema_list(
 pub async fn timeseries_query(
  cptestctx: &ControlPlaneTestContext<omicron_nexus::Server>,
  query: impl ToString,
-) -> Vec<oximeter_db::oxql::Table> {
+) -> Vec<oxql_types::Table> {
  // first, make sure the latest timeseries have been collected.
  cptestctx.oximeter.force_collect().await;
 
@@ -429,11 +429,11 @@ async fn test_instance_watcher_metrics(
 
  #[track_caller]
  fn count_state(
- table: &oximeter_db::oxql::Table,
+ table: &oxql_types::Table,
  instance_id: InstanceUuid,
  state: &'static str,
  ) -> i64 {
- use oximeter_db::oxql::point::ValueArray;
+ use oxql_types::point::ValueArray;
  let uuid = FieldValue::Uuid(instance_id.into_untyped_uuid());
  let state = FieldValue::String(state.into());
  let mut timeserieses = table.timeseries().filter(|ts| {

openapi/nexus.json

@@ -20131,10 +20131,20 @@
  "type": "object",
  "properties": {
  "metric_type": {
- "$ref": "#/components/schemas/MetricType"
+ "description": "The type of this metric.",
+ "allOf": [
+ {
+ "$ref": "#/components/schemas/MetricType"
+ }
+ ]
  },
  "values": {
- "$ref": "#/components/schemas/ValueArray"
+ "description": "The data values.",
+ "allOf": [
+ {
+ "$ref": "#/components/schemas/ValueArray"
+ }
+ ]
  }
  },
  "required": [

oximeter/db/Cargo.toml

@@ -24,6 +24,7 @@ num.workspace = true
 omicron-common.workspace = true
 omicron-workspace-hack.workspace = true
 oximeter.workspace = true
+oxql-types.workspace = true
 regex.workspace = true
 serde.workspace = true
 serde_json.workspace = true

oximeter/db/src/client/mod.rs

@@ -22,16 +22,16 @@ use crate::Error;
 use crate::Metric;
 use crate::Target;
 use crate::Timeseries;
-use crate::TimeseriesKey;
-use crate::TimeseriesName;
 use crate::TimeseriesPageSelector;
 use crate::TimeseriesScanParams;
 use crate::TimeseriesSchema;
 use dropshot::EmptyScanParams;
 use dropshot::PaginationOrder;
 use dropshot::ResultsPage;
 use dropshot::WhichPage;
+use oximeter::schema::TimeseriesKey;
 use oximeter::types::Sample;
+use oximeter::TimeseriesName;
 use regex::Regex;
 use regex::RegexBuilder;
 use slog::debug;

oximeter/db/src/client/oxql.rs

@@ -18,7 +18,7 @@ use crate::query::field_table_name;
 use crate::Error;
 use crate::Metric;
 use crate::Target;
-use crate::TimeseriesKey;
+use oximeter::schema::TimeseriesKey;
 use oximeter::TimeseriesSchema;
 use slog::debug;
 use slog::trace;
@@ -68,7 +68,7 @@ pub struct OxqlResult {
  pub query_summaries: Vec<QuerySummary>,
 
  /// The list of OxQL tables returned from the query.
- pub tables: Vec<oxql::Table>,
+ pub tables: Vec<oxql_types::Table>,
 }
 
 /// The maximum number of data values fetched from the database for an OxQL
@@ -479,7 +479,9 @@ impl Client {
  query_id,
  total_duration: query_start.elapsed(),
  query_summaries,
- tables: vec![oxql::Table::new(schema.timeseries_name.as_str())],
+ tables: vec![oxql_types::Table::new(
+ schema.timeseries_name.as_str(),
+ )],
  };
  return Ok(result);
  }
@@ -503,7 +505,7 @@ impl Client {
 
  // At this point, let's construct a set of tables and run the results
  // through the transformation pipeline.
- let mut tables = vec![oxql::Table::from_timeseries(
+ let mut tables = vec![oxql_types::Table::from_timeseries(
  schema.timeseries_name.as_str(),
  timeseries_by_key.into_values(),
  )?];
@@ -553,7 +555,7 @@ impl Client {
  limit: Option<Limit>,
  total_rows_fetched: &mut u64,
  ) -> Result<
- (Vec<QuerySummary>, BTreeMap<TimeseriesKey, oxql::Timeseries>),
+ (Vec<QuerySummary>, BTreeMap<TimeseriesKey, oxql_types::Timeseries>),
  Error,
  > {
  // We'll create timeseries for each key on the fly. To enable computing
@@ -624,25 +626,25 @@ impl Client {
  for (key, measurements) in measurements_by_key.into_iter() {
  // Constuct a new timeseries, from the target/metric info.
  let (target, metric) = info.get(&key).unwrap();
- let mut timeseries = oxql::Timeseries::new(
+ let mut timeseries = oxql_types::Timeseries::new(
  target
  .fields
  .iter()
  .chain(metric.fields.iter())
  .map(|field| (field.name.clone(), field.value.clone())),
- oxql::point::DataType::try_from(schema.datum_type)?,
+ oxql_types::point::DataType::try_from(schema.datum_type)?,
  if schema.datum_type.is_cumulative() {
- oxql::point::MetricType::Delta
+ oxql_types::point::MetricType::Delta
  } else {
- oxql::point::MetricType::Gauge
+ oxql_types::point::MetricType::Gauge
  },
  )?;
 
  // Covert its oximeter measurements into OxQL data types.
  let points = if schema.datum_type.is_cumulative() {
- oxql::point::Points::delta_from_cumulative(&measurements)?
+ oxql_types::point::Points::delta_from_cumulative(&measurements)?
  } else {
- oxql::point::Points::gauge_from_gauge(&measurements)?
+ oxql_types::point::Points::gauge_from_gauge(&measurements)?
  };
  timeseries.points = points;
  debug!(
@@ -1108,17 +1110,15 @@ fn update_total_rows_and_check(
 mod tests {
  use super::ConsistentKeyGroup;
  use crate::client::oxql::chunk_consistent_key_groups_impl;
- use crate::{
- oxql::{point::Points, Table, Timeseries},
- Client, DbWrite,
- };
+ use crate::{Client, DbWrite};
  use crate::{Metric, Target};
  use chrono::{DateTime, Utc};
  use dropshot::test_util::LogContext;
  use omicron_test_utils::dev::clickhouse::ClickHouseInstance;
  use omicron_test_utils::dev::test_setup_log;
  use oximeter::{types::Cumulative, FieldValue};
  use oximeter::{DatumType, Sample};
+ use oxql_types::{point::Points, Table, Timeseries};
  use std::collections::BTreeMap;
  use std::time::Duration;
 

oximeter/db/src/lib.rs

@@ -14,6 +14,7 @@ use dropshot::EmptyScanParams;
 use dropshot::PaginationParams;
 pub use oximeter::schema::FieldSchema;
 pub use oximeter::schema::FieldSource;
+use oximeter::schema::TimeseriesKey;
 pub use oximeter::schema::TimeseriesName;
 pub use oximeter::schema::TimeseriesSchema;
 pub use oximeter::DatumType;
@@ -267,8 +268,6 @@ pub async fn make_client(
  Ok(client)
 }
 
-pub(crate) type TimeseriesKey = u64;
-
 // TODO-cleanup: Add the timeseries version in to the computation of the key.
 // This will require a full drop of the database, since we're changing the
 // sorting key and the timeseries key on each past sample. See

oximeter/db/src/model.rs

@@ -11,13 +11,13 @@ use crate::FieldSchema;
 use crate::FieldSource;
 use crate::Metric;
 use crate::Target;
-use crate::TimeseriesKey;
 use crate::TimeseriesSchema;
 use bytes::Bytes;
 use chrono::DateTime;
 use chrono::Utc;
 use num::traits::Zero;
 use oximeter::histogram::Histogram;
+use oximeter::schema::TimeseriesKey;
 use oximeter::traits;
 use oximeter::types::Cumulative;
 use oximeter::types::Datum;

oximeter/db/src/oxql/ast/table_ops/align.rs

@@ -6,19 +6,19 @@
 
 // Copyright 2024 Oxide Computer Company
 
-use crate::oxql::point::DataType;
-use crate::oxql::point::MetricType;
-use crate::oxql::point::Points;
-use crate::oxql::point::ValueArray;
-use crate::oxql::point::Values;
-use crate::oxql::query::Alignment;
-use crate::oxql::Error;
-use crate::oxql::Table;
-use crate::oxql::Timeseries;
 use anyhow::Context;
+use anyhow::Error;
 use chrono::DateTime;
 use chrono::TimeDelta;
 use chrono::Utc;
+use oxql_types::point::DataType;
+use oxql_types::point::MetricType;
+use oxql_types::point::Points;
+use oxql_types::point::ValueArray;
+use oxql_types::point::Values;
+use oxql_types::Alignment;
+use oxql_types::Table;
+use oxql_types::Timeseries;
 use std::time::Duration;
 
 // The maximum factor by which an alignment operation may upsample data.
@@ -144,7 +144,7 @@ fn align_mean_within(
  "Alignment by mean requires a gauge or delta metric, not {}",
  metric_type,
  );
- verify_max_upsampling_ratio(&points.timestamps, &period)?;
+ verify_max_upsampling_ratio(points.timestamps(), &period)?;
 
  // Always convert the output to doubles, when computing the mean. The
  // output is always a gauge, so we do not need the start times of the
@@ -179,7 +179,7 @@ fn align_mean_within(
  // - Compute the mean of those.
  let period_ =
  TimeDelta::from_std(*period).context("time delta out of range")?;
- let first_timestamp = points.timestamps[0];
+ let first_timestamp = points.timestamps()[0];
  let mut ix: u32 = 0;
  loop {
  // Compute the next output timestamp, by shifting the query end time
@@ -220,15 +220,15 @@ fn align_mean_within(
  // entries.
  let output_value = if matches!(metric_type, MetricType::Gauge) {
  mean_gauge_value_in_window(
- &points.timestamps,
+ points.timestamps(),
  &input_points,
  window_start,
  output_time,
  )
  } else {
  mean_delta_value_in_window(
- points.start_times.as_ref().unwrap(),
- &points.timestamps,
+ points.start_times().unwrap(),
+ points.timestamps(),
  &input_points,
  window_start,
  output_time,
@@ -255,10 +255,9 @@ fn align_mean_within(
  ValueArray::Double(output_values.into_iter().rev().collect());
  let timestamps = output_timestamps.into_iter().rev().collect();
  let values = Values { values, metric_type: MetricType::Gauge };
- new_timeseries.points =
- Points { start_times: None, timestamps, values: vec![values] };
- new_timeseries.alignment =
- Some(Alignment { end_time: *query_end, period: *period });
+ new_timeseries.points = Points::new(None, timestamps, vec![values]);
+ new_timeseries
+ .set_alignment(Alignment { end_time: *query_end, period: *period });
  output_table.insert(new_timeseries).unwrap();
  }
  Ok(output_table)