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