feat: metrics instrumentation via typed EventEmitter

src/adapter/ember/adapter/emberAdapter.ts

@@ -6,6 +6,7 @@ import equals from "fast-deep-equal/es6";
 import type {Backup} from "../../../models";
 import {BackupUtils, Queue, wait} from "../../../utils";
 import {logger} from "../../../utils/logger";
+import {metrics} from "../../../utils/metrics";
 import * as ZSpec from "../../../zspec";
 import type {Eui64, ExtendedPanId, NodeId, PanId} from "../../../zspec/tstypes";
 import * as Zcl from "../../../zspec/zcl";
@@ -2007,6 +2008,8 @@ export class EmberAdapter extends Adapter {
                 }
 
                 if (status === SLStatus.ZIGBEE_MAX_MESSAGE_LIMIT_REACHED || status === SLStatus.BUSY) {
+                    /* v8 ignore next */
+                    metrics.emit("adapterRetry", {adapterType: "ember", ieeeAddr, reason: SLStatus[status] ?? String(status)});
                     await wait(QUEUE_BUSY_DEFER_MSEC);
                 } else if (status === SLStatus.NETWORK_DOWN) {
                     await wait(QUEUE_NETWORK_DOWN_DEFER_MSEC);

src/adapter/z-stack/adapter/zStackAdapter.ts

@@ -3,6 +3,7 @@ import debounce from "debounce";
 import type * as Models from "../../../models";
 import {Queue, Waitress, wait} from "../../../utils";
 import {logger} from "../../../utils/logger";
+import {metrics} from "../../../utils/metrics";
 import * as ZSpec from "../../../zspec";
 import type {BroadcastAddress} from "../../../zspec/enums";
 import type {Eui64} from "../../../zspec/tstypes";
@@ -584,6 +585,12 @@ export class ZStackAdapter extends Adapter {
                  * MAC_NO_RESOURCES: Operation could not be completed because no memory resources are available,
                  * wait some time and retry.
                  */
+                metrics.emit("adapterRetry", {
+                    adapterType: "zstack",
+                    ieeeAddr,
+                    /* v8 ignore next */
+                    reason: ZnpCommandStatus[dataConfirmResult] ?? String(dataConfirmResult),
+                });
                 await wait(2000);
                 return await this.sendZclFrameToEndpointInternal(
                     ieeeAddr,
@@ -665,6 +672,8 @@ export class ZStackAdapter extends Adapter {
                 await wait(2000);
             }
 
+            /* v8 ignore next */
+            metrics.emit("adapterRetry", {adapterType: "zstack", ieeeAddr, reason: ZnpCommandStatus[dataConfirmResult] ?? String(dataConfirmResult)});
             return await this.sendZclFrameToEndpointInternal(
                 ieeeAddr,
                 networkAddress,
@@ -716,6 +725,7 @@ export class ZStackAdapter extends Adapter {
                         });
                     }
                     // No response could be of invalid route, e.g. when message is send to wrong parent of end device.
+                    metrics.emit("adapterRetry", {adapterType: "zstack", ieeeAddr, reason: "no_response"});
                     await this.discoverRoute(networkAddress);
                     return await this.sendZclFrameToEndpointInternal(
                         ieeeAddr,

src/adapter/zboss/adapter/zbossAdapter.ts

@@ -4,6 +4,7 @@ import assert from "node:assert";
 import type {Backup} from "../../../models";
 import {Queue, Waitress} from "../../../utils";
 import {logger} from "../../../utils/logger";
+import {metrics} from "../../../utils/metrics";
 import * as ZSpec from "../../../zspec";
 import * as Zcl from "../../../zspec/zcl";
 import * as Zdo from "../../../zspec/zdo";
@@ -387,6 +388,7 @@ export class ZBOSSAdapter extends Adapter {
                 } catch (error) {
                     logger.debug(`Response timeout (${ieeeAddr}:${networkAddress},${responseAttempt})`, NS);
                     if (responseAttempt < 1 && !disableRecovery) {
+                        metrics.emit("adapterRetry", {adapterType: "zboss", ieeeAddr, reason: "no_response"});
                         return await this.sendZclFrameToEndpointInternal(
                             ieeeAddr,
                             networkAddress,

src/adapter/zigate/adapter/zigateAdapter.ts

@@ -3,6 +3,7 @@
 import type * as Models from "../../../models";
 import {Queue, Waitress, wait} from "../../../utils";
 import {logger} from "../../../utils/logger";
+import {metrics} from "../../../utils/metrics";
 import * as ZSpec from "../../../zspec";
 import type {BroadcastAddress} from "../../../zspec/enums";
 import * as Zcl from "../../../zspec/zcl";
@@ -368,6 +369,7 @@ export class ZiGateAdapter extends Adapter {
         } catch {
             if (responseAttempt < 1 && !disableRecovery) {
                 // @todo discover route
+                metrics.emit("adapterRetry", {adapterType: "zigate", ieeeAddr, reason: "send_failure"});
                 return await this.sendZclFrameToEndpointInternal(
                     ieeeAddr,
                     networkAddress,
@@ -395,6 +397,7 @@ export class ZiGateAdapter extends Adapter {
             } catch (error) {
                 logger.error(`Response error ${(error as Error).message} (${ieeeAddr}:${networkAddress},${responseAttempt})`, NS);
                 if (responseAttempt < 1 && !disableRecovery) {
+                    metrics.emit("adapterRetry", {adapterType: "zigate", ieeeAddr, reason: "no_response"});
                     return await this.sendZclFrameToEndpointInternal(
                         ieeeAddr,
                         networkAddress,

src/adapter/zoh/adapter/zohAdapter.ts

@@ -8,6 +8,7 @@ import type {ZigbeeAPSHeader, ZigbeeAPSPayload} from "zigbee-on-host/dist/zigbee
 import type {ZigbeeNWKGPHeader} from "zigbee-on-host/dist/zigbee/zigbee-nwkgp";
 import type {Backup} from "../../../models/backup";
 import {logger} from "../../../utils/logger";
+import {metrics} from "../../../utils/metrics";
 import {Queue} from "../../../utils/queue";
 import {wait} from "../../../utils/wait";
 import {Waitress} from "../../../utils/waitress";
@@ -646,6 +647,7 @@ export class ZoHAdapter extends Adapter {
                     if (disableRecovery || i === 1) {
                         throw error;
                     } // else retry
+                    metrics.emit("adapterRetry", {adapterType: "zoh", ieeeAddr, reason: "send_failure"});
                 }
                 /* v8 ignore start */
             } // coverage detection failure

src/controller/controller.ts

@@ -5,6 +5,7 @@ import {Adapter, type Events as AdapterEvents, type TsType as AdapterTsType} fro
 import type {ZclPayload} from "../adapter/events";
 import {BackupUtils, wait} from "../utils";
 import {logger} from "../utils/logger";
+import {metrics, type SendStatus} from "../utils/metrics.js";
 import {isNumberArrayOfLength} from "../utils/utils";
 import * as ZSpec from "../zspec";
 import type {Eui64} from "../zspec/tstypes";
@@ -28,6 +29,18 @@ import type {DeviceType, GreenPowerDeviceJoinedPayload, RawPayload} from "./tsty
 
 const NS = "zh:controller";
 
+async function instrumentSend<T>(fn: () => Promise<T>, record: (status: SendStatus, durationSeconds: number) => void): Promise<T> {
+    const start = Date.now();
+    try {
+        const result = await fn();
+        record("success", (Date.now() - start) / 1000);
+        return result;
+    } catch (e) {
+        record("failure", (Date.now() - start) / 1000);
+        throw e;
+    }
+}
+
 interface Options {
     network: AdapterTsType.NetworkOptions;
     serialPort: AdapterTsType.SerialPortOptions;
@@ -280,7 +293,11 @@ export class Controller extends events.EventEmitter<ControllerEventMap> {
             // will fail if args are incorrect for request
             const buf = Zdo.Buffalo.buildRequest(this.adapter.hasZdoMessageOverhead, clusterKey, ...zdoParams);
 
-            return (await this.adapter.sendZdo(ieeeAddress, networkAddress, clusterKey, buf, disableResponse)) as ZdoTypes.GenericZdoResponse;
+            return (await instrumentSend(
+                () => this.adapter.sendZdo(ieeeAddress, networkAddress, clusterKey, buf, disableResponse),
+                (status, durationSeconds) =>
+                    metrics.emit("adapterSendZdo", {ieeeAddr: ieeeAddress, clusterId: clusterKey as number, status, durationSeconds}),
+            )) as ZdoTypes.GenericZdoResponse;
         }
 
         assert(zcl);
@@ -330,7 +347,10 @@ export class Controller extends events.EventEmitter<ControllerEventMap> {
         if (groupId !== undefined) {
             assert(groupId >= 0x0000 && groupId <= 0xffff);
 
-            await this.adapter.sendZclFrameToGroup(groupId, zclFrame, srcEndpoint, profileId);
+            await instrumentSend(
+                () => this.adapter.sendZclFrameToGroup(groupId, zclFrame, srcEndpoint, profileId),
+                (status, durationSeconds) => metrics.emit("adapterSendZclGroup", {groupId, status, durationSeconds}),
+            );
             return;
         }
 
@@ -339,22 +359,29 @@ export class Controller extends events.EventEmitter<ControllerEventMap> {
         assert(networkAddress !== undefined && networkAddress >= 0x0000 && networkAddress <= 0xffff);
 
         if (networkAddress >= ZSpec.BROADCAST_MIN) {
-            await this.adapter.sendZclFrameToAll(dstEndpoint, zclFrame, srcEndpoint, networkAddress, profileId);
+            await instrumentSend(
+                () => this.adapter.sendZclFrameToAll(dstEndpoint, zclFrame, srcEndpoint, networkAddress, profileId),
+                (status, durationSeconds) => metrics.emit("adapterSendZclBroadcast", {status, durationSeconds}),
+            );
             return;
         }
 
         assert(ieeeAddress);
 
-        return await this.adapter.sendZclFrameToEndpoint(
-            ieeeAddress,
-            networkAddress,
-            dstEndpoint,
-            zclFrame,
-            timeout,
-            disableResponse,
-            false,
-            srcEndpoint,
-            profileId,
+        return await instrumentSend(
+            () =>
+                this.adapter.sendZclFrameToEndpoint(
+                    ieeeAddress,
+                    networkAddress,
+                    dstEndpoint,
+                    zclFrame,
+                    timeout,
+                    disableResponse,
+                    false,
+                    srcEndpoint,
+                    profileId,
+                ),
+            (status, durationSeconds) => metrics.emit("adapterSendZclUnicast", {ieeeAddr: ieeeAddress, status, durationSeconds}),
         );
     }
 
@@ -623,7 +650,10 @@ export class Controller extends events.EventEmitter<ControllerEventMap> {
             undefined,
         );
 
-        await this.adapter.sendZdo(ZSpec.BLANK_EUI64, ZSpec.BroadcastAddress.SLEEPY, clusterId, zdoPayload, true);
+        await instrumentSend(
+            () => this.adapter.sendZdo(ZSpec.BLANK_EUI64, ZSpec.BroadcastAddress.SLEEPY, clusterId, zdoPayload, true),
+            (status, durationSeconds) => metrics.emit("adapterSendZdo", {ieeeAddr: ZSpec.BLANK_EUI64, clusterId, status, durationSeconds}),
+        );
         logger.info(`Channel changed to '${newChannel}'`, NS);
 
         this.networkParametersCached = undefined; // invalidate cache
@@ -644,7 +674,10 @@ export class Controller extends events.EventEmitter<ControllerEventMap> {
         const zdoPayload = Zdo.Buffalo.buildRequest(this.adapter.hasZdoMessageOverhead, clusterId, nwkAddress, false, 0);
 
         try {
-            const response = await this.adapter.sendZdo(ZSpec.BLANK_EUI64, nwkAddress, clusterId, zdoPayload, false);
+            const response = await instrumentSend(
+                () => this.adapter.sendZdo(ZSpec.BLANK_EUI64, nwkAddress, clusterId, zdoPayload, false),
+                (status, durationSeconds) => metrics.emit("adapterSendZdo", {ieeeAddr: ZSpec.BLANK_EUI64, clusterId, status, durationSeconds}),
+            );
 
             if (Zdo.Buffalo.checkStatus<Zdo.ClusterId.IEEE_ADDRESS_RESPONSE>(response)) {
                 const payload = response[1];
@@ -904,6 +937,7 @@ export class Controller extends events.EventEmitter<ControllerEventMap> {
     }
 
     private onZdoResponse(clusterId: Zdo.ClusterId, response: ZdoTypes.GenericZdoResponse): void {
+        metrics.emit("adapterReceiveZdoResponse", {clusterId});
         logger.debug(
             `Received ZDO response: clusterId=${Zdo.ClusterId[clusterId]}, status=${Zdo.Status[response[0]]}, payload=${JSON.stringify(response[1])}`,
             NS,
@@ -943,6 +977,12 @@ export class Controller extends events.EventEmitter<ControllerEventMap> {
             return;
         }
 
+        metrics.emit("adapterReceiveZclPayload", {
+            ieeeAddr: typeof payload.address === "string" ? payload.address : undefined,
+            clusterID: payload.clusterID,
+            wasBroadcast: payload.wasBroadcast,
+        });
+
         if (payload.clusterID === Zcl.Clusters.greenPower.ID) {
             try {
                 // Custom clusters are not supported for Green Power since we need to parse the frame to get the device.

src/controller/helpers/request.ts

@@ -32,6 +32,7 @@ export class Request<Type = any> {
     private func: () => Promise<Type>;
     frame: Zcl.Frame;
     expires: number;
+    enqueuedAt: number | undefined;
     sendPolicy: SendPolicy | undefined;
     private resolveQueue: Array<(value: Type) => void>;
     private rejectQueue: Array<(error: Error) => void>;
@@ -49,6 +50,7 @@ export class Request<Type = any> {
         this.func = func;
         this.frame = frame;
         this.expires = timeout + Date.now();
+        this.enqueuedAt = undefined;
         this.sendPolicy = sendPolicy ?? (!frame.command ? undefined : Request.defaultSendPolicy[frame.command.ID]);
         this.resolveQueue = resolve === undefined ? ([] as ((value: Type) => void)[]) : new Array<(value: Type) => void>(resolve);
         this.rejectQueue = reject === undefined ? ([] as ((error: Error) => void)[]) : new Array<(error: Error) => void>(reject);

src/controller/helpers/requestQueue.ts

@@ -1,5 +1,6 @@
 import equal from "fast-deep-equal/es6";
 import {logger} from "../../utils/logger";
+import {metrics} from "../../utils/metrics";
 import type * as Zcl from "../../zspec/zcl";
 import type {Endpoint} from "../model";
 import type Request from "./request";
@@ -36,6 +37,15 @@ export class RequestQueue extends Set<Request> {
                 logger.debug(`Request Queue (${this.deviceIeeeAddress}/${this.id}): discard after timeout. Size before: ${this.size}`, NS);
                 request.reject();
                 this.delete(request);
+                metrics.emit("requestQueueLength", {ieeeAddr: this.deviceIeeeAddress, endpointId: this.id, length: this.size});
+                if (request.enqueuedAt !== undefined) {
+                    metrics.emit("requestQueueDuration", {
+                        ieeeAddr: this.deviceIeeeAddress,
+                        endpointId: this.id,
+                        outcome: "expired",
+                        durationSeconds: (now - request.enqueuedAt) / 1000,
+                    });
+                }
             }
         }
 
@@ -48,6 +58,15 @@ export class RequestQueue extends Set<Request> {
                     logger.debug(`Request Queue (${this.deviceIeeeAddress}/${this.id}): send success`, NS);
                     request.resolve(result);
                     this.delete(request);
+                    metrics.emit("requestQueueLength", {ieeeAddr: this.deviceIeeeAddress, endpointId: this.id, length: this.size});
+                    if (request.enqueuedAt !== undefined) {
+                        metrics.emit("requestQueueDuration", {
+                            ieeeAddr: this.deviceIeeeAddress,
+                            endpointId: this.id,
+                            outcome: "sent",
+                            durationSeconds: (Date.now() - request.enqueuedAt) / 1000,
+                        });
+                    }
                 } catch (error) {
                     logger.debug(
                         `Request Queue (${this.deviceIeeeAddress}/${this.id}): send failed, expires in ` +
@@ -65,6 +84,8 @@ export class RequestQueue extends Set<Request> {
         return await new Promise((resolve, reject): void => {
             request.addCallbacks(resolve, reject);
             this.add(request);
+            request.enqueuedAt = Date.now();
+            metrics.emit("requestQueueLength", {ieeeAddr: this.deviceIeeeAddress, endpointId: this.id, length: this.size});
         });
     }
 

src/index.ts

@@ -8,6 +8,8 @@ export {getOtaFirmware, getOtaIndex, parseOtaHeader, parseOtaImage, parseOtaSubE
 export type * as Models from "./controller/model";
 export type * as Types from "./controller/tstype";
 export {setLogger} from "./utils/logger";
+export type {MetricsEventMap} from "./utils/metrics";
+export {metrics} from "./utils/metrics";
 export {getTimeClusterAttributes} from "./utils/timeService";
 export * as ZSpec from "./zspec";
 export * as Zcl from "./zspec/zcl";

src/utils/metrics.ts

@@ -0,0 +1,70 @@
+import {EventEmitter} from "node:events";
+
+export type SendStatus = "success" | "failure";
+
+export interface AdapterSendZclUnicastPayload {
+    ieeeAddr: string;
+    status: SendStatus;
+    durationSeconds: number;
+}
+
+export interface AdapterSendZdoPayload {
+    ieeeAddr: string;
+    clusterId: number;
+    status: SendStatus;
+    durationSeconds: number;
+}
+
+export interface AdapterSendZclGroupPayload {
+    groupId: number;
+    status: SendStatus;
+    durationSeconds: number;
+}
+
+export interface AdapterSendZclBroadcastPayload {
+    status: SendStatus;
+    durationSeconds: number;
+}
+
+export interface AdapterRetryPayload {
+    adapterType: string;
+    ieeeAddr: string | undefined;
+    reason: string;
+}
+
+export interface AdapterReceiveZclPayloadPayload {
+    ieeeAddr: string | undefined;
+    clusterID: number;
+    wasBroadcast: boolean;
+}
+
+export interface AdapterReceiveZdoResponsePayload {
+    clusterId: number;
+}
+
+export interface RequestQueueLengthPayload {
+    ieeeAddr: string;
+    endpointId: number;
+    length: number;
+}
+
+export interface RequestQueueDurationPayload {
+    ieeeAddr: string;
+    endpointId: number;
+    outcome: "sent" | "expired";
+    durationSeconds: number;
+}
+
+export interface MetricsEventMap {
+    adapterSendZclUnicast: [data: AdapterSendZclUnicastPayload];
+    adapterSendZdo: [data: AdapterSendZdoPayload];
+    adapterSendZclGroup: [data: AdapterSendZclGroupPayload];
+    adapterSendZclBroadcast: [data: AdapterSendZclBroadcastPayload];
+    adapterRetry: [data: AdapterRetryPayload];
+    adapterReceiveZclPayload: [data: AdapterReceiveZclPayloadPayload];
+    adapterReceiveZdoResponse: [data: AdapterReceiveZdoResponsePayload];
+    requestQueueLength: [data: RequestQueueLengthPayload];
+    requestQueueDuration: [data: RequestQueueDurationPayload];
+}
+
+export const metrics = new EventEmitter<MetricsEventMap>();