[03/11] telemetry: add client feature and sockets

diff --git a/lib/librte_telemetry/meson.build b/lib/librte_telemetry/meson.build
index 7716076..0ccfa36 100644
--- a/lib/librte_telemetry/meson.build
+++ b/lib/librte_telemetry/meson.build
@@ -5,3 +5,5 @@  sources = files('rte_telemetry.c')
 headers = files('rte_telemetry.h', 'rte_telemetry_internal.h')
 deps += ['metrics', 'ethdev']
 cflags += '-DALLOW_EXPERIMENTAL_API'
+jansson = cc.find_library('jansson', required: true)
+ext_deps += jansson
diff --git a/lib/librte_telemetry/rte_telemetry.c b/lib/librte_telemetry/rte_telemetry.c
index f984929..e9dd022 100644
--- a/lib/librte_telemetry/rte_telemetry.c
+++ b/lib/librte_telemetry/rte_telemetry.c
@@ -6,6 +6,7 @@ 
 #include <fcntl.h>
 #include <sys/socket.h>
 #include <sys/un.h>
+#include <jansson.h>
 
 #include <rte_eal.h>
 #include <rte_ethdev.h>
@@ -15,6 +16,8 @@ 
 #include "rte_telemetry.h"
 #include "rte_telemetry_internal.h"
 
+#define BUF_SIZE 1024
+#define ACTION_POST 1
 #define SLEEP_TIME 10
 
 #define DEFAULT_DPDK_PATH "/var/run/.rte_telemetry"
@@ -36,6 +39,88 @@  rte_telemetry_check_port_activity(int port_id)
 	return 0;
 }
 
+int32_t
+rte_telemetry_write_to_socket(struct telemetry_impl *telemetry,
+	const char *json_string)
+{
+	int ret;
+
+	if (!telemetry) {
+		TELEMETRY_LOG_ERR("Error, could not initialise "
+			"TELEMETRY_API\n");
+		return -1;
+	}
+	if (!telemetry->request_client) {
+		TELEMETRY_LOG_ERR("Error - No client has been chosen to"
+			" write to\n");
+		return -1;
+	}
+	if (!json_string) {
+		TELEMETRY_LOG_ERR("Error - Invalid JSON string!\n");
+		return -1;
+	}
+	ret = send(telemetry->request_client->fd,
+			 json_string, strlen(json_string), 0);
+	if (ret < 0) {
+		TELEMETRY_LOG_ERR("Error - Failed to write to socket for "
+			"client: %s\n", telemetry->request_client->file_path);
+		return -1;
+	}
+	return 0;
+}
+
+int32_t
+rte_telemetry_send_error_response(struct telemetry_impl *telemetry,
+	int error_type)
+{
+	int ret;
+	const char *status_code, *json_buffer;
+	json_t *root;
+
+	if (error_type == -EPERM)
+		status_code = "Status Error: Unknown";
+	else if (error_type == -EINVAL)
+		status_code = "Status Error: Invalid Argument 404";
+	else if (error_type == -ENOMEM)
+		status_code = "Status Error: Memory Allocation Error";
+	else {
+		TELEMETRY_LOG_ERR("Error - invalid error type\n");
+		return -EINVAL;
+	}
+
+	root = json_object();
+
+	if (!root) {
+		TELEMETRY_LOG_ERR("Error - Could not create root JSON "
+			"object\n");
+		return -EPERM;
+	}
+
+	ret = json_object_set_new(root, "status_code",
+		json_string(status_code));
+
+	if (ret < 0) {
+		TELEMETRY_LOG_ERR("Error - Status code field cannot be set\n");
+		return -EPERM;
+	}
+
+	ret = json_object_set_new(root, "data", json_null());
+	if (ret < 0) {
+		TELEMETRY_LOG_ERR("Error - Data field cannot be set\n");
+		return -EPERM;
+	}
+
+	json_buffer = json_dumps(root, JSON_INDENT(2));
+	json_decref(root);
+
+	ret = rte_telemetry_write_to_socket(telemetry, json_buffer);
+	if (ret < 0) {
+		TELEMETRY_LOG_ERR("Error - Could not write to socket\n");
+		return -EPERM;
+	}
+	return 0;
+}
+
 static int32_t
 rte_telemetry_reg_ethdev_to_metrics(uint16_t port_id)
 {
@@ -127,6 +212,31 @@  rte_telemetry_initial_accept(struct telemetry_impl *telemetry)
 }
 
 static int32_t
