feat: Added consumer lagging API and metrics. (#147)

Signed-off-by: Paolo Insogna <[email protected]>

Author: ShogunPanda

docs/consumer.md

@@ -7,7 +7,7 @@ Support for diagnostic and instrumentation is provided via [Node.js Diagnostic C
 The presence of the `result` and `error` properties and the order of tracing channels events follows the current pseudo-code flow:
 
 ```javascript
-function operation(options, callback) {
+function operation (options, callback) {
   const channel = tracingChannel('plt:kafka:example')
   const context = { operationId: 0n, options }
 
@@ -55,9 +55,10 @@ Each tracing channel publishes events with the following common properties:
 
 ## Published channels
 
-| Name                  | Description                                                                                                                                                                                        |
-| --------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `plt:kafka:instances` | Notifies any creation of a `Connection`, `ConnectionPool`, `Base`, `Admin`, `Producer`, `Consumer` or `MessagesStream`. This channel will publish objects with the `type` and `instance` property. |
+| Name                     | Description                                                                                                                                                                                        |
+| ------------------------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `plt:kafka:instances`    | Notifies any creation of a `Connection`, `ConnectionPool`, `Base`, `Admin`, `Producer`, `Consumer` or `MessagesStream`. This channel will publish objects with the `type` and `instance` property. |
+| `plt:kafka:consumer:lag` | Notifies any `Consumer` lag obtained via `Consumer.getLag` (including the one triggered via `Consumer.startLagMonitoring`).                                                                          |
 
 ## Published tracing channels
 

docs/diagnostic.md

@@ -6,14 +6,15 @@ To ensure maximum compatibility, no `prom-client` version is shipped with the pa
 
 If you provide both the `metrics.client` and `metrics.registry` (an instance of `Registry`) options, then `@platformatic/kafka` will register and provide the following metrics:
 
-| Name                      | Type      | Description                               |
-| ------------------------- | --------- | ----------------------------------------- |
-| `kafka_producers`         | `Gauge`   | Number of active Kafka producers.         |
-| `kafka_produced_messages` | `Counter` | Number of produced Kafka messages.        |
-| `kafka_consumers`         | `Gauge`   | Number of active Kafka consumers.         |
-| `kafka_consumers_streams` | `Gauge`   | Number of active Kafka consumers streams. |
-| `kafka_consumers_topics`  | `Gauge`   | Number of topics being consumed.          |
-| `kafka_consumed_messages` | `Counter` | Number of consumed Kafka messages.        |
+| Name                      | Type        | Description                               |
+| ------------------------- | ----------- | ----------------------------------------- |
+| `kafka_producers`         | `Gauge`     | Number of active Kafka producers.         |
+| `kafka_produced_messages` | `Counter`   | Number of produced Kafka messages.        |
+| `kafka_consumers`         | `Gauge`     | Number of active Kafka consumers.         |
+| `kafka_consumers_streams` | `Gauge`     | Number of active Kafka consumers streams. |
+| `kafka_consumers_topics`  | `Gauge`     | Number of topics being consumed.          |
+| `kafka_consumers_lags`    | `Histogram` | Lag of active Kafka consumers             |
+| `kafka_consumed_messages` | `Counter`   | Number of consumed Kafka messages.        |
 
 Optionally, you can provide labels with the `metrics.label` option.
 

docs/metrics.md

@@ -1,14 +1,8 @@
-import { type DescribeLogDirsRequest, type DescribeLogDirsResponse } from '../../apis/admin/describe-log-dirs-v4.ts'
 import {
   type AlterClientQuotasRequest,
   type AlterClientQuotasResponse,
   type AlterClientQuotasResponseEntries
 } from '../../apis/admin/alter-client-quotas-v1.ts'
-import {
-  type DescribeClientQuotasRequest,
-  type DescribeClientQuotasResponse,
-  type DescribeClientQuotasResponseEntry
-} from '../../apis/admin/describe-client-quotas-v0.ts'
 import {
   type CreateTopicsRequest,
   type CreateTopicsRequestTopic,
@@ -21,7 +15,13 @@ import {
   type DeleteTopicsRequestTopic,
   type DeleteTopicsResponse
 } from '../../apis/admin/delete-topics-v6.ts'
+import {
+  type DescribeClientQuotasRequest,
+  type DescribeClientQuotasResponse,
+  type DescribeClientQuotasResponseEntry
+} from '../../apis/admin/describe-client-quotas-v0.ts'
 import { type DescribeGroupsRequest, type DescribeGroupsResponse } from '../../apis/admin/describe-groups-v5.ts'
+import { type DescribeLogDirsRequest, type DescribeLogDirsResponse } from '../../apis/admin/describe-log-dirs-v4.ts'
 import { type ListGroupsRequest as ListGroupsRequestV4 } from '../../apis/admin/list-groups-v4.ts'
 import {
   type ListGroupsRequest as ListGroupsRequestV5,
@@ -37,14 +37,14 @@ import { type Callback } from '../../apis/definitions.ts'
 import { FindCoordinatorKeyTypes, type ConsumerGroupState } from '../../apis/enumerations.ts'
 import { type FindCoordinatorRequest, type FindCoordinatorResponse } from '../../apis/metadata/find-coordinator-v6.ts'
 import { type MetadataRequest, type MetadataResponse } from '../../apis/metadata/metadata-v12.ts'
-import { MultipleErrors } from '../../errors.ts'
 import {
   adminClientQuotasChannel,
   adminGroupsChannel,
   adminLogDirsChannel,
   adminTopicsChannel,
   createDiagnosticContext
 } from '../../diagnostic.ts'
+import { MultipleErrors } from '../../errors.ts'
 import { Reader } from '../../protocol/reader.ts'
 import {
   Base,
@@ -62,31 +62,31 @@ import {
 import { type BaseOptions } from '../base/types.ts'
 import { type GroupAssignment } from '../consumer/types.ts'
 import {
+  alterClientQuotasOptionsValidator,
   createTopicsOptionsValidator,
   deleteGroupsOptionsValidator,
   deleteTopicsOptionsValidator,
+  describeClientQuotasOptionsValidator,
   describeGroupsOptionsValidator,
+  describeLogDirsOptionsValidator,
   listGroupsOptionsValidator,
-  listTopicsOptionsValidator,
-  describeClientQuotasOptionsValidator,
-  alterClientQuotasOptionsValidator,
-  describeLogDirsOptionsValidator
+  listTopicsOptionsValidator
 } from './options.ts'
 import {
   type AdminOptions,
+  type AlterClientQuotasOptions,
+  type BrokerLogDirDescription,
   type CreatedTopic,
   type CreateTopicsOptions,
   type DeleteGroupsOptions,
   type DeleteTopicsOptions,
   type DescribeClientQuotasOptions,
-  type AlterClientQuotasOptions,
-  type DescribeLogDirsOptions,
   type DescribeGroupsOptions,
+  type DescribeLogDirsOptions,
   type Group,
   type GroupBase,
   type ListGroupsOptions,
-  type ListTopicsOptions,
-  type BrokerLogDirDescription
+  type ListTopicsOptions
 } from './types.ts'
 
 export class Admin extends Base<AdminOptions> {
@@ -368,6 +368,7 @@ export class Admin extends Base<AdminOptions> {
       callback = createPromisifiedCallback()
     }
 
+    /* c8 ignore next 3 - Hard to test */
     if (this[kCheckNotClosed](callback)) {
       return callback[kCallbackPromise]
     }
@@ -927,6 +928,7 @@ export class Admin extends Base<AdminOptions> {
 
   #describeLogDirs (options: DescribeLogDirsOptions, callback: CallbackWithPromise<BrokerLogDirDescription[]>): void {
     this[kMetadata]({ topics: [] }, (error, metadata) => {
+      /* c8 ignore next 4 - Hard to test */
       if (error) {
         callback(error, undefined as unknown as BrokerLogDirDescription[])
         return