[Go Functions] No way to register custom metric collectors: every user metric is a summary

### Search before reporting

- [X] I searched in the [issues](https://github.com/apache/pulsar/issues) and found nothing similar that is still open for the Go runtime.

**Prior art:**

- **#9772 "[Go Functions] Allow user metrics"** — closed as *completed* in February 2022, and the change it asked for is what added `FunctionContext.RecordMetric`. It opened by observing that "the registry is inaccessible to the user" and proposed `recordMetric` as a first step. The first step landed; the observation still holds. This issue is the remaining half.
- **#24853 "Expose FunctionCollectorRegistry through Context API for custom metrics"** — the same gap for the **Java** runtime. Its proposed API is Java-specific (`public interface Context` returning a `CollectorRegistry`) and its body does not mention Go, so a fix there would not reach a Go function. Filed separately rather than as a comment on that issue because the two runtimes share no code here and the Go surface would be a different API.

### Motivation

A Go function has exactly one way to emit a custom metric:

```go
// pulsar-function-go/pf/context.go:189
func (c *FunctionContext) RecordMetric(metricName string, metricValue float64)
```

Every value passed to it is observed into a single `SummaryVec` with fixed quantile objectives (`pulsar-function-go/pf/stats.go:130-141`):

```go
userMetricSummary = prometheus.NewSummaryVec(
prometheus.SummaryOpts{
Name: PulsarFunctionMetricsPrefix + UserMetric,
Help: "User defined metric.",
Objectives: map[float64]float64{0.5: 0.01, 0.9: 0.01, 0.99: 0.01, 0.999: 0.01},
}, userMetricLabelNames)
```

So every user metric is a summary, whatever it actually represents. A monotonically increasing count of processed records becomes a quantile distribution of the value `1`. A gauge of queue depth becomes a summary whose quantiles are meaningless. A latency histogram cannot be expressed at all, and neither can a metric with custom labels — the label set is fixed to tenant/namespace/name/instance/cluster plus the metric name.

The registry those collectors are registered into is a package-level unexported variable:

```go
// pulsar-function-go/pf/stats.go:148
var reg *prometheus.Registry
```

`FunctionContext` exposes `GetMetricsPort()` but no accessor for `reg`, and user function code lives outside package `pf`, so there is no supported way to register a collector of one's own. The workaround — running a second `promhttp` listener on another port from inside the function — means a second scrape target per instance that the function worker knows nothing about, which defeats the point of the instance already exposing one.

This is not a Go-only limitation in kind — Java's `BaseContext.recordMetric(String, double)` and Python's `record_metric` have the same single-summary shape. But the *consequence* differs sharply by runtime, and Go is the one where it actually bites.

#### A Python function can already do this; a Go function cannot

The Python instance serves `prometheus_client`'s process-global registry:

```python
# prometheus_client_fix.py:50
def start_http_server(port, addr='', registry=core.REGISTRY):

# python_instance_main.py:307 - called with the default, i.e. core.REGISTRY
prometheus_client_fix.start_http_server(args.metrics_port)
```

and `ContextImpl` registers its own summary into that same global by omitting the `registry` argument (`contextimpl.py:67`). Since `prometheus_client` defaults every collector to `core.REGISTRY`, a Python function author gets custom collectors today with no SDK change at all:

```python
from prometheus_client import Counter

orders = Counter('my_orders_total', 'Orders processed', ['region'])

def process(input, context):
orders.labels(region='us-east').inc()
```

That works because the Python client library has a process-global registry and the instance happens to serve exactly it. `prometheus/client_golang` has no equivalent global that the instance serves — `pulsar-function-go` creates a private `prometheus.NewRegistry()` and keeps it in an unexported package variable — so the same three lines are impossible in Go.

So the practical state is:

| Runtime | Registry the instance serves | Reachable from user code? |
| --- | --- | --- |
| Python | `prometheus_client.core.REGISTRY` (library global) | **Yes** — any collector, any type, custom labels |
| Go | `var reg *prometheus.Registry`, unexported in package `pf` | **No** |
| Java | internal `FunctionCollectorRegistry` | **No** — #24853 |

Two caveats on the Python path, since it holds by convention rather than by contract: it is undocumented, so nothing stops it being changed without that counting as a break; and the instance pins a patched `prometheus_client` (`python_instance_main.py:302-306`, carrying prometheus/client_python#356 for a thread leak). Neither affects registration, but neither is a guarantee either.

This is a parity argument rather than a novelty one: the capability already exists in one runtime, and the Go SDK is the reason it is unavailable in another. The fix has to be Go-shaped — the runtimes share no code here, and #24853 proposes a Java type with no Go equivalent.

### Solution

Expose the runtime's Prometheus registry through `FunctionContext`, so a function can register its own collectors alongside the ones the SDK maintains and have them served on the existing metrics endpoint:

```go
// GetMetricsRegistry returns the Prometheus registry the instance serves on its metrics port,
// so a function can register collectors of its own.
func (c *FunctionContext) GetMetricsRegistry() prometheus.Registerer
```

Returning `prometheus.Registerer` rather than `*prometheus.Registry` keeps the gather side out of the user's reach while allowing `MustRegister`/`Register`/`Unregister`, which is the whole requirement.

Points worth settling in review:

1. **Collision with SDK metric names.** A user registering a collector named `pulsar_function_user_metric` would fail `MustRegister` at runtime. Whether to wrap the registry in one that rejects the `pulsar_function_` prefix, or simply document it, is a design call.
2. **Cardinality.** Custom labels are the main reason to want this, and also the main way to melt a Prometheus server. Worth a documentation note at least.
3. **Whether `Registerer` is enough**, or whether the ask is really a set of typed helpers (`NewCounter`, `NewGauge`, `NewHistogram`) that pre-apply the standard function labels. The typed-helper shape is friendlier and harder to misuse; the registry shape is smaller and composes with existing Prometheus code the user already has.
4. **Alignment with #24853**, if that lands for Java — the two need not share an API, but they should not contradict each other on the questions above.

### Alternatives

- **Run a separate `promhttp` handler inside the function.** Works today, and is what people do, but it adds a scrape target per instance outside the worker's knowledge and duplicates the listener the instance already runs.
- **Encode structure into the metric name** (`orders_processed_us_east`) to fake labels. Unbounded name cardinality, and no way to aggregate.
- **Wait for OpenTelemetry** (#25885, which asks for the status of Functions OTel support generally). Reasonable if OTel is close, but it is an open question with no owner and would be a larger change; this is a small addition to an existing, working metrics path.

### Anything else?

Happy to submit a PR. I would want a steer on point 3 first — `Registerer` versus typed helpers — since that decides the shape of the change.

### Are you willing to submit a PR?

- [X] I'm willing to submit a PR!

Start by reading pulsar-function-go/pf/context.go around RecordMetric and GetMetricsPort, then inspect pulsar-function-go/pf/stats.go around the private registry and user metric collector. Review how the existing metrics endpoint gathers from that registry. Done means an agreed API lets Go functions register custom collectors on the existing endpoint, with the collision and cardinality behavior documented or covered by tests.

Written by the indexing model from the issue text.