Fix several oximeter bugs (#7126)

- Only update producer information if it's actually changed. This fixes
#7120, since we no longer indefinitely delay a collection timer just by
refreshing our list of producers from Nexus.
- Bail from a producer refresh on any error contacting Nexus. This
should fix #7124. If we get a progenitor-level error refreshing our list
from Nexus, which should mostly be a communcation error like
`ECONNREFUSED`, we don't want to blow away our entire state. Instead,
break out of the refresh function entirely, and continue collecting from
our last known-good set of producers. Only proceed with the update when
we are confident we have a full, valid list from Nexus.
- Change the missed tick behavior for refreshing producer list from
"burst" to "skip". This isn't strictly necessary, but better reflects
what we want, which is "reasonably often" updates, without causing undue
churn.

common/src/api/internal/nexus.rs

@@ -218,7 +218,7 @@ pub enum ProducerKind {
 
 /// Information announced by a metric server, used so that clients can contact it and collect
 /// available metric data from it.
-#[derive(Clone, Debug, Deserialize, JsonSchema, Serialize, PartialEq)]
+#[derive(Clone, Copy, Debug, Deserialize, JsonSchema, Serialize, PartialEq)]
 pub struct ProducerEndpoint {
     /// A unique ID for this producer.
     pub id: Uuid,

oximeter/collector/src/agent.rs

@@ -183,10 +183,20 @@ async fn collection_task(
                         perform_collection(&log, &mut stats, &client, &producer, &outbox, Some(token)).await;
                     },
                     Some(CollectionMessage::Update(new_info)) => {
+                        // If the collection interval is shorter than the
+                        // interval on which we receive these update messages,
+                        // we'll never actually collect anything! Instead, only
+                        // do the update if the information has changed. This
+                        // should also be guarded against by the main agent, but
+                        // we're being cautious here.
+                        if producer == new_info {
+                            continue;
+                        }
                         producer = new_info;
                         debug!(
                             log,
-                            "collection task received request to update its producer information";
+                            "collection task received request to update \
+                            its producer information";
                             "interval" => ?producer.interval,
                             "address" => producer.address,
                         );
@@ -606,30 +616,43 @@ impl OximeterAgent {
                     "component" => "collection-task",
                     "producer_id" => id.to_string(),
                 ));
-                let info_clone = info.clone();
+                let info_clone = info;
                 let target = self.collection_target;
                 let task = tokio::spawn(async move {
                     collection_task(log, target, info_clone, rx, q).await;
                 });
                 value.insert((info, CollectionTask { inbox: tx, task }));
             }
             Entry::Occupied(mut value) => {
-                debug!(
-                    self.log,
-                    "received request to register existing metric \
-                    producer, updating collection information";
-                   "producer_id" => id.to_string(),
-                   "interval" => ?info.interval,
-                   "address" => info.address,
-                );
-                value.get_mut().0 = info.clone();
-                value
-                    .get()
-                    .1
-                    .inbox
-                    .send(CollectionMessage::Update(info))
-                    .await
-                    .unwrap();
+                // Only update the endpoint information if it's actually
+                // different, to avoid indefinitely delaying the collection
+                // timer from expiring.
+                if value.get().0 == info {
+                    trace!(
+                        self.log,
+                        "ignoring request to update existing metric \
+                        producer, since the endpoint information is \
+                        the same as the existing";
+                        "producer_id" => %id,
+                    );
+                } else {
+                    debug!(
+                        self.log,
+                        "received request to register existing metric \
+                        producer, updating collection information";
+                        "producer_id" => id.to_string(),
+                        "interval" => ?info.interval,
+                        "address" => info.address,
+                    );
+                    value.get_mut().0 = info;
+                    value
+                        .get()
+                        .1
+                        .inbox
+                        .send(CollectionMessage::Update(info))
+                        .await
+                        .unwrap();
+                }
             }
         }
     }
@@ -675,7 +698,7 @@ impl OximeterAgent {
             .await
             .range((start, Bound::Unbounded))
             .take(limit)
-            .map(|(_id, (info, _t))| info.clone())
+            .map(|(_id, (info, _t))| *info)
             .collect()
     }
 
@@ -766,7 +789,14 @@ async fn refresh_producer_list(
     agent: OximeterAgent,
     nexus_pool: Pool<NexusClient>,
 ) {
+    // Setup our refresh timer.
+    //
+    // If we miss a tick, we'll skip until the next multiple of the interval
+    // from the start time. This is a good compromise between taking a bunch of
+    // quick updates (burst) and shifting our interval (delay).
     let mut interval = tokio::time::interval(agent.refresh_interval);
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+
     loop {
         interval.tick().await;
         info!(agent.log, "refreshing list of producers from Nexus");
@@ -783,11 +813,28 @@ async fn refresh_producer_list(
         loop {
             match stream.try_next().await {
                 Err(e) => {
+                    // TODO-robustness: Some errors here may not be "fatal", in
+                    // the sense that we can continue to process the list we
+                    // receive from Nexus. It's not clear which ones though,
+                    // since most of these are either impossible (pre-hook
+                    // errors) or indicate we've made a serious programming
+                    // error or updated to incompatible versions of Nexus /
+                    // Oximeter. One that we might be able to handle is a
+                    // communication error, say failing to fetch the last page
+                    // when we've already fetched the first few. But for now,
+                    // we'll simply keep the list we have and try again on the
+                    // next refresh.
+                    //
+                    // For now, let's just avoid doing anything at all here, on
+                    // the theory that if we hit this, we should just continue
+                    // collecting from the last known-good set of producers.
                     error!(
                         agent.log,
-                        "error fetching next assigned producer";
+                        "error fetching next assigned producer, \
+                        abandoning this refresh attempt";
                         "err" => ?e,
                     );
+                    return;
                 }
                 Ok(Some(p)) => {
                     let endpoint = match ProducerEndpoint::try_from(p) {

oximeter/collector/src/standalone.rs

@@ -112,7 +112,7 @@ impl StandaloneNexus {
                     ));
                 };
                 let assignment =
-                    ProducerAssignment { producer: info.clone(), collector_id };
+                    ProducerAssignment { producer: *info, collector_id };
                 assignment
             }
             Some(existing_assignment) => {
@@ -126,7 +126,7 @@ impl StandaloneNexus {
                 // changed its IP address. The collector will learn of this when
                 // it next fetches its list.
                 let collector_id = existing_assignment.collector_id;
-                ProducerAssignment { producer: info.clone(), collector_id }
+                ProducerAssignment { producer: *info, collector_id }
             }
         };
         inner.producers.insert(info.id, assignment);

oximeter/producer/src/lib.rs

@@ -130,7 +130,7 @@ impl Server {
     ) -> Result<Self, Error> {
         Self::new_impl(
             registry,
-            config.server_info.clone(),
+            config.server_info,
             config.registration_address.as_ref(),
             config.request_body_max_bytes,
             &config.log,