feat(worker-proxy): instrument extension gRPC calls

src/Functions.WorkerProxy/ExtensionGrpcActivity.cs

@@ -0,0 +1,38 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System.Diagnostics;
+
+namespace Azure.Functions.WorkerProxy;
+
+/// <summary>
+/// Enriches worker-facing ASP.NET Core activities with extension RPC correlation and concurrency data.
+/// </summary>
+internal static class ExtensionGrpcActivity
+{
+    private const string TagPrefix = "azure.functions.worker_proxy.extension_rpc";
+
+    /// <summary>
+    /// Enriches the current activity after an extension call acquires a host stream.
+    /// </summary>
+    /// <param name="callId">The logical extension call identifier.</param>
+    /// <param name="streamId">The physical extension stream identifier.</param>
+    /// <param name="activeCallCount">The active-call count when the call opens.</param>
+    public static void CallOpened(string callId, string streamId, int activeCallCount)
+    {
+        Activity? activity = Activity.Current;
+        activity?.SetTag($"{TagPrefix}.call_id", callId);
+        activity?.SetTag($"{TagPrefix}.stream_id", streamId);
+        activity?.SetTag($"{TagPrefix}.active_calls_at_open", activeCallCount);
+    }
+
+    /// <summary>
+    /// Enriches the current activity with an extension call's final concurrency snapshot.
+    /// </summary>
+    /// <param name="activeCallCount">The active-call count when the call completes.</param>
+    public static void CallCompleted(int activeCallCount)
+    {
+        Activity? activity = Activity.Current;
+        activity?.SetTag($"{TagPrefix}.active_calls_at_completion", activeCallCount);
+    }
+}

src/Functions.WorkerProxy/ExtensionGrpcIngress.cs

@@ -24,10 +24,12 @@ namespace Azure.Functions.WorkerProxy;
 /// </summary>
 /// <param name="endpoints">The WorkerProxy listener configuration.</param>
 /// <param name="streamCoordinator">The extension RPC stream coordinator.</param>