+rte_telemetry_read_client(struct telemetry_impl *telemetry)
+{
+	char buf[BUF_SIZE];
+	int buffer_read = 0;
+	errno = 0;
+
+	buffer_read = read(telemetry->accept_fd, buf, BUF_SIZE-1);
+	buf[buffer_read] = '\0';
+	if (buffer_read == -1) {
+		TELEMETRY_LOG_ERR("Error - Read error\n");
+		return -1;
+	} else if (buffer_read == 0) {
+		close(telemetry->accept_fd);
+		telemetry->accept_fd = 0;
+	} else {
+		int ret = rte_telemetry_parse_client_message(telemetry, buf);
+		if (ret < 0)
+			TELEMETRY_LOG_WARN("Warning - parse message failed\n");
+		close(telemetry->accept_fd);
+		telemetry->accept_fd = 0;
+	}
+	return 0;
+}
+
+static int32_t
 rte_telemetry_accept_new_client(struct telemetry_impl *telemetry)
 {
 	int ret;
@@ -148,6 +258,28 @@  rte_telemetry_accept_new_client(struct telemetry_impl *telemetry)
 				return -1;
 			}
 		}
+	} else {
+		ret = rte_telemetry_read_client(telemetry);
+		if (ret < 0) {
+			TELEMETRY_LOG_ERR("Error - failed to read socket "
+				"buffer\n");
+			return -1;
+		}
+	}
+	return 0;
+}
+
+static int32_t
+rte_telemetry_read_client_sockets(struct telemetry_impl *telemetry)
+{
+	telemetry_client *client;
+	TAILQ_FOREACH(client, &telemetry->client_list_head, client_list) {
+		char client_buf[BUF_SIZE];
+		int bytes = read(client->fd, client_buf, BUF_SIZE-1);
+		client_buf[bytes] = '\0';
+		if (bytes > 0) {
+			telemetry->request_client = client;
+		}
 	}
 	return 0;
 }
@@ -169,6 +301,12 @@  rte_telemetry_run(void *userdata)
 			"failed\n");
 		return -1;
 	}
+
+	ret = rte_telemetry_read_client_sockets(telemetry);
+	if (ret < 0) {
+		TELEMETRY_LOG_ERR("Error - client socket read failed\n");
+		return -1;
+	}
 	return 0;
 }
 
@@ -267,6 +405,7 @@  rte_telemetry_init(uint32_t socket_id)
 			TELEMETRY_LOG_ERR("Error - TELEMETRY cleanup failed\n");
 		return -EPERM;
 	}
+	TAILQ_INIT(&static_telemetry->client_list_head);
 
 	pthread_attr_init(&attr);
 	ret = rte_ctrl_thread_create(&static_telemetry->thread_id,
@@ -282,12 +421,38 @@  rte_telemetry_init(uint32_t socket_id)
 	return 0;
 }
 
