From 8657f3160d8f1f9c0527b5cef076d78b43cd66a4 Mon Sep 17 00:00:00 2001
From: Shannon Klaus <sklaus@aerospike.com>
Date: Tue, 27 Feb 2024 19:10:33 -0500
Subject: [PATCH] CLIENT-2699 Support client metrics.

Co-authored-by: Brian Nichols <bnichols@aerospike.com>
---
 Makefile                                  |   3 +
 src/include/aerospike/aerospike_stats.h   |  30 +-
 src/include/aerospike/as_async.h          |  13 +-
 src/include/aerospike/as_cdt_ctx.h        |   2 +-
 src/include/aerospike/as_cluster.h        | 184 +++++-
 src/include/aerospike/as_command.h        |   3 +-
 src/include/aerospike/as_event_internal.h |  11 +-
 src/include/aerospike/as_latency.h        |  66 +++
 src/include/aerospike/as_metrics.h        | 191 ++++++
 src/include/aerospike/as_metrics_writer.h |  81 +++
 src/include/aerospike/as_node.h           |  92 ++-
 src/main/aerospike/aerospike.c            |  11 +-
 src/main/aerospike/aerospike_batch.c      |  16 +-
 src/main/aerospike/aerospike_key.c        |  23 +-
 src/main/aerospike/aerospike_query.c      |  14 +-
 src/main/aerospike/aerospike_scan.c       |  12 +-
 src/main/aerospike/aerospike_stats.c      |  78 ++-
 src/main/aerospike/as_cluster.c           | 156 ++++-
 src/main/aerospike/as_command.c           |  33 +-
 src/main/aerospike/as_event.c             |  62 +-
 src/main/aerospike/as_event_ev.c          |  38 +-
 src/main/aerospike/as_event_event.c       |  38 +-
 src/main/aerospike/as_event_uv.c          |   6 +-
 src/main/aerospike/as_latency.c           |  42 ++
 src/main/aerospike/as_metrics.c           |  52 ++
 src/main/aerospike/as_metrics_writer.c    | 688 ++++++++++++++++++++++
 src/main/aerospike/as_node.c              | 113 +++-
 src/main/aerospike/as_pipe.c              |   6 +-
 src/test/aerospike_test.c                 |  17 +-
 vs/aerospike/aerospike.vcxproj            |   6 +
 vs/aerospike/aerospike.vcxproj.filters    |  18 +
 vs/props/base.props                       |   2 +-
 xcode/aerospike.xcodeproj/project.pbxproj |  24 +
 33 files changed, 1995 insertions(+), 136 deletions(-)
 create mode 100644 src/include/aerospike/as_latency.h
 create mode 100644 src/include/aerospike/as_metrics.h
 create mode 100644 src/include/aerospike/as_metrics_writer.h
 create mode 100644 src/main/aerospike/as_latency.c
 create mode 100644 src/main/aerospike/as_metrics.c
 create mode 100644 src/main/aerospike/as_metrics_writer.c

diff --git a/Makefile b/Makefile
index da78c30a6a..49b21ef60b 100644
--- a/Makefile
+++ b/Makefile
@@ -133,9 +133,12 @@ AEROSPIKE += as_host.o
 AEROSPIKE += as_info.o
 AEROSPIKE += as_job.o
 AEROSPIKE += as_key.o
+AEROSPIKE += as_latency.o
 AEROSPIKE += as_list_operations.o
 AEROSPIKE += as_lookup.o
 AEROSPIKE += as_map_operations.o
+AEROSPIKE += as_metrics.o
+AEROSPIKE += as_metrics_writer.o
 AEROSPIKE += as_node.o
 AEROSPIKE += as_operations.o
 AEROSPIKE += as_partition.o
diff --git a/src/include/aerospike/aerospike_stats.h b/src/include/aerospike/aerospike_stats.h
index 431f12266e..cfb3e81114 100644
--- a/src/include/aerospike/aerospike_stats.h
+++ b/src/include/aerospike/aerospike_stats.h
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2021 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -88,9 +88,16 @@ typedef struct as_node_stats_s {
 	as_conn_stats pipeline;
 
 	/**
-	 * Node error count within current window.
+	 * Transaction error count since node was initialized. If the error is retryable, multiple errors per
+	 * transaction may occur.
 	 */
-	uint32_t error_count;
+	uint64_t error_count;
+
+	/**
+	 * Transaction timeout count since node was initialized. If the timeout is retryable (ie socket timeout),
+	 * multiple timeouts per transaction may occur.
+	 */
+	uint64_t timeout_count;
 
 } as_node_stats;
 
@@ -128,6 +135,11 @@ typedef struct as_cluster_stats_s {
 	 */
 	as_event_loop_stats* event_loops;
 
+	/**
+	* Count of transaction retries since cluster was started.
+	*/
+	uint64_t retry_count;
+
 	/**
 	 * Node count.
 	 */
@@ -242,6 +254,18 @@ aerospike_event_loop_stats(as_event_loop* event_loop, as_event_loop_stats* stats
 AS_EXTERN char*
 aerospike_stats_to_string(as_cluster_stats* stats);
 
+static inline void
+as_conn_stats_init(as_conn_stats* stats)
+{
+	stats->in_pool = 0;
+	stats->in_use = 0;
+	stats->opened = 0;
+	stats->closed = 0;
+}
+
+void
+as_conn_stats_sum(as_conn_stats* stats, as_async_conn_pool* pool);
+
 #ifdef __cplusplus
 } // end extern "C"
 #endif
diff --git a/src/include/aerospike/as_async.h b/src/include/aerospike/as_async.h
index e4ca1f5551..2a1e2699c9 100644
--- a/src/include/aerospike/as_async.h
+++ b/src/include/aerospike/as_async.h
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -110,7 +110,9 @@ as_async_write_command_create(
 	cmd->flags = 0;
 	cmd->replica_size = pi->replica_size;
 	cmd->replica_index = 0;
+	cmd->latency_type = AS_LATENCY_TYPE_WRITE;
 	wcmd->listener = listener;
+	as_cluster_add_tran(cluster);
 	return cmd;
 }
 	
@@ -119,7 +121,8 @@ as_async_record_command_create(
 	as_cluster* cluster, const as_policy_base* policy, as_partition_info* pi,
 	as_policy_replica replica, uint8_t replica_index, bool deserialize, bool heap_rec,
 	uint8_t flags, as_async_record_listener listener, void* udata, as_event_loop* event_loop,
-	as_pipe_listener pipe_listener, size_t size, as_event_parse_results_fn parse_results
+	as_pipe_listener pipe_listener, size_t size, as_event_parse_results_fn parse_results, 
+	as_latency_type latency_type
 	)
 {
 	// Allocate enough memory to cover: struct size + write buffer size + auth max buffer size
@@ -158,7 +161,9 @@ as_async_record_command_create(
 
 	cmd->replica_size = pi->replica_size;
 	cmd->replica_index = replica_index;
+	cmd->latency_type = latency_type;
 	rcmd->listener = listener;
+	as_cluster_add_tran(cluster);
 	return cmd;
 }
 
@@ -197,7 +202,9 @@ as_async_value_command_create(
 	cmd->flags = 0;
 	cmd->replica_size = pi->replica_size;
 	cmd->replica_index = 0;
+	cmd->latency_type = AS_LATENCY_TYPE_WRITE;
 	vcmd->listener = listener;
+	as_cluster_add_tran(cluster);
 	return cmd;
 }
 
@@ -233,7 +240,9 @@ as_async_info_command_create(
 	cmd->flags = 0;
 	cmd->replica_size = 1;
 	cmd->replica_index = 0;
+	cmd->latency_type = AS_LATENCY_TYPE_NONE;
 	icmd->listener = listener;
+	as_cluster_add_tran(node->cluster);
 	return cmd;
 }
 
diff --git a/src/include/aerospike/as_cdt_ctx.h b/src/include/aerospike/as_cdt_ctx.h
index 9e5104b78d..8317a1cc76 100644
--- a/src/include/aerospike/as_cdt_ctx.h
+++ b/src/include/aerospike/as_cdt_ctx.h
@@ -211,7 +211,7 @@ as_cdt_ctx_add_list_value(as_cdt_ctx* ctx, as_val* val)
 
 /**
  * Lookup map by index offset.
- * <p>
+ *
  * If the index is negative, the resolved index starts backwards from end of list.
  * If an index is out of bounds, a parameter error will be returned.  Examples:
  * <ul>
diff --git a/src/include/aerospike/as_cluster.h b/src/include/aerospike/as_cluster.h
index b48b2568a8..d999b6a7c3 100644
--- a/src/include/aerospike/as_cluster.h
+++ b/src/include/aerospike/as_cluster.h
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -18,6 +18,7 @@
 
 #include <aerospike/as_atomic.h>
 #include <aerospike/as_config.h>
+#include <aerospike/as_metrics.h>
 #include <aerospike/as_node.h>
 #include <aerospike/as_partition.h>
 #include <aerospike/as_policy.h>
@@ -203,6 +204,12 @@ typedef struct as_cluster_s {
 	 */
 	pthread_mutex_t seed_lock;
 
+	/**
+	 * @private
+	 * Lock for metrics operations.
+	 */
+	pthread_mutex_t metrics_lock;
+
 	/**
 	 * @private
 	 * Lock for the tend thread to wait on with the tend interval as timeout.
@@ -379,6 +386,75 @@ typedef struct as_cluster_s {
 	 * Should continue to tend cluster.
 	 */
 	volatile bool valid;
+
+	/**
+	 * @private
+	 * Is metrics colleciton enabled.
+	 */
+	bool metrics_enabled;
+
+	/**
+	 * @private
+	 * Number of cluster tend iterations between metrics notification events. One tend iteration
+	 * is defined as as_config.tender_interval (default 1 second) plus the time to tend all
+	 * nodes. This is set using as_policy_metrics.
+	 */
+	uint32_t metrics_interval;
+
+	/**
+	 * @private
+	 * Number of elapsed time range buckets in latency histograms. This is set using as_policy_metrics.
+	 */
+	uint32_t metrics_latency_columns;
+
+	/**
+	 * @private
+	 * Power of 2 multiple between each range bucket in latency histograms starting at column 3. The bucket units
+	 * are in milliseconds. The first 2 buckets are "<=1ms" and ">1ms". Examples:
+	 * 
+	 * ~~~~~~~~~~{.c}
+	 * // latencyColumns=7 latencyShift=1
+	 * <=1ms >1ms >2ms >4ms >8ms >16ms >32ms
+	 *
+	 * // latencyColumns=5 latencyShift=3
+	 * <=1ms >1ms >8ms >64ms >512ms
+	 * ~~~~~~~~~~
+	 * 
+	 * This is set using as_policy_metrics.
+	 */
+	uint32_t metrics_latency_shift;
+
+	/**
+	 * @private
+	 * Listeners that handles metrics notification events. The default listener implementation
+	 * writes the metrics snapshot to a file which will later be read and forwarded to
+	 * OpenTelemetry by a separate offline application.
+	 *
+	 * The listener could be overridden to send the metrics snapshot directly to OpenTelemetry.
+	 * 
+	 * This is set using as_policy_metrics.
+	 */
+	as_metrics_listeners metrics_listeners;
+
+	/**
+	 * @private
+	 * Transaction retry count. There can be multiple retries for a single transaction.
+	 * The value is cumulative and not reset per metrics interval.
+	 */
+	uint64_t retry_count;
+
+	/**
+	 * @private
+	 * Transaction count. The value is cumulative and not reset per metrics interval.
+	 */
+	uint64_t tran_count;
+
+	/**
+	 * @private
+	 * Delay queue timeout count. The value is cumulative and not reset per metrics interval.
+	 */
+	uint64_t delay_queue_timeout_count;
+
 } as_cluster;
 
 /******************************************************************************
@@ -518,6 +594,92 @@ as_partition_shm_get_node(
 	as_node* prev_node, as_policy_replica replica, uint8_t replica_size, uint8_t* replica_index
 	);
 
+/**
+ * @private
+ * Enable the collection of metrics
+ */
+as_status
+as_cluster_enable_metrics(as_error* err, as_cluster* cluster, as_metrics_policy* policy);
+
+/**
+ * @private
+ * Disable the collection of metrics
+ */
+as_status
+as_cluster_disable_metrics(as_error* err, as_cluster* cluster);
+
+/**
+ * @private
+ * Increment transaction count when metrics are enabled.
+ */
+static inline void
+as_cluster_add_tran(as_cluster* cluster)
+{
+	if (cluster->metrics_enabled) {
+		as_incr_uint64(&cluster->tran_count);
+	}
+}
+
+/**
+ * @private
+ * Return transaction count. The value is cumulative and not reset per metrics interval.
+ */
+static inline uint64_t
+as_cluster_get_tran_count(const as_cluster* cluster)
+{
+	return as_load_uint64(&cluster->tran_count);
+}
+
+/**
+ * @private
+ * Increment async delay queue timeout count.
+ */
+static inline void
+as_cluster_add_retry(as_cluster* cluster)
+{
+	as_incr_uint64(&cluster->retry_count);
+}
+
+/**
+ * @private
+ * Add transaction retry count. There can be multiple retries for a single transaction.
+ */
+static inline void
+as_cluster_add_retries(as_cluster* cluster, uint32_t count)
+{
+	as_faa_uint64(&cluster->retry_count, count);
+}
+
+/**
+ * @private
+ * Return transaction retry count. The value is cumulative and not reset per metrics interval.
+ */
+static inline uint64_t
+as_cluster_get_retry_count(const as_cluster* cluster)
+{
+	return as_load_uint64(&cluster->retry_count);
+}
+
+/**
+ * @private
+ * Increment async delay queue timeout count.
+ */
+static inline void
+as_cluster_add_delay_queue_timeout(as_cluster* cluster)
+{
+	as_incr_uint64(&cluster->delay_queue_timeout_count);
+}
+
+/**
+ * @private
+ * Return async delay queue timeout count.
+ */
+static inline uint64_t
+as_cluster_get_delay_queue_timeout_count(const as_cluster* cluster)
+{
+	return as_load_uint64(&cluster->delay_queue_timeout_count);
+}
+
 /**
  * @private
  * Get mapped node given partition and replica.  This function does not reserve the node.
@@ -544,10 +706,10 @@ as_partition_get_node(
  * Increment node's error count.
  */
 static inline void
-as_node_incr_error_count(as_node* node)
+as_node_incr_error_rate(as_node* node)
 {
 	if (node->cluster->max_error_rate > 0) {
-		as_incr_uint32(&node->error_count);
+		as_incr_uint32(&node->error_rate);
 	}
 }
 
@@ -556,9 +718,9 @@ as_node_incr_error_count(as_node* node)
  * Reset node's error count.
  */
 static inline void
-as_node_reset_error_count(as_node* node)
+as_node_reset_error_rate(as_node* node)
 {
-	as_store_uint32(&node->error_count, 0);
+	as_store_uint32(&node->error_rate, 0);
 }
 
 /**
@@ -566,9 +728,9 @@ as_node_reset_error_count(as_node* node)
  * Get node's error count.
  */
 static inline uint32_t
-as_node_get_error_count(as_node* node)
+as_node_get_error_rate(as_node* node)
 {
-	return as_load_uint32(&node->error_count);
+	return as_load_uint32(&node->error_rate);
 }
 
 /**
@@ -576,10 +738,10 @@ as_node_get_error_count(as_node* node)
  * Validate node's error count.
  */
 static inline bool
-as_node_valid_error_count(as_node* node)
+as_node_valid_error_rate(as_node* node)
 {
 	uint32_t max = node->cluster->max_error_rate;
-	return max == 0 || max >= as_load_uint32(&node->error_count);
+	return max == 0 || max >= as_load_uint32(&node->error_rate);
 }
 
 /**
@@ -590,7 +752,7 @@ static inline void
 as_node_close_conn_error(as_node* node, as_socket* sock, as_conn_pool* pool)
 {
 	as_node_close_connection(node, sock, pool);
-	as_node_incr_error_count(node);
+	as_node_incr_error_rate(node);
 }
 
 /**
@@ -601,7 +763,7 @@ static inline void
 as_node_put_conn_error(as_node* node, as_socket* sock)
 {
 	as_node_put_connection(node, sock);
-	as_node_incr_error_count(node);
+	as_node_incr_error_rate(node);
 }
 
 #ifdef __cplusplus
diff --git a/src/include/aerospike/as_command.h b/src/include/aerospike/as_command.h
index c81806197c..32b4c4ef0d 100644
--- a/src/include/aerospike/as_command.h
+++ b/src/include/aerospike/as_command.h
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -190,6 +190,7 @@ typedef struct as_command_s {
 	uint8_t replica_size;
 	uint8_t replica_index;
 	uint8_t replica_index_sc; // Used in batch only.
+	as_latency_type latency_type;
 } as_command;
 
 /**
diff --git a/src/include/aerospike/as_event_internal.h b/src/include/aerospike/as_event_internal.h
index 52b4fe7413..40a5be7cf7 100644
--- a/src/include/aerospike/as_event_internal.h
+++ b/src/include/aerospike/as_event_internal.h
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -140,6 +140,7 @@ typedef struct as_event_command {
 	cf_ll_element pipe_link;
 	
 	uint8_t* buf;
+	uint64_t begin; // Used for metrics
 	uint32_t command_sent_counter;
 	uint32_t write_offset;
 	uint32_t write_len;
@@ -155,6 +156,7 @@ typedef struct as_event_command {
 	uint8_t replica_size;
 	uint8_t replica_index;
 	uint8_t replica_index_sc; // Used in batch only.
+	as_latency_type latency_type;
 } as_event_command;
 
 typedef struct {
@@ -189,6 +191,9 @@ as_event_command_execute(as_event_command* cmd, as_error* err);
 void
 as_event_command_schedule(as_event_command* cmd);
 
+void
+as_event_connection_complete(as_event_command* cmd);
+
 bool
 as_event_proto_parse(as_event_command* cmd, as_proto* proto);
 
@@ -761,7 +766,7 @@ as_event_release_async_connection(as_event_command* cmd)
 {
 	as_async_conn_pool* pool = &cmd->node->async_conn_pools[cmd->event_loop->index];
 	as_event_release_connection(cmd->conn, pool);
-	as_node_incr_error_count(cmd->node);
+	as_node_incr_error_rate(cmd->node);
 }
 
 static inline void
@@ -783,7 +788,7 @@ as_event_connection_timeout(as_event_command* cmd, as_async_conn_pool* pool)
 		if (conn->watching > 0) {
 			as_event_stop_watcher(cmd, conn);
 			as_event_release_connection(conn, pool);
-			as_node_incr_error_count(cmd->node);
+			as_node_incr_error_rate(cmd->node);
 		}
 		else {
 			cf_free(conn);
diff --git a/src/include/aerospike/as_latency.h b/src/include/aerospike/as_latency.h
new file mode 100644
index 0000000000..60d964e8a3
--- /dev/null
+++ b/src/include/aerospike/as_latency.h
@@ -0,0 +1,66 @@
+/*
+ * Copyright 2008-2024 Aerospike, Inc.
+ *
+ * Portions may be licensed to Aerospike, Inc. under one or more contributor
+ * license agreements.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+#pragma once
+
+#include <aerospike/as_atomic.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+//---------------------------------
+// Types
+//---------------------------------
+
+typedef uint8_t as_latency_type;
+
+#define AS_LATENCY_TYPE_CONN 0
+#define AS_LATENCY_TYPE_WRITE 1
+#define AS_LATENCY_TYPE_READ 2
+#define AS_LATENCY_TYPE_BATCH 3
+#define AS_LATENCY_TYPE_QUERY 4
+#define AS_LATENCY_TYPE_NONE 5
+
+/**
+ * Latency buckets for a transaction group.
+ * Latency bucket counts are cumulative and not reset on each metrics snapshot interval
+ */
+typedef struct as_latency_buckets_s {
+	uint64_t* buckets;
+	uint32_t latency_shift;
+	uint32_t latency_columns;
+} as_latency_buckets;
+
+//---------------------------------
+// Functions
+//---------------------------------
+
+static inline uint64_t
+as_latency_get_bucket(as_latency_buckets* buckets, uint32_t i)
+{
+	return as_load_uint64(&buckets->buckets[i]);
+}
+
+/**
+ * Convert latency_type to string version for printing to the output file
+ */
+AS_EXTERN char*
+as_latency_type_to_string(as_latency_type type);
+
+#ifdef __cplusplus
+} // end extern "C"
+#endif
diff --git a/src/include/aerospike/as_metrics.h b/src/include/aerospike/as_metrics.h
new file mode 100644
index 0000000000..3f6c9a2404
--- /dev/null
+++ b/src/include/aerospike/as_metrics.h
@@ -0,0 +1,191 @@
+/*
+ * Copyright 2008-2024 Aerospike, Inc.
+ *
+ * Portions may be licensed to Aerospike, Inc. under one or more contributor
+ * license agreements.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+#pragma once
+
+#include <stddef.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <aerospike/aerospike.h>
+#include <aerospike/as_error.h>
+#include <aerospike/as_string.h>
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+//---------------------------------
+// Types
+//---------------------------------
+
+struct as_policy_metrics_s;
+struct as_node_s;
+struct as_cluster_s;
+
+/**
+ * Callbacks for metrics listener operations.
+ */
+typedef as_status(*as_metrics_enable_listener)(as_error* err, void* udata);
+
+typedef as_status(*as_metrics_snapshot_listener)(as_error* err, struct as_cluster_s* cluster, void* udata);
+
+typedef as_status(*as_metrics_node_close_listener)(as_error* err, struct as_node_s* node, void* udata);
+
+typedef as_status(*as_metrics_disable_listener)(as_error* err, struct as_cluster_s* cluster, void* udata);
+
+/**
+ * Metrics listener callbacks.
+ */
+typedef struct as_metrics_listeners_s {
+	/**
+	 * Periodic extended metrics has been enabled for the given cluster.
+	 */
+	as_metrics_enable_listener enable_listener;
+
+	/**
+	 * A metrics snapshot has been requested for the given cluster.
+	 */
+	as_metrics_snapshot_listener snapshot_listener;
+
+	/**
+	 * A node is being dropped from the cluster.
+	 */
+	as_metrics_node_close_listener node_close_listener;
+
+	/**
+	 * Periodic extended metrics has been disabled for the given cluster.
+	 */
+	as_metrics_disable_listener disable_listener;
+
+	/**
+	 * User defined data.
+	 */
+	void* udata;
+} as_metrics_listeners;
+
+/**
+ * Client periodic metrics configuration.
+ */
+typedef struct as_metrics_policy_s {
+	/**
+	 * Listeners that handles metrics notification events. The default listener implementation
+	 * writes the metrics snapshot to a file which will later be read and forwarded to
+	 * OpenTelemetry by a separate offline application.
+	 *
+	 * The listener could be overridden to send the metrics snapshot directly to OpenTelemetry.
+	 */
+	as_metrics_listeners metrics_listeners;
+	
+	/**
+	 * Directory path to write metrics log files for listeners that write logs.
+	 *
+	 * Default: . (current directory)
+	 */
+	char report_dir[256];
+
+	/**
+	 * Metrics file size soft limit in bytes for listeners that write logs.
+	 *
+	 * When report_size_limit is reached or exceeded, the current metrics file is closed and a new
+	 * metrics file is created with a new timestamp. If report_size_limit is zero, the metrics file
+	 * size is unbounded and the file will only be closed when aerospike_disable_metrics() or
+	 * aerospike_close() is called.
+	 *
+	 * Default: 0
+	 */
+	uint64_t report_size_limit;
+
+	/**
+	 * Number of cluster tend iterations between metrics notification events. One tend iteration
+	 * is defined as as_config.tender_interval (default 1 second) plus the time to tend all
+	 * nodes.
+	 *
+	 * Default: 30
+	 */
+	uint32_t interval;
+
+	/**
+	 * Number of elapsed time range buckets in latency histograms.
+	 *
+	 * Default: 7
+	 */
+	uint32_t latency_columns;
+
+	/**
+	 * Power of 2 multiple between each range bucket in latency histograms starting at column 3. The bucket units
+	 * are in milliseconds. The first 2 buckets are "<=1ms" and ">1ms". Examples:
+	 * 
+	 * ~~~~~~~~~~{.c}
+	 * // latencyColumns=7 latencyShift=1
+	 * <=1ms >1ms >2ms >4ms >8ms >16ms >32ms
+	 *
+	 * // latencyColumns=5 latencyShift=3
+	 * <=1ms >1ms >8ms >64ms >512ms
+	 * ~~~~~~~~~~
+	 *
+	 * Default: 1
+	 */
+	uint32_t latency_shift;
+} as_metrics_policy;
+
+//---------------------------------
+// Functions
+//---------------------------------
+
+/**
+ * Initalize metrics policy
+ */
+AS_EXTERN void
+as_metrics_policy_init(as_metrics_policy* policy);
+
+/**
+ * Initalize metrics policy
+ */
+static inline void
+as_metrics_policy_set_report_dir(as_metrics_policy* policy, const char* report_dir)
+{
+	as_strncpy(policy->report_dir, report_dir, sizeof(policy->report_dir));
+}
+
+static inline void
+as_metrics_policy_set_listeners(
+	as_metrics_policy* policy, as_metrics_enable_listener enable,
+	as_metrics_disable_listener disable, as_metrics_node_close_listener node_close,
+	as_metrics_snapshot_listener snapshot, void* udata
+	)
+{
+	policy->metrics_listeners.enable_listener = enable;
+	policy->metrics_listeners.disable_listener = disable;
+	policy->metrics_listeners.node_close_listener = node_close;
+	policy->metrics_listeners.snapshot_listener = snapshot;
+	policy->metrics_listeners.udata = udata;
+}
+
+/**
+ * Enable extended periodic cluster and node latency metrics.
+ */
+AS_EXTERN as_status
+aerospike_enable_metrics(aerospike* as, as_error* err, as_metrics_policy* policy);
+
+/**
+ * Disable extended periodic cluster and node latency metrics.
+ */
+AS_EXTERN as_status
+aerospike_disable_metrics(aerospike* as, as_error* err);
+
+#ifdef __cplusplus
+} // end extern "C"
+#endif
diff --git a/src/include/aerospike/as_metrics_writer.h b/src/include/aerospike/as_metrics_writer.h
new file mode 100644
index 0000000000..b7fcf55490
--- /dev/null
+++ b/src/include/aerospike/as_metrics_writer.h
@@ -0,0 +1,81 @@
+/*
+ * Copyright 2008-2024 Aerospike, Inc.
+ *
+ * Portions may be licensed to Aerospike, Inc. under one or more contributor
+ * license agreements.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+#pragma once
+
+#include <aerospike/as_cluster.h>
+#include <aerospike/as_error.h>
+#include <aerospike/as_metrics.h>
+#include <aerospike/as_status.h>
+#include <stdio.h>
+
+#if defined(_MSC_VER)
+#define WIN32_LEAN_AND_MEAN
+#include <windows.h>
+#endif
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+//---------------------------------
+// Types
+//---------------------------------
+
+/**
+ * Default metrics listener. This implementation writes periodic metrics snapshots to a file which
+ * will later be read and forwarded to OpenTelemetry by a separate offline application.
+ */
+typedef struct as_metrics_writer_s {
+	char report_dir[256];
+	FILE* file;
+	uint64_t max_size;
+	uint64_t size;
+	uint32_t latency_columns;
+	uint32_t latency_shift;
+#ifdef _MSC_VER
+	FILETIME prev_process_times_kernel;
+	FILETIME prev_system_times_kernel;
+	FILETIME prev_process_times_user;
+	FILETIME prev_system_times_user;
+	HANDLE process;
+	DWORD pid;
+#endif
+	bool enable;
+} as_metrics_writer;
+
+//---------------------------------
+// Functions
+//---------------------------------
+
+AS_EXTERN as_status
+as_metrics_writer_create(as_error* err, const as_metrics_policy* policy, as_metrics_listeners* listeners);
+
+AS_EXTERN as_status
+as_metrics_writer_enable(as_error* err, void* udata);
+
+AS_EXTERN as_status
+as_metrics_writer_snapshot(as_error* err, as_cluster* cluster, void* udata);
+
+AS_EXTERN as_status
+as_metrics_writer_node_close(as_error* err, struct as_node_s* node, void* udata);
+
+AS_EXTERN as_status
+as_metrics_writer_disable(as_error* err, struct as_cluster_s* cluster, void* udata);
+
+#ifdef __cplusplus
+} // end extern "C"
+#endif
diff --git a/src/include/aerospike/as_node.h b/src/include/aerospike/as_node.h
index 1bab9c76ab..b2c18230e2 100644
--- a/src/include/aerospike/as_node.h
+++ b/src/include/aerospike/as_node.h
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2022 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -21,6 +21,7 @@
 #include <aerospike/as_conn_pool.h>
 #include <aerospike/as_error.h>
 #include <aerospike/as_event.h>
