feat(telemetry): standardize emitted metric names and add messenger metric

- rename metrics to namespaced keys: flow.etl.rows.*, flow.cache.*,
flow.filesystem.*, flow.phpunit.*, db.client.*
- emit messaging metrics for consumed/sent messages and process duration
- add messenger `metrics` and `metrics_duration_unit` config (s|ms)
- remove messenger `link_to_worker` and the `flow.messenger.worker` span
link
- add histogram bucket boundaries to postgresql duration/row metrics
- document new metrics and add upgrade notes

Author: norberttech

documentation/components/bridges/phpunit-telemetry-bridge.md

@@ -269,10 +269,10 @@ Each span includes attributes like:
 
 When enabled, the extension records:
 
-- `phpunit.suite.duration` - Histogram of suite execution time
-- `phpunit.suite.test_count` - Counter of tests per suite
-- `phpunit.test.duration` - Histogram of individual test execution time
-- `phpunit.test.count` - Counter of tests by status
+- `flow.phpunit.suite.duration` - Histogram of suite execution time
+- `flow.phpunit.suite.test_count` - Counter of tests per suite
+- `flow.phpunit.test.duration` - Histogram of individual test execution time
+- `flow.phpunit.test.count` - Counter of tests by status
 
 ## Running with Docker Compose
 

documentation/components/bridges/symfony-telemetry-bundle.md

@@ -122,11 +122,11 @@ flow_telemetry:
   runtime_mode: auto  # auto|classic|worker
 ```
 
-| Mode      | Behaviour                                                                                          |
-|-----------|----------------------------------------------------------------------------------------------------|
-| `auto`    | Detect the runtime per request and pick `worker` or `classic` accordingly (default).               |
-| `classic` | One process per request: shut telemetry down on terminate (full flush + transport close).          |
-| `worker`  | Long-running runtime: flush on terminate, drain async transports, never shut the transport down.   |
+| Mode      | Behaviour                                                                                        |
+|-----------|--------------------------------------------------------------------------------------------------|
+| `auto`    | Detect the runtime per request and pick `worker` or `classic` accordingly (default).             |
+| `classic` | One process per request: shut telemetry down on terminate (full flush + transport close).        |
+| `worker`  | Long-running runtime: flush on terminate, drain async transports, never shut the transport down. |
 
 `auto` detection looks for the Symfony Runtime worker signal (`APP_RUNTIME_MODE` containing `worker=1`),
 FrankenPHP (`FRANKENPHP_WORKER`), and RoadRunner (`RR_MODE`); when none are present it falls back to `classic`.
@@ -1260,9 +1260,10 @@ flow_telemetry:
   instrumentation:
     messenger:
       enabled: true
-      context_propagation: true  # Propagate context across message boundaries
-      propagation_style: link     # How the consumer span relates to the producer span
-      link_to_worker: true        # Link each message trace back to the messenger:consume worker span
+      context_propagation: true   # Propagate context across message boundaries
+      propagation_style: link      # How the consumer span relates to the producer span
+      metrics: true                # Emit messaging metrics (default true)
+      metrics_duration_unit: s     # process.duration histogram unit: 's' (OTEL semconv, default) or 'ms'
 ```
 
 When `context_propagation` is enabled, `propagation_style` controls how a consumed message's span
@@ -1279,10 +1280,6 @@ relates to the producing (publishing) span:
 `propagation_style` has no effect when `context_propagation` is `false` (there is nothing to relate to).
 The producer side is identical in both modes — the telemetry stamp is always written on dispatch.
 
