lib/librte_telemetry/telemetry.c

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  * Copyright(c) 2020 Intel Corporation
   3  */
   4
   5 #include <unistd.h>
   6 #include <pthread.h>
   7 #include <sys/socket.h>
   8 #include <sys/un.h>
   9 #include <dlfcn.h>
  10
  11 /* we won't link against libbsd, so just always use DPDKs-specific strlcpy */
  12 #undef RTE_USE_LIBBSD
  13 #include <rte_string_fns.h>
  14 #include <rte_common.h>
  15 #include <rte_spinlock.h>
  16 #include <rte_version.h>
  17
  18 #include "rte_telemetry.h"
  19 #include "telemetry_json.h"
  20 #include "telemetry_data.h"
  21 #include "rte_telemetry_legacy.h"
  22
  23 #define MAX_CMD_LEN 56
  24 #define MAX_HELP_LEN 64
  25 #define MAX_OUTPUT_LEN (1024 * 16)
  26 #define MAX_CONNECTIONS 10
  27
  28 static void *
  29 client_handler(void *socket);
  30
  31 struct cmd_callback {
  32         char cmd[MAX_CMD_LEN];
  33         telemetry_cb fn;
  34         char help[MAX_HELP_LEN];
  35 };
  36
  37 struct socket {
  38         int sock;
  39         char path[sizeof(((struct sockaddr_un *)0)->sun_path)];
  40         handler fn;
  41         uint16_t *num_clients;
  42 };
  43 static struct socket v2_socket; /* socket for v2 telemetry */
  44 static struct socket v1_socket; /* socket for v1 telemetry */
  45 static char telemetry_log_error[1024]; /* Will contain error on init failure */
  46 /* list of command callbacks, with one command registered by default */
  47 static struct cmd_callback callbacks[TELEMETRY_MAX_CALLBACKS];
  48 static int num_callbacks; /* How many commands are registered */
  49 /* Used when accessing or modifying list of command callbacks */
  50 static rte_spinlock_t callback_sl = RTE_SPINLOCK_INITIALIZER;
  51 static uint16_t v2_clients;
  52
  53 int
  54 rte_telemetry_register_cmd(const char *cmd, telemetry_cb fn, const char *help)
  55 {
  56         int i = 0;
  57
  58         if (strlen(cmd) >= MAX_CMD_LEN || fn == NULL || cmd[0] != '/'
  59                         || strlen(help) >= MAX_HELP_LEN)
  60                 return -EINVAL;
  61         if (num_callbacks >= TELEMETRY_MAX_CALLBACKS)
  62                 return -ENOENT;
  63
  64         rte_spinlock_lock(&callback_sl);
  65         while (i < num_callbacks && strcmp(cmd, callbacks[i].cmd) > 0)
  66                 i++;
  67         if (i != num_callbacks)
  68                 /* Move elements to keep the list alphabetical */
  69                 memmove(callbacks + i + 1, callbacks + i,
  70                         sizeof(struct cmd_callback) * (num_callbacks - i));
  71
  72         strlcpy(callbacks[i].cmd, cmd, MAX_CMD_LEN);
  73         callbacks[i].fn = fn;
  74         strlcpy(callbacks[i].help, help, MAX_HELP_LEN);
  75         num_callbacks++;
  76         rte_spinlock_unlock(&callback_sl);
  77
  78         return 0;
  79 }
  80
  81 static int
  82 list_commands(const char *cmd __rte_unused, const char *params __rte_unused,
  83                 struct rte_tel_data *d)
  84 {
  85         int i;
  86
  87         rte_tel_data_start_array(d, RTE_TEL_STRING_VAL);
  88         for (i = 0; i < num_callbacks; i++)
  89                 rte_tel_data_add_array_string(d, callbacks[i].cmd);
  90         return 0;
  91 }
  92
  93 static int
  94 json_info(const char *cmd __rte_unused, const char *params __rte_unused,
  95                 struct rte_tel_data *d)
  96 {
  97         rte_tel_data_start_dict(d);
  98         rte_tel_data_add_dict_string(d, "version", rte_version());
  99         rte_tel_data_add_dict_int(d, "pid", getpid());
 100         rte_tel_data_add_dict_int(d, "max_output_len", MAX_OUTPUT_LEN);
 101         return 0;
 102 }
 103
 104 static int
 105 command_help(const char *cmd __rte_unused, const char *params,
 106                 struct rte_tel_data *d)
 107 {
 108         int i;
 109
 110         if (!params)
 111                 return -1;
 112         rte_tel_data_start_dict(d);
 113         rte_spinlock_lock(&callback_sl);
 114         for (i = 0; i < num_callbacks; i++)
 115                 if (strcmp(params, callbacks[i].cmd) == 0) {
 116                         rte_tel_data_add_dict_string(d, params,
 117                                         callbacks[i].help);
 118                         break;
 119                 }
 120         rte_spinlock_unlock(&callback_sl);
 121         if (i == num_callbacks)
 122                 return -1;
 123         return 0;
 124 }
 125
 126 static void
 127 output_json(const char *cmd, const struct rte_tel_data *d, int s)
 128 {
 129         char out_buf[MAX_OUTPUT_LEN];
 130
 131         char *cb_data_buf;
 132         size_t buf_len, prefix_used, used = 0;
 133         unsigned int i;
 134
 135         RTE_BUILD_BUG_ON(sizeof(out_buf) < MAX_CMD_LEN +
 136                         RTE_TEL_MAX_SINGLE_STRING_LEN + 10);
 137         switch (d->type) {
 138         case RTE_TEL_NULL:
 139                 used = snprintf(out_buf, sizeof(out_buf), "{\"%.*s\":null}",
 140                                 MAX_CMD_LEN, cmd ? cmd : "none");
 141                 break;
 142         case RTE_TEL_STRING:
 143                 used = snprintf(out_buf, sizeof(out_buf), "{\"%.*s\":\"%.*s\"}",
 144                                 MAX_CMD_LEN, cmd,
 145                                 RTE_TEL_MAX_SINGLE_STRING_LEN, d->data.str);
 146                 break;
 147         case RTE_TEL_DICT:
 148                 prefix_used = snprintf(out_buf, sizeof(out_buf), "{\"%.*s\":",
 149                                 MAX_CMD_LEN, cmd);
 150                 cb_data_buf = &out_buf[prefix_used];
 151                 buf_len = sizeof(out_buf) - prefix_used - 1; /* space for '}' */
 152
 153                 used = rte_tel_json_empty_obj(cb_data_buf, buf_len, 0);
 154                 for (i = 0; i < d->data_len; i++) {
 155                         const struct tel_dict_entry *v = &d->data.dict[i];
 156                         switch (v->type) {
 157                         case RTE_TEL_STRING_VAL:
 158                                 used = rte_tel_json_add_obj_str(cb_data_buf,
 159                                                 buf_len, used,
 160                                                 v->name, v->value.sval);
 161                                 break;
 162                         case RTE_TEL_INT_VAL:
 163                                 used = rte_tel_json_add_obj_int(cb_data_buf,
 164                                                 buf_len, used,
 165                                                 v->name, v->value.ival);
 166                                 break;
 167                         case RTE_TEL_U64_VAL:
 168                                 used = rte_tel_json_add_obj_u64(cb_data_buf,
 169                                                 buf_len, used,
 170                                                 v->name, v->value.u64val);
 171                                 break;
 172                         }
 173                 }
 174                 used += prefix_used;
 175                 used += strlcat(out_buf + used, "}", sizeof(out_buf) - used);
 176                 break;
 177         case RTE_TEL_ARRAY_STRING:
 178         case RTE_TEL_ARRAY_INT:
 179         case RTE_TEL_ARRAY_U64:
 180                 prefix_used = snprintf(out_buf, sizeof(out_buf), "{\"%.*s\":",
 181                                 MAX_CMD_LEN, cmd);
 182                 cb_data_buf = &out_buf[prefix_used];
 183                 buf_len = sizeof(out_buf) - prefix_used - 1; /* space for '}' */
 184
 185                 used = rte_tel_json_empty_array(cb_data_buf, buf_len, 0);
 186                 for (i = 0; i < d->data_len; i++)
 187                         if (d->type == RTE_TEL_ARRAY_STRING)
 188                                 used = rte_tel_json_add_array_string(
 189                                                 cb_data_buf,
 190                                                 buf_len, used,
 191                                                 d->data.array[i].sval);
 192                         else if (d->type == RTE_TEL_ARRAY_INT)
 193                                 used = rte_tel_json_add_array_int(cb_data_buf,
 194                                                 buf_len, used,
 195                                                 d->data.array[i].ival);
 196                         else if (d->type == RTE_TEL_ARRAY_U64)
 197                                 used = rte_tel_json_add_array_u64(cb_data_buf,
 198                                                 buf_len, used,
 199                                                 d->data.array[i].u64val);
 200                 used += prefix_used;
 201                 used += strlcat(out_buf + used, "}", sizeof(out_buf) - used);
 202                 break;
 203         }
 204         if (write(s, out_buf, used) < 0)
 205                 perror("Error writing to socket");
 206 }
 207
 208 static void
 209 perform_command(telemetry_cb fn, const char *cmd, const char *param, int s)
 210 {
 211         struct rte_tel_data data;
 212
 213         int ret = fn(cmd, param, &data);
 214         if (ret < 0) {
 215                 char out_buf[MAX_CMD_LEN + 10];
 216                 int used = snprintf(out_buf, sizeof(out_buf), "{\"%.*s\":null}",
 217                                 MAX_CMD_LEN, cmd ? cmd : "none");
 218                 if (write(s, out_buf, used) < 0)
 219                         perror("Error writing to socket");
 220                 return;
 221         }
 222         output_json(cmd, &data, s);
 223 }
 224
 225 static int
 226 unknown_command(const char *cmd __rte_unused, const char *params __rte_unused,
 227                 struct rte_tel_data *d)
 228 {
 229         return d->type = RTE_TEL_NULL;
 230 }
 231
 232 static void *
 233 client_handler(void *sock_id)
 234 {
 235         int s = (int)(uintptr_t)sock_id;
 236         char buffer[1024];
 237         char info_str[1024];
 238         snprintf(info_str, sizeof(info_str),
 239                         "{\"version\":\"%s\",\"pid\":%d,\"max_output_len\":%d}",
 240                         rte_version(), getpid(), MAX_OUTPUT_LEN);
 241         if (write(s, info_str, strlen(info_str)) < 0) {
 242                 close(s);
 243                 return NULL;
 244         }
 245
 246         /* receive data is not null terminated */
 247         int bytes = read(s, buffer, sizeof(buffer) - 1);
 248         while (bytes > 0) {
 249                 buffer[bytes] = 0;
 250                 const char *cmd = strtok(buffer, ",");
 251                 const char *param = strtok(NULL, ",");
 252                 telemetry_cb fn = unknown_command;
 253                 int i;
 254
 255                 if (cmd && strlen(cmd) < MAX_CMD_LEN) {
 256                         rte_spinlock_lock(&callback_sl);
 257                         for (i = 0; i < num_callbacks; i++)
 258                                 if (strcmp(cmd, callbacks[i].cmd) == 0) {
 259                                         fn = callbacks[i].fn;
 260                                         break;
 261                                 }
 262                         rte_spinlock_unlock(&callback_sl);
 263                 }
 264                 perform_command(fn, cmd, param, s);
 265
 266                 bytes = read(s, buffer, sizeof(buffer) - 1);
 267         }
 268         close(s);
 269         __atomic_sub_fetch(&v2_clients, 1, __ATOMIC_RELAXED);
 270         return NULL;
 271 }
 272
 273 static void *
 274 socket_listener(void *socket)
 275 {
 276         while (1) {
 277                 pthread_t th;
 278                 struct socket *s = (struct socket *)socket;
 279                 int s_accepted = accept(s->sock, NULL, NULL);
 280                 if (s_accepted < 0) {
 281                         snprintf(telemetry_log_error,
 282                                 sizeof(telemetry_log_error),
 283                                 "Error with accept, telemetry thread quitting");
 284                         return NULL;
 285                 }
 286                 if (s->num_clients != NULL) {
 287                         uint16_t conns = __atomic_load_n(s->num_clients,
 288                                         __ATOMIC_RELAXED);
 289                         if (conns >= MAX_CONNECTIONS) {
 290                                 close(s_accepted);
 291                                 continue;
 292                         }
 293                         __atomic_add_fetch(s->num_clients, 1,
 294                                         __ATOMIC_RELAXED);
 295                 }
 296                 pthread_create(&th, NULL, s->fn, (void *)(uintptr_t)s_accepted);
 297                 pthread_detach(th);
 298         }
 299         return NULL;
 300 }
 301
 302 static inline char *
 303 get_socket_path(const char *runtime_dir, const int version)
 304 {
 305         static char path[PATH_MAX];
 306         snprintf(path, sizeof(path), "%s/dpdk_telemetry.v%d",
 307                         strlen(runtime_dir) ? runtime_dir : "/tmp", version);
 308         return path;
 309 }
 310
 311 static void
 312 unlink_sockets(void)
 313 {
 314         if (v2_socket.path[0])
 315                 unlink(v2_socket.path);
 316         if (v1_socket.path[0])
 317                 unlink(v1_socket.path);
 318 }
 319
 320 static int
 321 create_socket(char *path)
 322 {
 323         int sock = socket(AF_UNIX, SOCK_SEQPACKET, 0);
 324         if (sock < 0) {
 325                 snprintf(telemetry_log_error, sizeof(telemetry_log_error),
 326                                 "Error with socket creation, %s",
 327                                 strerror(errno));
 328                 return -1;
 329         }
 330
 331         struct sockaddr_un sun = {.sun_family = AF_UNIX};
 332         strlcpy(sun.sun_path, path, sizeof(sun.sun_path));
 333         unlink(sun.sun_path);
 334         if (bind(sock, (void *) &sun, sizeof(sun)) < 0) {
 335                 snprintf(telemetry_log_error, sizeof(telemetry_log_error),
 336                                 "Error binding socket: %s",
 337                                 strerror(errno));
 338                 sun.sun_path[0] = 0;
 339                 goto error;
 340         }
 341
 342         if (listen(sock, 1) < 0) {
 343                 snprintf(telemetry_log_error, sizeof(telemetry_log_error),
 344                                 "Error calling listen for socket: %s",
 345                                 strerror(errno));
 346                 goto error;
 347         }
 348
 349         return sock;
 350
 351 error:
 352         close(sock);
 353         unlink_sockets();
 354         return -1;
 355 }
 356
 357 static int
 358 telemetry_legacy_init(const char *runtime_dir, rte_cpuset_t *cpuset)
 359 {
 360         pthread_t t_old;
 361
 362         if (num_legacy_callbacks == 1) {
 363                 snprintf(telemetry_log_error, sizeof(telemetry_log_error),
 364                          "No legacy callbacks, legacy socket not created");
 365                 return -1;
 366         }
 367
 368         v1_socket.fn = legacy_client_handler;
 369         if ((size_t) snprintf(v1_socket.path, sizeof(v1_socket.path),
 370                         "%s/telemetry", runtime_dir)
 371                         >= sizeof(v1_socket.path)) {
 372                 snprintf(telemetry_log_error, sizeof(telemetry_log_error),
 373                                 "Error with socket binding, path too long");
 374                 return -1;
 375         }
 376         v1_socket.sock = create_socket(v1_socket.path);
 377         if (v1_socket.sock < 0)
 378                 return -1;
 379         pthread_create(&t_old, NULL, socket_listener, &v1_socket);
 380         pthread_setaffinity_np(t_old, sizeof(*cpuset), cpuset);
 381
 382         return 0;
 383 }
 384
 385 static int
 386 telemetry_v2_init(const char *runtime_dir, rte_cpuset_t *cpuset)
 387 {
 388         pthread_t t_new;
 389
 390         v2_socket.num_clients = &v2_clients;
 391         rte_telemetry_register_cmd("/", list_commands,
 392                         "Returns list of available commands, Takes no parameters");
 393         rte_telemetry_register_cmd("/info", json_info,
 394                         "Returns DPDK Telemetry information. Takes no parameters");
 395         rte_telemetry_register_cmd("/help", command_help,
 396                         "Returns help text for a command. Parameters: string command");
 397         v2_socket.fn = client_handler;
 398         if (strlcpy(v2_socket.path, get_socket_path(runtime_dir, 2),
 399                         sizeof(v2_socket.path)) >= sizeof(v2_socket.path)) {
 400                 snprintf(telemetry_log_error, sizeof(telemetry_log_error),
 401                                 "Error with socket binding, path too long");
 402                 return -1;
 403         }
 404
 405         v2_socket.sock = create_socket(v2_socket.path);
 406         if (v2_socket.sock < 0)
 407                 return -1;
 408         pthread_create(&t_new, NULL, socket_listener, &v2_socket);
 409         pthread_setaffinity_np(t_new, sizeof(*cpuset), cpuset);
 410         atexit(unlink_sockets);
 411
 412         return 0;
 413 }
 414
 415 int32_t
 416 rte_telemetry_init(const char *runtime_dir, rte_cpuset_t *cpuset,
 417                 const char **err_str)
 418 {
 419         if (telemetry_v2_init(runtime_dir, cpuset) != 0) {
 420                 *err_str = telemetry_log_error;
 421                 return -1;
 422         }
 423         if (telemetry_legacy_init(runtime_dir, cpuset) != 0) {
 424                 *err_str = telemetry_log_error;
 425         }
 426         return 0;
 427 }