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