+/// <param name="metrics">The extension gRPC metrics.</param>
 /// <param name="logger">The logger used for ingress diagnostics.</param>
 internal sealed partial class ExtensionGrpcIngress(
     WorkerProxyEndpointConfiguration endpoints,
     ExtensionRpcStreamCoordinator streamCoordinator,
+    ExtensionGrpcMetrics metrics,
     ILogger<ExtensionGrpcIngress> logger)
 {
     internal const string FunctionRpcEventStreamPath = "/AzureFunctionsRpcMessages.FunctionRpc/EventStream";
@@ -55,6 +57,8 @@ internal sealed partial class ExtensionGrpcIngress(
     private readonly ExtensionRpcStreamCoordinator _streamCoordinator = streamCoordinator
         ?? throw new ArgumentNullException(nameof(streamCoordinator));
 
+    private readonly ExtensionGrpcMetrics _metrics = metrics ?? throw new ArgumentNullException(nameof(metrics));
+
     private readonly ILogger _logger = logger ?? throw new ArgumentNullException(nameof(logger));
 
     /// <summary>
@@ -153,6 +157,9 @@ await CompleteWithStatusAsync(
 
         int activeCallCountAtOpen = call.ActiveCallCount;
         double openDurationMilliseconds = Stopwatch.GetElapsedTime(callStart).TotalMilliseconds;
+        ExtensionGrpcActivity.CallOpened(call.CallId, call.StreamId, activeCallCountAtOpen);
+        _metrics.CallOpenDuration.Record(openDurationMilliseconds);
+        _metrics.ActiveCalls.Increment();
 
         // Remove this per-call log when WorkerProxy metric exporting is wired up.
         Log.CallOpened(_logger, start.Method, call.CallId, activeCallCountAtOpen, openDurationMilliseconds);
@@ -207,6 +214,9 @@ await CompleteWithStatusAsync(
             await StopRelayTasksAsync(cancellationTokenSource, requestTask, responseTask);
             int activeCallCountAtCompletion = call.ActiveCallCount;
             double callDurationMilliseconds = Stopwatch.GetElapsedTime(callStart).TotalMilliseconds;
+            ExtensionGrpcActivity.CallCompleted(activeCallCountAtCompletion);
+            _metrics.CallDuration.Record(callDurationMilliseconds);
+            _metrics.ActiveCalls.Decrement();
 
             // Remove this per-call log when WorkerProxy metric exporting is wired up.
             Log.CallCompleted(

src/Functions.WorkerProxy/ExtensionGrpcMetrics.cs

@@ -0,0 +1,119 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using System.Diagnostics.Metrics;
+
+namespace Azure.Functions.WorkerProxy;
+
+/// <summary>
+/// Records transport-level measurements for worker-facing extension gRPC calls.
+/// </summary>
+internal sealed class ExtensionGrpcMetrics
+{
+    internal const string MeterName = "Microsoft.Azure.Functions.WorkerProxy.ExtensionGrpc";
+    internal const string MeterVersion = "1.0.0";
+    internal const string ActiveCallsInstrumentName = "azure.functions.worker_proxy.extension_rpc.calls.active";
+    internal const string CallDurationInstrumentName = "azure.functions.worker_proxy.extension_rpc.call.duration";
+    internal const string CallOpenDurationInstrumentName =
+        "azure.functions.worker_proxy.extension_rpc.call.open.duration";
+
+    /// <summary>
+    /// Initializes extension gRPC metrics using a factory-owned meter.
+    /// </summary>
+    /// <param name="meterFactory">The factory that owns the meter lifetime.</param>
+    public ExtensionGrpcMetrics(IMeterFactory meterFactory)
+    {
+        ArgumentNullException.ThrowIfNull(meterFactory);
+
+#pragma warning disable CA2000 // IMeterFactory owns the meter lifetime.
+        Meter meter = meterFactory.Create(MeterName, MeterVersion);
+#pragma warning restore CA2000
+
+        ActiveCalls = new(meter);
+        CallDuration = new(meter);
+        CallOpenDuration = new(meter);
+    }
+
+    /// <summary>
+    /// Gets the active-call counter.
+    /// </summary>
+    public ActiveCallsCounter ActiveCalls { get; }
+
+    /// <summary>
+    /// Gets the total call-duration histogram.
+    /// </summary>
+    public CallDurationHistogram CallDuration { get; }
+
+    /// <summary>
+    /// Gets the call-open-duration histogram.
+    /// </summary>
+    public CallOpenDurationHistogram CallOpenDuration { get; }
+
+    /// <summary>
+    /// Records the number of active extension gRPC calls.
+    /// </summary>
+    /// <remarks>
+    /// Initializes the active-call counter.
+    /// </remarks>
+    /// <param name="meter">The meter used to create the counter.</param>
+    internal sealed class ActiveCallsCounter(Meter meter)
+    {
+        private readonly UpDownCounter<long> _counter = meter.CreateUpDownCounter<long>(
+            ActiveCallsInstrumentName,
+            unit: "{call}",
+            description: "Number of worker-facing extension gRPC calls currently relayed by this proxy.");
+
+        /// <summary>
+        /// Records that an extension gRPC call started relaying.
+        /// </summary>
+        public void Increment() => _counter.Add(1);
+
+        /// <summary>
+        /// Records that an extension gRPC call stopped relaying.
+        /// </summary>
+        public void Decrement() => _counter.Add(-1);
+    }
+
+    /// <summary>
+    /// Records extension gRPC call durations.
+    /// </summary>
+    /// <remarks>
+    /// Initializes the call-duration histogram.
+    /// </remarks>
+    /// <param name="meter">The meter used to create the histogram.</param>
+    internal sealed class CallDurationHistogram(Meter meter)
+    {
+        private readonly Histogram<double> _histogram = meter.CreateHistogram<double>(
+            CallDurationInstrumentName,
+            unit: "ms",
+            description: "Time spent relaying a worker-facing extension gRPC call, including stream assignment.");
+
+        /// <summary>
+        /// Records a call duration.
+        /// </summary>
+        /// <param name="durationMilliseconds">The total relay duration in milliseconds.</param>
+        public void Record(double durationMilliseconds) => _histogram.Record(durationMilliseconds);
+    }
+
+    /// <summary>
+    /// Records extension gRPC call-open durations.
+    /// </summary>
+    /// <remarks>
+    /// Initializes the call-open-duration histogram.
+    /// </remarks>
+    /// <param name="meter">The meter used to create the histogram.</param>
+    internal sealed class CallOpenDurationHistogram(Meter meter)
+    {
+        private readonly Histogram<double> _histogram = meter.CreateHistogram<double>(
+            CallOpenDurationInstrumentName,
+            unit: "ms",
+            description: "Time spent assigning a worker-facing extension gRPC call to the host stream.");
+
+        /// <summary>
+        /// Records a call-open duration.
+        /// </summary>
+        /// <param name="durationMilliseconds">The stream-assignment latency in milliseconds.</param>
+        public void Record(double durationMilliseconds) => _histogram.Record(durationMilliseconds);
+    }
+}

src/Functions.WorkerProxy/WorkerProxyApplication.cs

@@ -53,6 +53,8 @@ public static WebApplication Build(string[] args)
         builder.Services.AddHostedService(static services => services.GetRequiredService<FunctionRpcRelay>());
         builder.Services.AddSingleton<ExtensionRpcStreamCoordinator>();
         builder.Services.AddSingleton<ExtensionRpcRelay>();
+        builder.Services.AddMetrics();
+        builder.Services.AddSingleton<ExtensionGrpcMetrics>();
         builder.Services.AddSingleton<ExtensionGrpcIngress>();
         ConfigureHttpForwarding(builder);
 

test/Functions.WorkerProxy.Tests/ExtensionGrpcIngressTests.cs

@@ -3,6 +3,8 @@
 
 using System;
 using System.Buffers.Binary;
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
 using System.IO;
 using System.Linq;
 using System.Text;
@@ -23,6 +25,7 @@ namespace Azure.Functions.WorkerProxy.Tests;
 public class ExtensionGrpcIngressTests
 {
     private const int WorkerGrpcPort = 50052;
+    private static readonly ExtensionGrpcMetrics Metrics = new(new TestMeterFactory());
 
     [Fact]
     public async Task HandleAsync_RelaysOpaqueGrpcFramesAndMetadata()
@@ -42,6 +45,7 @@ public async Task HandleAsync_RelaysOpaqueGrpcFramesAndMetadata()
         context.Features.Set<IHttpResponseTrailersFeature>(trailersFeature);
 
         ExtensionGrpcIngress ingress = CreateIngress(streamCoordinator);
+        using var activity = new Activity("extension-rpc-test").Start();
         Task ingressTask = ingress.HandleAsync(context);
 
         ExtensionRpcMessage start = await lease.Stream.Outbound.ReadAsync();
@@ -121,6 +125,12 @@ await lease.Stream.HandleInboundAsync(new ExtensionRpcMessage
         Assert.Equal("CQo=", context.Response.Headers["response-bin"]);
         Assert.Equal("Cww=", trailersFeature.Trailers["trailer-bin"]);
         Assert.Equal("0", trailersFeature.Trailers["grpc-status"]);
+        Assert.Equal(start.CallId, activity.GetTagItem("azure.functions.worker_proxy.extension_rpc.call_id"));
+        Assert.Equal(hello.ShardId, activity.GetTagItem("azure.functions.worker_proxy.extension_rpc.stream_id"));
+        Assert.Equal(1, activity.GetTagItem("azure.functions.worker_proxy.extension_rpc.active_calls_at_open"));
+        Assert.Equal(0, activity.GetTagItem("azure.functions.worker_proxy.extension_rpc.active_calls_at_completion"));
+        Assert.Null(activity.GetTagItem("azure.functions.worker_proxy.extension_rpc.call.open.duration_ms"));
+        Assert.Null(activity.GetTagItem("azure.functions.worker_proxy.extension_rpc.call.duration_ms"));
     }
 
     [Fact]
@@ -371,6 +381,7 @@ private static ExtensionGrpcIngress CreateIngress(ExtensionRpcStreamCoordinator
         return new ExtensionGrpcIngress(
             endpoints,
             streamCoordinator,
+            Metrics,
             NullLogger<ExtensionGrpcIngress>.Instance);
     }
 
@@ -417,4 +428,13 @@ private sealed class TestResponseTrailersFeature : IHttpResponseTrailersFeature
     {
         public IHeaderDictionary Trailers { get; set; } = new HeaderDictionary();
     }
+
+    private sealed class TestMeterFactory : IMeterFactory
+    {
+        public Meter Create(MeterOptions options) => new(options);
+
+        public void Dispose()
+        {
+        }
+    }
 }

test/Functions.WorkerProxy.Tests/ExtensionGrpcMetricsTests.cs

@@ -0,0 +1,93 @@
+// Copyright (c) .NET Foundation. All rights reserved.
+// Licensed under the MIT License. See License.txt in the project root for license information.
+
+using System;
+using System.Collections.Concurrent;
+using System.Diagnostics.Metrics;
+using Xunit;
+
+namespace Azure.Functions.WorkerProxy.Tests;
+
+public class ExtensionGrpcMetricsTests
+{
+    [Fact]
+    public void RecordsCallMeasurements()
+    {
+        using var meterFactory = new TestMeterFactory();
+        var metrics = new ExtensionGrpcMetrics(meterFactory);
+        var measurements = new ConcurrentQueue<(string Name, double Value)>();
+        using var listener = new MeterListener
+        {
+            InstrumentPublished = (instrument, meterListener) =>
+            {
+                if (string.Equals(
+                    instrument.Meter.Name,
+                    ExtensionGrpcMetrics.MeterName,
+                    StringComparison.Ordinal))
+                {
+                    meterListener.EnableMeasurementEvents(instrument);
+                }
+            },
+        };
+        listener.SetMeasurementEventCallback<double>(
+            (instrument, value, _, _) => measurements.Enqueue((instrument.Name, value)));
+        listener.SetMeasurementEventCallback<long>(
+            (instrument, value, _, _) => measurements.Enqueue((instrument.Name, value)));
+        listener.Start();
+
+        metrics.CallOpenDuration.Record(12.5);
+        metrics.ActiveCalls.Increment();
+        metrics.CallDuration.Record(25.5);
+        metrics.ActiveCalls.Decrement();
+
+        Assert.Contains(
+            measurements,
+            measurement => string.Equals(
+                measurement.Name,
+                ExtensionGrpcMetrics.CallOpenDurationInstrumentName,
+                StringComparison.Ordinal)
+                && measurement.Value == 12.5);
+        Assert.Contains(
+            measurements,
+            measurement => string.Equals(
+                measurement.Name,
+                ExtensionGrpcMetrics.CallDurationInstrumentName,
+                StringComparison.Ordinal)
+                && measurement.Value == 25.5);
+        Assert.Contains(
+            measurements,
+            measurement => string.Equals(
+                measurement.Name,
+                ExtensionGrpcMetrics.ActiveCallsInstrumentName,
+                StringComparison.Ordinal)
+                && measurement.Value == 1);
+        Assert.Contains(
+            measurements,
+            measurement => string.Equals(
+                measurement.Name,
+                ExtensionGrpcMetrics.ActiveCallsInstrumentName,
+                StringComparison.Ordinal)
+                && measurement.Value == -1);
+    }
+
+    private sealed class TestMeterFactory : IMeterFactory
+    {
+        private readonly ConcurrentBag<Meter> _meters = [];
+
+        public Meter Create(MeterOptions options)
+        {
+            var meter = new Meter(options);
+            _meters.Add(meter);
+
+            return meter;
+        }
+
+        public void Dispose()
+        {
+            foreach (Meter meter in _meters)
+            {
+                meter.Dispose();
+            }
+        }
+    }
+}