peardata/native/peardata-ebpf/main.c

/**
 * peardata-ebpf — embedded host helper for eBPF-family charts.
 *
 * Emits NDJSON lines on stdout every interval (default 1000ms):
 *   {"chart":"ebpf.cachestat","ts":...,"values":{...}}
 *
 * Uses /proc + /sys (works without CAP_BPF). When compiled with
 * -DPEARDATA_HAVE_LIBBPF and libbpf is linked, additional BPF-backed
 * series can be enabled later without changing the agent protocol.
 *
 * Build:
 *   cc -O2 -o peardata-ebpf main.c
 */
#define _GNU_SOURCE
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <time.h>
#include <stdint.h>
#include <errno.h>
#include <dirent.h>
#include <sys/stat.h>

static long long now_ms(void) {
  struct timespec ts;
  clock_gettime(CLOCK_REALTIME, &ts);
  return (long long)ts.tv_sec * 1000LL + ts.tv_nsec / 1000000LL;
}

static int read_file(const char *path, char *buf, size_t n) {
  FILE *f = fopen(path, "r");
  if (!f) return -1;
  size_t r = fread(buf, 1, n - 1, f);
  fclose(f);
  buf[r] = 0;
  return (int)r;
}

static long long read_ll(const char *path) {
  char buf[64];
  if (read_file(path, buf, sizeof(buf)) < 0) return -1;
  return atoll(buf);
}

static long long vmstat(const char *key) {
  FILE *f = fopen("/proc/vmstat", "r");
  if (!f) return 0;
  char line[256];
  char k[64];
  long long v = 0;
  while (fgets(line, sizeof(line), f)) {
    if (sscanf(line, "%63s %lld", k, &v) == 2 && strcmp(k, key) == 0) {
      fclose(f);
      return v;
    }
  }
  fclose(f);
  return 0;
}

static void emit(const char *chart, const char *values_json) {
  printf("{\"chart\":\"%s\",\"ts\":%lld,\"values\":%s}\n", chart, now_ms(), values_json);
  fflush(stdout);
}

static void collect_cachestat(
  long long *prev_hit,
  long long *prev_miss,
  long long *prev_in,
  long long *prev_out,
  double dt
) {
  /* Page-cache efficiency proxy from vmstat (no CAP_BPF required). */
  long long pgfault = vmstat("pgfault");
  long long pgmajfault = vmstat("pgmajfault");
  long long pgpgin = vmstat("pgpgin");
  long long pgpgout = vmstat("pgpgout");
  long long hit = pgfault > pgmajfault ? pgfault - pgmajfault : 0;
  long long miss = pgmajfault;
  double hits = 0, misses = 0, ratio = 0, rin = 0, rout = 0;
  if (*prev_hit >= 0 && dt > 0) {
    hits = (hit - *prev_hit) / dt;
    misses = (miss - *prev_miss) / dt;
    rin = (pgpgin - *prev_in) / dt;
    rout = (pgpgout - *prev_out) / dt;
    double den = hits + misses;
    ratio = den > 0 ? (hits / den) * 100.0 : 0;
  }
  *prev_hit = hit;
  *prev_miss = miss;
  *prev_in = pgpgin;
  *prev_out = pgpgout;
  char buf[256];
  snprintf(
    buf,
    sizeof(buf),
    "{\"hits\":%.3f,\"misses\":%.3f,\"ratio\":%.3f,\"pgpgin\":%.3f,\"pgpgout\":%.3f}",
    hits,
    misses,
    ratio,
    rin,
    rout
  );
  emit("ebpf.cachestat", buf);
}

static void collect_fd(void) {
  char buf[128];
  long long alloc = 0, max = 0;
  if (read_file("/proc/sys/fs/file-nr", buf, sizeof(buf)) > 0) {
    sscanf(buf, "%lld %*s %lld", &alloc, &max);
  }
  char out[128];
  snprintf(out, sizeof(out), "{\"open\":%lld,\"max\":%lld}", alloc, max);
  emit("ebpf.fd", out);
}

static void collect_oom(long long *prev, double dt) {
  long long kills = vmstat("oom_kill");
  double rate = 0;
  if (*prev >= 0 && dt > 0) rate = (kills - *prev) / dt;
  *prev = kills;
  char out[64];
  snprintf(out, sizeof(out), "{\"kills\":%.3f,\"total\":%lld}", rate, kills);
  emit("ebpf.oom", out);
}

static void collect_process(long long *prev_forks, long long *prev_ctxt, double dt) {
  FILE *f = fopen("/proc/stat", "r");
  long long forks = 0, ctxt = 0;
  if (f) {
    char line[256];
    while (fgets(line, sizeof(line), f)) {
      if (strncmp(line, "processes ", 10) == 0) forks = atoll(line + 10);
      else if (strncmp(line, "ctxt ", 5) == 0) ctxt = atoll(line + 5);
    }
    fclose(f);
  }
  double fr = 0, cr = 0;
  if (*prev_forks >= 0 && dt > 0) {
    fr = (forks - *prev_forks) / dt;
    cr = (ctxt - *prev_ctxt) / dt;
  }
  *prev_forks = forks;
  *prev_ctxt = ctxt;
  char out[128];
  snprintf(out, sizeof(out), "{\"forks\":%.3f,\"ctxt\":%.3f}", fr, cr);
  emit("ebpf.process", out);
}

