From a1118077319cfda52b95ee498c858b4936ab466d Mon Sep 17 00:00:00 2001
From: Kritharth Shetty <kritharth16@gmail.com>
Date: Thu, 3 Sep 2026 19:03:39 +0530
Subject: [PATCH] [feat][fn] Expose the Prometheus metrics registry to Go
 functions via FunctionContext

Go functions have exactly one way to emit a custom metric:
FunctionContext.RecordMetric(name, value), which funnels every value into a
single fixed-shape SummaryVec (pulsar_function_user_metric). There's no way
to register a Counter, Gauge, Histogram, or any other prometheus.Collector,
and the registry the SDK already runs and serves on the metrics port has no
exported accessor.

This adds FunctionContext.GetMetricsRegistry() prometheus.Registerer,
returning the existing package-level registry typed as the Registerer
interface (Register/MustRegister/Unregister, not Gather), so a function can
register its own collectors alongside the SDK's. RecordMetric and
userMetricSummary are unchanged; typed convenience helpers (Option B from
the issue) are left as a follow-on.

Covered by two new tests in stats_test.go: registering and scraping a
custom Counter, and the collision behavior on a name colliding with a
built-in metric (Register errors, MustRegister panics).

Fixes #26403
---
 pulsar-function-go/pf/context.go    | 11 +++++++
 pulsar-function-go/pf/stats_test.go | 49 +++++++++++++++++++++++++++++
 2 files changed, 60 insertions(+)

diff --git a/pulsar-function-go/pf/context.go b/pulsar-function-go/pf/context.go
index 69959cb3c1ac9..6408e967d6d1f 100644
--- a/pulsar-function-go/pf/context.go
+++ b/pulsar-function-go/pf/context.go
@@ -202,6 +202,17 @@ func (c *FunctionContext) RecordMetric(metricName string, metricValue float64) {
 	v.(prometheus.Observer).Observe(metricValue)
 }
 
+// GetMetricsRegistry returns the Prometheus registry that backs the function's
+// metrics endpoint, so a function can register additional collectors (counters,
+// gauges, histograms, ...) that are then exposed on the same endpoint alongside
+// the SDK's own metrics. Metric names prefixed with "pulsar_function_" are
+// reserved for SDK metrics. A collector whose fully-qualified name collides with
+// an already-registered metric fails to register: Register returns an error,
+// while MustRegister panics.
+func (c *FunctionContext) GetMetricsRegistry() prometheus.Registerer {
+	return reg
+}
+
 // An unexported type to be used as the key for types in this package. This
 // prevents collisions with keys defined in other packages.
 type key struct{}
diff --git a/pulsar-function-go/pf/stats_test.go b/pulsar-function-go/pf/stats_test.go
index 138dc91cd9cd3..c6ac272830eca 100644
--- a/pulsar-function-go/pf/stats_test.go
+++ b/pulsar-function-go/pf/stats_test.go
@@ -216,3 +216,52 @@ func TestInstanceControlMetrics(t *testing.T) {
 		assert.EqualValuesf(t, value+1, metrics.UserMetrics[label], "user metric %s != %d", label, value+1)
 	}
 }
+
+func TestGetMetricsRegistry_CustomCollector(t *testing.T) {
+	gi := newGoInstance()
+	metricsServicer := NewMetricsServicer(gi)
+	metricsServicer.serve()
+
+	// A Counter is something the user_metric Summary cannot express.
+	customCounter := prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "pulsar_function_go_test_custom_counter_total",
+		Help: "A user-registered counter exposed via FunctionContext.GetMetricsRegistry.",
+	})
+	err := gi.context.GetMetricsRegistry().Register(customCounter)
+	assert.NoError(t, err)
+	defer gi.context.GetMetricsRegistry().Unregister(customCounter)
+	customCounter.Add(42)
+
+	time.Sleep(time.Second * 1)
+	resp, err := http.Get(fmt.Sprintf("http://localhost:%d/metrics", gi.context.GetMetricsPort()))
+	assert.Equal(t, nil, err)
+	assert.NotEqual(t, nil, resp)
+	assert.Equal(t, 200, resp.StatusCode)
+	body, err := io.ReadAll(resp.Body)
+	assert.Equal(t, nil, err)
+	assert.Containsf(t, string(body), "\npulsar_function_go_test_custom_counter_total 42\n",
+		"custom collector should be exposed on /metrics")
+	resp.Body.Close()
+
+	gi.close()
+	metricsServicer.close()
+}
+
+func TestGetMetricsRegistry_NameCollision(t *testing.T) {
+	gi := newGoInstance()
+
+	// Collides with the built-in received_total gauge registered on reg in init().
+	colliding := prometheus.NewCounter(prometheus.CounterOpts{
+		Name: PulsarFunctionMetricsPrefix + TotalReceived,
+		Help: "collector whose name collides with a built-in SDK metric",
+	})
+
+	// Register is the documented safe path: it returns an error, it does not panic.
+	err := gi.context.GetMetricsRegistry().Register(colliding)
+	assert.Error(t, err)
+
+	// MustRegister panics on the same collision.
+	assert.Panics(t, func() {
+		gi.context.GetMetricsRegistry().MustRegister(colliding)
+	})
+}