+#include <aerospike/as_latency.h>
 #include <aerospike/as_socket.h>
 #include <aerospike/as_partition.h>
 #include <aerospike/as_queue.h>
@@ -205,6 +206,13 @@ typedef struct as_async_conn_pool_s {
 
 } as_async_conn_pool;
 
+/**
+ * Node metrics latency bucket struct
+ */
+typedef struct as_node_metrics_s {
+	as_latency_buckets* latency;
+} as_node_metrics;
+
 struct as_cluster_s;
 
 /**
@@ -297,11 +305,28 @@ typedef struct as_node_s {
 	 */
 	as_racks* racks;
 
+	/**
+	 * Node metrics 
+	 */
+	as_node_metrics* metrics;
+
 	/**
 	 * Socket used exclusively for cluster tend thread info requests.
 	 */
 	as_socket info_socket;
 
+	/**
+	 * Transaction error count since node was initialized. If the error is retryable, multiple errors per
+	 * transaction may occur.
+	 */
+	uint64_t error_count;
+
+	/**
+	 * Transaction timeout count since node was initialized. If the timeout is retryable (ie socketTimeout),
+	 * multiple timeouts per transaction may occur.
+	 */
+	uint64_t timeout_count;
+
 	/**
 	 * Connection queue iterator.  Not atomic by design.
 	 */
