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