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