<commit_before>#include <stddef.h>
#define _GNU_SOURCE
#include <sys/uio.h>
#include <sys/socket.h>
#include "brubeck.h"

#ifdef __GLIBC__
#	if ((__GLIBC__ > 2) || ((__GLIBC__ == 2) && (__GLIBC_MINOR__ >= 12)))
#		define HAVE_RECVMMSG 1
#	endif
#endif

#define MAX_PACKET_SIZE 512

#ifdef HAVE_RECVMMSG
static void statsd_run_recvmmsg(struct brubeck_statsd *statsd, int sock)
{
	const unsigned int SIM_PACKETS = statsd->mmsg_count;
	struct brubeck_server *server = statsd->sampler.server;

	struct brubeck_statsd_msg msg;
	struct brubeck_metric *metric;
	unsigned int i;

	struct iovec iovecs[SIM_PACKETS];
	struct mmsghdr msgs[SIM_PACKETS];

	memset(msgs, 0x0, sizeof(msgs));

	for (i = 0; i < SIM_PACKETS; ++i) {
		iovecs[i].iov_base = xmalloc(MAX_PACKET_SIZE);
		iovecs[i].iov_len = MAX_PACKET_SIZE;
		msgs[i].msg_hdr.msg_iov = &iovecs[i];
		msgs[i].msg_hdr.msg_iovlen = 1;
	}

	log_splunk("sampler=statsd event=worker_online syscall=recvmmsg socket=%d", sock);

	for (;;) {
		int res = recvmmsg(sock, msgs, SIM_PACKETS, 0, NULL);

		if (res < 0) {
			if (errno == EAGAIN || errno == EINTR)
				continue;

			log_splunk_errno("sampler=statsd event=failed_read");
			brubeck_server_mark_dropped(server);
			continue;
		}

		/* store stats */
		brubeck_atomic_add(&server->stats.metrics, SIM_PACKETS);
		brubeck_atomic_add(&statsd->sampler.inflow, SIM_PACKETS);

		for (i = 0; i < SIM_PACKETS; ++i) {
			char *buf = msgs[i].msg_hdr.msg_iov->iov_base;
			int len = msgs[i].msg_len;

			if (brubeck_statsd_msg_parse(&msg, buf, len) < 0) {
				brubeck_server_mark_dropped(server);
				log_splunk("sampler=statsd event=packet_drop");<commit_msg>log the actual bad key in case of an error in statsd_run_recvmmsg<commit_after>#include <stddef.h>
#define _GNU_SOURCE
#include <sys/uio.h>
#include <sys/socket.h>
#include "brubeck.h"

#ifdef __GLIBC__
#	if ((__GLIBC__ > 2) || ((__GLIBC__ == 2) && (__GLIBC_MINOR__ >= 12)))
#		define HAVE_RECVMMSG 1
#	endif
#endif

#define MAX_PACKET_SIZE 512

#ifdef HAVE_RECVMMSG
static void statsd_run_recvmmsg(struct brubeck_statsd *statsd, int sock)
{
	const unsigned int SIM_PACKETS = statsd->mmsg_count;
	struct brubeck_server *server = statsd->sampler.server;

	struct brubeck_statsd_msg msg;
	struct brubeck_metric *metric;
	unsigned int i;

	struct iovec iovecs[SIM_PACKETS];
	struct mmsghdr msgs[SIM_PACKETS];

	memset(msgs, 0x0, sizeof(msgs));

	for (i = 0; i < SIM_PACKETS; ++i) {
		iovecs[i].iov_base = xmalloc(MAX_PACKET_SIZE);
		iovecs[i].iov_len = MAX_PACKET_SIZE;
		msgs[i].msg_hdr.msg_iov = &iovecs[i];
		msgs[i].msg_hdr.msg_iovlen = 1;
	}

	log_splunk("sampler=statsd event=worker_online syscall=recvmmsg socket=%d", sock);

	for (;;) {
		int res = recvmmsg(sock, msgs, SIM_PACKETS, 0, NULL);

		if (res < 0) {
			if (errno == EAGAIN || errno == EINTR)
				continue;

			log_splunk_errno("sampler=statsd event=failed_read");
			brubeck_server_mark_dropped(server);
			continue;
		}

		/* store stats */
		brubeck_atomic_add(&server->stats.metrics, SIM_PACKETS);
		brubeck_atomic_add(&statsd->sampler.inflow, SIM_PACKETS);

		for (i = 0; i < SIM_PACKETS; ++i) {
			char *buf = msgs[i].msg_hdr.msg_iov->iov_base;
			int len = msgs[i].msg_len;

			if (brubeck_statsd_msg_parse(&msg, buf, len) < 0) {
				if (msg.key_len > 0)
					buf[msg.key_len] = ':';

				log_splunk("sampler=statsd event=bad_key key='%.*s'", len, buf);

				brubeck_server_mark_dropped(server);
{'fraction_non_alphanumeric': 0.13963039014373715, 'fraction_numerical': 0.009680258140217073, 'mean_word_length': 4.879310344827586, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 24, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7448515', 'n_tokens_mistral': 1572, 'n_tokens_neox': 1415, 'n_words': 275}