Refresh oximeter producer list more aggressively (#6926)

- Remove `oximeter` producer HTTP endpoint for registering individual
producers.
- Dramatically reduce interval on which `oximeter` collector refreshes
its list of producers. This is now the only way the collector learns of
producers. The interval is also much smaller in tests to ensure pretty
snappy registrations
- Remove calls from both Nexus and the `oximeter` standalone mock Nexus
for registering producers
- Have `oximeter` collector start polling producers immediately, rather
than waiting for the first polling interval to expire.
- Closes #6916, #6895, and #6901

nexus/src/app/oximeter.rs

@@ -15,7 +15,6 @@ use omicron_common::api::internal::nexus::{self, ProducerEndpoint};
 use oximeter_client::Client as OximeterClient;
 use oximeter_db::query::Timestamp;
 use oximeter_db::Measurement;
-use std::convert::TryInto;
 use std::net::SocketAddr;
 use std::num::NonZeroU32;
 use std::time::Duration;
@@ -62,6 +61,9 @@ impl super::Nexus {
     }
 
     /// Assign a newly-registered metric producer to an oximeter collector server.
+    ///
+    /// Note that we don't send the registration to the collector, the collector
+    /// polls for its list of producers periodically.
     pub(crate) async fn assign_producer(
         &self,
         opctx: &OpContext,
@@ -71,20 +73,6 @@ impl super::Nexus {
             .db_datastore
             .producer_endpoint_upsert_and_assign(opctx, &producer_info)
             .await?;
-
-        let address = SocketAddr::from((
-            collector_info.ip.ip(),
-            collector_info.port.try_into().unwrap(),
-        ));
-        let collector =
-            build_oximeter_client(&self.log, &collector_info.id, address);
-
-        collector
-            .producers_post(&oximeter_client::types::ProducerEndpoint::from(
-                &producer_info,
-            ))
-            .await
-            .map_err(Error::from)?;
         info!(
             self.log,
             "assigned collector to new producer";

nexus/test-utils/src/lib.rs

@@ -1470,7 +1470,11 @@ pub async fn start_oximeter(
     let config = oximeter_collector::Config {
         nexus_address: Some(nexus_address),
         db,
-        refresh_interval: oximeter_collector::default_refresh_interval(),
+        // The collector only learns about producers when it refreshes its list
+        // from Nexus. This interval is quite short, and much smaller than the
+        // one we use in production. That's important for test latency, but not
+        // strictly required for correctness.
+        refresh_interval: Duration::from_secs(2),
         log: ConfigLogging::StderrTerminal { level: ConfigLoggingLevel::Error },
     };
     let args = oximeter_collector::OximeterArguments {

nexus/tests/integration_tests/disks.rs

@@ -4,6 +4,8 @@
 
 //! Tests basic disk support in the API
 
+use crate::integration_tests::metrics::wait_for_producer;
+
 use super::instances::instance_wait_for_state;
 use super::metrics::{get_latest_silo_metric, query_for_metrics};
 use chrono::Utc;
@@ -34,14 +36,14 @@ use nexus_test_utils::SLED_AGENT_UUID;
 use nexus_test_utils_macros::nexus_test;
 use nexus_types::external_api::params;
 use nexus_types::silo::DEFAULT_SILO_ID;
-use omicron_common::api::external::ByteCount;
 use omicron_common::api::external::Disk;
 use omicron_common::api::external::DiskState;
 use omicron_common::api::external::IdentityMetadataCreateParams;
 use omicron_common::api::external::Instance;
 use omicron_common::api::external::InstanceState;
 use omicron_common::api::external::Name;
 use omicron_common::api::external::NameOrId;
+use omicron_common::api::external::{ByteCount, SimpleIdentity as _};
 use omicron_nexus::app::{MAX_DISK_SIZE_BYTES, MIN_DISK_SIZE_BYTES};
 use omicron_nexus::Nexus;
 use omicron_nexus::TestInterfaces as _;
@@ -1802,7 +1804,6 @@ async fn test_disk_metrics(cptestctx: &ControlPlaneTestContext) {
     DiskTest::new(&cptestctx).await;
     let project_id = create_project_and_pool(client).await;
     let disk = create_disk(&client, PROJECT_NAME, DISK_NAME).await;
-    oximeter.force_collect().await;
 
     // When grabbing a metric, we look for data points going back to the
     // start of this test all the way up to the current time.
@@ -1826,6 +1827,7 @@ async fn test_disk_metrics(cptestctx: &ControlPlaneTestContext) {
             .await;
     assert!(measurements.items.is_empty());
 
+    oximeter.force_collect().await;
     assert_eq!(
         get_latest_silo_metric(
             cptestctx,
@@ -1838,6 +1840,7 @@ async fn test_disk_metrics(cptestctx: &ControlPlaneTestContext) {
 
     // Create an instance, attach the disk to it.
     create_instance_with_disk(client).await;
+    wait_for_producer(&cptestctx.oximeter, disk.id()).await;
     oximeter.force_collect().await;
 
     for metric in &ALL_METRICS {
@@ -1870,8 +1873,9 @@ async fn test_disk_metrics_paginated(cptestctx: &ControlPlaneTestContext) {
     let client = &cptestctx.external_client;
     DiskTest::new(&cptestctx).await;
     create_project_and_pool(client).await;
-    create_disk(&client, PROJECT_NAME, DISK_NAME).await;
+    let disk = create_disk(&client, PROJECT_NAME, DISK_NAME).await;
     create_instance_with_disk(client).await;
+    wait_for_producer(&cptestctx.oximeter, disk.id()).await;
 
     let oximeter = &cptestctx.oximeter;
     oximeter.force_collect().await;

openapi/oximeter.json

@@ -81,31 +81,6 @@
         "x-dropshot-pagination": {
           "required": []
         }
-      },
-      "post": {
-        "summary": "Handle a request from Nexus to register a new producer with this collector.",
-        "operationId": "producers_post",
-        "requestBody": {
-          "content": {
-            "application/json": {
-              "schema": {
-                "$ref": "#/components/schemas/ProducerEndpoint"
-              }
-            }
-          },
-          "required": true
-        },
-        "responses": {
-          "204": {
-            "description": "resource updated"
-          },
-          "4XX": {
-            "$ref": "#/components/responses/Error"
-          },
-          "5XX": {
-            "$ref": "#/components/responses/Error"
-          }
-        }
       }
     },
     "/producers/{producer_id}": {

oximeter/api/src/lib.rs

@@ -5,8 +5,7 @@
 use chrono::{DateTime, Utc};
 use dropshot::{
     EmptyScanParams, HttpError, HttpResponseDeleted, HttpResponseOk,
-    HttpResponseUpdatedNoContent, PaginationParams, Query, RequestContext,
-    ResultsPage, TypedBody,
+    PaginationParams, Query, RequestContext, ResultsPage,
 };
 use omicron_common::api::internal::nexus::ProducerEndpoint;
 use schemars::JsonSchema;
@@ -17,16 +16,6 @@ use uuid::Uuid;
 pub trait OximeterApi {
     type Context;
 
-    /// Handle a request from Nexus to register a new producer with this collector.
-    #[endpoint {
-        method = POST,
-        path = "/producers",
-    }]
-    async fn producers_post(
-        request_context: RequestContext<Self::Context>,
-        body: TypedBody<ProducerEndpoint>,
-    ) -> Result<HttpResponseUpdatedNoContent, HttpError>;
-
     /// List all producers.
     #[endpoint {
         method = GET,

oximeter/collector/src/agent.rs

@@ -155,7 +155,6 @@ async fn collection_task(
     let mut log = orig_log.new(o!("address" => producer.address));
     let client = reqwest::Client::new();
     let mut collection_timer = interval(producer.interval);
-    collection_timer.tick().await; // completes immediately
     debug!(
         log,
         "starting oximeter collection task";
@@ -885,8 +884,7 @@ mod tests {
     // Period these tests wait using `tokio::time::advance()` before checking
     // their test conditions.
     const TEST_WAIT_PERIOD: Duration = Duration::from_millis(
-        COLLECTION_INTERVAL.as_millis() as u64 * N_COLLECTIONS
-            + COLLECTION_INTERVAL.as_millis() as u64 / 2,
+        COLLECTION_INTERVAL.as_millis() as u64 * N_COLLECTIONS,
     );
 
     // Test that we count successful collections from a target correctly.

oximeter/collector/src/http_entrypoints.rs

@@ -4,20 +4,18 @@
 
 //! Oximeter collector server HTTP API
 
-// Copyright 2023 Oxide Computer Company
+// Copyright 2024 Oxide Computer Company
 
 use crate::OximeterAgent;
 use dropshot::ApiDescription;
 use dropshot::EmptyScanParams;
 use dropshot::HttpError;
 use dropshot::HttpResponseDeleted;
 use dropshot::HttpResponseOk;
-use dropshot::HttpResponseUpdatedNoContent;
 use dropshot::PaginationParams;
 use dropshot::Query;
 use dropshot::RequestContext;
 use dropshot::ResultsPage;
-use dropshot::TypedBody;
 use dropshot::WhichPage;
 use omicron_common::api::internal::nexus::ProducerEndpoint;
 use oximeter_api::*;
@@ -34,19 +32,6 @@ enum OximeterApiImpl {}
 impl OximeterApi for OximeterApiImpl {
     type Context = Arc<OximeterAgent>;
 
-    async fn producers_post(
-        request_context: RequestContext<Self::Context>,
-        body: TypedBody<ProducerEndpoint>,
-    ) -> Result<HttpResponseUpdatedNoContent, HttpError> {
-        let agent = request_context.context();
-        let producer_info = body.into_inner();
-        agent
-            .register_producer(producer_info)
-            .await
-            .map_err(HttpError::from)
-            .map(|_| HttpResponseUpdatedNoContent())
-    }
-
     async fn producers_list(
         request_context: RequestContext<Arc<OximeterAgent>>,
         query: Query<PaginationParams<EmptyScanParams, ProducerPage>>,

oximeter/collector/src/lib.rs

@@ -127,7 +127,7 @@ impl DbConfig {
 
 /// Default interval on which we refresh our list of producers from Nexus.
 pub const fn default_refresh_interval() -> Duration {
-    Duration::from_secs(60 * 10)
+    Duration::from_secs(15)
 }
 
 /// Configuration used to initialize an oximeter server

oximeter/collector/src/standalone.rs

@@ -22,7 +22,6 @@ use nexus_types::internal_api::params::OximeterInfo;
 use omicron_common::api::internal::nexus::ProducerEndpoint;
 use omicron_common::api::internal::nexus::ProducerRegistrationResponse;
 use omicron_common::FileKv;
-use oximeter_client::Client;
 use rand::seq::IteratorRandom;
 use slog::debug;
 use slog::error;
@@ -104,21 +103,14 @@ impl StandaloneNexus {
                 //
                 // Select a random collector, and assign it to the producer.
                 // We'll return the assignment from this match block.
-                let Some((collector_id, collector_info)) =
+                let Some((collector_id, _collector_info)) =
                     inner.random_collector()
                 else {
                     return Err(HttpError::for_unavail(
                         None,
                         String::from("No collectors available"),
                     ));
                 };
-                let client = Client::new(
-                    format!("http://{}", collector_info.address).as_str(),
-                    self.log.clone(),
-                );
-                client.producers_post(&info.into()).await.map_err(|e| {
-                    HttpError::for_internal_error(e.to_string())
-                })?;
                 let assignment =
                     ProducerAssignment { producer: info.clone(), collector_id };
                 assignment
@@ -131,18 +123,9 @@ impl StandaloneNexus {
                 }
 
                 // This appears to be a re-registration, e.g., the producer
-                // changed its IP address. Re-register it with the collector to
-                // which it's already assigned.
+                // changed its IP address. The collector will learn of this when
+                // it next fetches its list.
                 let collector_id = existing_assignment.collector_id;
-                let collector_info =
-                    inner.collectors.get(&collector_id).unwrap();
-                let client = Client::new(
-                    format!("http://{}", collector_info.address).as_str(),
-                    self.log.clone(),
-                );
-                client.producers_post(&info.into()).await.map_err(|e| {
-                    HttpError::for_internal_error(e.to_string())
-                })?;
                 ProducerAssignment { producer: info.clone(), collector_id }
             }
         };
@@ -154,27 +137,9 @@ impl StandaloneNexus {
         &self,
         info: OximeterInfo,
     ) -> Result<(), HttpError> {
-        // If this is being registered again, send all its assignments again.
-        let mut inner = self.inner.lock().await;
-        if inner.collectors.insert(info.collector_id, info).is_some() {
-            let client = Client::new(
-                format!("http://{}", info.address).as_str(),
-                self.log.clone(),
-            );
-            for producer_info in
-                inner.producers.values().filter_map(|assignment| {
-                    if assignment.collector_id == info.collector_id {
-                        Some(&assignment.producer)
-                    } else {
-                        None
-                    }
-                })
-            {
-                client.producers_post(&producer_info.into()).await.map_err(
-                    |e| HttpError::for_internal_error(e.to_string()),
-                )?;
-            }
-        }
+        // No-op if this is being re-registered. It will fetch its list of
+        // producers again if needed.
+        self.inner.lock().await.collectors.insert(info.collector_id, info);
         Ok(())
     }
 }