src/dpdkstat.c

   1 /*-
   2  * collectd - src/dpdkstat.c
   3  * MIT License
   4  *
   5  * Copyright(c) 2016 Intel Corporation. All rights reserved.
   6  *
   7  * Permission is hereby granted, free of charge, to any person obtaining a copy
   8  * of this software and associated documentation files (the "Software"), to deal
   9  * in the Software without restriction, including without limitation the rights
  10  * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  11  * copies of the Software, and to permit persons to whom the Software is
  12  * furnished to do so, subject to the following conditions:
  13  *
  14  * The above copyright notice and this permission notice shall be included in
  15  * all copies or substantial portions of the Software.
  16  *
  17  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  18  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  19  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  20  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  21  * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  22  * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  23  * SOFTWARE.
  24  *
  25  * Authors:
  26  *   Maryam Tahhan <maryam.tahhan@intel.com>
  27  *   Harry van Haaren <harry.van.haaren@intel.com>
  28  *   Taras Chornyi <tarasx.chornyi@intel.com>
  29  */
  30
  31 #include "collectd.h"
  32
  33 #include "common.h" /* auxiliary functions */
  34 #include "plugin.h" /* plugin_register_*, plugin_dispatch_values */
  35 #include "utils_time.h"
  36
  37 #include <getopt.h>
  38 #include <poll.h>
  39 #include <semaphore.h>
  40 #include <sys/mman.h>
  41 #include <sys/queue.h>
  42
  43 #include <rte_atomic.h>
  44 #include <rte_branch_prediction.h>
  45 #include <rte_common.h>
  46 #include <rte_config.h>
  47 #include <rte_debug.h>
  48 #include <rte_debug.h>
  49 #include <rte_eal.h>
  50 #include <rte_ethdev.h>
  51 #include <rte_launch.h>
  52 #include <rte_lcore.h>
  53 #include <rte_log.h>
  54 #include <rte_malloc.h>
  55 #include <rte_memory.h>
  56 #include <rte_memzone.h>
  57 #include <rte_per_lcore.h>
  58 #include <rte_string_fns.h>
  59 #include <rte_tailq.h>
  60 #include <rte_version.h>
  61
  62 #define DPDK_DEFAULT_RTE_CONFIG "/var/run/.rte_config"
  63 #define DPDK_MAX_ARGC 8
  64 #define DPDKSTAT_MAX_BUFFER_SIZE (4096 * 4)
  65 #define DPDK_SHM_NAME "dpdk_collectd_stats_shm"
  66 #define ERR_BUF_SIZE 1024
  67 #define REINIT_SHM 1
  68 #define RESET 1
  69 #define NO_RESET 0
  70
  71 #define RTE_VERSION_16_07 RTE_VERSION_NUM(16, 7, 0, 16)
  72
  73 #if RTE_VERSION < RTE_VERSION_16_07
  74 #define DPDK_STATS_XSTAT_GET_VALUE(ctx, index) ctx->xstats[index].value
  75 #define DPDK_STATS_XSTAT_GET_NAME(ctx, index) ctx->xstats[index].name
  76 #define DPDK_STATS_CTX_GET_XSTAT_SIZE sizeof(struct rte_eth_xstats)
  77 #define DPDK_STATS_CTX_INIT(ctx)                                               \
  78   do {                                                                         \
  79     ctx->xstats = (struct rte_eth_xstats *)&ctx->raw_data[0];                  \
  80   } while (0)
  81 #else
  82 #define DPDK_STATS_XSTAT_GET_VALUE(ctx, index) ctx->xstats[index].value
  83 #define DPDK_STATS_XSTAT_GET_NAME(ctx, index) ctx->xnames[index].name
  84 #define DPDK_STATS_CTX_GET_XSTAT_SIZE                                          \
  85   (sizeof(struct rte_eth_xstat) + sizeof(struct rte_eth_xstat_name))
  86 #define DPDK_STATS_CTX_INIT(ctx)                                               \
  87   do {                                                                         \
  88     ctx->xstats = (struct rte_eth_xstat *)&ctx->raw_data[0];                   \
  89     ctx->xnames =                                                              \
  90         (struct rte_eth_xstat_name *)&ctx                                      \
  91             ->raw_data[ctx->num_xstats * sizeof(struct rte_eth_xstat)];        \
  92   } while (0)
  93 #endif
  94
  95 enum DPDK_HELPER_ACTION {
  96   DPDK_HELPER_ACTION_COUNT_STATS,
  97   DPDK_HELPER_ACTION_SEND_STATS,
  98 };
  99
 100 enum DPDK_HELPER_STATUS {
 101   DPDK_HELPER_NOT_INITIALIZED = 0,
 102   DPDK_HELPER_WAITING_ON_PRIMARY,
 103   DPDK_HELPER_INITIALIZING_EAL,
 104   DPDK_HELPER_ALIVE_SENDING_STATS,
 105   DPDK_HELPER_GRACEFUL_QUIT,
 106 };
 107
 108 struct dpdk_config_s {
 109   /* General DPDK params */
 110   char coremask[DATA_MAX_NAME_LEN];
 111   char memory_channels[DATA_MAX_NAME_LEN];
 112   char socket_memory[DATA_MAX_NAME_LEN];
 113   char process_type[DATA_MAX_NAME_LEN];
 114   char file_prefix[DATA_MAX_NAME_LEN];
 115   cdtime_t interval;
 116   uint32_t eal_initialized;
 117   uint32_t enabled_port_mask;
 118   char port_name[RTE_MAX_ETHPORTS][DATA_MAX_NAME_LEN];
 119   uint32_t eal_argc;
 120   /* Helper info */
 121   int collectd_reinit_shm;
 122   pid_t helper_pid;
 123   sem_t sema_helper_get_stats;
 124   sem_t sema_stats_in_shm;
 125   int helper_pipes[2];
 126   enum DPDK_HELPER_STATUS helper_status;
 127   enum DPDK_HELPER_ACTION helper_action;
 128   /* xstats info */
 129   uint32_t num_ports;
 130   uint32_t num_xstats;
 131   cdtime_t port_read_time[RTE_MAX_ETHPORTS];
 132   uint32_t num_stats_in_port[RTE_MAX_ETHPORTS];
 133   struct rte_eth_link link_status[RTE_MAX_ETHPORTS];
 134 #if RTE_VERSION < RTE_VERSION_16_07
 135   struct rte_eth_xstats *xstats;
 136 #else
 137   struct rte_eth_xstat *xstats;
 138   struct rte_eth_xstat_name *xnames;
 139 #endif
 140   char *raw_data;
 141   /* rte_eth_xstats from here on until the end of the SHM */
 142 };
 143 typedef struct dpdk_config_s dpdk_config_t;
 144
 145 static int g_configured;
 146 static dpdk_config_t *g_configuration;
 147
 148 static void dpdk_config_init_default(void);
 149 static int dpdk_config(oconfig_item_t *ci);
 150 static int dpdk_helper_init_eal(void);
 151 static int dpdk_helper_run(void);
 152 static int dpdk_helper_spawn(enum DPDK_HELPER_ACTION action);
 153 static int dpdk_init(void);
 154 static int dpdk_read(user_data_t *ud);
 155 static int dpdk_shm_cleanup(void);
 156 static int dpdk_shm_init(size_t size);
 157
 158 /* Write the default configuration to the g_configuration instances */
 159 static void dpdk_config_init_default(void) {
 160   g_configuration->interval = plugin_get_interval();
 161   if (g_configuration->interval == cf_get_default_interval())
 162     WARNING("dpdkstat: No time interval was configured, default value %lu ms "
 163             "is set",
 164             CDTIME_T_TO_MS(g_configuration->interval));
 165   /* Default is all ports enabled */
 166   g_configuration->enabled_port_mask = ~0;
 167   g_configuration->eal_argc = DPDK_MAX_ARGC;
 168   g_configuration->eal_initialized = 0;
 169   ssnprintf(g_configuration->coremask, DATA_MAX_NAME_LEN, "%s", "0xf");
 170   ssnprintf(g_configuration->memory_channels, DATA_MAX_NAME_LEN, "%s", "1");
 171   ssnprintf(g_configuration->process_type, DATA_MAX_NAME_LEN, "%s",
 172             "secondary");
 173   ssnprintf(g_configuration->file_prefix, DATA_MAX_NAME_LEN, "%s",
 174             DPDK_DEFAULT_RTE_CONFIG);
 175
 176   for (int i = 0; i < RTE_MAX_ETHPORTS; i++)
 177     g_configuration->port_name[i][0] = 0;
 178 }
 179
 180 static int dpdk_config(oconfig_item_t *ci) {
 181   int port_counter = 0;
 182   /* Allocate g_configuration and
 183    * initialize a POSIX SHared Memory (SHM) object.
 184    */
 185   int err = dpdk_shm_init(sizeof(dpdk_config_t));
 186   if (err) {
 187     char errbuf[ERR_BUF_SIZE];
 188     ERROR("dpdkstat: error in shm_init, %s",
 189           sstrerror(errno, errbuf, sizeof(errbuf)));
 190     return -1;
 191   }
 192
 193   /* Set defaults for config, overwritten by loop if config item exists */
 194   dpdk_config_init_default();
 195
 196   for (int i = 0; i < ci->children_num; i++) {
 197     oconfig_item_t *child = ci->children + i;
 198
 199     if (strcasecmp("Coremask", child->key) == 0) {
 200       cf_util_get_string_buffer(child, g_configuration->coremask,
 201                                 sizeof(g_configuration->coremask));
 202       DEBUG("dpdkstat:COREMASK %s ", g_configuration->coremask);
 203     } else if (strcasecmp("MemoryChannels", child->key) == 0) {
 204       cf_util_get_string_buffer(child, g_configuration->memory_channels,
 205                                 sizeof(g_configuration->memory_channels));
 206       DEBUG("dpdkstat:Memory Channels %s ", g_configuration->memory_channels);
 207     } else if (strcasecmp("SocketMemory", child->key) == 0) {
 208       cf_util_get_string_buffer(child, g_configuration->socket_memory,
 209                                 sizeof(g_configuration->memory_channels));
 210       DEBUG("dpdkstat: socket mem %s ", g_configuration->socket_memory);
 211     } else if (strcasecmp("ProcessType", child->key) == 0) {
 212       cf_util_get_string_buffer(child, g_configuration->process_type,
 213                                 sizeof(g_configuration->process_type));
 214       DEBUG("dpdkstat: proc type %s ", g_configuration->process_type);
 215     } else if ((strcasecmp("FilePrefix", child->key) == 0) &&
 216                (child->values[0].type == OCONFIG_TYPE_STRING)) {
 217       ssnprintf(g_configuration->file_prefix, DATA_MAX_NAME_LEN,
 218                 "/var/run/.%s_config", child->values[0].value.string);
 219       DEBUG("dpdkstat: file prefix %s ", g_configuration->file_prefix);
 220     } else if ((strcasecmp("EnabledPortMask", child->key) == 0) &&
 221                (child->values[0].type == OCONFIG_TYPE_NUMBER)) {
 222       g_configuration->enabled_port_mask =
 223           (uint32_t)child->values[0].value.number;
 224       DEBUG("dpdkstat: Enabled Port Mask %u",
 225             g_configuration->enabled_port_mask);
 226     } else if (strcasecmp("PortName", child->key) == 0) {
 227       cf_util_get_string_buffer(
 228           child, g_configuration->port_name[port_counter],
 229           sizeof(g_configuration->port_name[port_counter]));
 230       DEBUG("dpdkstat: Port %d Name: %s ", port_counter,
 231             g_configuration->port_name[port_counter]);
 232       port_counter++;
 233     } else {
 234       WARNING("dpdkstat: The config option \"%s\" is unknown.", child->key);
 235     }
 236   }                 /* End for (int i = 0; i < ci->children_num; i++)*/
 237   g_configured = 1; /* Bypass configuration in dpdk_shm_init(). */
 238
 239   return 0;
 240 }
 241
 242 /*
 243  * Allocate g_configuration and initialize SHared Memory (SHM)
 244  * for config and helper process
 245  */
 246 static int dpdk_shm_init(size_t size) {
 247   /*
 248    * Check if SHM is already configured: when config items are provided, the
 249    * config function initializes SHM. If there is no config, then init() will
 250    * just return.
 251    */
 252   if (g_configuration)
 253     return 0;
 254
 255   char errbuf[ERR_BUF_SIZE];
 256
 257   /* Create and open a new object, or open an existing object. */
 258   int fd = shm_open(DPDK_SHM_NAME, O_CREAT | O_TRUNC | O_RDWR, 0666);
 259   if (fd < 0) {
 260     WARNING("dpdkstat:Failed to open %s as SHM:%s", DPDK_SHM_NAME,
 261             sstrerror(errno, errbuf, sizeof(errbuf)));
 262     goto fail;
 263   }
 264   /* Set the size of the shared memory object. */
 265   int ret = ftruncate(fd, size);
 266   if (ret != 0) {
 267     WARNING("dpdkstat:Failed to resize SHM:%s",
 268             sstrerror(errno, errbuf, sizeof(errbuf)));
 269     goto fail_close;
 270   }
 271   /* Map the shared memory object into this process' virtual address space. */
 272   g_configuration = mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
 273   if (g_configuration == MAP_FAILED) {
 274     WARNING("dpdkstat:Failed to mmap SHM:%s",
 275             sstrerror(errno, errbuf, sizeof(errbuf)));
 276     goto fail_close;
 277   }
 278   /*
 279    * Close the file descriptor, the shared memory object still exists
 280    * and can only be removed by calling shm_unlink().
 281    */
 282   close(fd);
 283
 284   /* Initialize g_configuration. */
 285   memset(g_configuration, 0, size);
 286
 287   /* Initialize the semaphores for SHM use */
 288   int err = sem_init(&g_configuration->sema_helper_get_stats, 1, 0);
 289   if (err) {
 290     ERROR("dpdkstat semaphore init failed: %s",
 291           sstrerror(errno, errbuf, sizeof(errbuf)));
 292     goto fail_close;
 293   }
 294   err = sem_init(&g_configuration->sema_stats_in_shm, 1, 0);
 295   if (err) {
 296     ERROR("dpdkstat semaphore init failed: %s",
 297           sstrerror(errno, errbuf, sizeof(errbuf)));
 298     goto fail_close;
 299   }
 300
 301   g_configuration->xstats = NULL;
 302
 303   return 0;
 304
 305 fail_close:
 306   close(fd);
 307 fail:
 308   /* Reset to zero, as it was set to MAP_FAILED aka: (void *)-1. Avoid
 309    * an issue if collectd attempts to run this plugin failure.
 310    */
 311   g_configuration = 0;
 312   return -1;
 313 }
 314
 315 static int dpdk_re_init_shm() {
 316   dpdk_config_t temp_config;
 317   memcpy(&temp_config, g_configuration, sizeof(dpdk_config_t));
 318   DEBUG("dpdkstat: %s: ports %" PRIu32 ", xstats %" PRIu32, __func__,
 319         temp_config.num_ports, temp_config.num_xstats);
 320
 321   size_t shm_xstats_size =
 322       sizeof(dpdk_config_t) +
 323       (DPDK_STATS_CTX_GET_XSTAT_SIZE * g_configuration->num_xstats);
 324   DEBUG("=== SHM new size for %" PRIu32 " xstats", g_configuration->num_xstats);
 325
 326   int err = dpdk_shm_cleanup();
 327   if (err) {
 328     ERROR("dpdkstat: Error in shm_cleanup in %s", __func__);
 329     return err;
 330   }
 331   err = dpdk_shm_init(shm_xstats_size);
 332   if (err) {
 333     WARNING("dpdkstat: Error in shm_init in %s", __func__);
 334     return err;
 335   }
 336   /* If the XML config() function has been run, don't re-initialize defaults */
 337   if (!g_configured)
 338     dpdk_config_init_default();
 339
 340   memcpy(g_configuration, &temp_config, sizeof(dpdk_config_t));
 341   g_configuration->collectd_reinit_shm = 0;
 342   g_configuration->raw_data = (char *)(g_configuration + 1);
 343   DPDK_STATS_CTX_INIT(g_configuration);
 344   return 0;
 345 }
 346
 347 static int dpdk_init(void) {
 348   int err = dpdk_shm_init(sizeof(dpdk_config_t));
 349   if (err) {
 350     ERROR("dpdkstat: %s : error %d in shm_init()", __func__, err);
 351     return err;
 352   }
 353
 354   /* If the XML config() function has been run, dont re-initialize defaults */
 355   if (!g_configured) {
 356     dpdk_config_init_default();
 357   }
 358
 359   return 0;
 360 }
 361
 362 static int dpdk_helper_stop(int reset) {
 363   g_configuration->helper_status = DPDK_HELPER_GRACEFUL_QUIT;
 364   if (reset) {
 365     g_configuration->eal_initialized = 0;
 366     g_configuration->num_ports = 0;
 367     g_configuration->xstats = NULL;
 368     g_configuration->num_xstats = 0;
 369     for (int i = 0; i < RTE_MAX_ETHPORTS; i++)
 370       g_configuration->num_stats_in_port[i] = 0;
 371   }
 372   close(g_configuration->helper_pipes[1]);
 373   int err = kill(g_configuration->helper_pid, SIGKILL);
 374   if (err) {
 375     char errbuf[ERR_BUF_SIZE];
 376     WARNING("dpdkstat: error sending kill to helper: %s",
 377             sstrerror(errno, errbuf, sizeof(errbuf)));
 378   }
 379
 380   return 0;
 381 }
 382
 383 static int dpdk_helper_spawn(enum DPDK_HELPER_ACTION action) {
 384   char errbuf[ERR_BUF_SIZE];
 385   g_configuration->eal_initialized = 0;
 386   g_configuration->helper_action = action;
 387   /*
 388    * Create a pipe for helper stdout back to collectd. This is necessary for
 389    * logging EAL failures, as rte_eal_init() calls rte_panic().
 390    */
 391   if (pipe(g_configuration->helper_pipes) != 0) {
 392     DEBUG("dpdkstat: Could not create helper pipe: %s",
 393           sstrerror(errno, errbuf, sizeof(errbuf)));
 394     return -1;
 395   }
 396
 397   int pipe0_flags = fcntl(g_configuration->helper_pipes[0], F_GETFL, 0);
 398   int pipe1_flags = fcntl(g_configuration->helper_pipes[1], F_GETFL, 0);
 399   if (pipe0_flags == -1 || pipe1_flags == -1) {
 400     WARNING("dpdkstat: Failed setting up pipe flags: %s",
 401             sstrerror(errno, errbuf, sizeof(errbuf)));
 402   }
 403   int pipe0_err = fcntl(g_configuration->helper_pipes[0], F_SETFL,
 404                         pipe1_flags | O_NONBLOCK);
 405   int pipe1_err = fcntl(g_configuration->helper_pipes[1], F_SETFL,
 406                         pipe0_flags | O_NONBLOCK);
 407   if (pipe0_err == -1 || pipe1_err == -1) {
 408     WARNING("dpdkstat: Failed setting up pipes: %s",
 409             sstrerror(errno, errbuf, sizeof(errbuf)));
 410   }
 411
 412   pid_t pid = fork();
 413   if (pid > 0) {
 414     close(g_configuration->helper_pipes[1]);
 415     g_configuration->helper_pid = pid;
 416     DEBUG("dpdkstat: helper pid %lu", (long)g_configuration->helper_pid);
 417     /* Kick helper once its alive to have it start processing */
 418     sem_post(&g_configuration->sema_helper_get_stats);
 419   } else if (pid == 0) {
 420     /* Replace stdout with a pipe to collectd. */
 421     close(g_configuration->helper_pipes[0]);
 422     close(STDOUT_FILENO);
 423     dup2(g_configuration->helper_pipes[1], STDOUT_FILENO);
 424     dpdk_helper_run();
 425     exit(0);
 426   } else {
 427     ERROR("dpdkstat: Failed to fork helper process: %s",
 428           sstrerror(errno, errbuf, sizeof(errbuf)));
 429     return -1;
 430   }
 431   return 0;
 432 }
 433
 434 /*
 435  * Initialize the DPDK EAL, if this returns, EAL is successfully initialized.
 436  * On failure, the EAL prints an error message, and the helper process exits.
 437  */
 438 static int dpdk_helper_init_eal(void) {
 439   g_configuration->helper_status = DPDK_HELPER_INITIALIZING_EAL;
 440   char *argp[(g_configuration->eal_argc) + 1];
 441   int i = 0;
 442
 443   argp[i++] = "collectd-dpdk";
 444   if (strcasecmp(g_configuration->coremask, "") != 0) {
 445     argp[i++] = "-c";
 446     argp[i++] = g_configuration->coremask;
 447   }
 448   if (strcasecmp(g_configuration->memory_channels, "") != 0) {
 449     argp[i++] = "-n";
 450     argp[i++] = g_configuration->memory_channels;
 451   }
 452   if (strcasecmp(g_configuration->socket_memory, "") != 0) {
 453     argp[i++] = "--socket-mem";
 454     argp[i++] = g_configuration->socket_memory;
 455   }
 456   if (strcasecmp(g_configuration->file_prefix, "") != 0 &&
 457       strcasecmp(g_configuration->file_prefix, DPDK_DEFAULT_RTE_CONFIG) != 0) {
 458     argp[i++] = "--file-prefix";
 459     argp[i++] = g_configuration->file_prefix;
 460   }
 461   if (strcasecmp(g_configuration->process_type, "") != 0) {
 462     argp[i++] = "--proc-type";
 463     argp[i++] = g_configuration->process_type;
 464   }
 465   g_configuration->eal_argc = i;
 466
 467   g_configuration->eal_initialized = 1;
 468   int ret = rte_eal_init(g_configuration->eal_argc, argp);
 469   if (ret < 0) {
 470     g_configuration->eal_initialized = 0;
 471     return ret;
 472   }
 473   return 0;
 474 }
 475
 476 static int dpdk_helper_run(void) {
 477   char errbuf[ERR_BUF_SIZE];
 478   pid_t ppid = getppid();
 479   g_configuration->helper_status = DPDK_HELPER_WAITING_ON_PRIMARY;
 480
 481   while (1) {
 482     /* sem_timedwait() to avoid blocking forever */
 483     cdtime_t now = cdtime();
 484     cdtime_t safety_period = MS_TO_CDTIME_T(1500);
 485     int ret =
 486         sem_timedwait(&g_configuration->sema_helper_get_stats,
 487                       &CDTIME_T_TO_TIMESPEC(now + safety_period +
 488                                             g_configuration->interval * 2));
 489
 490     if (ret == -1 && errno == ETIMEDOUT) {
 491       ERROR("dpdkstat-helper: sem timedwait()"
 492             " timeout, did collectd terminate?");
 493       dpdk_helper_stop(RESET);
 494     }
 495     /* Parent PID change means collectd died so quit the helper process. */
 496     if (ppid != getppid()) {
 497       WARNING("dpdkstat-helper: parent PID changed, quitting.");
 498       dpdk_helper_stop(RESET);
 499     }
 500
 501     /* Checking for DPDK primary process. */
 502     if (!rte_eal_primary_proc_alive(g_configuration->file_prefix)) {
 503       if (g_configuration->eal_initialized) {
 504         WARNING("dpdkstat-helper: no primary alive but EAL initialized:"
 505                 " quitting.");
 506         dpdk_helper_stop(RESET);
 507       }
 508       g_configuration->helper_status = DPDK_HELPER_WAITING_ON_PRIMARY;
 509       /* Back to start of while() - waiting for primary process */
 510       continue;
 511     }
 512
 513     if (!g_configuration->eal_initialized) {
 514       /* Initialize EAL. */
 515       int ret = dpdk_helper_init_eal();
 516       if (ret != 0) {
 517         WARNING("ERROR INITIALIZING EAL");
 518         dpdk_helper_stop(RESET);
 519       }
 520     }
 521
 522     g_configuration->helper_status = DPDK_HELPER_ALIVE_SENDING_STATS;
 523
 524     uint8_t nb_ports = rte_eth_dev_count();
 525     if (nb_ports == 0) {
 526       DEBUG("dpdkstat-helper: No DPDK ports available. "
 527             "Check bound devices to DPDK driver.");
 528       dpdk_helper_stop(RESET);
 529     }
 530
 531     if (nb_ports > RTE_MAX_ETHPORTS)
 532       nb_ports = RTE_MAX_ETHPORTS;
 533
 534     int len = 0, enabled_port_count = 0, num_xstats = 0;
 535     for (uint8_t i = 0; i < nb_ports; i++) {
 536       if (!(g_configuration->enabled_port_mask & (1 << i)))
 537         continue;
 538
 539       if (g_configuration->helper_action == DPDK_HELPER_ACTION_COUNT_STATS) {
 540 #if RTE_VERSION >= RTE_VERSION_16_07
 541         len = rte_eth_xstats_get_names(i, NULL, 0);
 542 #else
 543         len = rte_eth_xstats_get(i, NULL, 0);
 544 #endif
 545         if (len < 0) {
 546           ERROR("dpdkstat-helper: Cannot get xstats count on port %" PRIu8, i);
 547           break;
 548         }
 549         num_xstats += len;
 550         g_configuration->num_stats_in_port[enabled_port_count] = len;
 551         enabled_port_count++;
 552         continue;
 553       } else {
 554         len = g_configuration->num_stats_in_port[enabled_port_count];
 555         g_configuration->port_read_time[enabled_port_count] = cdtime();
 556         ret = rte_eth_xstats_get(
 557             i, g_configuration->xstats + num_xstats,
 558             g_configuration->num_stats_in_port[enabled_port_count]);
 559         if (ret < 0 || ret != len) {
 560           DEBUG("dpdkstat-helper: Error reading xstats on port %" PRIu8
 561                 " len = %d",
 562                 i, len);
 563           break;
 564         }
 565 #if RTE_VERSION >= RTE_VERSION_16_07
 566         ret = rte_eth_xstats_get_names(i, g_configuration->xnames + num_xstats,
 567                                        len);
 568         if (ret < 0 || ret != len) {
 569           ERROR("dpdkstat-helper: Error reading xstat names (port=%d; len=%d)",
 570                 i, len);
 571           break;
 572         }
 573 #endif
 574         num_xstats += g_configuration->num_stats_in_port[enabled_port_count];
 575         enabled_port_count++;
 576       }
 577     } /* for (nb_ports) */
 578
 579     if (g_configuration->helper_action == DPDK_HELPER_ACTION_COUNT_STATS) {
 580       g_configuration->num_ports = enabled_port_count;
 581       g_configuration->num_xstats = num_xstats;
 582       DEBUG("dpdkstat-helper ports: %" PRIu32 ", num stats: %" PRIu32,
 583             g_configuration->num_ports, g_configuration->num_xstats);
 584       /* Exit, allowing collectd to re-init SHM to the right size */
 585       g_configuration->collectd_reinit_shm = REINIT_SHM;
 586       dpdk_helper_stop(NO_RESET);
 587     }
 588     /* Now kick collectd send thread to send the stats */
 589     int err = sem_post(&g_configuration->sema_stats_in_shm);
 590     if (err) {
 591       WARNING("dpdkstat: error posting semaphore to helper %s",
 592               sstrerror(errno, errbuf, sizeof(errbuf)));
 593       dpdk_helper_stop(RESET);
 594     }
 595   } /* while(1) */
 596
 597   return 0;
 598 }
 599
 600 static void dpdk_submit_xstats(const char *dev_name, int count,
 601                                uint32_t counters, cdtime_t port_read_time) {
 602   for (uint32_t j = 0; j < counters; j++) {
 603     value_list_t vl = VALUE_LIST_INIT;
 604     char *counter_name;
 605     char *type_end;
 606
 607     vl.values = &(value_t){.derive = (derive_t)DPDK_STATS_XSTAT_GET_VALUE(
 608                                g_configuration, count + j)};
 609     vl.values_len = 1; /* Submit stats one at a time */
 610     vl.time = port_read_time;
 611     sstrncpy(vl.plugin, "dpdkstat", sizeof(vl.plugin));
 612     sstrncpy(vl.plugin_instance, dev_name, sizeof(vl.plugin_instance));
 613     counter_name = DPDK_STATS_XSTAT_GET_NAME(g_configuration, count + j);
 614     if (counter_name == NULL) {
 615       WARNING("dpdkstat: Failed to get counter name.");
 616       return;
 617     }
 618
 619     type_end = strrchr(counter_name, '_');
 620
 621     if ((type_end != NULL) &&
 622         (strncmp(counter_name, "rx_", strlen("rx_")) == 0)) {
 623       if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
 624         sstrncpy(vl.type, "if_rx_errors", sizeof(vl.type));
 625       } else if (strncmp(type_end, "_dropped", strlen("_dropped")) == 0) {
 626         sstrncpy(vl.type, "if_rx_dropped", sizeof(vl.type));
 627       } else if (strncmp(type_end, "_bytes", strlen("_bytes")) == 0) {
 628         sstrncpy(vl.type, "if_rx_octets", sizeof(vl.type));
 629       } else if (strncmp(type_end, "_packets", strlen("_packets")) == 0) {
 630         sstrncpy(vl.type, "if_rx_packets", sizeof(vl.type));
 631       } else if (strncmp(type_end, "_placement", strlen("_placement")) == 0) {
 632         sstrncpy(vl.type, "if_rx_errors", sizeof(vl.type));
 633       } else if (strncmp(type_end, "_buff", strlen("_buff")) == 0) {
 634         sstrncpy(vl.type, "if_rx_errors", sizeof(vl.type));
 635       } else {
 636         /* Does not fit obvious type: use a more generic one */
 637         sstrncpy(vl.type, "derive", sizeof(vl.type));
 638       }
 639
 640     } else if ((type_end != NULL) &&
 641                (strncmp(counter_name, "tx_", strlen("tx_"))) == 0) {
 642       if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
 643         sstrncpy(vl.type, "if_tx_errors", sizeof(vl.type));
 644       } else if (strncmp(type_end, "_dropped", strlen("_dropped")) == 0) {
 645         sstrncpy(vl.type, "if_tx_dropped", sizeof(vl.type));
 646       } else if (strncmp(type_end, "_bytes", strlen("_bytes")) == 0) {
 647         sstrncpy(vl.type, "if_tx_octets", sizeof(vl.type));
 648       } else if (strncmp(type_end, "_packets", strlen("_packets")) == 0) {
 649         sstrncpy(vl.type, "if_tx_packets", sizeof(vl.type));
 650       } else {
 651         /* Does not fit obvious type: use a more generic one */
 652         sstrncpy(vl.type, "derive", sizeof(vl.type));
 653       }
 654     } else if ((type_end != NULL) &&
 655                (strncmp(counter_name, "flow_", strlen("flow_"))) == 0) {
 656
 657       if (strncmp(type_end, "_filters", strlen("_filters")) == 0) {
 658         sstrncpy(vl.type, "operations", sizeof(vl.type));
 659       } else if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
 660         sstrncpy(vl.type, "errors", sizeof(vl.type));
 661       } else if (strncmp(type_end, "_filters", strlen("_filters")) == 0) {
 662         sstrncpy(vl.type, "filter_result", sizeof(vl.type));
 663       }
 664     } else if ((type_end != NULL) &&
 665                (strncmp(counter_name, "mac_", strlen("mac_"))) == 0) {
 666       if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
 667         sstrncpy(vl.type, "errors", sizeof(vl.type));
 668       }
 669     } else {
 670       /* Does not fit obvious type, or strrchr error:
 671        *   use a more generic type */
 672       sstrncpy(vl.type, "derive", sizeof(vl.type));
 673     }
 674
 675     sstrncpy(vl.type_instance, counter_name, sizeof(vl.type_instance));
 676     plugin_dispatch_values(&vl);
 677   }
 678 }
 679
 680 static int dpdk_read(user_data_t *ud) {
 681   int ret = 0;
 682
 683   /*
 684    * Check if SHM flag is set to be re-initialized. AKA DPDK ports have been
 685    * counted, so re-init SHM to be large enough to fit all the statistics.
 686    */
 687   if (g_configuration->collectd_reinit_shm) {
 688     DEBUG("dpdkstat: read() now reinit SHM then launching send-thread");
 689     dpdk_re_init_shm();
 690   }
 691
 692   /*
 693    * Check if DPDK proc is alive, and has already counted port / stats. This
 694    * must be done in dpdk_read(), because the DPDK primary process may not be
 695    * alive at dpdk_init() time.
 696    */
 697   if (g_configuration->helper_status == DPDK_HELPER_NOT_INITIALIZED ||
 698       g_configuration->helper_status == DPDK_HELPER_GRACEFUL_QUIT) {
 699     int action = DPDK_HELPER_ACTION_SEND_STATS;
 700     if (g_configuration->num_xstats == 0)
 701       action = DPDK_HELPER_ACTION_COUNT_STATS;
 702     /* Spawn the helper thread to count stats or to read stats. */
 703     int err = dpdk_helper_spawn(action);
 704     if (err) {
 705       char errbuf[ERR_BUF_SIZE];
 706       ERROR("dpdkstat: error spawning helper %s",
 707             sstrerror(errno, errbuf, sizeof(errbuf)));
 708       return -1;
 709     }
 710   }
 711
 712   pid_t ws = waitpid(g_configuration->helper_pid, NULL, WNOHANG);
 713   /*
 714    * Conditions under which to respawn helper:
 715    *  waitpid() fails, helper process died (or quit), so respawn
 716    */
 717   _Bool respawn_helper = 0;
 718   if (ws != 0) {
 719     respawn_helper = 1;
 720   }
 721
 722   char buf[DPDKSTAT_MAX_BUFFER_SIZE];
 723   char out[DPDKSTAT_MAX_BUFFER_SIZE];
 724
 725   /* non blocking check on helper logging pipe */
 726   struct pollfd fds = {
 727       .fd = g_configuration->helper_pipes[0], .events = POLLIN,
 728   };
 729   int data_avail = poll(&fds, 1, 0);
 730   if (data_avail < 0) {
 731     char errbuf[ERR_BUF_SIZE];
 732     if (errno != EINTR || errno != EAGAIN)
 733       ERROR("dpdkstats: poll(2) failed: %s",
 734             sstrerror(errno, errbuf, sizeof(errbuf)));
 735   }
 736   while (data_avail) {
 737     int nbytes = read(g_configuration->helper_pipes[0], buf, sizeof(buf));
 738     if (nbytes <= 0)
 739       break;
 740     ssnprintf(out, nbytes, "%s", buf);
 741     DEBUG("dpdkstat: helper-proc: %s", out);
 742   }
 743
 744   if (respawn_helper) {
 745     if (g_configuration->helper_pid)
 746       dpdk_helper_stop(RESET);
 747     dpdk_helper_spawn(DPDK_HELPER_ACTION_COUNT_STATS);
 748   }
 749
 750   /* Kick helper process through SHM */
 751   sem_post(&g_configuration->sema_helper_get_stats);
 752
 753   cdtime_t now = cdtime();
 754   ret = sem_timedwait(&g_configuration->sema_stats_in_shm,
 755                       &CDTIME_T_TO_TIMESPEC(now + g_configuration->interval));
 756   if (ret == -1) {
 757     if (errno == ETIMEDOUT)
 758       DEBUG(
 759           "dpdkstat: timeout in collectd thread: is a DPDK Primary running? ");
 760     return 0;
 761   }
 762
 763   /* Dispatch the stats.*/
 764   uint32_t count = 0, port_num = 0;
 765
 766   for (uint32_t i = 0; i < g_configuration->num_ports; i++) {
 767     char dev_name[64];
 768     cdtime_t port_read_time = g_configuration->port_read_time[i];
 769     uint32_t counters_num = g_configuration->num_stats_in_port[i];
 770     size_t ports_max = CHAR_BIT * sizeof(g_configuration->enabled_port_mask);
 771     for (size_t j = port_num; j < ports_max; j++) {
 772       if ((g_configuration->enabled_port_mask & (1 << j)) != 0)
 773         break;
 774       port_num++;
 775     }
 776
 777     if (g_configuration->port_name[i][0] != 0)
 778       ssnprintf(dev_name, sizeof(dev_name), "%s",
 779                 g_configuration->port_name[i]);
 780     else
 781       ssnprintf(dev_name, sizeof(dev_name), "port.%" PRIu32, port_num);
 782     dpdk_submit_xstats(dev_name, count, counters_num, port_read_time);
 783     count += counters_num;
 784     port_num++;
 785   } /* for each port */
 786   return 0;
 787 }
 788
 789 static int dpdk_shm_cleanup(void) {
 790   int ret = munmap(g_configuration, sizeof(dpdk_config_t));
 791   g_configuration = 0;
 792   if (ret) {
 793     ERROR("dpdkstat: munmap returned %d", ret);
 794     return ret;
 795   }
 796   ret = shm_unlink(DPDK_SHM_NAME);
 797   if (ret) {
 798     ERROR("dpdkstat: shm_unlink returned %d", ret);
 799     return ret;
 800   }
 801   return 0;
 802 }
 803
 804 static int dpdk_shutdown(void) {
 805   int ret = 0;
 806   char errbuf[ERR_BUF_SIZE];
 807   close(g_configuration->helper_pipes[1]);
 808   int err = kill(g_configuration->helper_pid, SIGKILL);
 809   if (err) {
 810     ERROR("dpdkstat: error sending sigkill to helper %s",
 811           sstrerror(errno, errbuf, sizeof(errbuf)));
 812     ret = -1;
 813   }
 814   err = dpdk_shm_cleanup();
 815   if (err) {
 816     ERROR("dpdkstat: error cleaning up SHM: %s",
 817           sstrerror(errno, errbuf, sizeof(errbuf)));
 818     ret = -1;
 819   }
 820
 821   return ret;
 822 }
 823
 824 void module_register(void) {
 825   plugin_register_complex_config("dpdkstat", dpdk_config);
 826   plugin_register_init("dpdkstat", dpdk_init);
 827   plugin_register_complex_read(NULL, "dpdkstat", dpdk_read, 0, NULL);
 828   plugin_register_shutdown("dpdkstat", dpdk_shutdown);
 829 }