Refactor SourceMetrics with LabelGuarded ones

[fuyufjh]

See background in the tracking issue #14838

risingwave/src/connector/src/source/monitor/metrics.rs

Lines 58 to 72 in 20fc879

	#[derive(Debug, Clone)]
	pub struct SourceMetrics {
	pub partition_input_count: GenericCounterVec<AtomicU64>,
	// **Note**: for normal messages, the metric is the message's payload size.
	// For messages from load generator, the metric is the size of stream chunk.
	pub partition_input_bytes: GenericCounterVec<AtomicU64>,
	/// Report latest message id
	pub latest_message_id: GenericGaugeVec<AtomicI64>,
	pub rdkafka_native_metric: Arc<RdKafkaStats>,
	pub connector_source_rows_received: GenericCounterVec<AtomicU64>,
	pub direct_cdc_event_lag_latency: HistogramVec,
	}