@@ -320,8 +345,8 @@ typedef struct as_node_s {
 	/**
 	 * Error count for this node's error_rate_window.
 	 */
-	uint32_t error_count;
-
+	uint32_t error_rate;
+	
 	/**
 	 * Server's generation count for peers.
 	 */
@@ -429,6 +454,13 @@ as_node_create(struct as_cluster_s* cluster, as_node_info* node_info);
 AS_EXTERN void
 as_node_destroy(as_node* node);
 
+/**
+ * @private
+ * Destroy node metrics.
+ */
+void
+as_node_destroy_metrics(as_node* node);
+
 /**
  * @private
  * Create configured minimum number of connections.
@@ -637,6 +669,60 @@ as_node_signal_login(as_node* node);
 bool
 as_node_has_rack(as_node* node, const char* ns, int rack_id);
 
+/**
+ * @private
+ * Record latency of type latency_type for node
+ */
+void
+as_node_add_latency(as_node* node, as_latency_type latency_type, uint64_t elapsed);
+
+struct as_metrics_policy_s;
+
+/**
+ * @private
+ * Enable metrics at the node level
+ */
+void
+as_node_enable_metrics(as_node* node, const struct as_metrics_policy_s* policy);
+
+/**
+ * Return transaction error count. The value is cumulative and not reset per metrics interval.
+ */
+static inline uint64_t
+as_node_get_error_count(as_node* node)
+{
+	return as_load_uint64(&node->error_count);
+}
+
+/**
+ * Increment transaction error count. If the error is retryable, multiple errors per
+ * transaction may occur.
+ */
+static inline void
+as_node_add_error(as_node* node)
+{
+	as_incr_uint64(&node->error_count);
+}
+
+/**
+ * Return transaction timeout count. The value is cumulative and not reset per metrics interval.
+ */
+static inline uint64_t
+as_node_get_timeout_count(as_node* node)
+{
+	return as_load_uint64(&node->timeout_count);
+}
+
+/**
+ * Increment transaction timeout count. If the timeout is retryable (ie socketTimeout),
+ * multiple timeouts per transaction may occur.
+ */
+static inline void
+as_node_add_timeout(as_node* node)
+{
+	as_incr_uint64(&node->timeout_count);
+}
+
 /**
  * @private
  * Volatile read session pointer.
diff --git a/src/main/aerospike/aerospike.c b/src/main/aerospike/aerospike.c
index dc28dc19a6..0017ba9324 100644
--- a/src/main/aerospike/aerospike.c
+++ b/src/main/aerospike/aerospike.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -251,6 +251,15 @@ aerospike_close(aerospike* as, as_error* err)
 	as_cluster* cluster = as->cluster;
 	
 	if (cluster) {
+		if (cluster->metrics_enabled) {
+			as_status status = aerospike_disable_metrics(as, err);
+			
+			if (status != AEROSPIKE_OK) {
+				as_log_warn("Metrics error: %s %s", as_error_string(status), err->message);
+				as_error_reset(err);
+			}
+		}
+		
 		if (as_event_loop_size > 0 && !as_event_single_thread) {
 			// Async configurations will attempt to wait till pending async commands have completed.
 			as_event_close_cluster(cluster);
diff --git a/src/main/aerospike/aerospike_batch.c b/src/main/aerospike/aerospike_batch.c
index c8a6a722c3..e9fbc3fee1 100644
--- a/src/main/aerospike/aerospike_batch.c
+++ b/src/main/aerospike/aerospike_batch.c
@@ -1616,6 +1616,7 @@ as_batch_command_init(
 	cmd->buf_size = size;
 	cmd->partition_id = 0; // Not referenced when node set.
 	cmd->replica = task->replica;
+	cmd->latency_type = AS_LATENCY_TYPE_BATCH;
 
 	// Note: Do not set flags to AS_COMMAND_FLAGS_LINEARIZE because AP and SC replicas
 	// are tracked separately for batch (cmd->master and cmd->master_sc).
@@ -2080,6 +2081,8 @@ as_batch_keys_execute(
 	as_batch_base_record* rec, as_batch_attr* attr, as_batch_listener listener, void* udata
 	)
 {
+	as_cluster* cluster = as->cluster;
+	as_cluster_add_tran(cluster);
 	uint32_t n_keys = batch->keys.size;
 	
 	if (n_keys == 0) {
@@ -2089,7 +2092,6 @@ as_batch_keys_execute(
 		return AEROSPIKE_OK;
 	}
 	
-	as_cluster* cluster = as->cluster;
 	as_nodes* nodes = as_nodes_reserve(cluster);
 	uint32_t n_nodes = nodes->size;
 	as_nodes_release(nodes);
@@ -2429,6 +2431,7 @@ as_batch_command_create(
 	// cmd->replica_size = 1;
 	cmd->replica_index = rep->replica_index;
 	cmd->replica_index_sc = rep->replica_index_sc;
+	cmd->latency_type = AS_LATENCY_TYPE_BATCH;
 	return bc;
 }
 
@@ -2553,6 +2556,8 @@ as_batch_records_execute(
 	as_async_batch_executor* async_executor, bool has_write
 	)
 {
+	as_cluster* cluster = as->cluster;
+	as_cluster_add_tran(cluster);
 	as_vector* list = &records->list;
 	uint32_t n_keys = records->list.size;
 	
@@ -2560,7 +2565,6 @@ as_batch_records_execute(
 		return AEROSPIKE_OK;
 	}
 	
-	as_cluster* cluster = as->cluster;
 	as_nodes* nodes = as_nodes_reserve(cluster);
 	uint32_t n_nodes = nodes->size;
 	as_nodes_release(nodes);
@@ -2669,6 +2673,7 @@ as_batch_records_execute_async(
 	as_async_batch_listener listener, void* udata, as_event_loop* event_loop, bool has_write
 	)
 {
+	as_cluster_add_tran(as->cluster);
 	// Check for empty batch.
 	if (records->list.size == 0) {
 		listener(0, records, udata, event_loop);
@@ -2789,6 +2794,8 @@ as_batch_retry_records(as_batch_task_records* btr, as_command* parent, as_error*
 			return AEROSPIKE_USE_NORMAL_RETRY;
 		}
 	}
+	
+	as_cluster_add_retries(cluster, batch_nodes.size);
 	parent->flags |= AS_COMMAND_FLAGS_SPLIT_RETRY;
 
 	return as_batch_execute_sync(cluster, err, task->policy, btr->defs, task->has_write, &rep,
@@ -2874,6 +2881,8 @@ as_batch_retry_keys(as_batch_task_keys* btk, as_command* parent, as_error* err)
 			return AEROSPIKE_USE_NORMAL_RETRY;
 		}
 	}
+	
+	as_cluster_add_retries(cluster, batch_nodes.size);
 	parent->flags |= AS_COMMAND_FLAGS_SPLIT_RETRY;
 
 	// Run batch retries sequentially in same thread.
@@ -2949,6 +2958,7 @@ as_batch_retry_command_create(
 	// cmd->replica_size = 1;
 	cmd->replica_index = rep->replica_index;
 	cmd->replica_index_sc = rep->replica_index_sc;
+	cmd->latency_type = AS_LATENCY_TYPE_BATCH;
 	return bc;
 }
 
@@ -3256,6 +3266,8 @@ as_batch_retry_async(as_event_command* parent, bool timeout)
 		}
 	}
 
+	as_cluster_add_retries(cluster, bnodes.size);
+
 	as_event_executor* e = &be->executor;
 	pthread_mutex_lock(&e->lock);
 	e->max += bnodes.size - 1;
diff --git a/src/main/aerospike/aerospike_key.c b/src/main/aerospike/aerospike_key.c
index 927f1c80b9..d83d70c1f0 100644
--- a/src/main/aerospike/aerospike_key.c
+++ b/src/main/aerospike/aerospike_key.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -80,6 +80,8 @@ as_command_init_read(
 	cmd->udata = udata;
 	cmd->buf_size = size;
 	cmd->partition_id = pi->partition_id;
+	cmd->latency_type = AS_LATENCY_TYPE_READ;
+	as_cluster_add_tran(cluster);
 
 	if (pi->sc_mode) {
 		switch (read_mode_sc) {
@@ -143,6 +145,8 @@ as_command_init_write(
 	cmd->replica = as_command_write_replica(replica);
 	cmd->replica_size = pi->replica_size;
 	cmd->replica_index = 0;
+	cmd->latency_type = AS_LATENCY_TYPE_WRITE;
+	as_cluster_add_tran(cluster);
 }
 
 static inline void
@@ -272,7 +276,7 @@ aerospike_key_get_async(
 	as_event_command* cmd = as_async_record_command_create(
 		cluster, &policy->base, &pi, ri.replica, ri.replica_index, policy->deserialize,
 		policy->async_heap_rec, ri.flags, listener, udata, event_loop, pipe_listener, size,
-		as_event_command_parse_result);
+		as_event_command_parse_result, AS_LATENCY_TYPE_READ);
 
 	uint32_t timeout = as_command_server_timeout(&policy->base);
 	uint8_t* p = as_command_write_header_read(cmd->buf, &policy->base, policy->read_mode_ap,
@@ -384,7 +388,7 @@ aerospike_key_select_async(
 	as_event_command* cmd = as_async_record_command_create(
 		cluster, &policy->base, &pi, ri.replica, ri.replica_index, policy->deserialize,
 		policy->async_heap_rec, ri.flags, listener, udata, event_loop, pipe_listener, size,
-		as_event_command_parse_result);
+		as_event_command_parse_result, AS_LATENCY_TYPE_READ);
 
 	uint32_t timeout = as_command_server_timeout(&policy->base);
 	uint8_t* p = as_command_write_header_read(cmd->buf, &policy->base, policy->read_mode_ap,
@@ -474,7 +478,8 @@ aerospike_key_exists_async(
 
 	as_event_command* cmd = as_async_record_command_create(
 		cluster, &policy->base, &pi, ri.replica, ri.replica_index, false, policy->async_heap_rec,
-		ri.flags, listener, udata, event_loop, pipe_listener, size, as_event_command_parse_result);
+		ri.flags, listener, udata, event_loop, pipe_listener, size, as_event_command_parse_result,
+		AS_LATENCY_TYPE_READ);
 
 	uint8_t* p = as_command_write_header_read_header(cmd->buf, &policy->base, policy->read_mode_ap,
 		policy->read_mode_sc, n_fields, 0, AS_MSG_INFO1_READ | AS_MSG_INFO1_GET_NOBINDATA);
@@ -1035,13 +1040,13 @@ aerospike_key_operate_async(
 
 	as_event_command* cmd;
 
-	if (! (policy->base.compress && oper.size > AS_COMPRESS_THRESHOLD)) {
+	if (! (policy->base.compress && oper.size > AS_COMPRESS_THRESHOLD)) {		
 		// Send uncompressed command.
 		if (oper.write_attr & AS_MSG_INFO2_WRITE) {
 			cmd = as_async_record_command_create(
 				cluster, &policy->base, &pi, policy->replica, 0, policy->deserialize,
 				policy->async_heap_rec, 0, listener, udata, event_loop, pipe_listener, oper.size,
-				as_event_command_parse_result);
+				as_event_command_parse_result, AS_LATENCY_TYPE_WRITE);
 		}
 		else {
 			as_read_info ri;
@@ -1050,7 +1055,7 @@ aerospike_key_operate_async(
 			cmd = as_async_record_command_create(
 				cluster, &policy->base, &pi, ri.replica, ri.replica_index, policy->deserialize,
 				policy->async_heap_rec, ri.flags, listener, udata, event_loop, pipe_listener,
-				oper.size, as_event_command_parse_result);
+				oper.size, as_event_command_parse_result, AS_LATENCY_TYPE_READ);
 		}
 
 		cmd->write_len = (uint32_t)as_operate_write(&oper, cmd->buf);
@@ -1069,7 +1074,7 @@ aerospike_key_operate_async(
 			cmd = as_async_record_command_create(
 				cluster, &policy->base, &pi, policy->replica, 0, policy->deserialize,
 				policy->async_heap_rec, 0, listener, udata, event_loop, pipe_listener, comp_size,
-				as_event_command_parse_result);
+				as_event_command_parse_result, AS_LATENCY_TYPE_WRITE);
 		}
 		else {
 			as_read_info ri;
@@ -1078,7 +1083,7 @@ aerospike_key_operate_async(
 			cmd = as_async_record_command_create(
 				cluster, &policy->base, &pi, ri.replica, ri.replica_index, policy->deserialize,
 				policy->async_heap_rec, ri.flags, listener, udata, event_loop, pipe_listener,
-				comp_size, as_event_command_parse_result);
+				comp_size, as_event_command_parse_result, AS_LATENCY_TYPE_READ);
 		}
 
 		// Compress buffer and execute.
diff --git a/src/main/aerospike/aerospike_query.c b/src/main/aerospike/aerospike_query.c
index 348d84d3dd..087ff95e10 100644
--- a/src/main/aerospike/aerospike_query.c
+++ b/src/main/aerospike/aerospike_query.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -1132,6 +1132,7 @@ as_query_command_execute_old(as_query_task* task)
 	cmd.flags = flags;
 	cmd.replica_size = 1;
 	cmd.replica_index = 0;
+	cmd.latency_type = AS_LATENCY_TYPE_QUERY;
 
 	as_command_start_timer(&cmd);
 
@@ -1233,6 +1234,7 @@ as_query_command_execute_new(as_query_task* task)
 	cmd.flags = flags;
 	cmd.replica_size = 1;
 	cmd.replica_index = 0;
+	cmd.latency_type = AS_LATENCY_TYPE_QUERY;
 
 	as_command_start_timer(&cmd);
 
@@ -1301,6 +1303,7 @@ as_query_worker_new(void* data)
 static as_status
 as_query_execute(as_query_task* task, const as_query* query, as_nodes* nodes)
 {
+	as_cluster_add_tran(task->cluster);
 	as_status status = AEROSPIKE_OK;
 
 	if (task->query_policy && task->query_policy->fail_on_cluster_change) {
@@ -1468,6 +1471,7 @@ as_query_partitions(
 	as_cluster* cluster, as_error* err, const as_policy_query* policy, const as_query* query,
 	as_partition_tracker* pt, aerospike_query_foreach_callback callback, void* udata)
 {
+	as_cluster_add_tran(cluster);
 	uint64_t parent_id = as_random_get_uint64();
 	as_status status = AEROSPIKE_OK;
 
@@ -1482,6 +1486,10 @@ as_query_partitions(
 
 		uint32_t n_nodes = pt->node_parts.size;
 
+		if (pt->iteration > 1) {
+			as_cluster_add_retries(cluster, n_nodes);
+		}
+
 		// Initialize task.
 		uint32_t error_mutex = 0;
 
@@ -1717,6 +1725,7 @@ as_query_partition_execute_async(
 		cmd->flags = qe->deserialize ? AS_ASYNC_FLAGS_DESERIALIZE : 0;
 		cmd->replica_size = 1;
 		cmd->replica_index = 0;
+		cmd->latency_type = AS_LATENCY_TYPE_QUERY;
 		ee->commands[i] = cmd;
 	}
 
@@ -1755,6 +1764,7 @@ as_query_partition_async(
 	as_event_loop* event_loop
 	)
 {
+	as_cluster_add_tran(cluster);
 	pt->sleep_between_retries = 0;
 	as_status status = as_partition_tracker_assign(pt, cluster, query->ns, err);
 
@@ -1872,6 +1882,7 @@ as_query_partition_retry_async(as_async_query_executor* qe_old, as_error* err)
 	ee->queued = 0;
 	ee->notify = true;
 	ee->valid = true;
+	as_cluster_add_retry(qe->cluster);
 
 	return as_query_partition_execute_async(qe, qe->pt, err);
 }
@@ -2249,6 +2260,7 @@ aerospike_query_async(
 		cmd->flags = policy->deserialize ? AS_ASYNC_FLAGS_DESERIALIZE : 0;
 		cmd->replica_size = 1;
 		cmd->replica_index = 0;
+		cmd->latency_type = AS_LATENCY_TYPE_QUERY;
 		memcpy(cmd->buf, cmd_buf, size);
 		exec->commands[i] = cmd;
 	}
diff --git a/src/main/aerospike/aerospike_scan.c b/src/main/aerospike/aerospike_scan.c
index b303819c96..9d781ad2ab 100644
--- a/src/main/aerospike/aerospike_scan.c
+++ b/src/main/aerospike/aerospike_scan.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -687,6 +687,7 @@ as_scan_command_execute(as_scan_task* task)
 	cmd.flags = AS_COMMAND_FLAGS_READ;
 	cmd.replica_size = 1;
 	cmd.replica_index = 0;
+	cmd.latency_type = AS_LATENCY_TYPE_QUERY;
 
 	as_command_start_timer(&cmd);
 
@@ -760,6 +761,7 @@ as_scan_generic(
 	aerospike_scan_foreach_callback callback, void* udata, uint64_t* task_id_ptr
 	)
 {
+	as_cluster_add_tran(cluster);
 	as_status status = as_scan_validate(err, policy, scan);
 
 	if (status != AEROSPIKE_OK) {
@@ -876,6 +878,7 @@ as_scan_partitions(
 	as_cluster* cluster, as_error* err, const as_policy_scan* policy, const as_scan* scan,
 	as_partition_tracker* pt, aerospike_scan_foreach_callback callback, void* udata)
 {
+	as_cluster_add_tran(cluster);
 	uint64_t parent_id = as_random_get_uint64();
 	as_status status = AEROSPIKE_OK;
 
@@ -889,6 +892,10 @@ as_scan_partitions(
 		}
 
 		uint32_t n_nodes = pt->node_parts.size;
+		
+		if (pt->iteration > 1) {
+			as_cluster_add_retries(cluster, n_nodes);
+		}
 
 		// Initialize task.
 		uint32_t error_mutex = 0;
@@ -1091,6 +1098,7 @@ as_scan_partition_execute_async(as_async_scan_executor* se, as_partition_tracker
 		cmd->flags = se->deserialize_list_map ? AS_ASYNC_FLAGS_DESERIALIZE : 0;
 		cmd->replica_size = 1;
 		cmd->replica_index = 0;
+		cmd->latency_type = AS_LATENCY_TYPE_QUERY;
 		ee->commands[i] = cmd;
 	}
 
@@ -1163,6 +1171,7 @@ as_scan_partition_retry_async(as_async_scan_executor* se_old, as_error* err)
 	ee->queued = 0;
 	ee->notify = true;
 	ee->valid = true;
+	as_cluster_add_retry(se->cluster);
 
 	return as_scan_partition_execute_async(se, se->pt, err);
 }
@@ -1174,6 +1183,7 @@ as_scan_partition_async(
 	as_event_loop* event_loop
 	)
 {
+	as_cluster_add_tran(cluster);
 	pt->sleep_between_retries = 0;
 	as_status status = as_partition_tracker_assign(pt, cluster, scan->ns, err);
 
diff --git a/src/main/aerospike/aerospike_stats.c b/src/main/aerospike/aerospike_stats.c
index 1d48cf6e7e..b943167d23 100644
--- a/src/main/aerospike/aerospike_stats.c
+++ b/src/main/aerospike/aerospike_stats.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2021 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -32,36 +32,6 @@ extern uint32_t as_event_loop_size;
  *****************************************************************************/
 
 static inline void
-as_sum_init(as_conn_stats* stats)
-{
-	stats->in_pool = 0;
-	stats->in_use = 0;
-	stats->opened = 0;
-	stats->closed = 0;
-}
-
-static inline void
-as_sum_no_lock(as_async_conn_pool* pool, as_conn_stats* stats)
-{
-	// Warning: cross-thread reference without a lock.
-	int tmp = as_queue_size(&pool->queue);
-
-	// Timing issues may cause values to go negative. Adjust.
-	if (tmp < 0) {
-		tmp = 0;
-	}
-	stats->in_pool += tmp;
-	tmp = pool->queue.total - tmp;
-
-	if (tmp < 0) {
-		tmp = 0;
-	}
-	stats->in_use += tmp;
-	stats->opened += pool->opened;
-	stats->closed += pool->closed;
-}
-
-static void
 as_conn_stats_tostring(as_string_builder* sb, const char* title, as_conn_stats* cs)
 {
 	as_string_builder_append_char(sb, ' ');
@@ -110,6 +80,7 @@ aerospike_cluster_stats(as_cluster* cluster, as_cluster_stats* stats)
 
 	// cf_queue applies locks, so we are safe here.
 	stats->thread_pool_queued_tasks = cf_queue_sz(cluster->thread_pool.dispatch_queue);
+	stats->retry_count = cluster->retry_count;
 }
 
 void
@@ -134,10 +105,11 @@ aerospike_node_stats(as_node* node, as_node_stats* stats)
 	as_node_reserve(node); // Released in aerospike_node_stats_destroy()
 	stats->node = node;
 	stats->error_count = as_node_get_error_count(node);
+	stats->timeout_count = as_node_get_timeout_count(node);
 
-	as_sum_init(&stats->sync);
-	as_sum_init(&stats->async);
-	as_sum_init(&stats->pipeline);
+	as_conn_stats_init(&stats->sync);
+	as_conn_stats_init(&stats->async);
+	as_conn_stats_init(&stats->pipeline);
 
 	uint32_t max = node->cluster->conn_pools_per_node;
 
@@ -160,10 +132,10 @@ aerospike_node_stats(as_node* node, as_node_stats* stats)
 	if (as_event_loop_capacity > 0) {
 		for (uint32_t i = 0; i < as_event_loop_size; i++) {
 			// Regular async.
-			as_sum_no_lock(&node->async_conn_pools[i], &stats->async);
+			as_conn_stats_sum(&stats->async, &node->async_conn_pools[i]);
 
 			// Pipeline async.
-			as_sum_no_lock(&node->pipe_conn_pools[i], &stats->pipeline);
+			as_conn_stats_sum(&stats->pipeline, &node->pipe_conn_pools[i]);
 		}
 	}
 }
