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