<commit_before>/*****************************************************************************
 *
 * RIEMANN-SEND-TCP.C
 *
 *    $ indent -br -nut -l125 riemann-send-tcp.c
 *
 *    $ gcc -o riemann-send-tcp riemann-send-tcp.c riemann.pb-c.c $(apr-1-config --cflags --cppflags --includes --link-ld) -lprotobuf-c
 *
 *****************************************************************************/

#include <stdio.h>
#include <stdlib.h>
#include "riemann.pb-c.h"

#include <string.h>
#include <errno.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netdb.h>

#include <apr_general.h>
#include <apr_thread_proc.h>

int done = 0;

#define RIEMANN_CB_OPEN 0
#define RIEMANN_CB_HALF_OPEN 1
#define RIEMANN_CB_CLOSED 2

#define RIEMANN_TIMEOUT 60
#define RIEMANN_MAX_FAILURES 5

char *riemann_server = "127.0.0.1";
char *riemann_protocol = "tcp";
int riemann_port = 5555;
int riemann_circuit_breaker = RIEMANN_CB_CLOSED;
int riemann_reset_timeout = 0;
int riemann_failures = 0;

int riemann_tcp_socket;
struct sockaddr_in servaddr;

int
riemann_connect (const char *server, int port)
{
  printf ("Connecting to %s:%d\n", server, port);

  int sockfd = socket (AF_INET, SOCK_STREAM, 0);
  if (sockfd < 0) {
    perror ("Could not open socket");
    return -1;
  }
  else {
    printf ("Socket created!\n");
  }

  struct sockaddr_in remote_addr = { };
  remote_addr.sin_family = AF_INET;
  remote_addr.sin_port = htons (port);

  if (inet_aton (server, &remote_addr.sin_addr) <= 0) {
    perror ("inet_aton failed");
    return -1;
  }

  if (connect (sockfd, (struct sockaddr *) &remote_addr, sizeof (remote_addr)) < 0) {
    perror ("Connect failed");
    return -1;
  }
  else {
    printf ("Connected to %s:%d\n", server, port);
  }
  return sockfd;
}


int
tokenize (char *str, char *delim, char **tokens)
{
  char *p;
  int i = 0;

  p = strtok (str, delim);
  while (p != NULL) {
    printf ("> %s\n", p);
    tokens[i] = malloc (strlen (p) + 1);
    if (tokens[i])
      strcpy (tokens[i], p);
    i++;
    p = strtok (NULL, delim);
  }
  return i++;
}

