Merge pull request #185 from influxdata/crepererum/add_cache_invalidation_logs

feat: log reason about WHY we invalidate caches

Author: kodiakhq[bot]

src/client/controller.rs

@@ -2,7 +2,7 @@ use async_trait::async_trait;
 use std::ops::ControlFlow;
 use std::sync::Arc;
 use tokio::sync::Mutex;
-use tracing::{debug, error, info};
+use tracing::{error, info};
 
 use crate::{
     backoff::{Backoff, BackoffConfig, ErrorOrThrottle},
@@ -139,8 +139,8 @@ impl BrokerCache for &ControllerClient {
         Ok(broker)
     }
 
-    async fn invalidate(&self) {
-        debug!("Invalidating cached controller broker");
+    async fn invalidate(&self, reason: &'static str) {
+        info!(reason, "Invalidating cached controller broker",);
         self.current_broker.lock().await.take();
     }
 }
@@ -175,14 +175,20 @@ where
             match error {
                 // broken connection
                 Error::Request(RequestError::Poisoned(_) | RequestError::IO(_))
-                | Error::Connection(_) => broker_cache.invalidate().await,
+                | Error::Connection(_) => {
+                    broker_cache
+                        .invalidate("controller client: connection broken")
+                        .await
+                }
 
                 // our broker is actually not the controller
                 Error::ServerError {
                     protocol_error: ProtocolError::NotController,
                     ..
                 } => {
-                    broker_cache.invalidate().await;
+                    broker_cache
+                        .invalidate("controller client: server error: not controller")
+                        .await;
                 }
 
                 // fatal

src/client/metadata_cache.rs

@@ -1,5 +1,5 @@
 use parking_lot::Mutex;
-use tracing::debug;
+use tracing::{debug, info};
 
 use crate::protocol::messages::MetadataResponse;
 
@@ -39,7 +39,7 @@ impl MetadataCache {
                 // if a caller keeps requesting metadata for a non-existent
                 // topic.
                 debug!("cached metadata query for unknown topic");
-                self.invalidate();
+                self.invalidate("get from metadata cache: unknown topic");
                 return None;
             }
         }
@@ -49,9 +49,9 @@ impl MetadataCache {
         Some(m)
     }
 
-    pub(crate) fn invalidate(&self) {
+    pub(crate) fn invalidate(&self, reason: &'static str) {
         *self.cache.lock() = None;
-        debug!("invalidated metadata cache");
+        info!(reason, "invalidated metadata cache",);
     }
 
     pub(crate) fn update(&self, m: MetadataResponse) {
@@ -148,7 +148,7 @@ mod tests {
         });
 
         assert!(cache.get(&None).is_some());
-        cache.invalidate();
+        cache.invalidate("test");
         assert!(cache.get(&None).is_none());
     }
 }

src/client/partition.rs