@@ -173,7 +145,7 @@ aerospike_stats_to_string(as_cluster_stats* stats)
 {
 	as_string_builder sb;
 	as_string_builder_init(&sb, 4096, true);
-	as_string_builder_append(&sb, "nodes(inUse,inPool,opened,closed):");
+	as_string_builder_append(&sb, "nodes(inUse,inPool,opened,closed) error_count,timeout_count");
 	as_string_builder_append_newline(&sb);
 
 	for (uint32_t i = 0; i < stats->nodes_size; i++) {
@@ -182,9 +154,10 @@ aerospike_stats_to_string(as_cluster_stats* stats)
 		as_conn_stats_tostring(&sb, "sync", &node_stats->sync);
 		as_conn_stats_tostring(&sb, "async", &node_stats->async);
 		as_conn_stats_tostring(&sb, "pipeline", &node_stats->pipeline);
-		as_string_builder_append_newline(&sb);
-		as_string_builder_append(&sb, "error count: ");
-		as_string_builder_append_uint(&sb, node_stats->error_count);
+		as_string_builder_append_char(&sb, ' ');
+		as_string_builder_append_uint64(&sb, node_stats->error_count);
+		as_string_builder_append_char(&sb, ',');
+		as_string_builder_append_uint64(&sb, node_stats->timeout_count);
 		as_string_builder_append_newline(&sb);
 	}
 
@@ -205,5 +178,30 @@ aerospike_stats_to_string(as_cluster_stats* stats)
 		}
 		as_string_builder_append_newline(&sb);
 	}
+	
+	as_string_builder_append(&sb, "retry_count: ");
+	as_string_builder_append_uint64(&sb, stats->retry_count);
+
 	return sb.data;
 }
+
+void
+as_conn_stats_sum(as_conn_stats* stats, as_async_conn_pool* pool)
+{
+	// Warning: cross-thread reference without a lock.
+	int tmp = as_queue_size(&pool->queue);
+
+	// Timing issues may cause values to go negative. Adjust.
+	if (tmp < 0) {
+		tmp = 0;
+	}
+	stats->in_pool += tmp;
+	tmp = pool->queue.total - tmp;
+
+	if (tmp < 0) {
+		tmp = 0;
+	}
+	stats->in_use += tmp;
+	stats->opened += pool->opened;
+	stats->closed += pool->closed;
+}
diff --git a/src/main/aerospike/as_cluster.c b/src/main/aerospike/as_cluster.c
index cbb7e3cf21..811f742967 100644
--- a/src/main/aerospike/as_cluster.c
+++ b/src/main/aerospike/as_cluster.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -22,6 +22,7 @@
 #include <aerospike/as_info.h>
 #include <aerospike/as_log_macros.h>
 #include <aerospike/as_lookup.h>
+#include <aerospike/as_metrics_writer.h>
 #include <aerospike/as_password.h>
 #include <aerospike/as_peers.h>
 #include <aerospike/as_shm_cluster.h>
@@ -554,19 +555,134 @@ as_cluster_remove_nodes_copy(as_cluster* cluster, as_vector* /* <as_node*> */ no
 	as_vector_append(cluster->gc, &item);
 }
 
+static void
+as_cluster_destroy_node_metrics(as_cluster* cluster)
+{
+	as_nodes* nodes = as_nodes_reserve(cluster);
+	
+	for (uint32_t i = 0; i < nodes->size; i++) {
+		as_node_destroy_metrics(nodes->array[i]);
+	}
+	as_nodes_release(nodes);
+}
+
+as_status
+as_cluster_enable_metrics(as_error* err, as_cluster* cluster, as_metrics_policy* policy)
+{
+	bool custom_listener = policy->metrics_listeners.enable_listener != NULL;
+	
+	if (custom_listener) {
+		// Ensure all listeners and user data has been defined.
+		if (! (policy->metrics_listeners.enable_listener && policy->metrics_listeners.snapshot_listener &&
+			   policy->metrics_listeners.node_close_listener &&  policy->metrics_listeners.disable_listener &&
+			   policy->metrics_listeners.udata)) {
+			return as_error_set_message(err, AEROSPIKE_ERR_PARAM, "All metrics listeners and udata must be defined");
+		}
+	}
+	
+	pthread_mutex_lock(&cluster->metrics_lock);
+
+	as_status status = AEROSPIKE_OK;
+
+	if (cluster->metrics_enabled) {
+		cluster->metrics_enabled = false;
+		status = cluster->metrics_listeners.disable_listener(err, cluster, cluster->metrics_listeners.udata);
+		as_cluster_destroy_node_metrics(cluster);
+
+		if (status != AEROSPIKE_OK) {
+			// Disabling old metrics should not prevent new metrics from being created.
+			// Log error and continue processing.
+			as_log_warn("Metrics disable error: %s %s", as_error_string(status), err->message);
+		}
+	}
+
+	as_error_reset(err);
+
+	if (custom_listener) {
+		// Copy listeners from policy.
+		cluster->metrics_listeners = policy->metrics_listeners;
+	}
+	else {
+		// Create default metrics writer and set cluster llsteners.
+		status = as_metrics_writer_create(err, policy, &cluster->metrics_listeners);
+		
+		if (status != AEROSPIKE_OK) {
+			pthread_mutex_unlock(&cluster->metrics_lock);
+			return status;
+		}
+	}
+	
+	cluster->metrics_interval = policy->interval;
+	cluster->metrics_latency_columns = policy->latency_columns;
+	cluster->metrics_latency_shift = policy->latency_shift;
+
+	as_nodes* nodes = as_nodes_reserve(cluster);
+	
+	for (uint32_t i = 0; i < nodes->size; i++) {
+		as_node* node = nodes->array[i];
+		as_node_enable_metrics(node, policy);
+	}
+	as_nodes_release(nodes);
+
+	status = cluster->metrics_listeners.enable_listener(err, cluster->metrics_listeners.udata);
+	
+	if (status != AEROSPIKE_OK) {
+		as_cluster_destroy_node_metrics(cluster);
+		pthread_mutex_unlock(&cluster->metrics_lock);
+		return status;
+	}
+
+	cluster->metrics_enabled = true;
+	pthread_mutex_unlock(&cluster->metrics_lock);
+	return status;
+}
+
+as_status
+as_cluster_disable_metrics(as_error* err, as_cluster* cluster)
+{
+	as_status status = AEROSPIKE_OK;
+	as_error_reset(err);
+
+	pthread_mutex_lock(&cluster->metrics_lock);
+
+	if (cluster->metrics_enabled) {
+		cluster->metrics_enabled = false;
+		status = cluster->metrics_listeners.disable_listener(err, cluster, cluster->metrics_listeners.udata);
+		as_cluster_destroy_node_metrics(cluster);
+	}
+
+	pthread_mutex_unlock(&cluster->metrics_lock);
+	return status;
+}
+
 static void
 as_cluster_remove_nodes(as_cluster* cluster, as_vector* /* <as_node*> */ nodes_to_remove)
 {
 	// There is no need to delete nodes from partition tables because the nodes
 	// have already been set to inactive. Further connection requests will result
 	// in an exception and a different node will be tried.
+	as_error err;
 	
 	// Set node to inactive.
 	for (uint32_t i = 0; i < nodes_to_remove->size; i++) {
 		as_node* node = as_vector_get_ptr(nodes_to_remove, i);
+		as_status status = AEROSPIKE_OK;
+
+		pthread_mutex_lock(&cluster->metrics_lock);
+
+		if (cluster->metrics_enabled) {
+			status = cluster->metrics_listeners.node_close_listener(&err, node, node->cluster->metrics_listeners.udata);
+		}
+		pthread_mutex_unlock(&cluster->metrics_lock);
+		
+		if (status != AEROSPIKE_OK) {
+			// Metrics failures should not interrupt cluster tend.
+			// Log warning and continue processing.
+			as_log_warn("Metrics error: %s %s", as_error_string(status), err.message);
+		}
 		as_node_deactivate(node);
 	}
-			
+
 	// Remove all nodes at once to avoid copying entire array multiple times.
 	as_cluster_remove_nodes_copy(cluster, nodes_to_remove);
 	
@@ -635,12 +751,12 @@ as_cluster_balance_connections(as_cluster* cluster)
 }
 
 static void
-as_cluster_reset_error_count(as_cluster* cluster)
+as_cluster_reset_error_rate(as_cluster* cluster)
 {
 	as_nodes* nodes = cluster->nodes;
 
 	for (uint32_t i = 0; i < nodes->size; i++) {
-		as_node_reset_error_count(nodes->array[i]);
+		as_node_reset_error_rate(nodes->array[i]);
 	}
 }
 
@@ -656,7 +772,24 @@ as_cluster_manage(as_cluster* cluster)
 
 	// Reset connection error window for all nodes every error_rate_window tend iterations.
 	if (cluster->max_error_rate > 0 && cluster->tend_count % cluster->error_rate_window == 0) {
-		as_cluster_reset_error_count(cluster);
+		as_cluster_reset_error_rate(cluster);
+	}
+
+	// Call metrics listener every metrics_interval when enabled.
+	as_status status = AEROSPIKE_OK;
+	as_error err;
+	
+	pthread_mutex_lock(&cluster->metrics_lock);
+
+	if (cluster->metrics_enabled && cluster->tend_count % cluster->metrics_interval == 0) {
+		status = cluster->metrics_listeners.snapshot_listener(&err, cluster, cluster->metrics_listeners.udata);
+	}
+	pthread_mutex_unlock(&cluster->metrics_lock);
+
+	if (status != AEROSPIKE_OK) {
+		// Metrics failures should not interrupt cluster tend.
+		// Log warning and continue processing.
+		as_log_warn("Metrics error: %s %s", as_error_string(status), err.message);
 	}
 }
 
@@ -841,7 +974,7 @@ as_cluster_tend(as_cluster* cluster, as_error* err, bool is_init)
 		}
 	}
 
-	cluster->invalid_node_count = as_peers_invalid_count(&peers);
+	cluster->invalid_node_count += as_peers_invalid_count(&peers);
 
 	// Refresh partition map when necessary.
 	for (uint32_t i = 0; i < nodes->size; i++) {
@@ -1294,6 +1427,7 @@ as_cluster_create(as_config* config, as_error* err, as_cluster** cluster_out)
 	}
 	cluster->seeds = trg;
 	pthread_mutex_init(&cluster->seed_lock, NULL);
+	pthread_mutex_init(&cluster->metrics_lock, NULL);
 
 	// Initialize IP map translation if provided.
 	if (config->ip_map && config->ip_map_size > 0) {
@@ -1395,6 +1529,15 @@ as_cluster_create(as_config* config, as_error* err, as_cluster** cluster_out)
 		}
 		pthread_attr_destroy(&attr);
 	}
+
+	// Initialize metrics fields
+	cluster->metrics_enabled = false;
+	cluster->metrics_interval = 0;
+	cluster->metrics_latency_columns = 0;
+	cluster->metrics_latency_shift = 0;
+	cluster->tran_count = 0;
+	cluster->retry_count = 0;
+	cluster->delay_queue_timeout_count = 0;
 	*cluster_out = cluster;
 	return AEROSPIKE_OK;
 }
@@ -1465,6 +1608,7 @@ as_cluster_destroy(as_cluster* cluster)
 	as_vector_destroy(seeds);
 	pthread_mutex_unlock(&cluster->seed_lock);
 	pthread_mutex_destroy(&cluster->seed_lock);
+	pthread_mutex_destroy(&cluster->metrics_lock);
 
 	// Destroy tend lock and condition.
 	pthread_mutex_destroy(&cluster->tend_lock);
diff --git a/src/main/aerospike/as_command.c b/src/main/aerospike/as_command.c
index 5c80e014d6..81111976d2 100644
--- a/src/main/aerospike/as_command.c
+++ b/src/main/aerospike/as_command.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -590,6 +590,7 @@ as_command_execute(as_command* cmd, as_error* err)
 	as_node* node = NULL;
 	as_status status;
 	bool release_node;
+	as_latency_type latency_type = cmd->cluster->metrics_enabled ? cmd->latency_type : AS_LATENCY_TYPE_NONE;
 
 	// Execute command until successful, timed out or maximum iterations have been reached.
 	while (true) {
@@ -615,7 +616,7 @@ as_command_execute(as_command* cmd, as_error* err)
 			release_node = true;
 		}
 
-		if (! as_node_valid_error_count(node)) {
+		if (! as_node_valid_error_rate(node)) {
 			status = as_error_set_message(err, AEROSPIKE_MAX_ERROR_RATE, "Max error rate exceeded");
 			goto Retry;
 		}
@@ -635,6 +636,11 @@ as_command_execute(as_command* cmd, as_error* err)
 			goto Retry;
 		}
 		
+		uint64_t begin = 0;
+		if (latency_type != AS_LATENCY_TYPE_NONE) {
+			begin = cf_getns();
+		}
+		
 		// Send command.
 		status = as_socket_write_deadline(err, &socket, node, cmd->buf, cmd->buf_size,
 										  cmd->socket_timeout, cmd->deadline_ms);
@@ -656,6 +662,11 @@ as_command_execute(as_command* cmd, as_error* err)
 		}
 
 		if (status == AEROSPIKE_OK) {
+			if (latency_type != AS_LATENCY_TYPE_NONE) {
+				uint64_t elapsed = cf_getns() - begin;
+				as_node_add_latency(node, latency_type, elapsed);
+			}
+			
 			// Reset error code if retry had occurred.
 			if (cmd->iteration > 0) {
 				as_error_reset(err);
@@ -668,14 +679,18 @@ as_command_execute(as_command* cmd, as_error* err)
 			switch (status) {
 				case AEROSPIKE_ERR_CLUSTER:
 				case AEROSPIKE_ERR_DEVICE_OVERLOAD:
+					as_node_add_error(node);
 					as_node_put_conn_error(node, &socket);
 					goto Retry;
 
 				case AEROSPIKE_ERR_CONNECTION:
+					as_node_add_error(node);
 					as_node_close_conn_error(node, &socket, socket.pool);
 					goto Retry;
 
 				case AEROSPIKE_ERR_TIMEOUT:
+					as_node_add_timeout(node);
+					
 					if (is_server_timeout(err)) {
 						as_node_put_conn_error(node, &socket);
 					}
@@ -690,6 +705,7 @@ as_command_execute(as_command* cmd, as_error* err)
 				case AEROSPIKE_ERR_SCAN_ABORTED:
 				case AEROSPIKE_ERR_CLIENT_ABORT:
 				case AEROSPIKE_ERR_CLIENT:
+					as_node_add_error(node);
 					as_node_close_conn_error(node, &socket, socket.pool);
 					if (release_node) {
 						as_node_release(node);
@@ -697,7 +713,18 @@ as_command_execute(as_command* cmd, as_error* err)
 					as_error_set_in_doubt(err, cmd->flags & AS_COMMAND_FLAGS_READ, cmd->sent);
 					return status;
 				
+				case AEROSPIKE_ERR_RECORD_NOT_FOUND:
+					// Do not increment error count on record not found.
+					// Add latency metrics instead.
+					if (latency_type != AS_LATENCY_TYPE_NONE) {
+						uint64_t elapsed = cf_getns() - begin;
+						as_node_add_latency(node, latency_type, elapsed);
+					}
+					as_error_set_in_doubt(err, cmd->flags & AS_COMMAND_FLAGS_READ, cmd->sent);
+					break;
+
 				default:
+					as_node_add_error(node);
 					as_error_set_in_doubt(err, cmd->flags & AS_COMMAND_FLAGS_READ, cmd->sent);
 					break;
 			}
@@ -772,6 +799,8 @@ as_command_execute(as_command* cmd, as_error* err)
 				return status;
 			}
 		}
+
+		as_cluster_add_retry(cmd->cluster);
 	}
 
 	// Retries have been exhausted.
diff --git a/src/main/aerospike/as_event.c b/src/main/aerospike/as_event.c
index 4ad1b27b91..2d569fa8bf 100644
--- a/src/main/aerospike/as_event.c
+++ b/src/main/aerospike/as_event.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -433,6 +433,13 @@ void
 as_event_command_execute_in_loop(as_event_loop* event_loop, as_event_command* cmd)
 {
 	// Initialize read buffer (buf) to be located after write buffer.
+	if (cmd->cluster->metrics_enabled) {
+		cmd->begin = cf_getns();
+	}
+	else {
+		cmd->begin = 0;
+		cmd->latency_type = AS_LATENCY_TYPE_NONE;
+	}
 	cmd->write_offset = (uint32_t)(cmd->buf - (uint8_t*)cmd);
 	cmd->buf += cmd->write_len;
 	cmd->conn = NULL;
@@ -580,6 +587,15 @@ as_event_create_connection(as_event_command* cmd, as_async_conn_pool* pool)
 	as_event_connect(cmd, pool);
 }
 
