Use wrapper to add context to logging

Using a decorator pattern, add peer_id and channel_id to Record
stored on Logger.

Author: henghonglee

lightning/src/chain/chainmonitor.rs

@@ -29,7 +29,7 @@ use bitcoin::hash_types::{Txid, BlockHash};
 use crate::chain;
 use crate::chain::{ChannelMonitorUpdateStatus, Filter, WatchedOutput};
 use crate::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
-use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, Balance, MonitorEvent, TransactionOutputs, LATENCY_GRACE_PERIOD_BLOCKS};
+use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, Balance, MonitorEvent, TransactionOutputs, WithChannelMonitor, LATENCY_GRACE_PERIOD_BLOCKS};
 use crate::chain::transaction::{OutPoint, TransactionData};
 use crate::sign::WriteableEcdsaChannelSigner;
 use crate::events;
@@ -359,6 +359,7 @@ where C::Target: chain::Filter,
 		process: FN, funding_outpoint: &OutPoint, monitor_state: &MonitorHolder<ChannelSigner>
 	) -> Result<(), ()> where FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs> {
 		let monitor = &monitor_state.monitor;
+		let logger = WithChannelMonitor::from(&self.logger, &monitor);
 		let mut txn_outputs;
 		{
 			txn_outputs = process(monitor, txdata);
@@ -375,12 +376,12 @@ where C::Target: chain::Filter,
 				}
 			}
 
-			log_trace!(self.logger, "Syncing Channel Monitor for channel {}", log_funding_info!(monitor));
+			log_trace!(logger, "Syncing Channel Monitor for channel {}", log_funding_info!(monitor));
 			match self.persister.update_persisted_channel(*funding_outpoint, None, monitor, update_id) {
 				ChannelMonitorUpdateStatus::Completed =>
-					log_trace!(self.logger, "Finished syncing Channel Monitor for channel {}", log_funding_info!(monitor)),
+					log_trace!(logger, "Finished syncing Channel Monitor for channel {}", log_funding_info!(monitor)),
 				ChannelMonitorUpdateStatus::InProgress => {
-					log_debug!(self.logger, "Channel Monitor sync for channel {} in progress, holding events until completion!", log_funding_info!(monitor));
+					log_debug!(logger, "Channel Monitor sync for channel {} in progress, holding events until completion!", log_funding_info!(monitor));
 					pending_monitor_updates.push(update_id);
 				},
 				ChannelMonitorUpdateStatus::UnrecoverableError => {
@@ -619,8 +620,9 @@ where C::Target: chain::Filter,
 	pub fn rebroadcast_pending_claims(&self) {
 		let monitors = self.monitors.read().unwrap();
 		for (_, monitor_holder) in &*monitors {
+			let logger = WithChannelMonitor::from(&self.logger, &monitor_holder.monitor);
 			monitor_holder.monitor.rebroadcast_pending_claims(
-				&*self.broadcaster, &*self.fee_estimator, &*self.logger
+				&*self.broadcaster, &*self.fee_estimator, &logger
 			)
 		}
 	}
@@ -638,17 +640,19 @@ where
 	fn filtered_block_connected(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) {
 		log_debug!(self.logger, "New best block {} at height {} provided via block_connected", header.block_hash(), height);
 		self.process_chain_data(header, Some(height), &txdata, |monitor, txdata| {
+			let logger = WithChannelMonitor::from(&self.logger, &monitor);
 			monitor.block_connected(
-				header, txdata, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger)
+				header, txdata, height, &*self.broadcaster, &*self.fee_estimator, &logger)
 		});
 	}
 
 	fn block_disconnected(&self, header: &BlockHeader, height: u32) {
 		let monitor_states = self.monitors.read().unwrap();
 		log_debug!(self.logger, "Latest block {} at height {} removed via block_disconnected", header.block_hash(), height);
 		for monitor_state in monitor_states.values() {
+			let logger = WithChannelMonitor::from(&self.logger, &monitor_state.monitor);
 			monitor_state.monitor.block_disconnected(
-				header, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
+				header, height, &*self.broadcaster, &*self.fee_estimator, &logger);
 		}
 	}
 }
