[1.0.0] Metrics follow-up.

composer.json

@@ -17,7 +17,7 @@
   "require": {
     "php": ">=8.2",
     "freedsx/asn1": "dev-main#b57a38f",
-    "freedsx/socket": "dev-main#610bf80",
+    "freedsx/socket": "dev-main#287974b",
     "freedsx/sasl": "dev-main#85e1ef9",
     "psr/log": "^3"
   },

docs/Server/Monitoring.md

@@ -53,9 +53,17 @@ off rather than returned empty.
 | `connectionsTotal` | Connections accepted since start. |
 | `connectionsRejected` | Connections turned away at the connection limit. |
 | `connectionsWriteTimeouts`, `connectionsIdleTimeouts` | Connections closed by the write or idle timeout. |
+| `connectionsRequestSizeExceeded` | Connections dropped because a request exceeded `setMaxRequestSize`. |
 | `connectionsMax` | The configured connection limit (`0` is unlimited). |
 | `operationsCompleted`, `operationsFailed` | Total operations and the failed subset. |
 | `operationsByType` | Per-type counts, e.g. `search=1402, bind=210, add=8`. |
+| `operationsByResultCode` | Per-result-code counts, e.g. `0=1610, 49=3, 32=5`. |
+| `operationsAvgLatencyMsByType` | Mean latency in milliseconds per type, e.g. `search=1.83, bind=0.42`. |
+| `bindsByMethod` | Bind counts split by method, e.g. `anonymous=12, simple=200, sasl=8`. |
+| `searchesByScope` | Search counts split by scope, e.g. `base=4, one=10, sub=1388`. |
+| `operationsInProgressByType` | Operations currently executing, per type. Coroutine runner only (see [Runner Differences](#runner-differences)). |
+| `trafficBytesSent`, `trafficBytesReceived` | LDAP protocol bytes written to and read from clients. |
+| `trafficEntriesReturned` | Search-result entries returned to clients. |
 
 Counters are monotonic since start and are never reset, so sample and diff them to get rates. A restart starts them over.
 
@@ -67,13 +75,17 @@ authenticated-only. To restrict it further, add `RuleBasedAccessControl` rules t
 
 ## Runner Differences
 
-* **Swoole** runs in one process, so `cn=monitor` is fully live.
-* **PCNTL** forks per connection. Connection gauges are authoritative, and operation counts stay current to within about
-  one accept cycle (`setSocketAcceptTimeout`), including on long-lived connections. They are best-effort: a forcibly
-  killed worker may lose its most recent operations.
+**Swoole** (single process):
 
-Under PCNTL the monitor data is published to a JSON file, by default under the system temp directory keyed by port. Set
-`setMonitorSnapshotPath()` to relocate it or to avoid collisions when running several instances on one host.
+* `cn=monitor` is fully live.
+* `operationsInProgressByType` is reported.
+
+**PCNTL** (forks per connection):
+
+* Connection gauges are authoritative.
+* Operation counts, traffic totals, and breakdowns are best-effort and current to within about one accept cycle.
+* `operationsInProgressByType` is omitted: it is a per-child gauge the parent serving `cn=monitor` cannot aggregate.
+* Monitor data is published to a JSON file, by default under the system temp directory keyed by port; set `setMonitorSnapshotPath()` to relocate it or avoid collisions across instances.
 
 For per-operation aggregation that survives saturation or spans instances, prefer a push exporter.
 
@@ -85,8 +97,9 @@ Provide any `MetricsRecorderInterface` to receive events out-of-band:
 $options->setMetricsRecorder($myRecorder);
 ```
 
-It is notified of each operation (`operationObserved`), connection lifecycle event (`connectionObserved`), server start,
-and config reload. The recorder and `cn=monitor` are independent, so the two options compose:
+It is notified of each operation start (`operationStarted`) and completion (`operationObserved`), transport traffic
+(`trafficObserved`), connection lifecycle event (`connectionObserved`), server start, and config reload. The recorder and
+`cn=monitor` are independent, so the two options compose:
 
 | `setMonitorEnabled` | `setMetricsRecorder` | Result |
 | --- | --- | --- |

src/FreeDSx/Ldap/Protocol/LdapQueue.php

@@ -127,13 +127,16 @@ protected function unwrap(string $bytes): Buffer
     protected function decode(string $bytes): Message
     {
         try {
-            return parent::decode($bytes);
+            $message = parent::decode($bytes);
         } catch (PduLengthException $e) {
             throw new RequestSizeExceededException(
                 $e->getMessage(),
                 previous: $e,
             );
         }
+        $this->onMessageDecoded($message);
+
+        return $message;
     }
 
     /**
@@ -152,6 +155,7 @@ protected function sendLdapMessage(iterable $messages): static
 
         foreach ($messages as $message) {
             $encoded = $this->encoder->encode($message->toAsn1());
+            $this->onMessageEncoded($encoded);
             $buffer .= $this->messageWrapper !== null ? $this->messageWrapper->wrap($encoded) : $encoded;
             $bufferLen = strlen($buffer);
             if ($bufferLen >= self::BUFFER_SIZE) {
@@ -166,6 +170,16 @@ protected function sendLdapMessage(iterable $messages): static
         return $this;
     }
 
+    /**
+     * Extension point invoked with each message's encoded bytes as it is sent.
+     */
+    protected function onMessageEncoded(string $encoded): void {}
+
+    /**
+     * Extension point invoked with each message decoded off the socket.
+     */
+    protected function onMessageDecoded(Message $message): void {}
+
     public function isConnected(): bool
     {
         return $this->socket->isConnected();

src/FreeDSx/Ldap/Protocol/Queue/Response/MetricsResponseInterceptor.php

@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * This file is part of the FreeDSx LDAP package.
+ *
+ * (c) Chad Sikorra <Chad.Sikorra@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace FreeDSx\Ldap\Protocol\Queue\Response;
+
+use FreeDSx\Ldap\Operation\Response\SearchResultEntry;
+use FreeDSx\Ldap\Protocol\LdapMessageResponse;
+use FreeDSx\Ldap\Server\Metrics\MetricsRecorderInterface;
+use FreeDSx\Ldap\Server\Metrics\Observation\TrafficObservation;
+
+/**
+ * Counts search-result entries as they stream out to the client.
+ *
+ * @author Chad Sikorra <Chad.Sikorra@gmail.com>
+ */
+final readonly class MetricsResponseInterceptor implements ResponseInterceptor
+{
+    public function __construct(private MetricsRecorderInterface $recorder) {}
+
+    public function intercept(LdapMessageResponse $response): LdapMessageResponse
+    {
+        if ($response->getResponse() instanceof SearchResultEntry) {
+            $this->recorder->trafficObserved(new TrafficObservation(entriesReturned: 1));
+        }
+
+        return $response;
+    }
+}

src/FreeDSx/Ldap/Protocol/Queue/ServerQueue.php

@@ -27,11 +27,16 @@
 use FreeDSx\Ldap\Protocol\LdapMessageResponse;
 use FreeDSx\Ldap\Protocol\LdapQueue;
 use FreeDSx\Ldap\Protocol\Queue\Response\ResponseInterceptor;
+use FreeDSx\Ldap\Server\Metrics\MetricsRecorderInterface;
+use FreeDSx\Ldap\Server\Metrics\Observation\TrafficObservation;
+use FreeDSx\Ldap\Server\Metrics\Recorder\NullMetricsRecorder;
 use FreeDSx\Socket\Exception\ConnectionException;
 use FreeDSx\Socket\Queue\Message;
 use FreeDSx\Socket\Socket;
 use Generator;
 
+use function strlen;
+
 /**
  * The LDAP Queue class for sending and receiving messages for servers.
  *
@@ -57,6 +62,7 @@ public function __construct(
         ?EncoderInterface $encoder = null,
         int $maxReceiveSize = 0,
         array $interceptors = [],
+        private readonly MetricsRecorderInterface $metricsRecorder = new NullMetricsRecorder(),
     ) {
         parent::__construct(
             $socket,
@@ -66,6 +72,32 @@ public function __construct(
         $this->interceptors = $interceptors;
     }
 
+    /**
+     * Count the LDAP bytes received for a decoded request.
+     *
+     * @throws ProtocolException
+     */
+    protected function decode(string $bytes): Message
+    {
+        $message = parent::decode($bytes);
+
+        $this->metricsRecorder->trafficObserved(new TrafficObservation(
+            bytesReceived: (int) $message->getLastPosition(),
+        ));
+
+        return $message;
+    }
+
+    /**
+     * Count the LDAP bytes sent for each outgoing response.
+     */
+    protected function onMessageEncoded(string $encoded): void
+    {
+        $this->metricsRecorder->trafficObserved(new TrafficObservation(
+            bytesSent: strlen($encoded),
+        ));
+    }
+
     /**
      * @throws ProtocolException
      * @throws UnsolicitedNotificationException

src/FreeDSx/Ldap/Protocol/ServerProtocolHandler.php

@@ -16,6 +16,7 @@
 use FreeDSx\Asn1\Exception\EncoderException;
 use FreeDSx\Ldap\Exception\OperationException;
 use FreeDSx\Ldap\Exception\ProtocolException;
+use FreeDSx\Ldap\Exception\RequestSizeExceededException;
 use FreeDSx\Ldap\Exception\RequestValidationException;
 use FreeDSx\Ldap\Exception\ResponseAlreadySentException;
 use FreeDSx\Ldap\Operation\Response\ExtendedResponse;
@@ -26,9 +27,11 @@
 use FreeDSx\Ldap\Server\Logging\EventContext;
 use FreeDSx\Ldap\Server\Logging\EventLogger;
 use FreeDSx\Ldap\Server\Logging\ServerEvent;
+use FreeDSx\Ldap\Server\Metrics\Observation\ConnectionObservation;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\MiddlewareHandlerInterface;
 use FreeDSx\Ldap\Server\Middleware\Pipeline\ServerRequestContext;
 use FreeDSx\Socket\Exception\ConnectionException;
+use FreeDSx\Socket\Exception\IdleTimeoutException;
 use FreeDSx\Socket\Exception\WriteTimeoutException;
 use Throwable;
 
@@ -50,10 +53,14 @@ public function __construct(
     /**
      * Listens for messages from the socket and handles the responses/actions needed.
      *
+     * @return ?ConnectionObservation The connection-timeout that ended the session, or null for a normal close.
+     *
      * @throws EncoderException
      */
-    public function handle(): void
+    public function handle(): ?ConnectionObservation
     {
+        $closeReason = null;
+
         try {
             while ($message = $this->queue->getMessage()) {
                 $this->dispatchRequest($message);
@@ -72,12 +79,27 @@ public function handle(): void
                 ServerEvent::WriteTimeout,
                 [EventContext::REASON_MESSAGE => $e->getMessage()],
             );
+            $closeReason = ConnectionObservation::WriteTimeout;
+        } catch (IdleTimeoutException $e) {
+            # The client sent nothing within the read timeout. Record it and close; there is nothing to send back.
+            $this->eventLogger->record(
+                ServerEvent::IdleTimeout,
+                [EventContext::REASON_MESSAGE => $e->getMessage()],
+            );
+            $closeReason = ConnectionObservation::IdleTimeout;
         } catch (ConnectionException) {
             # Connection closure is recorded by the runner's lifecycle logging; no audit event for normal client disconnects.
+        } catch (RequestSizeExceededException $e) {
+            # The client sent a PDU larger than the configured maximum. Per RFC 4511 §4.1.1 answer with a Notice of
+            # Disconnection, passing the cause so the log identifies the size violation, then record it and close.
+            $this->sendNoticeOfDisconnect(
+                $e->getMessage(),
+                cause: $e,
+            );
+            $closeReason = ConnectionObservation::RequestSizeExceeded;
         } catch (EncoderException|ProtocolException) {
-            # Per RFC 4511 §4.1.1, a PDU that cannot be processed — malformed, or rejected for exceeding the configured
-            # size cap (RequestSizeExceededException) — warrants a disconnect with a protocol error. The
-            # NoticeOfDisconnectSent event records the specific reason.
+            # Per RFC 4511 §4.1.1, a PDU that cannot be processed (malformed) warrants a disconnect with a protocol
+            # error. The NoticeOfDisconnectSent event records the specific reason.
             $this->sendNoticeOfDisconnect('The message could not be processed.');
         } catch (Throwable $e) {
             if ($this->queue->isConnected()) {
@@ -88,6 +110,8 @@ public function handle(): void
                 $this->queue->close();
             }
         }
+
+        return $closeReason;
     }
 
     /**

src/FreeDSx/Ldap/Protocol/ServerProtocolHandler/ServerMonitorHandler.php

@@ -24,6 +24,7 @@
 use FreeDSx\Ldap\Server\Metrics\Snapshot\MetricsSnapshot;
 use FreeDSx\Ldap\Server\Operation\OperationOutcomeResult;
 use FreeDSx\Ldap\Server\Operation\OperationResult;
+use FreeDSx\Ldap\Server\ServerRunner\CoroutineServerRunnerInterface;
 use FreeDSx\Ldap\Server\ServerRunner\PcntlServerRunner;
 use FreeDSx\Ldap\Server\ServerRunner\SwooleServerRunner;
 use FreeDSx\Ldap\Server\Token\TokenInterface;
@@ -81,6 +82,7 @@ private function attributes(MetricsSnapshot $snapshot): array
         $lifecycle = $snapshot->lifecycle;
         $connections = $snapshot->connections;
         $operations = $snapshot->operations;
+        $traffic = $snapshot->traffic;
 
         return array_filter([
             'objectClass' => ['top', 'extensibleObject'],
@@ -97,13 +99,79 @@ private function attributes(MetricsSnapshot $snapshot): array
             'connectionsRejected' => [(string) $connections->rejected],
             'connectionsWriteTimeouts' => [(string) $connections->writeTimeouts],
             'connectionsIdleTimeouts' => [(string) $connections->idleTimeouts],
+            'connectionsRequestSizeExceeded' => [(string) $connections->requestSizeExceeded],
             'connectionsMax' => [(string) $this->options->getMaxConnections()],
             'operationsCompleted' => [(string) $operations->total()],
             'operationsFailed' => [(string) $operations->totalErrors()],
-            'operationsByType' => $this->operationsByType($operations->counts),
+            'operationsByType' => $this->formatCounts($operations->counts),
+            'operationsByResultCode' => $this->formatCounts($operations->resultCodeCounts),
+            'bindsByMethod' => $this->formatCounts($operations->bindCounts),
+            'searchesByScope' => $this->formatCounts($operations->searchScopeCounts),
+            'operationsAvgLatencyMsByType' => $this->avgLatencyMsByType(
+                $operations->counts,
+                $operations->durationSeconds,
+            ),
+            'operationsInProgressByType' => $this->operationsInProgress($snapshot->operationsInProgress),
+            'trafficBytesSent' => [(string) $traffic->bytesSent],
+            'trafficBytesReceived' => [(string) $traffic->bytesReceived],
+            'trafficEntriesReturned' => [(string) $traffic->entriesReturned],
         ]);
     }
 
+    /**
+     * The in-flight gauge is only meaningful under the single-process Swoole runner.
+     *
+     * @param array<string, int> $inProgress
+     * @return list<string>
+     */
+    private function operationsInProgress(array $inProgress): array
+    {
+        if (!$this->isCoroutineRunner()) {
+            return [];
+        }
+
+        return $this->formatCounts($inProgress);
+    }
+
+    private function isCoroutineRunner(): bool
+    {
+        $runner = $this->options->getServerRunner();
+
+        if ($runner !== null) {
+            return $runner instanceof CoroutineServerRunnerInterface;
+        }
+
+        return $this->options->getUseSwooleRunner();
+    }
+
+    /**
+     * Mean latency per operation type in milliseconds, derived from the summed duration and count.
+     *
+     * @param array<string, int> $counts
+     * @param array<string, float> $durationSeconds
+     * @return list<string>
+     */
+    private function avgLatencyMsByType(
+        array $counts,
+        array $durationSeconds,
+    ): array {
+        $values = [];
+
+        foreach ($counts as $operation => $count) {
+            if ($count <= 0) {
+                continue;
+            }
+
+            $averageMs = (($durationSeconds[$operation] ?? 0.0) / $count) * 1000;
+            $values[] = $operation . '=' . (string) round(
+                $averageMs,
+                2,
+            );
+        }
+
+        return $values;
+    }
+
     /**
      * @return list<string>
      */
@@ -135,15 +203,17 @@ private function runnerClass(): string
     }
 
     /**
-     * @param array<string, int> $counts
+     * Render a count map as multivalue "key=count" strings.
+     *
+     * @param array<array-key, int> $counts
      * @return list<string>
      */
-    private function operationsByType(array $counts): array
+    private function formatCounts(array $counts): array
     {
         $values = [];
 
-        foreach ($counts as $operation => $count) {
-            $values[] = $operation . '=' . $count;
+        foreach ($counts as $key => $count) {
+            $values[] = $key . '=' . $count;
         }
 
         return $values;