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