@@ -665,27 +669,30 @@ where
 	fn transactions_confirmed(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) {
 		log_debug!(self.logger, "{} provided transactions confirmed at height {} in block {}", txdata.len(), height, header.block_hash());
 		self.process_chain_data(header, None, txdata, |monitor, txdata| {
+			let logger = WithChannelMonitor::from(&self.logger, &monitor);
 			monitor.transactions_confirmed(
-				header, txdata, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger)
+				header, txdata, height, &*self.broadcaster, &*self.fee_estimator, &logger)
 		});
 	}
 
 	fn transaction_unconfirmed(&self, txid: &Txid) {
 		log_debug!(self.logger, "Transaction {} reorganized out of chain", txid);
 		let monitor_states = self.monitors.read().unwrap();
 		for monitor_state in monitor_states.values() {
-			monitor_state.monitor.transaction_unconfirmed(txid, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
+			let logger = WithChannelMonitor::from(&self.logger, &monitor_state.monitor);
+			monitor_state.monitor.transaction_unconfirmed(txid, &*self.broadcaster, &*self.fee_estimator, &logger);
 		}
 	}
 
 	fn best_block_updated(&self, header: &BlockHeader, height: u32) {
 		log_debug!(self.logger, "New best block {} at height {} provided via best_block_updated", header.block_hash(), height);
 		self.process_chain_data(header, Some(height), &[], |monitor, txdata| {
+			let logger = WithChannelMonitor::from(&self.logger, &monitor);
 			// While in practice there shouldn't be any recursive calls when given empty txdata,
 			// it's still possible if a chain::Filter implementation returns a transaction.
 			debug_assert!(txdata.is_empty());
 			monitor.best_block_updated(
-				header, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger)
+				header, height, &*self.broadcaster, &*self.fee_estimator, &logger)
 		});
 	}
 
@@ -711,29 +718,30 @@ where C::Target: chain::Filter,
 	    P::Target: Persist<ChannelSigner>,
 {
 	fn watch_channel(&self, funding_outpoint: OutPoint, monitor: ChannelMonitor<ChannelSigner>) -> Result<ChannelMonitorUpdateStatus, ()> {
+		let logger = WithChannelMonitor::from(&self.logger, &monitor);
 		let mut monitors = self.monitors.write().unwrap();
 		let entry = match monitors.entry(funding_outpoint) {
 			hash_map::Entry::Occupied(_) => {
-				log_error!(self.logger, "Failed to add new channel data: channel monitor for given outpoint is already present");
+				log_error!(logger, "Failed to add new channel data: channel monitor for given outpoint is already present");
 				return Err(());
 			},
 			hash_map::Entry::Vacant(e) => e,
 		};
-		log_trace!(self.logger, "Got new ChannelMonitor for channel {}", log_funding_info!(monitor));
+		log_trace!(logger, "Got new ChannelMonitor for channel {}", log_funding_info!(monitor));
 		let update_id = MonitorUpdateId::from_new_monitor(&monitor);
 		let mut pending_monitor_updates = Vec::new();
 		let persist_res = self.persister.persist_new_channel(funding_outpoint, &monitor, update_id);
 		match persist_res {
 			ChannelMonitorUpdateStatus::InProgress => {
-				log_info!(self.logger, "Persistence of new ChannelMonitor for channel {} in progress", log_funding_info!(monitor));
+				log_info!(logger, "Persistence of new ChannelMonitor for channel {} in progress", log_funding_info!(monitor));
 				pending_monitor_updates.push(update_id);
 			},
 			ChannelMonitorUpdateStatus::Completed => {
-				log_info!(self.logger, "Persistence of new ChannelMonitor for channel {} completed", log_funding_info!(monitor));
+				log_info!(logger, "Persistence of new ChannelMonitor for channel {} completed", log_funding_info!(monitor));
 			},
 			ChannelMonitorUpdateStatus::UnrecoverableError => {
 				let err_str = "ChannelMonitor[Update] persistence failed unrecoverably. This indicates we cannot continue normal operation and must shut down.";
-				log_error!(self.logger, "{}", err_str);
+				log_error!(logger, "{}", err_str);
 				panic!("{}", err_str);
 			},
 		}
@@ -765,7 +773,8 @@ where C::Target: chain::Filter,
 			},
 			Some(monitor_state) => {
 				let monitor = &monitor_state.monitor;
-				log_trace!(self.logger, "Updating ChannelMonitor for channel {}", log_funding_info!(monitor));
+				let logger = WithChannelMonitor::from(&self.logger, &monitor);
+				log_trace!(logger, "Updating ChannelMonitor for channel {}", log_funding_info!(monitor));
 				let update_res = monitor.update_monitor(update, &self.broadcaster, &self.fee_estimator, &self.logger);
 
 				let update_id = MonitorUpdateId::from_monitor_update(update);
@@ -776,18 +785,18 @@ where C::Target: chain::Filter,
 					// We don't want to persist a `monitor_update` which results in a failure to apply later
 					// while reading `channel_monitor` with updates from storage. Instead, we should persist
 					// the entire `channel_monitor` here.
-					log_warn!(self.logger, "Failed to update ChannelMonitor for channel {}. Going ahead and persisting the entire ChannelMonitor", log_funding_info!(monitor));
+					log_warn!(logger, "Failed to update ChannelMonitor for channel {}. Going ahead and persisting the entire ChannelMonitor", log_funding_info!(monitor));
 					self.persister.update_persisted_channel(funding_txo, None, monitor, update_id)
 				} else {
 					self.persister.update_persisted_channel(funding_txo, Some(update), monitor, update_id)
 				};
 				match persist_res {
 					ChannelMonitorUpdateStatus::InProgress => {
 						pending_monitor_updates.push(update_id);
-						log_debug!(self.logger, "Persistence of ChannelMonitorUpdate for channel {} in progress", log_funding_info!(monitor));
+						log_debug!(logger, "Persistence of ChannelMonitorUpdate for channel {} in progress", log_funding_info!(monitor));
 					},
 					ChannelMonitorUpdateStatus::Completed => {
-						log_debug!(self.logger, "Persistence of ChannelMonitorUpdate for channel {} completed", log_funding_info!(monitor));
+						log_debug!(logger, "Persistence of ChannelMonitorUpdate for channel {} completed", log_funding_info!(monitor));
 					},
 					ChannelMonitorUpdateStatus::UnrecoverableError => { /* we'll panic in a moment */ },
 				}
@@ -799,12 +808,16 @@ where C::Target: chain::Filter,
 			}
 		};
 		if let ChannelMonitorUpdateStatus::UnrecoverableError = ret {
+			let logger = WithChannelMonitor::from(
+				&self.logger, &monitors.get(&funding_txo).unwrap().monitor
+			);
+
 			// Take the monitors lock for writing so that we poison it and any future
 			// operations going forward fail immediately.
 			core::mem::drop(monitors);
 			let _poison = self.monitors.write().unwrap();
 			let err_str = "ChannelMonitor[Update] persistence failed unrecoverably. This indicates we cannot continue normal operation and must shut down.";
-			log_error!(self.logger, "{}", err_str);
+			log_error!(logger, "{}", err_str);
 			panic!("{}", err_str);
 		}
 		ret
