chore: use meter factory for Event Hubs cache metrics

src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/EventHubAdapterFactory.cs

@@ -265,7 +265,7 @@ protected virtual IEventHubQueueCacheFactory CreateCacheFactory(EventHubStreamCa
         {
            var eventHubPath = this.ehOptions.EventHubName;
             var sharedDimensions = new EventHubMonitorAggregationDimensions(eventHubPath);
-            return new EventHubQueueCacheFactory(eventHubCacheOptions, cacheEvictionOptions, statisticOptions, this.dataAdapter, sharedDimensions);
+            return new EventHubQueueCacheFactory(eventHubCacheOptions, cacheEvictionOptions, statisticOptions, this.dataAdapter, sharedDimensions, this.orleansInstruments);
         }
 
         private EventHubAdapterReceiver MakeReceiver(QueueId queueId)

src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/EventHubQueueCacheFactory.cs

@@ -2,6 +2,7 @@
 using Microsoft.Extensions.Logging;
 using Orleans.Configuration;
 using Orleans.Providers.Streams.Common;
+using Orleans.Runtime;
 using Orleans.Streams;
 using Orleans.Streaming.EventHubs.StatisticMonitors;
 
@@ -19,6 +20,7 @@ public class EventHubQueueCacheFactory : IEventHubQueueCacheFactory
         private readonly IEventHubDataAdapter dataAdater;
         private readonly TimePurgePredicate timePurge;
         private readonly EventHubMonitorAggregationDimensions sharedDimensions;
+        private readonly OrleansInstruments orleansInstruments;
         private IObjectPool<FixedSizeBuffer> bufferPool;
         private string bufferPoolId;
 
@@ -45,17 +47,37 @@ public EventHubQueueCacheFactory(
             EventHubMonitorAggregationDimensions sharedDimensions,
             Func<EventHubCacheMonitorDimensions, ILoggerFactory, ICacheMonitor> cacheMonitorFactory = null,
             Func<EventHubBlockPoolMonitorDimensions, ILoggerFactory, IBlockPoolMonitor> blockPoolMonitorFactory = null)
+            : this(cacheOptions, evictionOptions, statisticOptions, dataAdater, sharedDimensions, null, cacheMonitorFactory, blockPoolMonitorFactory)
+        {
+        }
+
+        internal EventHubQueueCacheFactory(
+            EventHubStreamCachePressureOptions cacheOptions,
+            StreamCacheEvictionOptions evictionOptions,
+            StreamStatisticOptions statisticOptions,
+            IEventHubDataAdapter dataAdater,
+            EventHubMonitorAggregationDimensions sharedDimensions,
+            OrleansInstruments instruments,
+            Func<EventHubCacheMonitorDimensions, ILoggerFactory, ICacheMonitor> cacheMonitorFactory = null,
+            Func<EventHubBlockPoolMonitorDimensions, ILoggerFactory, IBlockPoolMonitor> blockPoolMonitorFactory = null)
         {
             this.cacheOptions = cacheOptions;
             this.evictionOptions = evictionOptions;
             this.statisticOptions = statisticOptions;
             this.dataAdater = dataAdater;
             this.timePurge = new TimePurgePredicate(evictionOptions.DataMinTimeInCache, evictionOptions.DataMaxAgeInCache);
             this.sharedDimensions = sharedDimensions;
-            this.CacheMonitorFactory = cacheMonitorFactory ?? ((dimensions, logger) => new DefaultEventHubCacheMonitor(dimensions));
-            this.BlockPoolMonitorFactory = blockPoolMonitorFactory ?? ((dimensions, logger) => new DefaultEventHubBlockPoolMonitor(dimensions));
+            this.orleansInstruments = instruments;
+            this.CacheMonitorFactory = cacheMonitorFactory ?? CreateDefaultCacheMonitor;
+            this.BlockPoolMonitorFactory = blockPoolMonitorFactory ?? CreateDefaultBlockPoolMonitor;
         }
 
+        private ICacheMonitor CreateDefaultCacheMonitor(EventHubCacheMonitorDimensions dimensions, ILoggerFactory logger) =>
+            this.orleansInstruments is not null ? new DefaultEventHubCacheMonitor(dimensions, this.orleansInstruments) : new DefaultEventHubCacheMonitor(dimensions);
+
+        private IBlockPoolMonitor CreateDefaultBlockPoolMonitor(EventHubBlockPoolMonitorDimensions dimensions, ILoggerFactory logger) =>
+            this.orleansInstruments is not null ? new DefaultEventHubBlockPoolMonitor(dimensions, this.orleansInstruments) : new DefaultEventHubBlockPoolMonitor(dimensions);
+
         /// <summary>
         /// Function which create an EventHubQueueCache, which by default will configure the EventHubQueueCache using configuration in CreateBufferPool function
         /// and AddCachePressureMonitors function.

src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/StatisticMonitors/DefaultEventHubBlockPoolMonitor.cs

@@ -1,5 +1,6 @@
 using System.Collections.Generic;
 using Orleans.Providers.Streams.Common;
+using Orleans.Runtime;
 
 namespace Orleans.Streaming.EventHubs.StatisticMonitors
 {
@@ -15,5 +16,10 @@ public class DefaultEventHubBlockPoolMonitor : DefaultBlockPoolMonitor
         public DefaultEventHubBlockPoolMonitor(EventHubBlockPoolMonitorDimensions dimensions) : base(new KeyValuePair<string, object>[] { new("Path", dimensions.EventHubPath), new("ObjectPoolId", dimensions.BlockPoolId) })
         {
         }
+
+        internal DefaultEventHubBlockPoolMonitor(EventHubBlockPoolMonitorDimensions dimensions, OrleansInstruments instruments)
+            : base(new KeyValuePair<string, object>[] { new("Path", dimensions.EventHubPath), new("ObjectPoolId", dimensions.BlockPoolId) }, instruments)
+        {
+        }
     }
 }