+void
+as_event_connection_complete(as_event_command* cmd)
+{
+	if (cmd->cluster->metrics_enabled) {
+		uint64_t elapsed = cf_getns() - cmd->begin;
+		as_node_add_latency(cmd->node, AS_LATENCY_TYPE_CONN, elapsed);
+	}
+}
+
 static void
 as_event_command_begin(as_event_loop* event_loop, as_event_command* cmd)
 {
@@ -611,7 +627,7 @@ as_event_command_begin(as_event_loop* event_loop, as_event_command* cmd)
 		as_node_reserve(cmd->node);
 	}
 
-	if (! as_node_valid_error_count(cmd->node)) {
+	if (! as_node_valid_error_rate(cmd->node)) {
 		event_loop->errors++;
 
 		if (as_event_command_retry(cmd, true)) {
@@ -648,7 +664,7 @@ as_event_command_begin(as_event_loop* event_loop, as_event_command* cmd)
 		if (len != 0) {
 			as_log_debug("Invalid async socket from pool: %d", len);
 			as_event_release_connection(&conn->base, pool);
-			as_node_incr_error_count(cmd->node);
+			as_node_incr_error_rate(cmd->node);
 			continue;
 		}
 
@@ -805,6 +821,8 @@ as_event_socket_timeout(as_event_command* cmd)
 		return;
 	}
 
+	as_node_add_timeout(cmd->node);
+
 	if (cmd->pipe_listener) {
 		as_pipe_timeout(cmd, true);
 		return;
@@ -829,6 +847,10 @@ as_event_delay_timeout(as_event_command* cmd)
 {
 	cmd->state = AS_ASYNC_STATE_QUEUE_ERROR;
 
+	if (cmd->latency_type != AS_LATENCY_TYPE_NONE) {
+		as_cluster_add_delay_queue_timeout(cmd->cluster);
+	}
+
 	as_error err;
 	as_error_set_message(&err, AEROSPIKE_ERR_TIMEOUT, "Delay queue timeout");
 
@@ -865,12 +887,14 @@ as_event_process_timer(as_event_command* cmd)
 void
 as_event_total_timeout(as_event_command* cmd)
 {
+	// Node should not be null at this point.
+	as_node_add_timeout(cmd->node);
+	
 	if (cmd->pipe_listener) {
 		as_pipe_timeout(cmd, false);
 		return;
 	}
 
-	// Node should not be null at this point.
 	as_event_connection_timeout(cmd, &cmd->node->async_conn_pools[cmd->event_loop->index]);
 
 	as_error err;
@@ -962,6 +986,7 @@ as_event_execute_retry(as_event_command* cmd)
 	}
 
 	// Retry command.
+	as_cluster_add_retry(cmd->cluster);
 	as_event_command_begin(cmd->event_loop, cmd);
 }
 
@@ -978,6 +1003,11 @@ as_event_put_connection(as_event_command* cmd, as_async_conn_pool* pool)
 static inline void
 as_event_response_complete(as_event_command* cmd)
 {
+	if (cmd->latency_type != AS_LATENCY_TYPE_NONE) {
+		uint64_t elapsed = cf_getns() - cmd->begin;
+		as_node_add_latency(cmd->node, cmd->latency_type, elapsed);
+	}
+	
 	if (cmd->pipe_listener != NULL) {
 		as_pipe_response_complete(cmd);
 		return;
@@ -1263,8 +1293,9 @@ as_event_response_error(as_event_command* cmd, as_error* err)
 	switch (err->code) {
 		case AEROSPIKE_ERR_CLUSTER:
 		case AEROSPIKE_ERR_DEVICE_OVERLOAD:
+			as_node_add_error(cmd->node);
+			as_node_incr_error_rate(cmd->node);
 			as_event_put_connection(cmd, pool);
-			as_node_incr_error_count(cmd->node);
 			break;
 
 		case AEROSPIKE_ERR_QUERY_ABORTED:
@@ -1274,11 +1305,28 @@ as_event_response_error(as_event_command* cmd, as_error* err)
 		case AEROSPIKE_ERR_CLIENT_ABORT:
 		case AEROSPIKE_ERR_CLIENT:
 		case AEROSPIKE_NOT_AUTHENTICATED:
+			as_node_add_error(cmd->node);
+			as_node_incr_error_rate(cmd->node);
 			as_event_release_connection(cmd->conn, pool);
-			as_node_incr_error_count(cmd->node);
+			break;
+		
+		case AEROSPIKE_ERR_TIMEOUT:
+			as_node_add_timeout(cmd->node);
+			as_event_put_connection(cmd, pool);
 			break;
 			
+		case AEROSPIKE_ERR_RECORD_NOT_FOUND:
+			// Do not increment error count on record not found.
+			// Add latency metrics instead.
+			if (cmd->latency_type != AS_LATENCY_TYPE_NONE) {
+				uint64_t elapsed = cf_getns() - cmd->begin;
+				as_node_add_latency(cmd->node, cmd->latency_type, elapsed);
+			}
+			as_event_put_connection(cmd, pool);
+			break;
+
 		default:
+			as_node_add_error(cmd->node);
 			as_event_put_connection(cmd, pool);
 			break;
 	}
@@ -1794,7 +1842,7 @@ as_event_balance_connections_node(as_event_loop* event_loop, as_cluster* cluster
 		// Do not close idle pipeline connections because pipelines work better with a stable
 		// number of connections.
 	}
-	else if (excess < 0 && as_node_valid_error_count(node)) {
+	else if (excess < 0 && as_node_valid_error_rate(node)) {
 		create_connections(event_loop, node, pool, -excess);
 	}
 }
diff --git a/src/main/aerospike/as_event_ev.c b/src/main/aerospike/as_event_ev.c
index b234306a30..aad3a7391d 100644
--- a/src/main/aerospike/as_event_ev.c
+++ b/src/main/aerospike/as_event_ev.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2022 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -389,6 +389,21 @@ as_event_command_write_start(as_event_command* cmd)
 	as_ev_command_write(cmd);
 }
 
+static int
+as_ev_command_start(as_event_command* cmd)
+{
+	as_event_connection_complete(cmd);
+	
+	if (cmd->type == AS_ASYNC_TYPE_CONNECTOR) {
+		as_event_connector_success(cmd);
+		return AS_EVENT_COMMAND_DONE;
+	}
+	else {
+		as_event_command_write_start(cmd);
+		return AS_EVENT_READ_COMPLETE;
+	}
+}
+
 static inline void
 as_ev_command_auth_write(as_event_command* cmd)
 {
@@ -402,7 +417,7 @@ as_ev_command_auth_write(as_event_command* cmd)
 }
 
 static void
-as_ev_command_start(as_event_command* cmd)
+as_ev_connect_complete(as_event_command* cmd)
 {
 	if (cmd->cluster->auth_enabled) {
 		as_session* session = as_session_load(&cmd->node->session);
@@ -416,14 +431,11 @@ as_ev_command_start(as_event_command* cmd)
 			as_ev_command_auth_write(cmd);
 		}
 		else {
-			as_event_command_write_start(cmd);
+			as_ev_command_start(cmd);
 		}
 	}
-	else if (cmd->type == AS_ASYNC_TYPE_CONNECTOR) {
-		as_event_connector_success(cmd);
-	}
 	else {
-		as_event_command_write_start(cmd);
+		as_ev_command_start(cmd);
 	}
 }
 
@@ -528,13 +540,7 @@ as_ev_parse_authentication(as_event_command* cmd)
 		return AS_EVENT_READ_ERROR;
 	}
 
-	if (cmd->type == AS_ASYNC_TYPE_CONNECTOR) {
-		as_event_connector_success(cmd);
-		return AS_EVENT_COMMAND_DONE;
-	}
-
-	as_event_command_write_start(cmd);
-	return AS_EVENT_READ_COMPLETE;
+	return as_ev_command_start(cmd);
 }
 
 static int
@@ -629,7 +635,7 @@ as_ev_tls_connect(as_event_command* cmd, as_event_connection* conn)
 	}
 
 	// TLS connection established.
-	as_ev_command_start(cmd);
+	as_ev_connect_complete(cmd);
 	return false;
 }
 
@@ -637,7 +643,7 @@ static void
 as_ev_callback_common(as_event_command* cmd, as_event_connection* conn) {
 	switch (cmd->state) {
 	case AS_ASYNC_STATE_CONNECT:
-		as_ev_command_start(cmd);
+		as_ev_connect_complete(cmd);
 		break;
 
 	case AS_ASYNC_STATE_TLS_CONNECT:
diff --git a/src/main/aerospike/as_event_event.c b/src/main/aerospike/as_event_event.c
index 96afc87ee0..40107ef948 100644
--- a/src/main/aerospike/as_event_event.c
+++ b/src/main/aerospike/as_event_event.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2022 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -469,6 +469,21 @@ as_event_command_write_start(as_event_command* cmd)
 	as_event_command_write(cmd);
 }
 
+static int
+as_event_command_start(as_event_command* cmd)
+{
+	as_event_connection_complete(cmd);
+	
+	if (cmd->type == AS_ASYNC_TYPE_CONNECTOR) {
+		as_event_connector_success(cmd);
+		return AS_EVENT_COMMAND_DONE;
+	}
+	else {
+		as_event_command_write_start(cmd);
+		return AS_EVENT_READ_COMPLETE;
+	}
+}
+
 static inline void
 as_event_command_auth_write(as_event_command* cmd)
 {
@@ -482,7 +497,7 @@ as_event_command_auth_write(as_event_command* cmd)
 }
 
 static inline void
-as_event_command_start(as_event_command* cmd)
+as_event_connect_complete(as_event_command* cmd)
 {
 	if (cmd->cluster->auth_enabled) {
 		as_session* session = as_session_load(&cmd->node->session);
@@ -496,14 +511,11 @@ as_event_command_start(as_event_command* cmd)
 			as_event_command_auth_write(cmd);
 		}
 		else {
-			as_event_command_write_start(cmd);
+			as_event_command_start(cmd);
 		}
 	}
-	else if (cmd->type == AS_ASYNC_TYPE_CONNECTOR) {
-		as_event_connector_success(cmd);
-	}
 	else {
-		as_event_command_write_start(cmd);
+		as_event_command_start(cmd);
 	}
 }
 
@@ -608,13 +620,7 @@ as_event_parse_authentication(as_event_command* cmd)
 		return AS_EVENT_READ_ERROR;
 	}
 	
-	if (cmd->type == AS_ASYNC_TYPE_CONNECTOR) {
-		as_event_connector_success(cmd);
-		return AS_EVENT_COMMAND_DONE;
-	}
-
-	as_event_command_write_start(cmd);
-	return AS_EVENT_READ_COMPLETE;
+	return as_event_command_start(cmd);
 }
 
 static int
@@ -709,7 +715,7 @@ as_event_tls_connect(as_event_command* cmd, as_event_connection* conn)
 	}
 
 	// TLS connection established.
-	as_event_command_start(cmd);
+	as_event_connect_complete(cmd);
 	return false;
 }
 