@@ -414,14 +414,18 @@ impl BrokerCache for &PartitionClient {
         let broker = match self.brokers.connect(leader).await {
             Ok(Some(c)) => Ok(c),
             Ok(None) => {
-                self.brokers.invalidate_metadata_cache();
+                self.brokers.invalidate_metadata_cache(
+                    "partition client: broker that is leader is unknown",
+                );
                 Err(Error::InvalidResponse(format!(
                     "Partition leader {} not found in metadata response",
                     leader
                 )))
             }
             Err(e) => {
-                self.brokers.invalidate_metadata_cache();
+                self.brokers.invalidate_metadata_cache(
+                    "partition client: error connecting to partition leader",
+                );
                 Err(e.into())
             }
         }?;
@@ -442,7 +446,9 @@ impl BrokerCache for &PartitionClient {
             .await?;
         if leader != leader_self {
             // The cached metadata identified an incorrect leader - it is stale and should be refreshed.
-            self.brokers.invalidate_metadata_cache();
+            self.brokers.invalidate_metadata_cache(
+                "partition client: broker that should be leader does treat itself as a leader",
+            );
 
             // this might happen if the leader changed after we got the hint from a arbitrary broker and this specific
             // metadata call.
@@ -469,13 +475,14 @@ impl BrokerCache for &PartitionClient {
         Ok(broker)
     }
 
-    async fn invalidate(&self) {
+    async fn invalidate(&self, reason: &'static str) {
         info!(
             topic = self.topic.deref(),
             partition = self.partition,
+            reason,
             "Invaliding cached leader",
         );
-        self.brokers.invalidate_metadata_cache();
+        self.brokers.invalidate_metadata_cache(reason);
         *self.current_broker.lock().await = None
     }
 }
@@ -511,7 +518,9 @@ where
             let retry = match error {
                 Error::Request(RequestError::Poisoned(_) | RequestError::IO(_))
                 | Error::Connection(_) => {
-                    broker_cache.invalidate().await;
+                    broker_cache
+                        .invalidate("partition client: connection broken")
+                        .await;
                     true
                 }
                 Error::ServerError {
@@ -525,14 +534,18 @@ where
                     protocol_error: ProtocolError::NotLeaderOrFollower,
                     ..
                 } => {
-                    broker_cache.invalidate().await;
+                    broker_cache
+                        .invalidate("partition client: server error: not leader or follower")
+                        .await;
                     true
                 }
                 Error::ServerError {
                     protocol_error: ProtocolError::UnknownTopicOrPartition,
                     ..
                 } => {
-                    broker_cache.invalidate().await;
+                    broker_cache
+                        .invalidate("partition client: server error: unknown topic or partition")
+                        .await;
                     unknown_topic_handling == UnknownTopicHandling::Retry
                 }
                 _ => false,

src/connection.rs

@@ -4,7 +4,7 @@ use std::ops::ControlFlow;
 use std::sync::Arc;
 use thiserror::Error;
 use tokio::{io::BufStream, sync::Mutex};
-use tracing::{debug, error, info, warn};
+use tracing::{error, info, warn};
 
 use crate::backoff::ErrorOrThrottle;
 use crate::connection::topology::{Broker, BrokerTopology};
@@ -264,8 +264,8 @@ impl BrokerConnector {
         Ok(response)
     }
 
-    pub(crate) fn invalidate_metadata_cache(&self) {
-        self.cached_metadata.invalidate()
+    pub(crate) fn invalidate_metadata_cache(&self, reason: &'static str) {
+        self.cached_metadata.invalidate(reason)
     }
 
     /// Returns a new connection to the broker with the provided id
@@ -343,7 +343,7 @@ pub trait BrokerCache: Send + Sync {
 
     async fn get(&self) -> Result<Arc<Self::R>, Self::E>;
 
-    async fn invalidate(&self);
+    async fn invalidate(&self, reason: &'static str);
 }
 
 /// BrokerConnector caches an arbitrary broker that can successfully connect.
@@ -372,8 +372,8 @@ impl BrokerCache for &BrokerConnector {
         Ok(connection)
     }
 
-    async fn invalidate(&self) {
-        debug!("Invalidating cached arbitrary broker");
+    async fn invalidate(&self, reason: &'static str) {
+        info!(reason, "Invalidating cached arbitrary broker",);
         self.cached_arbitrary_broker.lock().await.take();
     }
 }
@@ -461,7 +461,11 @@ where
                 Err(e @ RequestError::Poisoned(_) | e @ RequestError::IO(_))
                     if !matches!(metadata_mode, MetadataLookupMode::SpecificBroker(_)) =>
                 {
-                    arbitrary_broker_cache.invalidate().await;
+                    arbitrary_broker_cache
+                        .invalidate(
+                            "metadata request: arbitrary/cached broker is connection is broken",
+                        )
+                        .await;
                     ControlFlow::Continue(ErrorOrThrottle::Error(e))
                 }
                 Err(error) => {
@@ -523,7 +527,7 @@ mod tests {
             (self.get)()
         }
 
-        async fn invalidate(&self) {
+        async fn invalidate(&self, _reason: &'static str) {
             (self.invalidate)()
         }
     }