src/Azure/Orleans.Streaming.EventHubs/Providers/Streams/EventHub/StatisticMonitors/DefaultEventHubCacheMonitor.cs

@@ -1,5 +1,6 @@
 using System.Collections.Generic;
 using Orleans.Providers.Streams.Common;
+using Orleans.Runtime;
 
 namespace Orleans.Streaming.EventHubs.StatisticMonitors
 {
@@ -16,5 +17,10 @@ public DefaultEventHubCacheMonitor(EventHubCacheMonitorDimensions dimensions)
             : base(new KeyValuePair<string, object>[] { new("Path", dimensions.EventHubPath), new("Partition", dimensions.EventHubPartition) })
         {
         }
+
+        internal DefaultEventHubCacheMonitor(EventHubCacheMonitorDimensions dimensions, OrleansInstruments instruments)
+            : base(new KeyValuePair<string, object>[] { new("Path", dimensions.EventHubPath), new("Partition", dimensions.EventHubPartition) }, instruments)
+        {
+        }
     }
 }

src/Orleans.Streaming/Common/Monitors/DefaultBlockPoolMonitor.cs

@@ -31,6 +31,11 @@ protected DefaultBlockPoolMonitor(KeyValuePair<string, object>[] dimensions)
         {
         }
 
+        protected DefaultBlockPoolMonitor(KeyValuePair<string, object>[] dimensions, OrleansInstruments instruments)
+            : this(dimensions, instruments.Meter)
+        {
+        }
+
         internal DefaultBlockPoolMonitor(BlockPoolMonitorDimensions dimensions, OrleansInstruments instruments)
             : this(new KeyValuePair<string, object>[] { new("BlockPoolId", dimensions.BlockPoolId) }, instruments.Meter)
         {

src/Orleans.Streaming/Common/Monitors/DefaultCacheMonitor.cs

@@ -43,6 +43,11 @@ protected DefaultCacheMonitor(KeyValuePair<string, object>[] dimensions)
         {
         }
 
+        protected DefaultCacheMonitor(KeyValuePair<string, object>[] dimensions, OrleansInstruments instruments)
+            : this(dimensions, instruments.Meter)
+        {
+        }
+
         internal DefaultCacheMonitor(CacheMonitorDimensions dimensions, OrleansInstruments instruments)
             : this(new KeyValuePair<string, object>[] { new("QueueId", dimensions.QueueId) }, instruments.Meter)
         {

src/api/Orleans.Streaming/Orleans.Streaming.cs

@@ -553,6 +553,8 @@ public DefaultBlockPoolMonitor(BlockPoolMonitorDimensions dimensions) { }
 
         protected DefaultBlockPoolMonitor(System.Collections.Generic.KeyValuePair<string, object>[] dimensions) { }
 
+        protected DefaultBlockPoolMonitor(System.Collections.Generic.KeyValuePair<string, object>[] dimensions, Runtime.OrleansInstruments instruments) { }
+
         public void Report(long totalMemoryInByte, long availableMemoryInByte, long claimedMemoryInByte) { }
 
         public void TrackMemoryAllocated(long allocatedMemoryInByte) { }
@@ -566,6 +568,8 @@ public DefaultCacheMonitor(CacheMonitorDimensions dimensions) { }
 
         protected DefaultCacheMonitor(System.Collections.Generic.KeyValuePair<string, object>[] dimensions) { }
 
+        protected DefaultCacheMonitor(System.Collections.Generic.KeyValuePair<string, object>[] dimensions, Runtime.OrleansInstruments instruments) { }
+
         public void ReportCacheSize(long totalCacheSizeInByte) { }
 
         public void ReportMessageStatistics(System.DateTime? oldestMessageEnqueueTimeUtc, System.DateTime? oldestMessageDequeueTimeUtc, System.DateTime? newestMessageEnqueueTimeUtc, long totalMessageCount) { }