static void collect_shm(void) {
  FILE *f = fopen("/proc/sysvipc/shm", "r");
  long long segs = 0, bytes = 0;
  if (f) {
    char line[512];
    int first = 1;
    while (fgets(line, sizeof(line), f)) {
      if (first) {
        first = 0;
        continue;
      }
      long long size = 0;
      /* key shmid perms size ... */
      if (sscanf(line, "%*s %*s %*s %lld", &size) >= 1) {
        segs++;
        bytes += size;
      }
    }
    fclose(f);
  }
  char out[128];
  snprintf(out, sizeof(out), "{\"segments\":%lld,\"bytes\":%lld}", segs, bytes);
  emit("ebpf.shm", out);
}

static void collect_swap(long long *prev_in, long long *prev_out, double dt) {
  long long in = vmstat("pswpin");
  long long out = vmstat("pswpout");
  double ri = 0, ro = 0;
  if (*prev_in >= 0 && dt > 0) {
    ri = (in - *prev_in) / dt;
    ro = (out - *prev_out) / dt;
  }
  *prev_in = in;
  *prev_out = out;
  char buf[128];
  snprintf(buf, sizeof(buf), "{\"in\":%.3f,\"out\":%.3f}", ri, ro);
  emit("ebpf.swap", buf);
}

static void collect_vfs(long long *prev_read, long long *prev_write, double dt) {
  long long r = vmstat("pgpgout"); /* proxy */
  long long w = vmstat("pgpgin");
  /* Prefer ext4-ish: use /proc/vmstat nr_dirty_threshold unrelated — keep simple */
  double rr = 0, ww = 0;
  if (*prev_read >= 0 && dt > 0) {
    rr = (r - *prev_read) / dt;
    ww = (w - *prev_write) / dt;
  }
  *prev_read = r;
  *prev_write = w;
  char buf[128];
  snprintf(buf, sizeof(buf), "{\"read_pages\":%.3f,\"write_pages\":%.3f}", rr, ww);
  emit("ebpf.vfs", buf);
}

static void collect_net(long long *prev_tcp, double dt) {
  FILE *f = fopen("/proc/net/sockstat", "r");
  long long tcp = 0;
  if (f) {
    char line[256];
    while (fgets(line, sizeof(line), f)) {
      if (strncmp(line, "TCP:", 4) == 0) {
        sscanf(line, "TCP: inuse %lld", &tcp);
      }
    }
    fclose(f);
  }
  double rate = 0;
  if (*prev_tcp >= 0 && dt > 0) rate = (tcp - *prev_tcp) / dt;
  *prev_tcp = tcp;
  char buf[128];
  snprintf(buf, sizeof(buf), "{\"tcp_inuse\":%lld,\"tcp_delta\":%.3f}", tcp, rate);
  emit("ebpf.socket", buf);
}

int main(int argc, char **argv) {
  int interval_ms = 1000;
  for (int i = 1; i < argc; i++) {
    if (strcmp(argv[i], "--interval") == 0 && i + 1 < argc) {
      interval_ms = atoi(argv[++i]);
      if (interval_ms < 200) interval_ms = 200;
    }
  }

  long long prev_hit = -1, prev_miss = -1, prev_cin = -1, prev_cout = -1;
  long long prev_oom = -1;
  long long prev_forks = -1, prev_ctxt = -1;
  long long prev_swap_in = -1, prev_swap_out = -1;
  long long prev_vfs_r = -1, prev_vfs_w = -1;
  long long prev_tcp = -1;
  long long last = now_ms();

  fprintf(stderr, "peardata-ebpf ready interval=%dms\n", interval_ms);

  for (;;) {
    long long t0 = now_ms();
    double dt = (t0 - last) / 1000.0;
    if (dt <= 0) dt = interval_ms / 1000.0;
    last = t0;

    collect_cachestat(&prev_hit, &prev_miss, &prev_cin, &prev_cout, dt);
    collect_fd();
    collect_oom(&prev_oom, dt);
    collect_process(&prev_forks, &prev_ctxt, dt);
    collect_shm();
    collect_swap(&prev_swap_in, &prev_swap_out, dt);
    collect_vfs(&prev_vfs_r, &prev_vfs_w, dt);
    collect_net(&prev_tcp, dt);

    long long elapsed = now_ms() - t0;
    long long sleep_ms = interval_ms - elapsed;
    if (sleep_ms < 50) sleep_ms = 50;
    usleep((useconds_t)(sleep_ms * 1000));
  }
  return 0;
}