int
send_data_to_riemann (const char *grid, const char *cluster, const char *host, const char *ip,
                      const char *metric, const char *value, const char *type, const char *units,
                      const char *state, unsigned int localtime, const char *tags_str,
                      const char *location, unsigned int ttl)
{

  int i;
  char *buffer = NULL;

  printf
    ("[riemann] grid=%s, cluster=%s, host=%s, ip=%s, metric=%s, value=%s %s, type=%s, state=%s, localtime=%u, tags=%s, location=%s, ttl=%u\n",
     grid, cluster, host, ip, metric, value, units, type, state, localtime, tags_str, location, ttl);

  Event evt = EVENT__INIT;

  evt.host = (char *) host;
  evt.service = (char *) metric;

  if (value) {
    if (!strcmp (type, "int")) {
      evt.has_metric_sint64 = 1;
      evt.metric_sint64 = strtol (value, (char **) NULL, 10);
    }
    else if (!strcmp (type, "float")) {
      evt.has_metric_d = 1;
      evt.metric_d = (double) strtod (value, (char **) NULL);
    }
    else {
      evt.state = (char *) value;
    }
  }
  evt.description = (char *) units;

  if (state)
    evt.state = (char *) state;

  if (localtime)
    evt.time = localtime;

  char *tags[64] = { NULL };
  buffer = strdup (tags_str);

  evt.n_tags = tokenize (buffer, ",", tags);    /* assume tags are comma-separated */
  evt.tags = tags;
  free (buffer);

  char attr_str[512];
  sprintf (attr_str, "grid=%s,cluster=%s,ip=%s,location=%s", grid, cluster, ip, location);

  int n_attrs;
  char *kv[64] = { NULL };
  buffer = strdup (attr_str);

  n_attrs = tokenize (buffer, ",", kv);
  free (buffer);

  Attribute **attrs;
  attrs = malloc (sizeof (Attribute *) * n_attrs);

  for (i = 0; i < n_attrs; i++) {

    char *pair[1] = { NULL };
    tokenize (kv[i], "=", pair);

    attrs[i] = malloc (sizeof (Attribute));
    attribute__init (attrs[i]);
    attrs[i]->key = pair[0];
    attrs[i]->value = pair[1];
  }
  evt.n_attributes = n_attrs;
  evt.attributes = attrs;

  evt.has_ttl = 1;
  evt.ttl = ttl;

  Msg riemann_msg = MSG__INIT;
  void *buf;
  unsigned len;

  riemann_msg.n_events = 1;
  riemann_msg.events = malloc (sizeof (Event) * riemann_msg.n_events);
  riemann_msg.events[0] = &evt;

  len = msg__get_packed_size (&riemann_msg);
  buf = malloc (len);
  msg__pack (&riemann_msg, buf);

  printf ("[riemann] %zu host=%s, service=%s, state=%s, metric_f=%f, metric_d=%lf, metric_sint64=%" PRId64
          ", description=%s, ttl=%f, tags(%zu), attributes(%zu)\n", evt.time, evt.host, evt.service, evt.state, evt.metric_f,
          evt.metric_d, evt.metric_sint64, evt.description, evt.ttl, evt.n_tags, evt.n_attributes);

  int nbytes = 0;

  if (riemann_circuit_breaker == RIEMANN_CB_CLOSED) {
    if (!strcmp (riemann_protocol, "udp")) {
      nbytes = sendto (riemann_tcp_socket, buf, len, 0, (struct sockaddr *) &servaddr, sizeof (servaddr));
    }
    else {
      printf ("[riemann] Sending metric via TCP...");
      nbytes = send (riemann_tcp_socket, buf, len, 0);
    }
<commit_msg>recv response<commit_after>/*****************************************************************************
 *
 * RIEMANN-SEND-TCP.C
 *
 *    $ indent -br -nut -l125 riemann-send-tcp.c
 *
 *    $ gcc -o riemann-send-tcp riemann-send-tcp.c riemann.pb-c.c $(apr-1-config --cflags --cppflags --includes --link-ld) -lprotobuf-c
 *
 *****************************************************************************/

#include <stdio.h>
#include <stdlib.h>
#include "riemann.pb-c.h"

#include <string.h>
#include <errno.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netdb.h>

#include <apr_general.h>
#include <apr_thread_proc.h>

int done = 0;

#define RIEMANN_CB_OPEN 0
#define RIEMANN_CB_HALF_OPEN 1
#define RIEMANN_CB_CLOSED 2

#define RIEMANN_TIMEOUT 60
#define RIEMANN_MAX_FAILURES 5

char *riemann_server = "127.0.0.1";
char *riemann_protocol = "tcp";
int riemann_port = 5555;
int riemann_circuit_breaker = RIEMANN_CB_CLOSED;
int riemann_reset_timeout = 0;
int riemann_failures = 0;

int riemann_tcp_socket;
struct sockaddr_in servaddr;

int
riemann_connect (const char *server, int port)
{
  printf ("Connecting to %s:%d\n", server, port);

  int sockfd = socket (AF_INET, SOCK_STREAM, 0);
  if (sockfd < 0) {
    perror ("Could not open socket");
    return -1;
  }
  else {
    printf ("Socket created!\n");
  }

  struct sockaddr_in remote_addr = { };
  remote_addr.sin_family = AF_INET;
  remote_addr.sin_port = htons (port);

  if (inet_aton (server, &remote_addr.sin_addr) <= 0) {
    perror ("inet_aton failed");
    return -1;
  }

  if (connect (sockfd, (struct sockaddr *) &remote_addr, sizeof (remote_addr)) < 0) {
    perror ("Connect failed");
    return -1;
  }
  else {
    printf ("Connected to %s:%d\n", server, port);
  }
  return sockfd;
}


int
tokenize (char *str, char *delim, char **tokens)
{
  char *p;
  int i = 0;

  p = strtok (str, delim);
  while (p != NULL) {
    printf ("> %s\n", p);
    tokens[i] = malloc (strlen (p) + 1);
    if (tokens[i])
      strcpy (tokens[i], p);
    i++;
    p = strtok (NULL, delim);
  }
  return i++;
}

int
send_data_to_riemann (const char *grid, const char *cluster, const char *host, const char *ip,
                      const char *metric, const char *value, const char *type, const char *units,
                      const char *state, unsigned int localtime, const char *tags_str,
                      const char *location, unsigned int ttl)
{

  int i;
  char *buffer = NULL;

  printf
    ("[riemann] grid=%s, cluster=%s, host=%s, ip=%s, metric=%s, value=%s %s, type=%s, state=%s, localtime=%u, tags=%s, location=%s, ttl=%u\n",
     grid, cluster, host, ip, metric, value, units, type, state, localtime, tags_str, location, ttl);

  Event evt = EVENT__INIT;

  evt.host = (char *) host;
  evt.service = (char *) metric;

  if (value) {
    if (!strcmp (type, "int")) {
      evt.has_metric_sint64 = 1;
      evt.metric_sint64 = strtol (value, (char **) NULL, 10);
    }
    else if (!strcmp (type, "float")) {
      evt.has_metric_d = 1;
      evt.metric_d = (double) strtod (value, (char **) NULL);
    }
    else {
      evt.state = (char *) value;
    }
  }
  evt.description = (char *) units;

  if (state)
    evt.state = (char *) state;

  if (localtime)
    evt.time = localtime;

  char *tags[64] = { NULL };
  buffer = strdup (tags_str);

  evt.n_tags = tokenize (buffer, ",", tags);    /* assume tags are comma-separated */
  evt.tags = tags;
  free (buffer);

  char attr_str[512];
  sprintf (attr_str, "grid=%s,cluster=%s,ip=%s,location=%s", grid, cluster, ip, location);

  int n_attrs;
  char *kv[64] = { NULL };
  buffer = strdup (attr_str);

  n_attrs = tokenize (buffer, ",", kv);
  free (buffer);

  Attribute **attrs;
  attrs = malloc (sizeof (Attribute *) * n_attrs);

  for (i = 0; i < n_attrs; i++) {

    char *pair[1] = { NULL };
    tokenize (kv[i], "=", pair);

    attrs[i] = malloc (sizeof (Attribute));
    attribute__init (attrs[i]);
    attrs[i]->key = pair[0];
    attrs[i]->value = pair[1];
  }
  evt.n_attributes = n_attrs;
  evt.attributes = attrs;

  evt.has_ttl = 1;
  evt.ttl = ttl;

  Msg riemann_msg = MSG__INIT;
  void *buf;
  unsigned len;

  riemann_msg.n_events = 1;
  riemann_msg.events = malloc (sizeof (Event) * riemann_msg.n_events);
  riemann_msg.events[0] = &evt;

  len = msg__get_packed_size (&riemann_msg);
  buf = malloc (len);
  msg__pack (&riemann_msg, buf);

  printf ("[riemann] %zu host=%s, service=%s, state=%s, metric_f=%f, metric_d=%lf, metric_sint64=%" PRId64
          ", description=%s, ttl=%f, tags(%zu), attributes(%zu)\n", evt.time, evt.host, evt.service, evt.state, evt.metric_f,
          evt.metric_d, evt.metric_sint64, evt.description, evt.ttl, evt.n_tags, evt.n_attributes);

  int nbytes = 0;

  if (riemann_circuit_breaker == RIEMANN_CB_CLOSED) {
    if (!strcmp (riemann_protocol, "udp")) {
      nbytes = sendto (riemann_tcp_socket, buf, len, 0, (struct sockaddr *) &servaddr, sizeof (servaddr));
    }
    else {
      printf ("[riemann] Sending metric via TCP...");
      nbytes = send (riemann_tcp_socket, buf, len, 0);

      Msg *msg;
      uint32_t header, len;
      uint8_t *buffer;
      ssize_t response;

      response = recv (riemann_tcp_socket, &header, sizeof(header), 0);
      len = ntohl(header);
      buffer = malloc(len);
      response = recv (riemann_tcp_socket, buffer, len, 0);
      msg = msg__unpack(NULL, len, buffer);
      printf("ok %d\n", msg->ok);

    }

{'fraction_non_alphanumeric': 0.1833887043189369, 'fraction_numerical': 0.012434741338395823, 'mean_word_length': 3.3074407195421096, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '477333', 'n_tokens_mistral': 4099, 'n_tokens_neox': 3890, 'n_words': 995}