@@ -813,12 +826,13 @@ where C::Target: chain::Filter,
 	fn release_pending_monitor_events(&self) -> Vec<(OutPoint, Vec<MonitorEvent>, Option<PublicKey>)> {
 		let mut pending_monitor_events = self.pending_monitor_events.lock().unwrap().split_off(0);
 		for monitor_state in self.monitors.read().unwrap().values() {
+			let logger = WithChannelMonitor::from(&self.logger, &monitor_state.monitor);
 			let is_pending_monitor_update = monitor_state.has_pending_chainsync_updates(&monitor_state.pending_monitor_updates.lock().unwrap());
 			if !is_pending_monitor_update || monitor_state.last_chain_persist_height.load(Ordering::Acquire) + LATENCY_GRACE_PERIOD_BLOCKS as usize <= self.highest_chain_height.load(Ordering::Acquire) {
 				if is_pending_monitor_update {
-					log_error!(self.logger, "A ChannelMonitor sync took longer than {} blocks to complete.", LATENCY_GRACE_PERIOD_BLOCKS);
-					log_error!(self.logger, "   To avoid funds-loss, we are allowing monitor updates to be released.");
-					log_error!(self.logger, "   This may cause duplicate payment events to be generated.");
+					log_error!(logger, "A ChannelMonitor sync took longer than {} blocks to complete.", LATENCY_GRACE_PERIOD_BLOCKS);
+					log_error!(logger, "   To avoid funds-loss, we are allowing monitor updates to be released.");
+					log_error!(logger, "   This may cause duplicate payment events to be generated.");
 				}
 				let monitor_events = monitor_state.monitor.get_and_clear_pending_monitor_events();
 				if monitor_events.len() > 0 {

lightning/src/chain/channelmonitor.rs

@@ -4167,11 +4167,11 @@ where
 	L::Target: Logger,
 {
 	fn filtered_block_connected(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) {
-		self.0.block_connected(header, txdata, height, &*self.1, &*self.2, &*self.3);
+		self.0.block_connected(header, txdata, height, &*self.1, &*self.2, &WithChannelMonitor::from(&self.3, &self.0));
 	}
 
 	fn block_disconnected(&self, header: &BlockHeader, height: u32) {
-		self.0.block_disconnected(header, height, &*self.1, &*self.2, &*self.3);
+		self.0.block_disconnected(header, height, &*self.1, &*self.2, &WithChannelMonitor::from(&self.3, &self.0));
 	}
 }
 
@@ -4183,15 +4183,15 @@ where
 	L::Target: Logger,
 {
 	fn transactions_confirmed(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) {
-		self.0.transactions_confirmed(header, txdata, height, &*self.1, &*self.2, &*self.3);
+		self.0.transactions_confirmed(header, txdata, height, &*self.1, &*self.2, &WithChannelMonitor::from(&self.3, &self.0));
 	}
 
 	fn transaction_unconfirmed(&self, txid: &Txid) {
-		self.0.transaction_unconfirmed(txid, &*self.1, &*self.2, &*self.3);
+		self.0.transaction_unconfirmed(txid, &*self.1, &*self.2, &WithChannelMonitor::from(&self.3, &self.0));
 	}
 
 	fn best_block_updated(&self, header: &BlockHeader, height: u32) {
-		self.0.best_block_updated(header, height, &*self.1, &*self.2, &*self.3);
+		self.0.best_block_updated(header, height, &*self.1, &*self.2, &WithChannelMonitor::from(&self.3, &self.0));
 	}
 
 	fn get_relevant_txids(&self) -> Vec<(Txid, Option<BlockHash>)> {

lightning/src/ln/channel.rs

@@ -34,7 +34,7 @@ use crate::ln::chan_utils;
 use crate::ln::onion_utils::HTLCFailReason;
 use crate::chain::BestBlock;
 use crate::chain::chaininterface::{FeeEstimator, ConfirmationTarget, LowerBoundedFeeEstimator};
-use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateStep, LATENCY_GRACE_PERIOD_BLOCKS, CLOSED_CHANNEL_UPDATE_ID};
+use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateStep, WithChannelMonitor, LATENCY_GRACE_PERIOD_BLOCKS, CLOSED_CHANNEL_UPDATE_ID};
 use crate::chain::transaction::{OutPoint, TransactionData};
 use crate::sign::{EcdsaChannelSigner, WriteableEcdsaChannelSigner, EntropySource, ChannelSigner, SignerProvider, NodeSigner, Recipient};
 use crate::events::ClosureReason;
@@ -2748,14 +2748,14 @@ impl<SP: Deref> Channel<SP> where
 		                                          funding_redeemscript.clone(), self.context.channel_value_satoshis,
 		                                          obscure_factor,
 		                                          holder_commitment_tx, best_block, self.context.counterparty_node_id);
-
+		let logger_with_chan_monitor = WithChannelMonitor::from(logger, &channel_monitor);
 		channel_monitor.provide_initial_counterparty_commitment_tx(
 			counterparty_initial_bitcoin_tx.txid, Vec::new(),
 			self.context.cur_counterparty_commitment_transaction_number,
 			self.context.counterparty_cur_commitment_point.unwrap(),
 			counterparty_initial_commitment_tx.feerate_per_kw(),
 			counterparty_initial_commitment_tx.to_broadcaster_value_sat(),
-			counterparty_initial_commitment_tx.to_countersignatory_value_sat(), logger);
+			counterparty_initial_commitment_tx.to_countersignatory_value_sat(), &&logger_with_chan_monitor);
 
 		assert_eq!(self.context.channel_state & (ChannelState::MonitorUpdateInProgress as u32), 0); // We have no had any monitor(s) yet to fail update!
 		if self.context.is_batch_funding() {
@@ -6910,13 +6910,13 @@ impl<SP: Deref> InboundV1Channel<SP> where SP::Target: SignerProvider {
 		                                          funding_redeemscript.clone(), self.context.channel_value_satoshis,
 		                                          obscure_factor,
 		                                          holder_commitment_tx, best_block, self.context.counterparty_node_id);
-
+		let logger_with_chan_monitor = WithChannelMonitor::from(logger, &channel_monitor);
 		channel_monitor.provide_initial_counterparty_commitment_tx(
 			counterparty_initial_commitment_tx.trust().txid(), Vec::new(),
 			self.context.cur_counterparty_commitment_transaction_number + 1,
 			self.context.counterparty_cur_commitment_point.unwrap(), self.context.feerate_per_kw,
 			counterparty_initial_commitment_tx.to_broadcaster_value_sat(),
-			counterparty_initial_commitment_tx.to_countersignatory_value_sat(), logger);
+			counterparty_initial_commitment_tx.to_countersignatory_value_sat(), &&logger_with_chan_monitor);
 
 		log_info!(logger, "{} funding_signed for peer for channel {}",
 			if funding_signed.is_some() { "Generated" } else { "Waiting for signature on" }, &self.context.channel_id());