lib/telemetry/telemetry.c

   1 /* SPDX-License-Identifier: BSD-3-Clause
   2  * Copyright(c) 2020 Intel Corporation
   3  */
   4
   5 #ifndef RTE_EXEC_ENV_WINDOWS
   6 #include <unistd.h>
   7 #include <pthread.h>
   8 #include <sys/socket.h>
   9 #include <sys/un.h>
  10 #include <dlfcn.h>
  11 #endif /* !RTE_EXEC_ENV_WINDOWS */
  12
  13 /* we won't link against libbsd, so just always use DPDKs-specific strlcpy */
  14 #undef RTE_USE_LIBBSD
  15 #include <rte_string_fns.h>
  16 #include <rte_common.h>
  17 #include <rte_spinlock.h>
  18 #include <rte_log.h>
  19
  20 #include "rte_telemetry.h"
  21 #include "telemetry_json.h"
  22 #include "telemetry_data.h"
  23 #include "telemetry_internal.h"
  24
  25 #define MAX_CMD_LEN 56
  26 #define MAX_HELP_LEN 64
  27 #define MAX_OUTPUT_LEN (1024 * 16)
  28 #define MAX_CONNECTIONS 10
  29
  30 #ifndef RTE_EXEC_ENV_WINDOWS
  31 static void *
  32 client_handler(void *socket);
  33 #endif /* !RTE_EXEC_ENV_WINDOWS */
  34
  35 struct cmd_callback {
  36         char cmd[MAX_CMD_LEN];
  37         telemetry_cb fn;
  38         char help[MAX_HELP_LEN];
  39 };
  40
  41 #ifndef RTE_EXEC_ENV_WINDOWS
  42 struct socket {
  43         int sock;
  44         char path[sizeof(((struct sockaddr_un *)0)->sun_path)];
  45         handler fn;
  46         uint16_t *num_clients;
  47 };
  48 static struct socket v2_socket; /* socket for v2 telemetry */
  49 static struct socket v1_socket; /* socket for v1 telemetry */
  50 #endif /* !RTE_EXEC_ENV_WINDOWS */
  51
  52 static const char *telemetry_version; /* save rte_version */
  53 static const char *socket_dir;        /* runtime directory */
  54 static rte_cpuset_t *thread_cpuset;
  55 static rte_log_fn rte_log_ptr;
  56 static uint32_t logtype;
  57
  58 #define TMTY_LOG(l, ...) \
  59         rte_log_ptr(RTE_LOG_ ## l, logtype, "TELEMETRY: " __VA_ARGS__)
  60
  61 /* list of command callbacks, with one command registered by default */
  62 static struct cmd_callback *callbacks;
  63 static int num_callbacks; /* How many commands are registered */
  64 /* Used when accessing or modifying list of command callbacks */
  65 static rte_spinlock_t callback_sl = RTE_SPINLOCK_INITIALIZER;
  66 #ifndef RTE_EXEC_ENV_WINDOWS
  67 static uint16_t v2_clients;
  68 #endif /* !RTE_EXEC_ENV_WINDOWS */
  69
  70 int
  71 rte_telemetry_register_cmd(const char *cmd, telemetry_cb fn, const char *help)
  72 {
  73         struct cmd_callback *new_callbacks;
  74         int i = 0;
  75
  76         if (strlen(cmd) >= MAX_CMD_LEN || fn == NULL || cmd[0] != '/'
  77                         || strlen(help) >= MAX_HELP_LEN)
  78                 return -EINVAL;
  79
  80         rte_spinlock_lock(&callback_sl);
  81         new_callbacks = realloc(callbacks, sizeof(callbacks[0]) * (num_callbacks + 1));
  82         if (new_callbacks == NULL) {
  83                 rte_spinlock_unlock(&callback_sl);
  84                 return -ENOMEM;
  85         }
  86         callbacks = new_callbacks;
  87
  88         while (i < num_callbacks && strcmp(cmd, callbacks[i].cmd) > 0)
  89                 i++;
  90         if (i != num_callbacks)
  91                 /* Move elements to keep the list alphabetical */
  92                 memmove(callbacks + i + 1, callbacks + i,
  93                         sizeof(struct cmd_callback) * (num_callbacks - i));
  94
  95         strlcpy(callbacks[i].cmd, cmd, MAX_CMD_LEN);
  96         callbacks[i].fn = fn;
  97         strlcpy(callbacks[i].help, help, MAX_HELP_LEN);
  98         num_callbacks++;
  99         rte_spinlock_unlock(&callback_sl);
 100
 101         return 0;
 102 }
 103
 104 #ifndef RTE_EXEC_ENV_WINDOWS
 105
 106 static int
 107 list_commands(const char *cmd __rte_unused, const char *params __rte_unused,
 108                 struct rte_tel_data *d)
 109 {
 110         int i;
 111
 112         rte_tel_data_start_array(d, RTE_TEL_STRING_VAL);
 113         rte_spinlock_lock(&callback_sl);
 114         for (i = 0; i < num_callbacks; i++)
 115                 rte_tel_data_add_array_string(d, callbacks[i].cmd);
 116         rte_spinlock_unlock(&callback_sl);
 117         return 0;
 118 }
 119
 120 static int
 121 json_info(const char *cmd __rte_unused, const char *params __rte_unused,
 122                 struct rte_tel_data *d)
 123 {
 124         rte_tel_data_start_dict(d);
 125         rte_tel_data_add_dict_string(d, "version", telemetry_version);
 126         rte_tel_data_add_dict_int(d, "pid", getpid());
 127         rte_tel_data_add_dict_int(d, "max_output_len", MAX_OUTPUT_LEN);
 128         return 0;
 129 }
 130
 131 static int
 132 command_help(const char *cmd __rte_unused, const char *params,
 133                 struct rte_tel_data *d)
 134 {
 135         int i;
 136
 137         if (!params)
 138                 return -1;
 139         rte_tel_data_start_dict(d);
 140         rte_spinlock_lock(&callback_sl);
 141         for (i = 0; i < num_callbacks; i++)
 142                 if (strcmp(params, callbacks[i].cmd) == 0) {
 143                         rte_tel_data_add_dict_string(d, params,
 144                                         callbacks[i].help);
 145                         break;
 146                 }
 147         rte_spinlock_unlock(&callback_sl);
 148         if (i == num_callbacks)
 149                 return -1;
 150         return 0;
 151 }
 152
 153 static int
 154 container_to_json(const struct rte_tel_data *d, char *out_buf, size_t buf_len)
 155 {
 156         size_t used = 0;
 157         unsigned int i;
 158
 159         if (d->type != RTE_TEL_ARRAY_U64 && d->type != RTE_TEL_ARRAY_INT
 160                         && d->type != RTE_TEL_ARRAY_STRING)
 161                 return snprintf(out_buf, buf_len, "null");
 162
 163         used = rte_tel_json_empty_array(out_buf, buf_len, 0);
 164         if (d->type == RTE_TEL_ARRAY_U64)
 165                 for (i = 0; i < d->data_len; i++)
 166                         used = rte_tel_json_add_array_u64(out_buf,
 167                                 buf_len, used,
 168                                 d->data.array[i].u64val);
 169         if (d->type == RTE_TEL_ARRAY_INT)
 170                 for (i = 0; i < d->data_len; i++)
 171                         used = rte_tel_json_add_array_int(out_buf,
 172                                 buf_len, used,
 173                                 d->data.array[i].ival);
 174         if (d->type == RTE_TEL_ARRAY_STRING)
 175                 for (i = 0; i < d->data_len; i++)
 176                         used = rte_tel_json_add_array_string(out_buf,
 177                                 buf_len, used,
 178                                 d->data.array[i].sval);
 179         return used;
 180 }
 181
 182 static void
 183 output_json(const char *cmd, const struct rte_tel_data *d, int s)
 184 {
 185         char out_buf[MAX_OUTPUT_LEN];
 186
 187         char *cb_data_buf;
 188         size_t buf_len, prefix_used, used = 0;
 189         unsigned int i;
 190
 191         RTE_BUILD_BUG_ON(sizeof(out_buf) < MAX_CMD_LEN +
 192                         RTE_TEL_MAX_SINGLE_STRING_LEN + 10);
 193         switch (d->type) {
 194         case RTE_TEL_NULL:
 195                 used = snprintf(out_buf, sizeof(out_buf), "{\"%.*s\":null}",
 196                                 MAX_CMD_LEN, cmd ? cmd : "none");
 197                 break;
 198         case RTE_TEL_STRING:
 199                 used = snprintf(out_buf, sizeof(out_buf), "{\"%.*s\":\"%.*s\"}",
 200                                 MAX_CMD_LEN, cmd,
 201                                 RTE_TEL_MAX_SINGLE_STRING_LEN, d->data.str);
 202                 break;
 203         case RTE_TEL_DICT:
 204                 prefix_used = snprintf(out_buf, sizeof(out_buf), "{\"%.*s\":",
 205                                 MAX_CMD_LEN, cmd);
 206                 cb_data_buf = &out_buf[prefix_used];
 207                 buf_len = sizeof(out_buf) - prefix_used - 1; /* space for '}' */
 208
 209                 used = rte_tel_json_empty_obj(cb_data_buf, buf_len, 0);
 210                 for (i = 0; i < d->data_len; i++) {
 211                         const struct tel_dict_entry *v = &d->data.dict[i];
 212                         switch (v->type) {
 213                         case RTE_TEL_STRING_VAL:
 214                                 used = rte_tel_json_add_obj_str(cb_data_buf,
 215                                                 buf_len, used,
 216                                                 v->name, v->value.sval);
 217                                 break;
 218                         case RTE_TEL_INT_VAL:
 219                                 used = rte_tel_json_add_obj_int(cb_data_buf,
 220                                                 buf_len, used,
 221                                                 v->name, v->value.ival);
 222                                 break;
 223                         case RTE_TEL_U64_VAL:
 224                                 used = rte_tel_json_add_obj_u64(cb_data_buf,
 225                                                 buf_len, used,
 226                                                 v->name, v->value.u64val);
 227                                 break;
 228                         case RTE_TEL_CONTAINER:
 229                         {
 230                                 char temp[buf_len];
 231                                 const struct container *cont =
 232                                                 &v->value.container;
 233                                 if (container_to_json(cont->data,
 234                                                 temp, buf_len) != 0)
 235                                         used = rte_tel_json_add_obj_json(
 236                                                         cb_data_buf,
 237                                                         buf_len, used,
 238                                                         v->name, temp);
 239                                 if (!cont->keep)
 240                                         rte_tel_data_free(cont->data);
 241                         }
 242                         }
 243                 }
 244                 used += prefix_used;
 245                 used += strlcat(out_buf + used, "}", sizeof(out_buf) - used);
 246                 break;
 247         case RTE_TEL_ARRAY_STRING:
 248         case RTE_TEL_ARRAY_INT:
 249         case RTE_TEL_ARRAY_U64:
 250         case RTE_TEL_ARRAY_CONTAINER:
 251                 prefix_used = snprintf(out_buf, sizeof(out_buf), "{\"%.*s\":",
 252                                 MAX_CMD_LEN, cmd);
 253                 cb_data_buf = &out_buf[prefix_used];
 254                 buf_len = sizeof(out_buf) - prefix_used - 1; /* space for '}' */
 255
 256                 used = rte_tel_json_empty_array(cb_data_buf, buf_len, 0);
 257                 for (i = 0; i < d->data_len; i++)
 258                         if (d->type == RTE_TEL_ARRAY_STRING)
 259                                 used = rte_tel_json_add_array_string(
 260                                                 cb_data_buf,
 261                                                 buf_len, used,
 262                                                 d->data.array[i].sval);
 263                         else if (d->type == RTE_TEL_ARRAY_INT)
 264                                 used = rte_tel_json_add_array_int(cb_data_buf,
 265                                                 buf_len, used,
 266                                                 d->data.array[i].ival);
 267                         else if (d->type == RTE_TEL_ARRAY_U64)
 268                                 used = rte_tel_json_add_array_u64(cb_data_buf,
 269                                                 buf_len, used,
 270                                                 d->data.array[i].u64val);
 271                         else if (d->type == RTE_TEL_ARRAY_CONTAINER) {
 272                                 char temp[buf_len];
 273                                 const struct container *rec_data =
 274                                                 &d->data.array[i].container;
 275                                 if (container_to_json(rec_data->data,
 276                                                 temp, buf_len) != 0)
 277                                         used = rte_tel_json_add_array_json(
 278                                                         cb_data_buf,
 279                                                         buf_len, used, temp);
 280                                 if (!rec_data->keep)
 281                                         rte_tel_data_free(rec_data->data);
 282                         }
 283                 used += prefix_used;
 284                 used += strlcat(out_buf + used, "}", sizeof(out_buf) - used);
 285                 break;
 286         }
 287         if (write(s, out_buf, used) < 0)
 288                 perror("Error writing to socket");
 289 }
 290
 291 static void
 292 perform_command(telemetry_cb fn, const char *cmd, const char *param, int s)
 293 {
 294         struct rte_tel_data data;
 295
 296         int ret = fn(cmd, param, &data);
 297         if (ret < 0) {
 298                 char out_buf[MAX_CMD_LEN + 10];
 299                 int used = snprintf(out_buf, sizeof(out_buf), "{\"%.*s\":null}",
 300                                 MAX_CMD_LEN, cmd ? cmd : "none");
 301                 if (write(s, out_buf, used) < 0)
 302                         perror("Error writing to socket");
 303                 return;
 304         }
 305         output_json(cmd, &data, s);
 306 }
 307
 308 static int
 309 unknown_command(const char *cmd __rte_unused, const char *params __rte_unused,
 310                 struct rte_tel_data *d)
 311 {
 312         return d->type = RTE_TEL_NULL;
 313 }
 314
 315 static void *
 316 client_handler(void *sock_id)
 317 {
 318         int s = (int)(uintptr_t)sock_id;
 319         char buffer[1024];
 320         char info_str[1024];
 321         snprintf(info_str, sizeof(info_str),
 322                         "{\"version\":\"%s\",\"pid\":%d,\"max_output_len\":%d}",
 323                         telemetry_version, getpid(), MAX_OUTPUT_LEN);
 324         if (write(s, info_str, strlen(info_str)) < 0) {
 325                 close(s);
 326                 return NULL;
 327         }
 328
 329         /* receive data is not null terminated */
 330         int bytes = read(s, buffer, sizeof(buffer) - 1);
 331         while (bytes > 0) {
 332                 buffer[bytes] = 0;
 333                 const char *cmd = strtok(buffer, ",");
 334                 const char *param = strtok(NULL, "\0");
 335                 telemetry_cb fn = unknown_command;
 336                 int i;
 337
 338                 if (cmd && strlen(cmd) < MAX_CMD_LEN) {
 339                         rte_spinlock_lock(&callback_sl);
 340                         for (i = 0; i < num_callbacks; i++)
 341                                 if (strcmp(cmd, callbacks[i].cmd) == 0) {
 342                                         fn = callbacks[i].fn;
 343                                         break;
 344                                 }
 345                         rte_spinlock_unlock(&callback_sl);
 346                 }
 347                 perform_command(fn, cmd, param, s);
 348
 349                 bytes = read(s, buffer, sizeof(buffer) - 1);
 350         }
 351         close(s);
 352         __atomic_sub_fetch(&v2_clients, 1, __ATOMIC_RELAXED);
 353         return NULL;
 354 }
 355
 356 static void *
 357 socket_listener(void *socket)
 358 {
 359         while (1) {
 360                 pthread_t th;
 361                 int rc;
 362                 struct socket *s = (struct socket *)socket;
 363                 int s_accepted = accept(s->sock, NULL, NULL);
 364                 if (s_accepted < 0) {
 365                         TMTY_LOG(ERR, "Error with accept, telemetry thread quitting\n");
 366                         return NULL;
 367                 }
 368                 if (s->num_clients != NULL) {
 369                         uint16_t conns = __atomic_load_n(s->num_clients,
 370                                         __ATOMIC_RELAXED);
 371                         if (conns >= MAX_CONNECTIONS) {
 372                                 close(s_accepted);
 373                                 continue;
 374                         }
 375                         __atomic_add_fetch(s->num_clients, 1,
 376                                         __ATOMIC_RELAXED);
 377                 }
 378                 rc = pthread_create(&th, NULL, s->fn,
 379                                     (void *)(uintptr_t)s_accepted);
 380                 if (rc != 0) {
 381                         TMTY_LOG(ERR, "Error with create client thread: %s\n",
 382                                  strerror(rc));
 383                         close(s_accepted);
 384                         if (s->num_clients != NULL)
 385                                 __atomic_sub_fetch(s->num_clients, 1,
 386                                                    __ATOMIC_RELAXED);
 387                         continue;
 388                 }
 389                 pthread_detach(th);
 390         }
 391         return NULL;
 392 }
 393
 394 static inline char *
 395 get_socket_path(const char *runtime_dir, const int version)
 396 {
 397         static char path[PATH_MAX];
 398         snprintf(path, sizeof(path), "%s/dpdk_telemetry.v%d",
 399                         strlen(runtime_dir) ? runtime_dir : "/tmp", version);
 400         return path;
 401 }
 402
 403 static void
 404 unlink_sockets(void)
 405 {
 406         if (v2_socket.path[0])
 407                 unlink(v2_socket.path);
 408         if (v1_socket.path[0])
 409                 unlink(v1_socket.path);
 410 }
 411
 412 static int
 413 create_socket(char *path)
 414 {
 415         int sock = socket(AF_UNIX, SOCK_SEQPACKET, 0);
 416         if (sock < 0) {
 417                 TMTY_LOG(ERR, "Error with socket creation, %s\n", strerror(errno));
 418                 return -1;
 419         }
 420
 421         struct sockaddr_un sun = {.sun_family = AF_UNIX};
 422         strlcpy(sun.sun_path, path, sizeof(sun.sun_path));
 423         unlink(sun.sun_path);
 424         if (bind(sock, (void *) &sun, sizeof(sun)) < 0) {
 425                 TMTY_LOG(ERR, "Error binding socket: %s\n", strerror(errno));
 426                 sun.sun_path[0] = 0;
 427                 goto error;
 428         }
 429
 430         if (listen(sock, 1) < 0) {
 431                 TMTY_LOG(ERR, "Error calling listen for socket: %s\n", strerror(errno));
 432                 goto error;
 433         }
 434
 435         return sock;
 436
 437 error:
 438         close(sock);
 439         unlink_sockets();
 440         return -1;
 441 }
 442
 443 static void
 444 set_thread_name(pthread_t id __rte_unused, const char *name __rte_unused)
 445 {
 446 #if defined RTE_EXEC_ENV_LINUX && defined __GLIBC__ && defined __GLIBC_PREREQ
 447 #if __GLIBC_PREREQ(2, 12)
 448         pthread_setname_np(id, name);
 449 #endif
 450 #elif defined RTE_EXEC_ENV_FREEBSD
 451         pthread_set_name_np(id, name);
 452 #endif
 453 }
 454
 455 static int
 456 telemetry_legacy_init(void)
 457 {
 458         pthread_t t_old;
 459         int rc;
 460
 461         if (num_legacy_callbacks == 1) {
 462                 TMTY_LOG(WARNING, "No legacy callbacks, legacy socket not created\n");
 463                 return -1;
 464         }
 465
 466         v1_socket.fn = legacy_client_handler;
 467         if ((size_t) snprintf(v1_socket.path, sizeof(v1_socket.path),
 468                         "%s/telemetry", socket_dir) >= sizeof(v1_socket.path)) {
 469                 TMTY_LOG(ERR, "Error with socket binding, path too long\n");
 470                 return -1;
 471         }
 472         v1_socket.sock = create_socket(v1_socket.path);
 473         if (v1_socket.sock < 0)
 474                 return -1;
 475         rc = pthread_create(&t_old, NULL, socket_listener, &v1_socket);
 476         if (rc != 0) {
 477                 TMTY_LOG(ERR, "Error with create legcay socket thread: %s\n",
 478                          strerror(rc));
 479                 close(v1_socket.sock);
 480                 v1_socket.sock = -1;
 481                 unlink(v1_socket.path);
 482                 v1_socket.path[0] = '\0';
 483                 return -1;
 484         }
 485         pthread_setaffinity_np(t_old, sizeof(*thread_cpuset), thread_cpuset);
 486         set_thread_name(t_old, "telemetry-v1");
 487         TMTY_LOG(DEBUG, "Legacy telemetry socket initialized ok\n");
 488         return 0;
 489 }
 490
 491 static int
 492 telemetry_v2_init(void)
 493 {
 494         pthread_t t_new;
 495         int rc;
 496
 497         v2_socket.num_clients = &v2_clients;
 498         rte_telemetry_register_cmd("/", list_commands,
 499                         "Returns list of available commands, Takes no parameters");
 500         rte_telemetry_register_cmd("/info", json_info,
 501                         "Returns DPDK Telemetry information. Takes no parameters");
 502         rte_telemetry_register_cmd("/help", command_help,
 503                         "Returns help text for a command. Parameters: string command");
 504         v2_socket.fn = client_handler;
 505         if (strlcpy(v2_socket.path, get_socket_path(socket_dir, 2),
 506                         sizeof(v2_socket.path)) >= sizeof(v2_socket.path)) {
 507                 TMTY_LOG(ERR, "Error with socket binding, path too long\n");
 508                 return -1;
 509         }
 510
 511         v2_socket.sock = create_socket(v2_socket.path);
 512         if (v2_socket.sock < 0)
 513                 return -1;
 514         rc = pthread_create(&t_new, NULL, socket_listener, &v2_socket);
 515         if (rc != 0) {
 516                 TMTY_LOG(ERR, "Error with create socket thread: %s\n",
 517                          strerror(rc));
 518                 close(v2_socket.sock);
 519                 v2_socket.sock = -1;
 520                 unlink(v2_socket.path);
 521                 v2_socket.path[0] = '\0';
 522                 return -1;
 523         }
 524         pthread_setaffinity_np(t_new, sizeof(*thread_cpuset), thread_cpuset);
 525         set_thread_name(t_new, "telemetry-v2");
 526         atexit(unlink_sockets);
 527
 528         return 0;
 529 }
 530
 531 #endif /* !RTE_EXEC_ENV_WINDOWS */
 532
 533 int32_t
 534 rte_telemetry_init(const char *runtime_dir, const char *rte_version, rte_cpuset_t *cpuset,
 535                 rte_log_fn log_fn, uint32_t registered_logtype)
 536 {
 537         telemetry_version = rte_version;
 538         socket_dir = runtime_dir;
 539         thread_cpuset = cpuset;
 540         rte_log_ptr = log_fn;
 541         logtype = registered_logtype;
 542
 543 #ifndef RTE_EXEC_ENV_WINDOWS
 544         if (telemetry_v2_init() != 0)
 545                 return -1;
 546         TMTY_LOG(DEBUG, "Telemetry initialized ok\n");
 547         telemetry_legacy_init();
 548 #endif /* RTE_EXEC_ENV_WINDOWS */
 549
 550         return 0;
 551 }