Simplifies types used in metric producer descriptions (#125)

- Removes the `ProducerId` type entirely. This was mostly there due to a
misunderstanding of how Dropshot deserializes path types. It's been
replaced with a Uuid
- Makes most fields of the `ProducerEndpoint` type public. This now
stores the base collection route and the ID, and concats them on-demand,
rather than storing the joined path and splitting it on demand.

Author: bnaecker

omicron-common/src/model.rs

@@ -24,7 +24,7 @@ use std::fmt::Debug;
 use std::fmt::Display;
 use std::fmt::Formatter;
 use std::fmt::Result as FormatResult;
-use std::net::{SocketAddr, ToSocketAddrs};
+use std::net::SocketAddr;
 use std::num::NonZeroU32;
 use std::time::Duration;
 use thiserror::Error;
@@ -1394,133 +1394,24 @@ pub struct BootstrapAgentShareResponse {
  * Oximeter producer/collector objects.
  */
 
-/**
- * Idenitifier for a producer.
- */
-#[derive(
-    Debug,
-    Clone,
-    Copy,
-    PartialEq,
-    PartialOrd,
-    Ord,
-    Eq,
-    JsonSchema,
-    Serialize,
-    Deserialize,
-)]
-pub struct ProducerId {
-    pub producer_id: Uuid,
-}
-
-impl ProducerId {
-    /**
-     * Construct a new producer ID.
-     */
-    pub fn new() -> Self {
-        Self { producer_id: Uuid::new_v4() }
-    }
-}
-
-impl Default for ProducerId {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl std::fmt::Display for ProducerId {
-    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
-        write!(f, "{}", self.producer_id.to_string())
-    }
-}
-
 /**
  * Information announced by a metric server, used so that clients can contact it and collect
  * available metric data from it.
  */
 #[derive(Debug, Clone, JsonSchema, Serialize, Deserialize)]
 pub struct ProducerEndpoint {
-    producer_id: ProducerId,
-    address: SocketAddr,
-    collection_route: String,
-    interval: Duration,
+    pub id: Uuid,
+    pub address: SocketAddr,
+    pub base_route: String,
+    pub interval: Duration,
 }
 
 impl ProducerEndpoint {
-    /**
-     * Generate info for a metric server listening on the given address and route.
-     *
-     * This will generate a new, random [`ProducerId`] for the server. The `base_route` should be
-     * a route stem, to which the producer ID will be appended. `interval` is the desired initial
-     * collection interval for the producers metrics.
-     *
-     * Example
-     * -------
-     * ```rust
-     * use std::time::Duration;
-     * use omicron_common::model::ProducerEndpoint;
-     *
-     * let info = ProducerEndpoint::new("127.0.0.1:4444", "/collect", Duration::from_secs(10));
-     * assert_eq!(info.collection_route(), format!("/collect/{}", info.producer_id()));
-     * ```
-     */
-    pub fn new<T>(address: T, base_route: &str, interval: Duration) -> Self
-    where
-        T: ToSocketAddrs,
-    {
-        Self::with_id(ProducerId::new(), address, base_route, interval)
-    }
-
-    /**
-     * Generate info for a metric server, listening on the given address and route, with a known
-     * ID.
-     */
-    pub fn with_id<T>(
-        producer_id: ProducerId,
-        address: T,
-        base_route: &str,
-        interval: Duration,
-    ) -> Self
-    where
-        T: ToSocketAddrs,
-    {
-        Self {
-            producer_id,
-            address: address.to_socket_addrs().unwrap().next().unwrap(),
-            collection_route: format!(
-                "{}/{}",
-                base_route, producer_id.producer_id
-            ),
-            interval,
-        }
-    }
-
-    /**
-     * Return the producer ID for this server.
-     */
-    pub fn producer_id(&self) -> ProducerId {
-        self.producer_id
-    }
-
-    /**
-     * Return the address on which this server listens.
-     */
-    pub fn address(&self) -> SocketAddr {
-        self.address
-    }
-
     /**
      * Return the route that can be used to request metric data.
      */
-    pub fn collection_route(&self) -> &str {
-        &self.collection_route
-    }
-
-    /**
-     * Return the interval on which metrics should be collected.
-     */
-    pub fn interval(&self) -> &Duration {
-        &self.interval
+    pub fn collection_route(&self) -> String {
+        format!("{}/{}", &self.base_route, &self.id)
     }
 }
 

omicron-nexus/src/nexus.rs

@@ -1244,7 +1244,7 @@ impl Nexus {
         info!(
             self.log,
             "assigned collector to new producer";
-            "producer_id" => ?producer_info.producer_id(),
+            "producer_id" => ?producer_info.id,
             "collector_id" => ?collector.id,
         );
         Ok(())

oximeter/oximeter/examples/producer.rs

@@ -88,8 +88,12 @@ async fn main() {
         ConfigLogging::StderrTerminal { level: ConfigLoggingLevel::Debug };
     let registration_info =
         RegistrationInfo::new("127.0.0.1:12221", "/metrics/producers");
-    let server_info =
-        ProducerEndpoint::new(address, "/collect", Duration::from_secs(10));
+    let server_info = ProducerEndpoint {
+        id: Uuid::new_v4().into(),
+        address,
+        base_route: "/collect".to_string(),
+        interval: Duration::from_secs(10),
+    };
     let config = ProducerServerConfig {
         server_info,
         registration_info,

oximeter/oximeter/src/collect.rs

@@ -11,11 +11,12 @@ use dropshot::{
     endpoint, ApiDescription, ConfigDropshot, ConfigLogging, HttpError,
     HttpResponseOk, HttpServer, HttpServerStarter, Path, RequestContext,
 };
-use omicron_common::model::{ProducerEndpoint, ProducerId};
+use omicron_common::model::ProducerEndpoint;
 use reqwest::Client;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 use slog::{debug, info, o};
+use uuid::Uuid;
 
 use crate::types;
 use crate::{Error, Producer};
@@ -57,7 +58,7 @@ pub type ProducerResults = Vec<Result<BTreeSet<types::Sample>, Error>>;
 #[derive(Clone)]
 pub struct Collector {
     producers: Arc<Mutex<ProducerList>>,
-    producer_id: ProducerId,
+    producer_id: Uuid,
 }
 
 impl Default for Collector {
@@ -69,11 +70,11 @@ impl Default for Collector {
 impl Collector {
     /// Construct a new `Collector`.
     pub fn new() -> Self {
-        Self::with_id(ProducerId::new())
+        Self::with_id(Uuid::new_v4())
     }
 
     /// Construct a new `Collector` with the given producer ID.
-    pub fn with_id(producer_id: ProducerId) -> Self {
+    pub fn with_id(producer_id: Uuid) -> Self {
         Self { producers: Arc::new(Mutex::new(vec![])), producer_id }
     }
 
@@ -101,7 +102,7 @@ impl Collector {
     }
 
     /// Return the producer ID associated with this collector.
-    pub fn producer_id(&self) -> ProducerId {
+    pub fn producer_id(&self) -> Uuid {
         self.producer_id
     }
 }
@@ -132,7 +133,7 @@ impl ProducerServer {
             .logging_config
             .to_logger("metric-server")
             .map_err(|msg| Error::ProducerServer(msg.to_string()))?;
-        let collector = Collector::with_id(config.server_info.producer_id());
+        let collector = Collector::with_id(config.server_info.id);
         let dropshot_log = log.new(o!("component" => "dropshot"));
         let server = HttpServerStarter::new(
             &config.dropshot_config,
@@ -160,7 +161,7 @@ impl ProducerServer {
             "starting oximeter metric server";
             "route" => config.server_info.collection_route(),
             "producer_id" => ?collector.producer_id(),
-            "address" => config.server_info.address(),
+            "address" => config.server_info.address,
         );
         Ok(Self { collector, server })
     }
@@ -190,17 +191,22 @@ fn metric_server_api() -> ApiDescription<Collector> {
     api
 }
 
+#[derive(Clone, Copy, Debug, Deserialize, JsonSchema, Serialize)]
+struct ProducerIdPathParams {
+    pub producer_id: Uuid,
+}
+
 // Implementation of the actual collection routine used by the `ProducerServer`.
 #[endpoint {
     method = GET,
     path = "/collect/{producer_id}",
 }]
 async fn collect_endpoint(
     request_context: Arc<RequestContext<Collector>>,
-    path_params: Path<ProducerId>,
+    path_params: Path<ProducerIdPathParams>,
 ) -> Result<HttpResponseOk<ProducerResults>, HttpError> {
     let collector = request_context.context();
-    let producer_id = path_params.into_inner();
+    let producer_id = path_params.into_inner().producer_id;
     collect(collector, producer_id).await
 }
 
@@ -231,7 +237,7 @@ pub async fn register(
 /// Handle a request to pull available metric data from a [`Collector`].
 pub async fn collect(
     collector: &Collector,
-    producer_id: ProducerId,
+    producer_id: Uuid,
 ) -> Result<HttpResponseOk<ProducerResults>, HttpError> {
     if producer_id == collector.producer_id() {
         Ok(HttpResponseOk(collector.collect()))

oximeter/oximeter/src/oximeter_server.rs

@@ -13,9 +13,7 @@ use dropshot::{
     RequestContext, TypedBody,
 };
 use omicron_common::backoff;
-use omicron_common::model::{
-    OximeterStartupInfo, ProducerEndpoint, ProducerId,
-};
+use omicron_common::model::{OximeterStartupInfo, ProducerEndpoint};
 use reqwest::Client;
 use serde::{Deserialize, Serialize};
 use slog::{debug, info, o, trace, warn, Logger};
@@ -49,13 +47,13 @@ async fn collection_task(
     outbox: mpsc::Sender<ProducerResults>,
 ) {
     let client = Client::new();
-    let mut collection_timer = interval(*producer.interval());
+    let mut collection_timer = interval(producer.interval);
     collection_timer.tick().await; // completes immediately
     debug!(
         log,
         "starting oximeter collection task";
         "collector_id" => ?id,
-        "interval" => ?producer.interval(),
+        "interval" => ?producer.interval,
     );
     loop {
         tokio::select! {
@@ -99,9 +97,9 @@ async fn collection_task(
                     log,
                     "collecting from producer";
                     "collector_id" => ?id,
-                    "producer_id" => ?producer.producer_id().producer_id,
+                    "producer_id" => ?producer.id,
                 );
-                let res = client.get(format!("http://{}{}", producer.address(), producer.collection_route()))
+                let res = client.get(format!("http://{}{}", producer.address, producer.collection_route()))
                     .send()
                     .await;
                 match res {
@@ -113,7 +111,7 @@ async fn collection_task(
                                     "collected {} total results",
                                     results.len();
                                     "collector_id" => ?id,
-                                    "producer_id" => ?producer.producer_id().producer_id,
+                                    "producer_id" => ?producer.id,
                                 );
                                 outbox.send(results).await.unwrap();
                             },
@@ -123,7 +121,7 @@ async fn collection_task(
                                     "failed to collect results from producer: {}",
                                     e.to_string();
                                     "collector_id" => ?id,
-                                    "producer_id" => ?producer.producer_id().producer_id,
+                                    "producer_id" => ?producer.id,
                                 );
                             }
                         }
@@ -134,7 +132,7 @@ async fn collection_task(
                             "failed to send collection request to producer: {}",
                             e.to_string();
                             "collector_id" => ?id,
-                            "producer_id" => ?producer.producer_id().producer_id,
+                            "producer_id" => ?producer.id,
                         );
                     }
                 }
@@ -245,7 +243,7 @@ struct OximeterAgent {
     // Handle to the TX-side of a channel for collecting results from the collection tasks
     result_sender: mpsc::Sender<ProducerResults>,
     // The actual tokio tasks running the collection on a timer.
-    collection_tasks: Arc<Mutex<BTreeMap<ProducerId, CollectionTask>>>,
+    collection_tasks: Arc<Mutex<BTreeMap<Uuid, CollectionTask>>>,
 }
 
 impl OximeterAgent {
@@ -287,12 +285,12 @@ impl OximeterAgent {
         &self,
         info: ProducerEndpoint,
     ) -> Result<(), Error> {
-        let id = info.producer_id();
+        let id = info.id;
         match self.collection_tasks.lock().unwrap().entry(id) {
             Entry::Vacant(value) => {
                 info!(self.log, "registered new metric producer";
-                      "producer_id" => info.producer_id().to_string(),
-                      "address" => info.address(),
+                      "producer_id" => id.to_string(),
+                      "address" => info.address,
                 );
 
                 // Build channel to control the task and receive results.