+static int32_t
+rte_telemetry_client_cleanup(struct telemetry_client *client)
+{
+	int ret;
+
+	ret = close(client->fd);
+	free(client->file_path);
+	free(client);
+
+	if (ret < 0) {
+		TELEMETRY_LOG_ERR("Error - Close client socket failed\n");
+		return -EPERM;
+	}
+	return 0;
+}
+
 int32_t
 rte_telemetry_cleanup(void)
 {
 	int ret;
 	struct telemetry_impl *telemetry = static_telemetry;
 
+	telemetry_client *client, *temp_client;
+	TAILQ_FOREACH_SAFE(client, &telemetry->client_list_head, client_list,
+		temp_client) {
+		TAILQ_REMOVE(&telemetry->client_list_head, client, client_list);
+		ret = rte_telemetry_client_cleanup(client);
+		if (ret < 0) {
+			TELEMETRY_LOG_ERR("Error - Client cleanup failed\n");
+			return -EPERM;
+		}
+	}
 	ret = close(telemetry->server_fd);
 	if (ret < 0) {
 		TELEMETRY_LOG_ERR("Error - Close TELEMETRY socket failed\n");
@@ -301,6 +466,199 @@  rte_telemetry_cleanup(void)
 	return 0;
 }
 
+int32_t
+rte_telemetry_unregister_client(struct telemetry_impl *telemetry,
+	const char *client_path)
+{
+	int ret;
+
+	if (!telemetry) {
+		TELEMETRY_LOG_WARN("Warning - TELEMETRY is not initialised\n");
+		return -ENODEV;
+	}
+
+	if (!client_path) {
+		TELEMETRY_LOG_ERR("Error - Invalid client path\n");
+		goto einval_fail;
+	}
+
+	if (TAILQ_EMPTY(&telemetry->client_list_head)) {
+		TELEMETRY_LOG_ERR("Error - there are no clients currently "
+			"registered\n");
+		return -EPERM;
+	}
+	telemetry_client *client, *temp_client;
+	TAILQ_FOREACH_SAFE(client, &telemetry->client_list_head, client_list,
+			temp_client) {
+		if (strcmp(client_path, client->file_path) == 0) {
+			TAILQ_REMOVE(&telemetry->client_list_head, client,
+				client_list);
+			ret = rte_telemetry_client_cleanup(client);
+			if (ret < 0) {
+				TELEMETRY_LOG_ERR("Error - Client cleanup "
+					"failed\n");
+				return -EPERM;
+			}
+			return 0;
+		}
+	}
+	TELEMETRY_LOG_WARN("Warning - couldn't find client, possibly not "
+		"registered yet.\n");
+	return -1;
+
+einval_fail:
+	ret = rte_telemetry_send_error_response(telemetry, -EINVAL);
+	if (ret < 0)
+		TELEMETRY_LOG_ERR("Error - Could not send error\n");
+	return -EINVAL;
+}
+
+int32_t
+rte_telemetry_register_client(struct telemetry_impl *telemetry,
+	const char *client_path)
+{
+	int ret, fd;
+	if (!telemetry) {
+		TELEMETRY_LOG_ERR("Error - could not initialize "
+			"TELEMETRY API\n");
+		return -ENODEV;
+	}
+
+	if (!client_path) {
+		TELEMETRY_LOG_ERR("Error - Invalid client path\n");
+		return -EINVAL;
+	}
+
+	telemetry_client *client;
+	TAILQ_FOREACH(client, &telemetry->client_list_head, client_list) {
+		if (strcmp(client_path, client->file_path) == 0) {
+			TELEMETRY_LOG_WARN("Warning - '%s' already "
+				"registered\n", client_path);
+			return -EINVAL;
+		}
+	}
+	fd = socket(AF_UNIX, SOCK_STREAM, 0);
+	if (fd == -1) {
+		TELEMETRY_LOG_ERR("Error - Client socket error\n");
+		return -EACCES;
+	}
+	ret = rte_telemetry_set_socket_nonblock(fd);
+	if (ret < 0) {
+		TELEMETRY_LOG_ERR("Error - Could not set socket to NONBLOCK\n");
+		return -EPERM;
+	}
+
+	struct sockaddr_un addrs = {0};
+	addrs.sun_family = AF_UNIX;
+	strlcpy(addrs.sun_path, client_path, sizeof(addrs.sun_path));
+	telemetry_client *new_client =
+		(telemetry_client *)malloc(sizeof(telemetry_client));
+	new_client->file_path = strdup(client_path);
+	new_client->fd = fd;
+
+	if (connect(fd, (struct sockaddr *)&addrs, sizeof(addrs)) == -1) {
+		TELEMETRY_LOG_ERR("Error - TELEMETRY client connect to %s "
+			"didn't work\n", client_path);
+		ret = rte_telemetry_client_cleanup(new_client);
+		if (ret < 0) {
+			TELEMETRY_LOG_ERR("Error - Client cleanup failed\n");
+			return -EPERM;
+		}
+		return -EINVAL;
+	}
+	TAILQ_INSERT_HEAD(&telemetry->client_list_head, new_client,
+		client_list);
+
+	return 0;
+}
+
+int32_t
+rte_telemetry_parse_client_message(struct telemetry_impl *telemetry, char *buf)
+{
+	int ret;
+	json_error_t error;
+	json_t *root = json_loads(buf, 0, &error);
+
+	if (!root) {
+		TELEMETRY_LOG_WARN("Warning - Could not load JSON object from "
+			"data passed in : %s\n", error.text);
+		goto fail;
+	} else if (!json_is_object(root)) {
+		TELEMETRY_LOG_WARN("Warning - JSON Request is not a JSON "
+			"object\n");
+		json_decref(root);
+		goto fail;
+	}
+
+	json_t *action = json_object_get(root, "action");
+	if (!action) {
+		TELEMETRY_LOG_WARN("Warning - Request does not have action "
+			"field\n");
+		goto fail;
+	} else if (!json_is_integer(action)) {
+		TELEMETRY_LOG_WARN("Warning - Action value is not an "
+			"integer\n");
+		goto fail;
+	}
+
+	json_t *command = json_object_get(root, "command");
+	if (!command) {
+		TELEMETRY_LOG_WARN("Warning - Request does not have command "
+			"field\n");
+		goto fail;
+	} else if (!json_is_string(command)) {
+		TELEMETRY_LOG_WARN("Warning - Command value is not a string\n");
+		goto fail;
+	}
+
+	int action_int = json_integer_value(action);
+	if (action_int != ACTION_POST) {
+		TELEMETRY_LOG_WARN("Warning - Invalid action code\n");
+		goto fail;
+	}
+
+	const char *command_string = json_string_value(command);
+	if (strcmp(command_string, "clients") != 0) {
+		TELEMETRY_LOG_WARN("Warning - Invalid command\n");
+		goto fail;
+	}
+
+	json_t *data = json_object_get(root, "data");
+	if (!data) {
+		TELEMETRY_LOG_WARN("Warning - Request does not have data "
+			"field\n");
+		goto fail;
+	}
+
+	json_t *client_path = json_object_get(data, "client_path");
+	if (!client_path) {
+		TELEMETRY_LOG_WARN("Warning - Request does not have "
+			"client_path field\n");
+		goto fail;
+	}
+	if (!json_is_string(client_path)) {
+		TELEMETRY_LOG_WARN("Warning - client_path value is not a "
+			"string\n");
+		goto fail;
+	}
+
+	const char *client_path_string = json_string_value(client_path);
+
+	ret = rte_telemetry_register_client(telemetry, client_path_string);
+	if (ret < 0) {
+		TELEMETRY_LOG_ERR("Error - could not register client\n");
+		telemetry->register_fail_count++;
+		goto fail;
+	}
+
+	return 0;
+
+fail:
+	TELEMETRY_LOG_WARN("Warning - Client attempted to register with "
+		"invalid message\n");
+	return -1;
+}
+
 int telemetry_log_level;
 RTE_INIT(rte_telemetry_log_init);
 
diff --git a/lib/librte_telemetry/rte_telemetry_internal.h b/lib/librte_telemetry/rte_telemetry_internal.h
index 569d56a..e3292cf 100644
--- a/lib/librte_telemetry/rte_telemetry_internal.h
+++ b/lib/librte_telemetry/rte_telemetry_internal.h
@@ -3,6 +3,7 @@ 
  */
 
 #include <rte_log.h>
+#include <rte_tailq.h>
 
 #ifndef _RTE_TELEMETRY_INTERNAL_H_
 #define _RTE_TELEMETRY_INTERNAL_H_
@@ -23,6 +24,12 @@  extern int telemetry_log_level;
 #define TELEMETRY_LOG_INFO(fmt, args...) \
 	TELEMETRY_LOG(INFO, fmt, ## args)
 
+typedef struct telemetry_client {
+	char *file_path;
+	int fd;
+	TAILQ_ENTRY(telemetry_client) client_list;
+} telemetry_client;
+
 typedef struct telemetry_impl {
 	int accept_fd;
 	int server_fd;
@@ -31,6 +38,24 @@  typedef struct telemetry_impl {
 	uint32_t socket_id;
 	int reg_index;
 	int metrics_register_done;
+	TAILQ_HEAD(, telemetry_client) client_list_head;
+	struct telemetry_client *request_client;
+	int register_fail_count;
 } telemetry_impl;
 
+int32_t
+rte_telemetry_parse_client_message(struct telemetry_impl *telemetry, char *buf);
+
+int32_t
+rte_telemetry_send_error_response(struct telemetry_impl *telemetry,
+	int error_type);
+
+int32_t
+rte_telemetry_register_client(struct telemetry_impl *telemetry,
+	const char *client_path);
+
+int32_t
+rte_telemetry_unregister_client(struct telemetry_impl *telemetry,
+	const char *client_path);
+
 #endif
diff --git a/mk/rte.app.mk b/mk/rte.app.mk
index 8551191..1963812 100644
--- a/mk/rte.app.mk
+++ b/mk/rte.app.mk
@@ -80,7 +80,7 @@  _LDLIBS-$(CONFIG_RTE_LIBRTE_SECURITY)       += -lrte_security
 _LDLIBS-$(CONFIG_RTE_LIBRTE_COMPRESSDEV)    += -lrte_compressdev
 _LDLIBS-$(CONFIG_RTE_LIBRTE_EVENTDEV)       += -lrte_eventdev
 _LDLIBS-$(CONFIG_RTE_LIBRTE_RAWDEV)         += -lrte_rawdev
-_LDLIBS-$(CONFIG_RTE_LIBRTE_TELEMETRY)      += -lrte_metrics -lrte_telemetry
+_LDLIBS-$(CONFIG_RTE_LIBRTE_TELEMETRY)      += -lrte_metrics -lrte_telemetry -ljansson
 _LDLIBS-$(CONFIG_RTE_LIBRTE_TIMER)          += -lrte_timer
 _LDLIBS-$(CONFIG_RTE_LIBRTE_MEMPOOL)        += -lrte_mempool
 _LDLIBS-$(CONFIG_RTE_DRIVER_MEMPOOL_RING)   += -lrte_mempool_ring