@@ -717,7 +723,7 @@ static void
 as_event_callback_common(as_event_command* cmd, as_event_connection* conn) {
 	switch (cmd->state) {
 	case AS_ASYNC_STATE_CONNECT:
-		as_event_command_start(cmd);
+		as_event_connect_complete(cmd);
 		break;
 
 	case AS_ASYNC_STATE_TLS_CONNECT:
diff --git a/src/main/aerospike/as_event_uv.c b/src/main/aerospike/as_event_uv.c
index 4ceb44c90f..d15d485da9 100644
--- a/src/main/aerospike/as_event_uv.c
+++ b/src/main/aerospike/as_event_uv.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2022 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -434,6 +434,8 @@ as_uv_command_write_start(as_event_command* cmd, uv_stream_t* stream)
 static inline void
 as_uv_command_start(as_event_command* cmd, uv_stream_t* stream)
 {
+	as_event_connection_complete(cmd);
+	
 	if (cmd->type == AS_ASYNC_TYPE_CONNECTOR) {
 		as_event_connector_success(cmd);
 		return;
@@ -935,6 +937,8 @@ as_uv_tls_command_write_start(as_event_command* cmd)
 static inline void
 as_uv_tls_command_start(as_event_command* cmd)
 {
+	as_event_connection_complete(cmd);
+
 	if (cmd->type == AS_ASYNC_TYPE_CONNECTOR) {
 		as_event_connector_success(cmd);
 		return;
diff --git a/src/main/aerospike/as_latency.c b/src/main/aerospike/as_latency.c
new file mode 100644
index 0000000000..5372a325cc
--- /dev/null
+++ b/src/main/aerospike/as_latency.c
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2008-2024 Aerospike, Inc.
+ *
+ * Portions may be licensed to Aerospike, Inc. under one or more contributor
+ * license agreements.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+#include <aerospike/as_latency.h>
+
+char*
+as_latency_type_to_string(as_latency_type type)
+{
+	switch (type) {
+	case AS_LATENCY_TYPE_CONN:
+		return "conn"; 
+
+	case AS_LATENCY_TYPE_WRITE:
+		return "write";
+
+	case AS_LATENCY_TYPE_READ:
+		return "read";
+
+	case AS_LATENCY_TYPE_BATCH:
+		return "batch";
+
+	case AS_LATENCY_TYPE_QUERY:
+		return "query";
+
+	default:
+	case AS_LATENCY_TYPE_NONE:
+		return "none";
+	}
+}
diff --git a/src/main/aerospike/as_metrics.c b/src/main/aerospike/as_metrics.c
new file mode 100644
index 0000000000..3daaa41a43
--- /dev/null
+++ b/src/main/aerospike/as_metrics.c
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2008-2024 Aerospike, Inc.
+ *
+ * Portions may be licensed to Aerospike, Inc. under one or more contributor
+ * license agreements.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+#include <aerospike/as_metrics.h>
+#include <aerospike/as_cluster.h>
+#include <aerospike/as_event.h>
+#include <aerospike/as_node.h>
+#include <aerospike/as_string_builder.h>
+
+//---------------------------------
+// Functions
+//---------------------------------
+
+as_status
+aerospike_enable_metrics(aerospike* as, as_error* err, as_metrics_policy* policy)
+{
+	return as_cluster_enable_metrics(err, as->cluster, policy);
+}
+
+as_status
+aerospike_disable_metrics(aerospike* as, as_error* err)
+{
+	return as_cluster_disable_metrics(err, as->cluster);
+}
+
+void
+as_metrics_policy_init(as_metrics_policy* policy)
+{
+	policy->report_size_limit = 0;
+	as_strncpy(policy->report_dir, ".", sizeof(policy->report_dir));
+	policy->interval = 30;
+	policy->latency_columns = 7;
+	policy->latency_shift = 1;
+	policy->metrics_listeners.enable_listener = NULL;
+	policy->metrics_listeners.snapshot_listener = NULL;
+	policy->metrics_listeners.node_close_listener = NULL;
+	policy->metrics_listeners.disable_listener = NULL;
+	policy->metrics_listeners.udata = NULL;
+}
diff --git a/src/main/aerospike/as_metrics_writer.c b/src/main/aerospike/as_metrics_writer.c
new file mode 100644
index 0000000000..4b6a9f1410
--- /dev/null
+++ b/src/main/aerospike/as_metrics_writer.c
@@ -0,0 +1,688 @@
+/*
+ * Copyright 2008-2024 Aerospike, Inc.
+ *
+ * Portions may be licensed to Aerospike, Inc. under one or more contributor
+ * license agreements.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+#include <aerospike/as_metrics_writer.h>
+#include <aerospike/aerospike_stats.h>
+#include <aerospike/as_event.h>
+#include <aerospike/as_string_builder.h>
+#include <time.h>
+
+//---------------------------------
+// Macros
+//---------------------------------
+
+#define MIN_FILE_SIZE 1000000
+
+#ifdef _MSC_VER
+static char as_dir_sep = '\\';
+#else
+static char as_dir_sep = '/';
+#endif
+
+//---------------------------------
+// Linux Static Functions
+//---------------------------------
+
+#if defined(__linux__)
+#include <unistd.h>
+#include <sys/sysinfo.h>
+
+static as_status
+as_metrics_proc_stat_mem_cpu(as_error* err, double* vm_usage, double* resident_set, double* cpu_usage)
+{
+	*vm_usage = 0.0;
+	*resident_set = 0.0;
+
+	FILE* proc_stat = fopen("/proc/self/stat", "r");
+	if (!proc_stat) {
+		return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+			"Error calculating memory and CPU usage");
+	}
+
+	// dummies
+	int dummy_d;
+	char dummy_c;
+	unsigned int dummy_u;
+	long unsigned int dummy_lu;
+	long int dummy_ld;
+
+	// the fields we want
+	uint64_t utime, stime;
+	long long unsigned int starttime;
+	uint64_t vsize;
+	int64_t rss;
+
+	int matched = fscanf(proc_stat, "%d %s %c %d %d %d %d %d %u %lu %lu %lu %lu %lu %lu %ld %ld %ld %ld %ld %ld %llu %lu %ld",
+		&dummy_d, &dummy_c, &dummy_c, &dummy_d, &dummy_d, &dummy_d, &dummy_d, &dummy_d, &dummy_u, &dummy_lu, &dummy_lu, &dummy_lu, &dummy_lu,
+		&utime, &stime, &dummy_ld, &dummy_ld, &dummy_ld, &dummy_ld, &dummy_ld, &dummy_ld, &starttime, &vsize, &rss);
+
+	if (matched == 0) {
+		return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+			"Error calculating memory and CPU usage");
+	}
+
+	int result = fclose(proc_stat);
+
+	if (result != 0) {
+		return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+			"Error closing /proc/self/stat");
+	}
+
+	int64_t page_size_kb = sysconf(_SC_PAGE_SIZE) / 1024; // in case x86-64 is configured to use 2MB pages
+	*vm_usage = vsize / 1024.0;
+	*resident_set = rss * page_size_kb;
+
+	float u_time_sec = utime / sysconf(_SC_CLK_TCK);
+	float s_time_sec = stime / sysconf(_SC_CLK_TCK);
+	float start_time_sec = starttime / sysconf(_SC_CLK_TCK);
+
+	struct sysinfo info;
+	int success = sysinfo(&info);
+	if (success != 0) {
+		return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+			"Error calculating CPU usage");
+	}
+
+	*cpu_usage = (u_time_sec + s_time_sec) / (info.uptime - start_time_sec) * 100;
+
+	return AEROSPIKE_OK;
+}
+
+static as_status
+as_metrics_process_cpu_load_mem_usage(as_error* err, as_metrics_writer* mw, uint32_t* cpu_usage, uint32_t* mem)
+{
+	double resident_set = 0.0;
+	double mem_d = 0.0;
+	double cpu_usage_d = 0.0;
+	as_status result = as_metrics_proc_stat_mem_cpu(err, &mem_d, &resident_set, &cpu_usage_d);
+	if (result != AEROSPIKE_OK) {
+		return result;
+	}
+
+	cpu_usage_d = cpu_usage_d + 0.5 - (cpu_usage_d < 0);
+	mem_d = mem_d + 0.5 - (mem_d < 0);
+	*cpu_usage = (uint32_t)cpu_usage_d;
+	*mem = (uint32_t)mem_d;
+
+	return AEROSPIKE_OK;
+}
+#endif
+
+//---------------------------------
+// MacOS Static Functions
+//---------------------------------
+
+#if defined(__APPLE__)
+#include<mach/mach.h>
+#include<unistd.h>
+#include<sys/syscall.h>
+
+static double
+as_metrics_process_mem_usage()
+{
+	struct task_basic_info t_info;
+	mach_msg_type_number_t t_info_count = TASK_BASIC_INFO_COUNT;
+
+	if (KERN_SUCCESS != task_info(mach_task_self(), TASK_BASIC_INFO, (task_info_t)&t_info, &t_info_count))
+	{
+		return -1.0;
+	}
+
+	return t_info.resident_size;
+}
+
+static double
+as_metrics_process_cpu_load()
+{
+	pid_t pid = getpid();
+
+	as_string_builder sb;
+	as_string_builder_inita(&sb, 128, false);
+	as_string_builder_append(&sb, "ps -p ");
+	as_string_builder_append_int(&sb, pid);
+	as_string_builder_append(&sb, " -o %cpu");
+
+	FILE* file = popen(sb.data, "r");
+
+	if (!file) {
+		return -1.0;
+	}
+	
+	char cpu_holder[5];
+	char cpu_percent[6];
+	
+	if (!fgets(cpu_holder, sizeof(cpu_holder), file)) {
+		pclose(file);
+		return 0.0;
+	}
+	
+	if (!fgets(cpu_percent, sizeof(cpu_percent), file)) {
+		pclose(file);
+		return 0.0;
+	}
+	
+	pclose(file);
+	return atof(cpu_percent);
+}
+
+static as_status
+as_metrics_process_cpu_load_mem_usage(as_error* err, as_metrics_writer* mw, uint32_t* cpu_usage, uint32_t* mem)
+{
+	double cpu_usage_d = as_metrics_process_cpu_load();
+	double mem_d = as_metrics_process_mem_usage();
+
+	if (cpu_usage_d < 0) {
+		return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+			"Error calculating CPU usage");
+	}
+
+	if (mem_d < 0) {
+		return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+			"Error calculating memory usage");
+	}
+
+	// Round values.
+	cpu_usage_d = cpu_usage_d + 0.5;
+	mem_d = mem_d + 0.5;
+	*cpu_usage = (uint32_t)cpu_usage_d;
+	*mem = (uint32_t)mem_d;
+
+	return AEROSPIKE_OK;
+}
+#endif
+
+//---------------------------------
+// Microsoft Static Functions
+//---------------------------------
+
+#if defined(_MSC_VER)
+#include <Psapi.h>
+
+static ULONGLONG
+as_metrics_filetime_difference(FILETIME* prev_kernel, FILETIME* prev_user, FILETIME* cur_kernel, FILETIME* cur_user) {
+	LARGE_INTEGER a1, a2;
+	a1.LowPart = prev_kernel->dwLowDateTime;
+	a1.HighPart = prev_kernel->dwHighDateTime;
+	a2.LowPart = prev_user->dwLowDateTime;
+	a2.HighPart = prev_user->dwHighDateTime;
+
+	LARGE_INTEGER b1, b2;
+	b1.LowPart = cur_kernel->dwLowDateTime;
+	b1.HighPart = cur_kernel->dwHighDateTime;
+	b2.LowPart = cur_user->dwLowDateTime;
+	b2.HighPart = cur_user->dwHighDateTime;
+
+	//a1 and b1 - contains kernel times
+	//a2 and b2 - contains user times
+	return (b1.QuadPart - a1.QuadPart) + (b2.QuadPart - a2.QuadPart);
+}
+
+static double
+as_metrics_process_cpu_load(as_metrics_writer* mw)
+{
+	if (mw->process == NULL) {
+		return -1;
+	}
+
+	FILETIME dummy;
+	FILETIME process_times_kernel, process_times_user, system_times_kernel, system_times_user;
+	
+	if (GetProcessTimes(mw->process, &dummy, &dummy, &process_times_kernel, &process_times_user) == 0) {
+		return -1;
+	}
+	if (GetSystemTimes(0, &system_times_kernel, &system_times_user) == 0) {
+		return -1;
+	}
+
+	// Get diffrence latest - previous times.
+	ULONGLONG proc = as_metrics_filetime_difference(&mw->prev_process_times_kernel, &mw->prev_process_times_user,
+		&process_times_kernel, &process_times_user);
+	ULONGLONG system = as_metrics_filetime_difference(&mw->prev_system_times_kernel, &mw->prev_system_times_user,
+		&system_times_kernel, &system_times_user);
+	double usage = 0.0;
+
+	// Calcualte percentage.
+	if (system != 0) {
+		usage = 100.0 * (proc / (double)system);
+	}
+
+	// Assign latest times to previous times for the next round of calculation.
+	mw->prev_process_times_kernel = process_times_kernel;
+	mw->prev_process_times_user = process_times_user;
+	mw->prev_system_times_kernel = system_times_kernel;
+	mw->prev_system_times_user = system_times_user;
+
+	return usage;
+}
+
+static uint32_t
+as_metrics_process_mem_usage()
+{
+	PROCESS_MEMORY_COUNTERS memCounter;
+	BOOL result = GetProcessMemoryInfo(GetCurrentProcess(),
+		&memCounter,
+		sizeof(memCounter));
+
+	return (uint32_t)memCounter.WorkingSetSize;
+}
+
+static as_status
+as_metrics_process_cpu_load_mem_usage(as_error* err, as_metrics_writer* mw, uint32_t* cpu_usage, uint32_t* mem)
+{
+	double cpu_usage_d = as_metrics_process_cpu_load(mw);
+	if (cpu_usage_d < 0) {
+		return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+			"Error calculating CPU usage");
+	}
+	cpu_usage_d = cpu_usage_d + 0.5 - (cpu_usage_d < 0);
+	*cpu_usage = (uint32_t)cpu_usage_d;
+	*mem = as_metrics_process_mem_usage();
+
+	return AEROSPIKE_OK;
+}
+#endif
+
+//---------------------------------
+// Static Functions
+//---------------------------------
+
+static as_status
+as_metrics_open_writer(as_metrics_writer* mw, as_error* err);
+
+static void
+timestamp_to_string(char* str, size_t str_size)
+{
+	time_t now = time(NULL);
+	struct tm* local = localtime(&now);
+	snprintf(str, str_size,
+		"%4d-%02d-%02d %02d:%02d:%02d",
+		1900 + local->tm_year, local->tm_mon + 1, local->tm_mday,
+		local->tm_hour, local->tm_min, local->tm_sec);
+}
+
+static void
+timestamp_to_string_filename(char* str, size_t str_size)
+{
+	time_t now = time(NULL);
+	struct tm* local = localtime(&now);
+	snprintf(str, str_size,
+		"%4d%02d%02d%02d%02d%02d",
+		1900 + local->tm_year, local->tm_mon + 1, local->tm_mday,
+		local->tm_hour, local->tm_min, local->tm_sec);
+}
+
+static as_status
+as_metrics_write_line(as_metrics_writer* mw, const char* data, as_error* err)
+{
+	int written = fprintf(mw->file, "%s", data);
+	if (written <= 0) {
+		return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+			"Failed to write metrics data: %d,%s", written, mw->report_dir);
+	}
+	mw->size += written;
+
+	if (mw->max_size > 0 && mw->size >= mw->max_size) {
+		uint32_t result = fclose(mw->file);
+
+		if (result != 0) {
+			return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+				"File stream did not close successfully: %s", mw->report_dir);
+		}
+		return as_metrics_open_writer(mw, err);
+	}
+
+	return AEROSPIKE_OK;
+}
+
+static as_status
+as_metrics_open_writer(as_metrics_writer* mw, as_error* err)
+{
+	as_error_reset(err);
+	char now_file_str[128];
+	timestamp_to_string_filename(now_file_str, sizeof(now_file_str));
+	
+	as_string_builder file_name;
+	as_string_builder_inita(&file_name, 256, false);
+	as_string_builder_append(&file_name, mw->report_dir);
+	char last_char = mw->report_dir[(strlen(mw->report_dir) - 1)];
+	if (last_char != '/' && last_char != '\\') {
+		as_string_builder_append_char(&file_name, as_dir_sep);
+	}
+	as_string_builder_append(&file_name, "metrics-");
+	as_string_builder_append(&file_name, now_file_str);
+	as_string_builder_append(&file_name, ".log");
+	mw->file = fopen(file_name.data, "w");
+
+	if (!mw->file) {
+		return as_error_update(err, AEROSPIKE_ERR_CLIENT, "Failed to open file: %s", file_name.data);
+	}
+
+	mw->size = 0;
+	char now_str[128];
+	timestamp_to_string(now_str, sizeof(now_str));
+	
+	char data[512];
+	int rv = snprintf(data, sizeof(data), "%s header(1) cluster[name,cpu,mem,invalidNodeCount,tranCount,retryCount,delayQueueTimeoutCount,eventloop[],node[]] eventloop[processSize,queueSize] node[name,address,port,syncConn,asyncConn,errors,timeouts,latency[]] conn[inUse,inPool,opened,closed] latency(%u,%u)[type[l1,l2,l3...]]\n",
+		now_str, mw->latency_columns, mw->latency_shift);
+	if (rv <= 0) {
+		fclose(mw->file);
+		return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+			"Failed to write metrics header: %d,%s", rv, file_name.data);
+	}
+
+	as_status status = as_metrics_write_line(mw, data, err);
+	
+	if (status != AEROSPIKE_OK) {
+		fclose(mw->file);
+	}
+	return status;
+}
+
+static void
+as_metrics_get_node_sync_conn_stats(const struct as_node_s* node, struct as_conn_stats_s* sync)
+{
+	uint32_t max = node->cluster->conn_pools_per_node;
+
+	// Sync connection summary.
+	for (uint32_t i = 0; i < max; i++) {
+		as_conn_pool* pool = &node->sync_conn_pools[i];
+
+		pthread_mutex_lock(&pool->lock);
+		uint32_t in_pool = as_queue_size(&pool->queue);
+		uint32_t total = pool->queue.total;
+		pthread_mutex_unlock(&pool->lock);
+
+		sync->in_pool += in_pool;
+		sync->in_use += total - in_pool;
+	}
+	sync->opened = node->sync_conns_opened;
+	sync->closed = node->sync_conns_closed;
+}
+
+static void
+as_metrics_get_node_async_conn_stats(const struct as_node_s* node, struct as_conn_stats_s* async)
+{
+	// Async connection summary.
+	for (uint32_t i = 0; i < as_event_loop_size; i++) {
+		// Regular async.
+		as_conn_stats_sum(async, &node->async_conn_pools[i]);
+	}
+}
+
+static void
+as_metrics_write_conn(as_metrics_writer* mw, as_string_builder* sb, const struct as_conn_stats_s* stats)
+{
+	as_string_builder_append_uint(sb, stats->in_use);
+	as_string_builder_append_char(sb, ',');
+	as_string_builder_append_uint(sb, stats->in_pool);
+	as_string_builder_append_char(sb, ',');
+	as_string_builder_append_uint(sb, stats->opened); // Cumulative. Not reset on each interval.
+	as_string_builder_append_char(sb, ',');
+	as_string_builder_append_uint(sb, stats->closed); // Cumulative. Not reset on each interval.
+}
+
+static void
+as_metrics_write_node(as_metrics_writer* mw, as_string_builder* sb, struct as_node_s* node)
+{
+	as_string_builder_append_char(sb, '[');
+	as_string_builder_append(sb, node->name);
+	as_string_builder_append_char(sb, ',');
+	
+	as_address* address = as_node_get_address(node);
+	struct sockaddr* addr = (struct sockaddr*)&address->addr;
+	
+	char address_name[AS_IP_ADDRESS_SIZE];
+	as_address_short_name(addr, address_name, sizeof(address_name));
+	as_string_builder_append(sb, address_name);
+	as_string_builder_append_char(sb, ',');
+
+	uint16_t port = as_address_port(addr);
+	as_string_builder_append_uint(sb, port);
+	as_string_builder_append_char(sb, ',');
+
+	struct as_conn_stats_s sync;
+	struct as_conn_stats_s async;
+	as_conn_stats_init(&sync);
+	as_conn_stats_init(&async);
+	as_metrics_get_node_sync_conn_stats(node, &sync);
+	as_metrics_write_conn(mw, sb, &sync);
+	as_string_builder_append_char(sb, ',');
+	as_metrics_get_node_async_conn_stats(node, &async);
+	as_metrics_write_conn(mw, sb, &async);
+	as_string_builder_append_char(sb, ',');
+
+	as_string_builder_append_uint64(sb, as_node_get_error_count(node));
+	as_string_builder_append_char(sb, ',');
+	as_string_builder_append_uint64(sb, as_node_get_timeout_count(node));
+	as_string_builder_append(sb, ",[");
+
+	as_node_metrics* node_metrics = node->metrics;
+	uint32_t max = AS_LATENCY_TYPE_NONE;
+
+	for (uint32_t i = 0; i < max; i++) {
+		if (i > 0) {
+			as_string_builder_append_char(sb, ',');
+		}
+		as_string_builder_append(sb, as_latency_type_to_string(i));
+		as_string_builder_append_char(sb, '[');
+
+		as_latency_buckets* buckets = &node_metrics->latency[i];
+		uint32_t bucket_max = buckets->latency_columns;
+
+		for (uint32_t j = 0; j < bucket_max; j++) {
+			if (j > 0) {
+				as_string_builder_append_char(sb, ',');
+			}
+			as_string_builder_append_uint64(sb, as_latency_get_bucket(buckets, j));
+		}
+		as_string_builder_append_char(sb, ']');
+	}
+	as_string_builder_append(sb, "]]");
+}
+
+static as_status
+as_metrics_write_cluster(as_error* err, as_metrics_writer* mw, as_cluster* cluster)
+{
+	char* cluster_name = cluster->cluster_name;
+	
+	if (cluster_name == NULL) {
+		cluster_name = "";
+	}
+
+	uint32_t cpu_load = 0;
+	uint32_t mem = 0;
+	as_status result = as_metrics_process_cpu_load_mem_usage(err, mw, &cpu_load, &mem);
+	if (result != AEROSPIKE_OK) {
+		return result;
+	}
+
+	char now_str[128];
+	timestamp_to_string(now_str, sizeof(now_str));
+	as_string_builder sb;
+	as_string_builder_inita(&sb, 16384, true);
+	as_string_builder_append(&sb, now_str);
+	as_string_builder_append(&sb, " cluster[");
+	as_string_builder_append(&sb, cluster_name);
+	as_string_builder_append_char(&sb, ',');
+	as_string_builder_append_int(&sb, cpu_load);
+	as_string_builder_append_char(&sb, ',');
+	as_string_builder_append_int(&sb, mem);
+	as_string_builder_append_char(&sb, ',');
+	as_string_builder_append_uint(&sb, cluster->invalid_node_count); // Cumulative. Not reset on each interval.
+	as_string_builder_append_char(&sb, ',');
+	as_string_builder_append_uint64(&sb, as_cluster_get_tran_count(cluster));  // Cumulative. Not reset on each interval.
+	as_string_builder_append_char(&sb, ',');
+	as_string_builder_append_uint64(&sb, as_cluster_get_retry_count(cluster)); // Cumulative. Not reset on each interval.
+	as_string_builder_append_char(&sb, ',');
+	as_string_builder_append_uint64(&sb, as_cluster_get_delay_queue_timeout_count(cluster)); // Cumulative. Not reset on each interval.
+	as_string_builder_append(&sb, ",[");
+
+	for (uint32_t i = 0; i < as_event_loop_size; i++) {
+		as_event_loop* loop = &as_event_loops[i];
+		if (i > 0) {
+			as_string_builder_append_char(&sb, ',');
+		}
+		as_string_builder_append_char(&sb, '[');
+		as_string_builder_append_int(&sb, as_event_loop_get_process_size(loop));
+		as_string_builder_append_char(&sb, ',');
+		as_string_builder_append_uint(&sb, as_event_loop_get_queue_size(loop));
+		as_string_builder_append_char(&sb, ']');
+	}
+	as_string_builder_append(&sb, "],[");
+
+	as_nodes* nodes = as_nodes_reserve(cluster);
+	
+	for (uint32_t i = 0; i < nodes->size; i++) {
+		as_node* node = nodes->array[i];
+		
+		if (i > 0) {
+			as_string_builder_append_char(&sb, ',');
+		}
+		as_metrics_write_node(mw, &sb, node);
+	}
+	as_nodes_release(nodes);
+	as_string_builder_append(&sb, "]]");
+
+	as_string_builder_append_newline(&sb);
+	as_status status = as_metrics_write_line(mw, sb.data, err);
+	as_string_builder_destroy(&sb);
+	return status;
+}
+
+static void
+as_metrics_writer_destroy(as_metrics_writer* mw)
+{
+	fclose(mw->file);
+	cf_free(mw);
+}
+
+//---------------------------------
+// Public Functions
+//---------------------------------
+
+as_status
+as_metrics_writer_create(as_error* err, const as_metrics_policy* policy, as_metrics_listeners* listeners)
+{
+	if (policy->report_size_limit != 0 && policy->report_size_limit < MIN_FILE_SIZE) {
+		return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+			"Metrics policy report_size_limit %" PRIu64 " must be at least %d", policy->report_size_limit, MIN_FILE_SIZE);
+	}
+
+	as_metrics_writer* mw = cf_calloc(1, sizeof(as_metrics_writer));
+	as_strncpy(mw->report_dir, policy->report_dir, sizeof(mw->report_dir));
+	mw->max_size = policy->report_size_limit;
+	mw->latency_columns = policy->latency_columns;
+	mw->latency_shift = policy->latency_shift;
+	mw->enable = false;
+
+#ifdef _MSC_VER
+	mw->pid = GetCurrentProcessId();
+	mw->process = OpenProcess(PROCESS_QUERY_INFORMATION, false, mw->pid);
+
+	FILETIME dummy;
+	if (mw->process != NULL)
+	{
+		GetProcessTimes(mw->process, &dummy, &dummy, &mw->prev_process_times_kernel, &mw->prev_process_times_user);
+		GetSystemTimes(0, &mw->prev_system_times_kernel, &mw->prev_system_times_user);
+	}
+#endif
+	
+	listeners->enable_listener = as_metrics_writer_enable;
+	listeners->snapshot_listener = as_metrics_writer_snapshot;
+	listeners->node_close_listener = as_metrics_writer_node_close;
+	listeners->disable_listener = as_metrics_writer_disable;
+	listeners->udata = mw;
+	return AEROSPIKE_OK;
+}
+
+as_status
+as_metrics_writer_enable(as_error* err, void* udata)
+{
+	as_metrics_writer* mw = udata;
+	as_status status = as_metrics_open_writer(mw, err);
+
+	if (status != AEROSPIKE_OK) {
+		return status;
+	}
+
+	mw->enable = true;
+	return AEROSPIKE_OK;
+}
+
+as_status
+as_metrics_writer_snapshot(as_error* err, as_cluster* cluster, void* udata)
+{
+	as_error_reset(err);
+	as_metrics_writer* mw = udata;
+
+	if (mw->enable && mw->file != NULL) {
+		as_status status = as_metrics_write_cluster(err, mw, cluster);
+		if (status != AEROSPIKE_OK) {
+			return status;
+		}
+		uint32_t result = fflush(mw->file);
+		if (result != 0) {
+			return as_error_update(err, AEROSPIKE_ERR_CLIENT,
+				"File stream did not flush successfully: %s", mw->report_dir);
+		}
+	}
+	return AEROSPIKE_OK;
+}
+
+as_status
+as_metrics_writer_node_close(as_error* err, as_node* node, void* udata)
+{
+	// write node info to file
+	as_error_reset(err);
+	as_metrics_writer* mw = udata;
+
+	if (mw->enable && mw->file != NULL) {
+		char now_str[128];
+		timestamp_to_string(now_str, sizeof(now_str));
+		
+		as_string_builder sb;
+		as_string_builder_inita(&sb, 16384, true);
+		as_string_builder_append(&sb, now_str);
+		as_string_builder_append_char(&sb, ' ');
+		as_metrics_write_node(mw, &sb, node);
+		as_string_builder_append_newline(&sb);
+		
+		as_status status = as_metrics_write_line(mw, sb.data, err);
+		
+		as_string_builder_destroy(&sb);
+		return status;
+	}
+	return AEROSPIKE_OK;
+}
+
+as_status
+as_metrics_writer_disable(as_error* err, as_cluster* cluster, void* udata)
+{
+	// write cluster into to file, disable
+	as_error_reset(err);
+	as_metrics_writer* mw = udata;
+	
+	if (mw != NULL) {
+		as_status status = AEROSPIKE_OK;
+		
+		if (mw->enable && mw->file != NULL) {
+			status = as_metrics_write_cluster(err, mw, cluster);
+		}
+		as_metrics_writer_destroy(mw);
+		return status;
+	}
+	return AEROSPIKE_OK;
+}
diff --git a/src/main/aerospike/as_node.c b/src/main/aerospike/as_node.c
index 08fedde8f1..e88d62dadb 100644
--- a/src/main/aerospike/as_node.c
+++ b/src/main/aerospike/as_node.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -23,6 +23,7 @@
 #include <aerospike/as_event_internal.h>
 #include <aerospike/as_info.h>
 #include <aerospike/as_log_macros.h>
+#include <aerospike/as_metrics.h>
 #include <aerospike/as_peers.h>
 #include <aerospike/as_queue.h>
 #include <aerospike/as_shm_cluster.h>
@@ -34,6 +35,9 @@
 // Replicas take ~2K per namespace, so this will cover most deployments:
 #define INFO_STACK_BUF_SIZE (16 * 1024)
 
+// Number of nanoseconds per millisecond
+#define NS_TO_MS 1000000
+
 /******************************************************************************
  * Function declarations.
  *****************************************************************************/
@@ -91,6 +95,48 @@ as_node_create_async_pools(uint32_t min_conns_per_node, uint32_t max_conns_per_n
 	return pools;
 }
 
+static void
+as_latency_buckets_init(as_latency_buckets* latency_buckets, uint32_t latency_columns, uint32_t latency_shift)
+{
+	latency_buckets->latency_columns = latency_columns;
+	latency_buckets->latency_shift = latency_shift;
+	latency_buckets->buckets = cf_malloc(sizeof(uint64_t) * latency_columns);
+	
+	for (uint32_t i = 0; i < latency_columns; i++) {
+		as_store_uint64(&latency_buckets->buckets[i], 0);
+	}
+}
+
+static as_node_metrics*
+as_node_metrics_init(uint32_t latency_columns, uint32_t latency_shift)
+{
+	as_node_metrics* node_metrics = (as_node_metrics*)cf_malloc(sizeof(as_node_metrics));
+	uint32_t max_latency_type = AS_LATENCY_TYPE_NONE;
+	node_metrics->latency = (as_latency_buckets*)cf_malloc(sizeof(as_latency_buckets) * max_latency_type);
+
+	for (uint32_t i = 0; i < max_latency_type; i++) {
+		as_latency_buckets_init(&node_metrics->latency[i], latency_columns, latency_shift);
+	}
+	return node_metrics;
+}
+
+void
+as_node_destroy_metrics(as_node* node)
+{
+	as_node_metrics* node_metrics = node->metrics;
+	
+	if (node_metrics) {
+		uint32_t max = AS_LATENCY_TYPE_NONE;
+		
+		for (uint32_t i = 0; i < max; i++) {
+			cf_free(node_metrics->latency[i].buckets);
+		}
+		cf_free(node_metrics->latency);
+		cf_free(node_metrics);
+		node->metrics = NULL;
+	}
+}
+
 as_node*
 as_node_create(as_cluster* cluster, as_node_info* node_info)
 {
@@ -135,12 +181,21 @@ as_node_create(as_cluster* cluster, as_node_info* node_info)
 	node->active = true;
 	node->partition_changed = true;
 	node->rebalance_changed = cluster->rack_aware;
+	node->error_rate = 0;
+	node->error_count = 0;
+	node->timeout_count = 0;
+
+	if (cluster->metrics_enabled) {
+		node->metrics = as_node_metrics_init(cluster->metrics_latency_columns, cluster->metrics_latency_shift);
+	}
+	else {
+		node->metrics = NULL;
+	}
 
 	// Create sync connection pools.
 	node->sync_conn_pools = cf_malloc(sizeof(as_conn_pool) * cluster->conn_pools_per_node);
 	node->sync_conns_opened = 1;
 	node->sync_conns_closed = 0;
-	node->error_count = 0;
 	node->conn_iter = 0;
 
 	uint32_t min = cluster->min_conns_per_node / cluster->conn_pools_per_node;
@@ -210,6 +265,8 @@ as_node_destroy(as_node* node)
 	if (racks) {
 		as_racks_release(racks);
 	}
+	
+	as_node_destroy_metrics(node);
 	cf_free(node);
 }
 
@@ -434,6 +491,12 @@ as_node_create_connection(
 	as_socket* sock
 	)
 {
+	uint64_t begin = 0;
+
+	if (node->cluster->metrics_enabled) {
+		begin = cf_getns();
+	}
+	
 	as_status status = as_node_create_socket(err, node, pool, sock, deadline_ms);
 
 	if (status) {
@@ -459,6 +522,11 @@ as_node_create_connection(
 			}
 		}
 	}
+
+	if (node->cluster->metrics_enabled) {
+		uint64_t elapsed = cf_getns() - begin;
+		as_node_add_latency(node, AS_LATENCY_TYPE_CONN, elapsed);
+	}
 	return AEROSPIKE_OK;
 }
 
@@ -635,7 +703,7 @@ as_node_balance_connections(as_node* node)
 		if (excess > 0) {
 			as_node_close_idle_connections(node, pool, excess);
 		}
-		else if (excess < 0 && as_node_valid_error_count(node)) {
+		else if (excess < 0 && as_node_valid_error_rate(node)) {
 			as_node_create_connections(node, pool, timeout_ms, -excess);
 		}
 	}
@@ -902,7 +970,7 @@ static void
 as_node_restart(as_cluster* cluster, as_node* node)
 {
 	if (cluster->max_error_rate > 0) {
-		as_node_reset_error_count(node);
+		as_node_reset_error_rate(node);
 	}
 
 	// Balance sync connections.
@@ -1299,6 +1367,43 @@ as_node_parse_racks(as_cluster* cluster, as_error* err, as_node* node, char* buf
 	return AEROSPIKE_OK;
 }
 
+static uint32_t
+as_latency_buckets_get_index(as_latency_buckets* latency_buckets, uint64_t elapsed_nanos)
+{
+	// Convert nanoseconds to milliseconds.
+	uint64_t elapsed = elapsed_nanos / NS_TO_MS;
+
+	// Round up elapsed to nearest millisecond.
+	if ((elapsed_nanos - (elapsed * NS_TO_MS)) > 0) {
+		elapsed++;
+	}
+
+	uint32_t last_bucket = latency_buckets->latency_columns - 1;
+	uint64_t limit = 1;
+
+	for (uint32_t i = 0; i < last_bucket; i++) {
+		if (elapsed <= limit) {
+			return i;
+		}
+		limit <<= latency_buckets->latency_shift;
+	}
+	return last_bucket;
+}
+
+void
+as_node_add_latency(as_node* node, as_latency_type latency_type, uint64_t elapsed)
+{
+	as_latency_buckets* latency_buckets = &node->metrics->latency[latency_type];
+	uint32_t index = as_latency_buckets_get_index(latency_buckets, elapsed);
+	as_incr_uint64(&latency_buckets->buckets[index]);
+}
+
+void
+as_node_enable_metrics(as_node* node, const as_metrics_policy* policy)
+{
+	 node->metrics = as_node_metrics_init(policy->latency_columns, policy->latency_shift);
+}
+
 static as_status
 as_node_process_racks(as_cluster* cluster, as_error* err, as_node* node, as_vector* values)
 {
diff --git a/src/main/aerospike/as_pipe.c b/src/main/aerospike/as_pipe.c
index be96c4b863..9a42a8ff7a 100644
--- a/src/main/aerospike/as_pipe.c
+++ b/src/main/aerospike/as_pipe.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2022 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -155,7 +155,7 @@ cancel_connection(as_event_command* cmd, as_error* err, int32_t source, bool ret
 		conn->canceled = true;
 		as_async_conn_pool* pool = &node->pipe_conn_pools[loop->index];
 		as_event_release_connection((as_event_connection*)conn, pool);
-		as_node_incr_error_count(node);
+		as_node_incr_error_rate(node);
 		as_node_release(node);
 		return;
 	}
@@ -349,7 +349,7 @@ as_pipe_get_connection(as_event_command* cmd)
 			if (len < 0) {
 				as_log_debug("Invalid pipeline socket from pool: %d", len);
 				release_connection(cmd, conn, pool);
-				as_node_incr_error_count(cmd->node);
+				as_node_incr_error_rate(cmd->node);
 				continue;
 			}
 
diff --git a/src/test/aerospike_test.c b/src/test/aerospike_test.c
index fdf9b618fe..b7b287b2ae 100644
--- a/src/test/aerospike_test.c
+++ b/src/test/aerospike_test.c
@@ -1,5 +1,5 @@
 /*
- * Copyright 2008-2023 Aerospike, Inc.
+ * Copyright 2008-2024 Aerospike, Inc.
  *
  * Portions may be licensed to Aerospike, Inc. under one or more contributor
  * license agreements.
@@ -24,6 +24,7 @@
 #include <aerospike/aerospike.h>
 #include <aerospike/aerospike_info.h>
 #include <aerospike/as_event.h>
+#include <aerospike/as_metrics.h>
 
 #include "test.h"
 #include "aerospike_test.h"
@@ -379,8 +380,20 @@ static bool before(atf_plan* plan)
 			return false;
 		}
 	}
-
 	cf_free(result);
+
+	/*
+	// Test metrics
+	as_metrics_policy policy;
+	as_metrics_policy_init(&policy);
+
+	as_status status = aerospike_enable_metrics(as, &err, &policy);
+
+	if (status != AEROSPIKE_OK) {
+		error("aerospike_enable_metrics() returned %d - %s", err.code, err.message);
+	}
+	*/
+	
 	return true;
 }
 
diff --git a/vs/aerospike/aerospike.vcxproj b/vs/aerospike/aerospike.vcxproj
index a8c0728ca9..7608c81a41 100644
--- a/vs/aerospike/aerospike.vcxproj
+++ b/vs/aerospike/aerospike.vcxproj
@@ -394,10 +394,13 @@
     <ClInclude Include="..\..\src\include\aerospike\as_info.h" />
     <ClInclude Include="..\..\src\include\aerospike\as_job.h" />
     <ClInclude Include="..\..\src\include\aerospike\as_key.h" />
+    <ClInclude Include="..\..\src\include\aerospike\as_latency.h" />
     <ClInclude Include="..\..\src\include\aerospike\as_listener.h" />
     <ClInclude Include="..\..\src\include\aerospike\as_list_operations.h" />
     <ClInclude Include="..\..\src\include\aerospike\as_lookup.h" />
     <ClInclude Include="..\..\src\include\aerospike\as_map_operations.h" />
+    <ClInclude Include="..\..\src\include\aerospike\as_metrics.h" />
+    <ClInclude Include="..\..\src\include\aerospike\as_metrics_writer.h" />
     <ClInclude Include="..\..\src\include\aerospike\as_node.h" />
     <ClInclude Include="..\..\src\include\aerospike\as_operations.h" />
     <ClInclude Include="..\..\src\include\aerospike\as_partition.h" />
@@ -546,9 +549,12 @@
     <ClCompile Include="..\..\src\main\aerospike\as_info.c" />
     <ClCompile Include="..\..\src\main\aerospike\as_job.c" />
     <ClCompile Include="..\..\src\main\aerospike\as_key.c" />
+    <ClCompile Include="..\..\src\main\aerospike\as_latency.c" />
     <ClCompile Include="..\..\src\main\aerospike\as_list_operations.c" />
     <ClCompile Include="..\..\src\main\aerospike\as_lookup.c" />
     <ClCompile Include="..\..\src\main\aerospike\as_map_operations.c" />
+    <ClCompile Include="..\..\src\main\aerospike\as_metrics.c" />
+    <ClCompile Include="..\..\src\main\aerospike\as_metrics_writer.c" />
     <ClCompile Include="..\..\src\main\aerospike\as_node.c" />
     <ClCompile Include="..\..\src\main\aerospike\as_operations.c" />
     <ClCompile Include="..\..\src\main\aerospike\as_partition.c" />
diff --git a/vs/aerospike/aerospike.vcxproj.filters b/vs/aerospike/aerospike.vcxproj.filters
index da07913095..fbb436a1e6 100644
--- a/vs/aerospike/aerospike.vcxproj.filters
+++ b/vs/aerospike/aerospike.vcxproj.filters
@@ -495,6 +495,15 @@
     <ClInclude Include="..\..\modules\lua\lzio.h">
       <Filter>Header Files\lua</Filter>
     </ClInclude>
+    <ClInclude Include="..\..\src\include\aerospike\as_metrics.h">
+      <Filter>Header Files</Filter>
+    </ClInclude>
+    <ClInclude Include="..\..\src\include\aerospike\as_metrics_writer.h">
+      <Filter>Header Files</Filter>
+    </ClInclude>
+    <ClInclude Include="..\..\src\include\aerospike\as_latency.h">
+      <Filter>Header Files</Filter>
+    </ClInclude>
   </ItemGroup>
   <ItemGroup>
     <ClCompile Include="..\..\src\main\aerospike\_bin.c">
@@ -941,6 +950,15 @@
     <ClCompile Include="..\..\modules\lua\lzio.c">
       <Filter>Source Files\lua</Filter>
     </ClCompile>
+    <ClCompile Include="..\..\src\main\aerospike\as_metrics.c">
+      <Filter>Source Files</Filter>
+    </ClCompile>
+    <ClCompile Include="..\..\src\main\aerospike\as_metrics_writer.c">
+      <Filter>Source Files</Filter>
+    </ClCompile>
+    <ClCompile Include="..\..\src\main\aerospike\as_latency.c">
+      <Filter>Source Files</Filter>
+    </ClCompile>
   </ItemGroup>
   <ItemGroup>
     <None Include="packages.config" />
diff --git a/vs/props/base.props b/vs/props/base.props
index 5f458bb22d..ad6ec7d48d 100644
--- a/vs/props/base.props
+++ b/vs/props/base.props
@@ -8,7 +8,7 @@
   <ItemDefinitionGroup>
     <ClCompile>
       <PreprocessorDefinitions>_CRT_SECURE_NO_DEPRECATE;_TIMESPEC_DEFINED;%(PreprocessorDefinitions)</PreprocessorDefinitions>
-      <DisableSpecificWarnings>4098;4996</DisableSpecificWarnings>
+      <DisableSpecificWarnings>4098;4996;6255;6262;26451</DisableSpecificWarnings>
     </ClCompile>
     <Link>
     </Link>
diff --git a/xcode/aerospike.xcodeproj/project.pbxproj b/xcode/aerospike.xcodeproj/project.pbxproj
index 31afd29387..c274396037 100644
--- a/xcode/aerospike.xcodeproj/project.pbxproj
+++ b/xcode/aerospike.xcodeproj/project.pbxproj
@@ -109,9 +109,13 @@
 		BF90C76C22AB154A0062D920 /* as_cdt_internal.c in Sources */ = {isa = PBXBuildFile; fileRef = BF90C76B22AB154A0062D920 /* as_cdt_internal.c */; };
 		BF90C77122AB30E40062D920 /* as_map_operations.c in Sources */ = {isa = PBXBuildFile; fileRef = BF90C77022AB30E40062D920 /* as_map_operations.c */; };
 		BF93AA061AE9E6EB003ECE3B /* as_thread_pool.c in Sources */ = {isa = PBXBuildFile; fileRef = BF93AA051AE9E6EB003ECE3B /* as_thread_pool.c */; };
+		BF94A3BD2B86A87800295885 /* as_latency.h in Headers */ = {isa = PBXBuildFile; fileRef = BF94A3BC2B86A87800295885 /* as_latency.h */; };
+		BF94A3BF2B86AA4300295885 /* as_latency.c in Sources */ = {isa = PBXBuildFile; fileRef = BF94A3BE2B86AA4300295885 /* as_latency.c */; };
 		BF986E001F466BEE0057802C /* version.h in Headers */ = {isa = PBXBuildFile; fileRef = BF986DFF1F466BEE0057802C /* version.h */; };
 		BFA5B21020FD3FA4002AF0BB /* as_cpu.h in Headers */ = {isa = PBXBuildFile; fileRef = BFA5B20F20FD3FA4002AF0BB /* as_cpu.h */; };
 		BFABF3311FCF85EC004745A1 /* as_queue_mt.c in Sources */ = {isa = PBXBuildFile; fileRef = BFABF3301FCF85EC004745A1 /* as_queue_mt.c */; };
+		BFAF276E2B6AB36A00A3858B /* as_metrics.h in Headers */ = {isa = PBXBuildFile; fileRef = BFAF276D2B6AB36A00A3858B /* as_metrics.h */; };
+		BFAF27702B6AB39100A3858B /* as_metrics.c in Sources */ = {isa = PBXBuildFile; fileRef = BFAF276F2B6AB39100A3858B /* as_metrics.c */; };
 		BFB0ED5522A72260007FEA9C /* as_cdt_ctx.h in Headers */ = {isa = PBXBuildFile; fileRef = BFB0ED5422A72260007FEA9C /* as_cdt_ctx.h */; };
 		BFB8A5D81D0F3F77007B4E22 /* as_tls.c in Sources */ = {isa = PBXBuildFile; fileRef = BFB8A5D71D0F3F77007B4E22 /* as_tls.c */; };
 		BFB8A5DA1D0F3F9E007B4E22 /* as_tls.h in Headers */ = {isa = PBXBuildFile; fileRef = BFB8A5D91D0F3F9E007B4E22 /* as_tls.h */; };
@@ -213,6 +217,8 @@
 		BFD8FE7C20CF6DFC000A80F1 /* as_query_validate.c in Sources */ = {isa = PBXBuildFile; fileRef = BFD8FE7B20CF6DFC000A80F1 /* as_query_validate.c */; };
 		BFE3C3991D6270C200AA7F20 /* as_address.h in Headers */ = {isa = PBXBuildFile; fileRef = BFE3C3981D6270C200AA7F20 /* as_address.h */; };
 		BFE3C39B1D62720800AA7F20 /* as_address.c in Sources */ = {isa = PBXBuildFile; fileRef = BFE3C39A1D62720800AA7F20 /* as_address.c */; };
+		BFE8EF472B7E9C0600D0C31B /* as_metrics_writer.h in Headers */ = {isa = PBXBuildFile; fileRef = BFE8EF462B7E9C0600D0C31B /* as_metrics_writer.h */; };
+		BFE8EF492B7E9C3A00D0C31B /* as_metrics_writer.c in Sources */ = {isa = PBXBuildFile; fileRef = BFE8EF482B7E9C3A00D0C31B /* as_metrics_writer.c */; };
 		BFEAF6322228638E00FB4248 /* as_conn_pool.h in Headers */ = {isa = PBXBuildFile; fileRef = BFEAF6312228638E00FB4248 /* as_conn_pool.h */; };
 		BFF344B01CDAC67700FD1976 /* as_map_operations.h in Headers */ = {isa = PBXBuildFile; fileRef = BFF344AF1CDAC67700FD1976 /* as_map_operations.h */; };
 		BFF344C31CEA7ACD00FD1976 /* as_list_operations.h in Headers */ = {isa = PBXBuildFile; fileRef = BFF344C21CEA7ACD00FD1976 /* as_list_operations.h */; };
@@ -322,9 +328,13 @@
 		BF90C76B22AB154A0062D920 /* as_cdt_internal.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = as_cdt_internal.c; path = ../src/main/aerospike/as_cdt_internal.c; sourceTree = "<group>"; };
 		BF90C77022AB30E40062D920 /* as_map_operations.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = as_map_operations.c; path = ../src/main/aerospike/as_map_operations.c; sourceTree = "<group>"; };
 		BF93AA051AE9E6EB003ECE3B /* as_thread_pool.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = as_thread_pool.c; path = ../modules/common/src/main/aerospike/as_thread_pool.c; sourceTree = "<group>"; };
+		BF94A3BC2B86A87800295885 /* as_latency.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; name = as_latency.h; path = ../src/include/aerospike/as_latency.h; sourceTree = "<group>"; };
+		BF94A3BE2B86AA4300295885 /* as_latency.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = as_latency.c; path = ../src/main/aerospike/as_latency.c; sourceTree = "<group>"; };
 		BF986DFF1F466BEE0057802C /* version.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; name = version.h; path = ../src/include/aerospike/version.h; sourceTree = "<group>"; };
 		BFA5B20F20FD3FA4002AF0BB /* as_cpu.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; name = as_cpu.h; path = ../src/include/aerospike/as_cpu.h; sourceTree = "<group>"; };
 		BFABF3301FCF85EC004745A1 /* as_queue_mt.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = as_queue_mt.c; path = ../modules/common/src/main/aerospike/as_queue_mt.c; sourceTree = "<group>"; };
+		BFAF276D2B6AB36A00A3858B /* as_metrics.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; name = as_metrics.h; path = ../src/include/aerospike/as_metrics.h; sourceTree = "<group>"; };
+		BFAF276F2B6AB39100A3858B /* as_metrics.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = as_metrics.c; path = ../src/main/aerospike/as_metrics.c; sourceTree = "<group>"; };
 		BFB0ED5422A72260007FEA9C /* as_cdt_ctx.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; name = as_cdt_ctx.h; path = ../src/include/aerospike/as_cdt_ctx.h; sourceTree = "<group>"; };
 		BFB8A5D71D0F3F77007B4E22 /* as_tls.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = as_tls.c; path = ../src/main/aerospike/as_tls.c; sourceTree = "<group>"; };
 		BFB8A5D91D0F3F9E007B4E22 /* as_tls.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; name = as_tls.h; path = ../src/include/aerospike/as_tls.h; sourceTree = "<group>"; };
@@ -428,6 +438,8 @@
 		BFD8FE7B20CF6DFC000A80F1 /* as_query_validate.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = as_query_validate.c; path = ../src/main/aerospike/as_query_validate.c; sourceTree = "<group>"; };
 		BFE3C3981D6270C200AA7F20 /* as_address.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; name = as_address.h; path = ../src/include/aerospike/as_address.h; sourceTree = "<group>"; };
 		BFE3C39A1D62720800AA7F20 /* as_address.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = as_address.c; path = ../src/main/aerospike/as_address.c; sourceTree = "<group>"; };
+		BFE8EF462B7E9C0600D0C31B /* as_metrics_writer.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; name = as_metrics_writer.h; path = ../src/include/aerospike/as_metrics_writer.h; sourceTree = "<group>"; };
+		BFE8EF482B7E9C3A00D0C31B /* as_metrics_writer.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; name = as_metrics_writer.c; path = ../src/main/aerospike/as_metrics_writer.c; sourceTree = "<group>"; };
 		BFEAF6312228638E00FB4248 /* as_conn_pool.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; name = as_conn_pool.h; path = ../src/include/aerospike/as_conn_pool.h; sourceTree = "<group>"; };
 		BFF344AF1CDAC67700FD1976 /* as_map_operations.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; name = as_map_operations.h; path = ../src/include/aerospike/as_map_operations.h; sourceTree = "<group>"; };
 		BFF344C21CEA7ACD00FD1976 /* as_list_operations.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; name = as_list_operations.h; path = ../src/include/aerospike/as_list_operations.h; sourceTree = "<group>"; };
@@ -548,9 +560,12 @@
 				BFBDAFDF191B0C5C007EB07C /* as_info.c */,
 				BF26C4661B45AE8F00E6929D /* as_job.c */,
 				BF2AA7C418BEBFA400E54AF3 /* as_key.c */,
+				BF94A3BE2B86AA4300295885 /* as_latency.c */,
 				BF90C76422AB0EB20062D920 /* as_list_operations.c */,
 				BFC002891901E08500CB9BC8 /* as_lookup.c */,
 				BF90C77022AB30E40062D920 /* as_map_operations.c */,
+				BFAF276F2B6AB39100A3858B /* as_metrics.c */,
+				BFE8EF482B7E9C3A00D0C31B /* as_metrics_writer.c */,
 				BFBB3C8E192D729A00251B15 /* as_node.c */,
 				BF2AA7C718BEBFA400E54AF3 /* as_operations.c */,
 				BFBA916A1914344B00AADA9A /* as_partition.c */,
@@ -745,10 +760,13 @@
 				BFC65B4D1C921E9E0079DF5A /* as_info.h */,
 				BFC65B4E1C921E9E0079DF5A /* as_job.h */,
 				BFC65B4F1C921E9E0079DF5A /* as_key.h */,
+				BF94A3BC2B86A87800295885 /* as_latency.h */,
 				BFF344C21CEA7ACD00FD1976 /* as_list_operations.h */,
 				BFC65B511C921E9E0079DF5A /* as_listener.h */,
 				BFC65B521C921E9E0079DF5A /* as_lookup.h */,
 				BFF344AF1CDAC67700FD1976 /* as_map_operations.h */,
+				BFAF276D2B6AB36A00A3858B /* as_metrics.h */,
+				BFE8EF462B7E9C0600D0C31B /* as_metrics_writer.h */,
 				BFC65B531C921E9E0079DF5A /* as_node.h */,
 				BFC65B541C921E9E0079DF5A /* as_operations.h */,
 				BFC65B551C921E9E0079DF5A /* as_partition.h */,
@@ -785,6 +803,7 @@
 				BF986E001F466BEE0057802C /* version.h in Headers */,
 				BFC65B8B1C921E9E0079DF5A /* as_udf.h in Headers */,
 				BFC65B811C921E9E0079DF5A /* as_pipe.h in Headers */,
+				BF94A3BD2B86A87800295885 /* as_latency.h in Headers */,
 				BF32146F23E8F630004A7E19 /* as_partition_tracker.h in Headers */,
 				BF457A8622B1AC6600409D04 /* as_bit_operations.h in Headers */,
 				BFC65B761C921E9E0079DF5A /* as_event_internal.h in Headers */,
@@ -792,12 +811,14 @@
 				BFC65B791C921E9E0079DF5A /* as_job.h in Headers */,
 				BF90C76A22AB143C0062D920 /* as_cdt_internal.h in Headers */,
 				BF1C2ADF20BE031B00868695 /* aerospike_stats.h in Headers */,
+				BFE8EF472B7E9C0600D0C31B /* as_metrics_writer.h in Headers */,
 				BFC65B731C921E9E0079DF5A /* as_command.h in Headers */,
 				BFF344B01CDAC67700FD1976 /* as_map_operations.h in Headers */,
 				BF809CDB24327E9300C16F3D /* as_hll_operations.h in Headers */,
 				BF65C9C6252D299D0026D9E2 /* as_exp.h in Headers */,
 				BFC65B821C921E9E0079DF5A /* as_policy.h in Headers */,
 				BF162EBE2413000B001B1747 /* as_cdt_order.h in Headers */,
+				BFAF276E2B6AB36A00A3858B /* as_metrics.h in Headers */,
 				BFC65B801C921E9E0079DF5A /* as_partition.h in Headers */,
 				BFC65B7D1C921E9E0079DF5A /* as_lookup.h in Headers */,
 				BF4E4E2A1D48213700BEEF94 /* as_host.h in Headers */,
@@ -920,6 +941,7 @@
 				BF93AA061AE9E6EB003ECE3B /* as_thread_pool.c in Sources */,
 				BFC38AE11948F7CA000C53D9 /* as_admin.c in Sources */,
 				BF8EF4CE2AE1B47B00FEEC3A /* lstrlib.c in Sources */,
+				BF94A3BF2B86AA4300295885 /* as_latency.c in Sources */,
 				BFBD205618BC3436009ED931 /* mod_lua_record.c in Sources */,
 				BF8EF4A52AE1B41100FEEC3A /* lauxlib.c in Sources */,
 				BF8EF4BA2AE1B44900FEEC3A /* lgc.c in Sources */,
@@ -938,6 +960,7 @@
 				BF8EF4A82AE1B41100FEEC3A /* lcorolib.c in Sources */,
 				BF32147123E8F9C6004A7E19 /* as_partition_tracker.c in Sources */,
 				BFBA04A91947AA8400F9924E /* cf_random.c in Sources */,
+				BFE8EF492B7E9C3A00D0C31B /* as_metrics_writer.c in Sources */,
 				BF2337A21B4DC8BD00670C64 /* as_double.c in Sources */,
 				BF8EF4AC2AE1B41100FEEC3A /* ldo.c in Sources */,
 				BF2AA7DC18BEBFA500E54AF3 /* aerospike_info.c in Sources */,
@@ -969,6 +992,7 @@
 				BFBA106F18B7DFA100A64E68 /* as_msgpack.c in Sources */,
 				BFCB38A71DFB764200C73D0F /* as_event_event.c in Sources */,
 				BF2AA7F018BEBFA500E54AF3 /* as_record_hooks.c in Sources */,
+				BFAF27702B6AB39100A3858B /* as_metrics.c in Sources */,
 				BFBD205918BC3436009ED931 /* mod_lua_val.c in Sources */,
 				BFBA106218B7D8B300A64E68 /* as_pair.c in Sources */,
 				BFBA105318B7D8B300A64E68 /* as_bytes.c in Sources */,