Refresh oximeter producer list more aggressively

- 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

Author: bnaecker

nexus/src/app/oximeter.rs

@@ -19,7 +19,6 @@ use oximeter_client::Client as OximeterClient;
 use oximeter_db::query::Timestamp;
 use oximeter_db::Measurement;
 use slog::Logger;
-use std::convert::TryInto;
 use std::net::SocketAddr;
 use std::num::NonZeroU32;
 use std::time::Duration;
@@ -118,6 +117,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,
@@ -127,20 +129,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

@@ -1477,7 +1477,7 @@ pub async fn start_oximeter(
     let config = oximeter_collector::Config {
         nexus_address: Some(nexus_address),
         db,
-        refresh_interval: oximeter_collector::default_refresh_interval(),
+        refresh_interval: Duration::from_secs(2),
         log: ConfigLogging::StderrTerminal { level: ConfigLoggingLevel::Error },
     };
     let args = oximeter_collector::OximeterArguments {

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";

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(())
     }
 }