Fixing bugs

Author: shannonklaus

Makefile

@@ -136,6 +136,7 @@ AEROSPIKE += as_key.o
 AEROSPIKE += as_list_operations.o
 AEROSPIKE += as_lookup.o
 AEROSPIKE += as_map_operations.o
+AEROSPIKE += as_metrics.o
 AEROSPIKE += as_node.o
 AEROSPIKE += as_operations.o
 AEROSPIKE += as_partition.o

src/include/aerospike/as_cluster.h

@@ -383,9 +383,13 @@ typedef struct as_cluster_s {
 
 	bool metrics_enabled;
 
-	as_policy_metrics* metrics_policy;
+	uint32_t metrics_interval;
 
-	as_metrics_listeners* metrics_listeners;
+	uint32_t metrics_latency_columns;
+
+	uint32_t metrics_latency_shift;
+
+	as_metrics_listeners metrics_listeners;
 
 	uint64_t retry_count;
 

src/include/aerospike/as_metrics.h

@@ -58,56 +58,54 @@ typedef uint8_t as_latency_type;
  * Latency bucket counts are cumulative and not reset on each metrics snapshot interval
  */
 typedef struct as_latency_buckets_s {
-	int32_t latency_shift;
-	int32_t latency_columns;
+	uint32_t latency_shift;
+	uint32_t latency_columns;
 	uint64_t* buckets;
 } as_latency_buckets;
 
-struct as_metrics_listeners_s;
-
-/**
-* Metrics Policy
-*/
-typedef struct as_policy_metrics_s {
-	const char* report_directory; // where the metrics file is output
-
-	int64_t report_size_limit; // default 0
-
-	int32_t interval; // default 30
-
-	int32_t latency_columns; // default 7
-
-	int32_t latency_shift; // default 1
-
-	struct as_metrics_listeners_s* metrics_listeners;
-
-	void* udata;
-} as_policy_metrics;
-
-struct as_cluster_s;
+struct as_policy_metrics_s;
 struct as_node_s;
+struct as_cluster_s;
 
 /**
  * Callbacks for metrics listener operations
  */
-typedef as_status (*as_metrics_enable_callback)(as_error* err, const struct as_policy_metrics_s* policy);
+typedef as_status(*as_metrics_enable_listener)(as_error* err, const struct as_policy_metrics_s* policy, void* udata);
 
-typedef as_status (*as_metrics_snapshot_callback)(as_error* err, struct as_cluster_s* cluster, 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_callback)(as_error* err, struct as_node_s* node, 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_callback)(as_error* err, struct as_cluster_s* cluster, void* udata);
+typedef as_status(*as_metrics_disable_listener)(as_error* err, struct as_cluster_s* cluster, void* udata);
 
 /**
  * Struct to hold required callbacks
  */
 typedef struct as_metrics_listeners_s {
-	as_metrics_enable_callback enable_callback;
-	as_metrics_snapshot_callback snapshot_callback;
-	as_metrics_node_close_callback node_close_callback;
-	as_metrics_disable_callback disable_callback;
+	as_metrics_enable_listener enable_listener;
+	as_metrics_snapshot_listener snapshot_listener;
+	as_metrics_node_close_listener node_close_listener;
+	as_metrics_disable_listener disable_listener;
+	void* udata;
 } as_metrics_listeners;
 
+/**
+* Metrics Policy
+*/
+typedef struct as_policy_metrics_s {
+	const char* report_directory; // where the metrics file is output
+
+	uint64_t report_size_limit; // default 0
+
+	uint32_t interval; // default 30
+
+	uint32_t latency_columns; // default 7
+
+	uint32_t latency_shift; // default 1
+
+	as_metrics_listeners metrics_listeners;
+} as_policy_metrics;
+
 /**
  * Node metrics latency bucket struct
  */
@@ -129,25 +127,32 @@ typedef struct as_metrics_writer_s {
 
 	uint64_t size;
 
-	int32_t latency_columns;
+	uint32_t latency_columns;
 
-	int32_t latency_shift;
+	uint32_t latency_shift;
 
 	const char* report_directory;
 } as_metrics_writer;
 
-/**
- * Format time into UTC string
- */
-const char* 
-utc_time_str(time_t t);
-
 /**
  * Initalize metrics policy
  */
-void
+AS_EXTERN void
 as_metrics_policy_init(as_policy_metrics* policy);
 
+static inline void
+as_metrics_set_listeners(
+	as_policy_metrics* policy, as_metrics_enable_listener enable, 
+	as_metrics_disable_listener disable, as_metrics_node_close_listener node_close,
+	as_metrics_snapshot_listener snapshot
+	)
+{
+	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;
+}
+
 /**
  * Convert latency_type to string version for printing to the output file
  */
@@ -158,7 +163,7 @@ as_latency_type_to_string(as_latency_type type);
  * Initalize latency bucket struct
  */
 void
-as_metrics_latency_buckets_init(as_latency_buckets* latency_buckets, int32_t latency_columns, int32_t latency_shift);
+as_metrics_latency_buckets_init(as_latency_buckets* latency_buckets, uint32_t latency_columns, uint32_t latency_shift);
 
 /**
  * Return cumulative count of a bucket.
@@ -181,72 +186,21 @@ as_metrics_get_index(as_latency_buckets* latency_buckets, uint64_t elapsed_nanos
 /**
  * Initalize node metrics struct
  */
-void
-as_node_metrics_init(as_node_metrics* node_metrics, const as_policy_metrics* policy);
+as_node_metrics*
+as_node_metrics_init(uint32_t latency_columns, uint32_t latency_shift);
 
 /**
  * Add latency to corresponding bucket type
  */
 void
 as_metrics_add_latency(as_node_metrics* node_metrics, as_latency_type latency_type, uint64_t elapsed);
 
-/**
- * Initalize metrics listener struct
- */
-void
-as_metrics_listeners_init(as_metrics_listeners* listeners);
-
-/**
- * Open output metrics file and write header 
- */
-as_status
-as_metrics_open_writer(as_metrics_writer* mw, as_error* err);
-
 /**
  * Calculate CPU and memory usage
  */
 void
 as_metrics_process_cpu_load_mem_usage(uint32_t* cpu_usage, uint32_t* mem);
 
-struct as_cluster_s;
-/**
- * Write cluster information to the metrics output file
- */
-as_status
-as_metrics_write_cluster(as_error* err, as_metrics_writer* mw, struct as_cluster_s* cluster);
-
-struct as_node_s;
-/**
- * Write node information to the metrics output file
- */
-void
-as_metrics_write_node(as_metrics_writer* mw, struct as_node_s* node_stats);
-
-struct as_conn_stats_s;
-/**
- * Write connection information to the metrics output file
- */
-void
-as_metrics_write_conn(as_metrics_writer* mw, const struct as_conn_stats_s* conn_stats);
-
-/**
- * Calculate sync conn stats data
- */
-void
-as_metrics_get_node_sync_conn_stats(const struct as_node_s* node, struct as_conn_stats_s* async);
-
-/**
- * Calculate async conn stats data
- */
-void
-as_metrics_get_node_async_conn_stats(const struct as_node_s* node, struct as_conn_stats_s* sync);
-
-/**
- * Write line to the metrics output file
- */
-as_status
-as_metrics_write_line(as_metrics_writer* mw, as_error* err);
-
 #if defined(__linux__)
 /**
  * Gets memory and CPU usage information from proc/stat

src/main/aerospike/aerospike_stats.c

@@ -183,14 +183,7 @@ aerospike_stats_to_string(as_cluster_stats* stats)
 as_status
 aerospike_enable_metrics(aerospike* as, as_error* err, struct as_policy_metrics_s* policy)
 {
-	as_cluster* cluster = as->cluster;
-	as_status status = as_cluster_enable_metrics(err, cluster, policy);
-	if (status != AEROSPIKE_OK)
-	{
-		return status;
-	}
-
-	return AEROSPIKE_OK;
+	return as_cluster_enable_metrics(err, as->cluster, policy);
 }
 
 as_status

src/main/aerospike/as_cluster.c

@@ -557,31 +557,28 @@ as_cluster_remove_nodes_copy(as_cluster* cluster, as_vector* /* <as_node*> */ no
 as_status
 as_cluster_enable_metrics(as_error* err, as_cluster* cluster, as_policy_metrics* policy)
 {
-	if (cluster->metrics_enabled)
-	{
-		cluster->metrics_listeners->disable_callback(err, cluster, policy->udata);
+	if (cluster->metrics_enabled) {
+		cluster->metrics_listeners.disable_listener(err, cluster, cluster->metrics_listeners.udata);
 	}
 
 	cluster->metrics_listeners = policy->metrics_listeners;
-	if (cluster->metrics_listeners == NULL)
-	{
-		as_metrics_listeners_init(cluster->metrics_listeners);
-	}
-
-	cluster->metrics_policy = policy;
+	cluster->metrics_interval = policy->interval;
+	cluster->metrics_latency_columns = policy->latency_columns;
+	cluster->metrics_latency_shift = policy->latency_shift;
 
-	as_nodes* nodes = cluster->nodes;
+	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_status status = cluster->metrics_listeners->enable_callback(err, policy);
-	if (status != AEROSPIKE_OK)
-	{
+	as_nodes_release(nodes);
+
+	as_status status = cluster->metrics_listeners.enable_listener(err, policy, cluster->metrics_listeners.udata);
+	if (status != AEROSPIKE_OK) {
 		return status;
 	}
-
+	cluster->metrics_enabled = true;
 	return AEROSPIKE_OK;
 }
 
@@ -591,11 +588,7 @@ as_cluster_disable_metrics(as_error* err, as_cluster* cluster)
 	if (cluster->metrics_enabled)
 	{
 		cluster->metrics_enabled = false;
-		as_status status = cluster->metrics_listeners->disable_callback(err, cluster, cluster->metrics_policy->udata);
-		if (status != AEROSPIKE_OK)
-		{
-			return status;
-		}
+		return cluster->metrics_listeners.disable_listener(err, cluster, cluster->metrics_listeners.udata);
 	}
 
 	return AEROSPIKE_OK;
@@ -659,7 +652,7 @@ as_cluster_remove_nodes(as_error* err, as_cluster* cluster, as_vector* /* <as_no
 		as_node_deactivate(node);
 
 		if (cluster->metrics_enabled) {
-			as_status status = cluster->metrics_listeners->node_close_callback(err, node, node->cluster->metrics_policy->udata);
+			as_status status = cluster->metrics_listeners.node_close_listener(err, node, node->cluster->metrics_listeners.udata);
 			if (status != AEROSPIKE_OK)
 			{
 				return status;
@@ -988,9 +981,9 @@ as_cluster_tend(as_cluster* cluster, as_error* err, bool is_init)
 		as_incr_uint32(&cluster->shm_info->cluster_shm->rebalance_gen);
 	}
 
-	if (cluster->metrics_enabled && (cluster->tend_count % cluster->metrics_policy->interval))
+	if (cluster->metrics_enabled && (cluster->tend_count % cluster->metrics_interval))
 	{
-		as_status status = cluster->metrics_listeners->snapshot_callback(err, cluster, cluster->metrics_policy->udata);
+		as_status status = cluster->metrics_listeners.snapshot_listener(err, cluster, cluster->metrics_listeners.udata);
 		if (status != AEROSPIKE_OK)
 		{
 			return status;
@@ -1510,6 +1503,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;
 }