-`link_to_worker` (default `true`) adds a span link from each consumed message's trace back to the active
-`messenger:consume` worker span, so you can pivot from a message trace to the worker that processed it.
-Set it to `false` to omit the link (e.g. if the consume command is not traced).
-
 Buffered telemetry is flushed **after each handled or failed message** while the worker keeps running, so
 consumer-side traces/logs/metrics export promptly instead of only when the `messenger:consume` worker
 stops. (Without it, signals emitted inside handlers sit in the batching processors — default batch size
@@ -1302,6 +1299,17 @@ to keep that sub-millisecond (see [Timeouts](#timeouts)).
 With the `async_curl` transport, the subscriber also pumps each transport's `tick()` on every `WorkerRunningEvent`, so
 in-flight requests complete in the background.
 
+**Metrics.** 
+
+| Metric                               | Instrument | Unit          | Emitted                       | Attributes                                                                                                                        |
+|--------------------------------------|------------|---------------|-------------------------------|-----------------------------------------------------------------------------------------------------------------------------------|
+| `messaging.client.consumed.messages` | Counter    | `{message}`   | once per **received** message | `messaging.system`, `messaging.operation.name=process`, `messaging.destination.name`, `messaging.consumer.group.name` (transport) |
+| `messaging.process.duration`         | Histogram  | `s` (or `ms`) | per **received** message      | same as above, plus `error.type` when the handler fails                                                                           |
+| `messaging.client.sent.messages`     | Counter    | `{message}`   | once per **sent** message     | `messaging.system`, `messaging.operation.name=send`, `messaging.destination.name`                                                 |
+
+`metrics_duration_unit` controls the `messaging.process.duration` histogram: `s` (default) uses seconds with the
+OTEL-recommended bucket boundaries; `ms` uses milliseconds with Flow's native histogram buckets.
+
 #### Twig
 
 Traces Twig template rendering.
@@ -1376,6 +1384,14 @@ flow_telemetry:
         - '/^cache\.validator.*/'
 ```
 
+Besides spans, the cache instrumentation emits hit/miss counters (a `meter_provider` must be configured for them
+to be exported):
+
+| Metric              | Instrument | Unit         | Emitted             |
+|---------------------|------------|--------------|---------------------|
+| `flow.cache.hits`   | Counter    | `operations` | once per cache hit  |
+| `flow.cache.misses` | Counter    | `operations` | once per cache miss |
+
 ### Web Profiler
 
 Adds a **Flow Telemetry** panel to the Symfony Web Profiler showing the signals captured during the
@@ -1719,10 +1735,12 @@ request continues the trace automatically (`context_propagation` extracts them):
 const headers = {};
 document
     .querySelectorAll('meta[name="traceparent"], meta[name="tracestate"], meta[name="baggage"]')
-    .forEach((meta) => { headers[meta.name] = meta.content; });
+    .forEach((meta) => {
+        headers[meta.name] = meta.content;
+    });
 
 // Only send to your own API origins so trace IDs do not leak to third parties.
-fetch('/api/orders', { headers });
+fetch('/api/orders', {headers});
 ```
 
 **Links / multi-step flows (query string).** A normal navigation cannot send headers, so carry the

documentation/components/libs/postgresql/client-telemetry.md

@@ -198,14 +198,14 @@ Query text is automatically truncated to `maxQueryLength` (default: 1000 charact
 
 When `collectMetrics` is enabled, two histograms are recorded:
 
-### `operation_duration`
+### `db.client.operation.duration`
 
 Duration of database operations in seconds.
 
 - **Unit**: `s` (seconds)
 - **Attributes**: `db.system.name`, `db.namespace`, `db.operation.name`, `server.address`
 
-### `response_returned_rows`
+### `db.client.response.returned_rows`
 
 Number of rows returned by database operations.
 

documentation/upgrading.md

@@ -78,7 +78,9 @@ subsequent spans in the same trace.
 
 `continue` mode still joins the producer's trace.
 
-### 5) `flow-php/telemetry-otlp-bridge`, `flow-php/symfony-telemetry-bundle` - curl is synchronous; async moved to `async_curl`
+### 5) `flow-php/telemetry-otlp-bridge`, `flow-php/symfony-telemetry-bundle` - curl is synchronous; async moved to
+
+`async_curl`
 
 | Before                                                         | After                                                                                           |
 |----------------------------------------------------------------|-------------------------------------------------------------------------------------------------|
@@ -107,12 +109,58 @@ flow_postgresql:
 
 ### 7) `flow-php/symfony-telemetry-bundle` - static resource cache file is keyed by kernel environment
 
-| Before                                                  | After                                                              |
-|---------------------------------------------------------|-------------------------------------------------------------------|
-| `sys_get_temp_dir()/flow_telemetry_resource.cache`      | `sys_get_temp_dir()/flow_telemetry_resource_<kernel.env>.cache`   |
+| Before                                             | After                                                           |
+|----------------------------------------------------|-----------------------------------------------------------------|
+| `sys_get_temp_dir()/flow_telemetry_resource.cache` | `sys_get_temp_dir()/flow_telemetry_resource_<kernel.env>.cache` |
 
 Delete the orphaned `flow_telemetry_resource.cache` from the temp dir; a per-env file is written on the next run.
 
+### 8) `flow-php/symfony-telemetry-bundle` - removed messenger `link_to_worker` and the `flow.messenger.worker` link
+
+| Before                                           | After   |
+|--------------------------------------------------|---------|
+| `instrumentation.messenger.link_to_worker: true` | removed |
+| consumer span link `flow.messenger.worker`       | removed |
+
+Remove the `link_to_worker` key from config — there is no replacement. Consumed messages remain their own trace
+linked to the producer (`link` mode). To keep a worker's trace list clean, give the Messenger transport a
+telemetry-disabled connection and/or exclude `messenger:consume` via `instrumentation.console.exclude_commands`.
+
+### 9) `flow-php/symfony-telemetry-bundle` - messenger instrumentation also emits messaging metrics
+
+| Before                                       | After                                                                                                           |
+|----------------------------------------------|-----------------------------------------------------------------------------------------------------------------|
+| messenger instrumentation emitted spans only | also emits `messaging.client.consumed.messages`, `messaging.client.sent.messages`, `messaging.process.duration` |
+
+Disable with `instrumentation.messenger.metrics: false`; set the `messaging.process.duration` unit via
+`instrumentation.messenger.metrics_duration_unit` (`s` default, or `ms`).
+
+### 10) `flow-php/etl`, `flow-php/filesystem`, `flow-php/postgresql`, `flow-php/symfony-telemetry-bundle`,
+`flow-php/phpunit-telemetry-bridge` - emitted metric names standardized
+
+| Before                      | After                              |
+|-----------------------------|------------------------------------|
+| `rows_processed`            | `flow.etl.rows.processed`          |
+| `rows_throughput`           | `flow.etl.rows.throughput`         |
+| `cache_hits`                | `flow.cache.hits`                  |
+| `cache_misses`              | `flow.cache.misses`                |
+| `cache.hits`                | `flow.cache.hits`                  |
+| `cache.misses`              | `flow.cache.misses`                |
+| `write_size`                | `flow.filesystem.write.size`       |
+| `write_operations`          | `flow.filesystem.write.operations` |
+| `read_size`                 | `flow.filesystem.read.size`        |
+| `read_operations`           | `flow.filesystem.read.operations`  |
+| `operation_duration`        | `db.client.operation.duration`     |
+| `response_returned_rows`    | `db.client.response.returned_rows` |
+| `phpunit.test.duration`     | `flow.phpunit.test.duration`       |
+| `phpunit.test.count`        | `flow.phpunit.test.count`          |
+| `phpunit.test.memory.peak`  | `flow.phpunit.test.memory.peak`    |
+| `phpunit.test.memory.delta` | `flow.phpunit.test.memory.delta`   |
+| `phpunit.suite.duration`    | `flow.phpunit.suite.duration`      |
+| `phpunit.suite.test_count`  | `flow.phpunit.suite.test_count`    |
+
+Rename these series in dashboards and alerts.
+
 ---
 
 ## Upgrading from 0.39.x to 0.40.x

src/bridge/phpunit/telemetry/src/Flow/Bridge/PHPUnit/Telemetry/Subscriber/TestFinishedSubscriber.php

@@ -43,13 +43,13 @@ public function notify(Finished $event): void
                 if ($this->config->emitMetrics) {
                     $meter = $this->telemetry->meter('phpunit', PackageVersion::get('phpunit/phpunit'));
 
-                    $meter->createCounter('phpunit.test.count')->add(1, ['test.status' => $status]);
-                    $meter->createHistogram('phpunit.test.memory.peak', 'bytes')->record($peakBytes, [
+                    $meter->createCounter('flow.phpunit.test.count')->add(1, ['test.status' => $status]);
+                    $meter->createHistogram('flow.phpunit.test.memory.peak', 'bytes')->record($peakBytes, [
                         'test.status' => $status,
                     ]);
 
                     if ($deltaBytes !== null) {
-                        $meter->createHistogram('phpunit.test.memory.delta', 'bytes')->record($deltaBytes, [
+                        $meter->createHistogram('flow.phpunit.test.memory.delta', 'bytes')->record($deltaBytes, [
                             'test.status' => $status,
                         ]);
                     }
@@ -98,19 +98,19 @@ public function notify(Finished $event): void
                 $meter = $this->telemetry->meter('phpunit', PackageVersion::get('phpunit/phpunit'));
 
                 if ($duration !== null) {
-                    $meter->createHistogram('phpunit.test.duration', 'ms')->record($duration, [
+                    $meter->createHistogram('flow.phpunit.test.duration', 'ms')->record($duration, [
                         'test.status' => $status,
                     ]);
 
-                    $meter->createCounter('phpunit.test.count')->add(1, ['test.status' => $status]);
+                    $meter->createCounter('flow.phpunit.test.count')->add(1, ['test.status' => $status]);
                 }
 
-                $meter->createHistogram('phpunit.test.memory.peak', 'bytes')->record($peakBytes, [
+                $meter->createHistogram('flow.phpunit.test.memory.peak', 'bytes')->record($peakBytes, [
                     'test.status' => $status,
                 ]);
 
                 if ($deltaBytes !== null) {
-                    $meter->createHistogram('phpunit.test.memory.delta', 'bytes')->record($deltaBytes, [
+                    $meter->createHistogram('flow.phpunit.test.memory.delta', 'bytes')->record($deltaBytes, [
                         'test.status' => $status,
                     ]);
                 }

src/bridge/phpunit/telemetry/src/Flow/Bridge/PHPUnit/Telemetry/Subscriber/TestSuiteFinishedSubscriber.php

@@ -55,11 +55,11 @@ public function notify(Finished $event): void
                 if ($this->config->emitMetrics && $duration !== null) {
                     $meter = $this->telemetry->meter('phpunit', $phpunitVersion);
 
-                    $meter->createHistogram('phpunit.suite.duration', 'ms')->record($duration, [
+                    $meter->createHistogram('flow.phpunit.suite.duration', 'ms')->record($duration, [
                         'test.suite' => $suiteName,
                     ]);
 
-                    $meter->createCounter('phpunit.suite.test_count')->add($suite->count(), [
+                    $meter->createCounter('flow.phpunit.suite.test_count')->add($suite->count(), [
                         'test.suite' => $suiteName,
                     ]);
                 }

src/bridge/phpunit/telemetry/tests/Flow/Bridge/PHPUnit/Telemetry/Tests/Unit/Subscriber/TestFinishedSubscriberTest.php

@@ -123,8 +123,8 @@ public function test_records_memory_metrics_when_metrics_enabled(): void
         $subscriber->notify($event);
         $telemetry->flush();
 
-        static::assertNotEmpty($metricProcessor->metricsWithName('phpunit.test.memory.peak'));
-        static::assertNotEmpty($metricProcessor->metricsWithName('phpunit.test.memory.delta'));
+        static::assertNotEmpty($metricProcessor->metricsWithName('flow.phpunit.test.memory.peak'));
+        static::assertNotEmpty($metricProcessor->metricsWithName('flow.phpunit.test.memory.delta'));
     }
 
     public function test_records_memory_delta_metric_when_only_metrics_enabled(): void
@@ -144,8 +144,8 @@ public function test_records_memory_delta_metric_when_only_metrics_enabled(): vo
         $subscriber->notify($event);
         $telemetry->flush();
 
-        static::assertNotEmpty($metricProcessor->metricsWithName('phpunit.test.memory.peak'));
-        static::assertNotEmpty($metricProcessor->metricsWithName('phpunit.test.memory.delta'));
+        static::assertNotEmpty($metricProcessor->metricsWithName('flow.phpunit.test.memory.peak'));
+        static::assertNotEmpty($metricProcessor->metricsWithName('flow.phpunit.test.memory.delta'));
     }
 
     public function test_sets_error_status_and_message_on_span_for_failed_test(): void