path: root/mico-store.c

#include <assert.h>
#include <stdint.h>
#include <stdio.h>
#include <string.h>

#include <zip.h>

#define STB_DS_IMPLEMENTATION
#include "stb_ds.h"

struct event {
	int64_t t;
	double v;
};

struct metric {
	char *key;
	struct event *value;  // array
};

struct metric *metrics;

int
timecmp(const void *a, const void *b)
{
        struct event *ea = (struct event *)a;
        struct event *eb = (struct event *)b;

	if (ea->t < eb->t)
		return -1;
	else if (ea->t > eb->t)
		return 1;
	else
		return 0;  // needed?
}

struct bitfile {
        FILE *output;
	char *mem;
	size_t memlen;

        uint64_t bitbuf;        // value of bits in write buffer
        int bitcount;   // number of bits in write buffer
};

static void
putbits1_msb(struct bitfile *bf, int count, uint32_t bits)
{
        assert(count >= 1 && count <= 57);

        bf->bitbuf <<= count;
        bf->bitcount += count;

        bf->bitbuf |= bits;   /* mask? */

        while (bf->bitcount > 8) {
                bf->bitcount -= 8;
                putc_unlocked((bf->bitbuf >> bf->bitcount) & 0xff, bf->output);
        }
}

static void
putbits1_flush(struct bitfile *bf)
{
        for (int i = bf->bitcount + 8; i >= 0; i--)
                putbits1_msb(bf, 1, 0);
        fflush(bf->output);
}

static void
putsigned(struct bitfile *bf, int count, int v)
{
        if (v < 0)
                putbits1_msb(bf, count, (1 << (count - 1)) - (v + 1));
        else
                putbits1_msb(bf, count, v);
}

void
put1(struct bitfile *bf, int64_t v)
{
        if (v == 0) {
                putbits1_msb(bf, 1, 0x0);
                return;
        }
        if (v >= -64 && v <= 63) {
                putbits1_msb(bf, 2, 0x2);
                putsigned(bf, 7, v);
                return;
        }
        if (v >= -256 && v <= 255) {
                putbits1_msb(bf, 3, 0x6);
                putsigned(bf, 9, v);
                return;
        }
        if (v >= -2048 && v <= 2047) {
                putbits1_msb(bf, 4, 0xe);
                putsigned(bf, 12, v);
                return;
        }

        putbits1_msb(bf, 4, 0xf);
        putsigned(bf, 32, v >> 32);
        putbits1_msb(bf, 32, v & 0xffffffff);
}

int
main(int argc, char *argv[])
{
	sh_new_arena(metrics);

	char name[255];
	char lastname[255];

	char *line = 0;
	size_t linebuflen = 0;
	ssize_t linelen = 0;
	int i = 0;

	if (argc < 2) {
		fprintf(stderr, "usage: %s output.zip\n", argv[0]);
		exit(-1);
	}

	while ((linelen = getdelim(&line, &linebuflen, '\n', stdin)) >= 0) {
		char *start = strchr(line, ' ');
		if (!start) break;
		char *start2 = strchr(start+1, ' ');
		if (!start2) break;

		struct event ev;
		char *eend;
		ev.v = strtod(start, &eend);
		ev.t = strtoll(eend+1, 0, 10);

		if (start - line > (ssize_t)sizeof name) {
			fprintf(stderr, "metric too long: %s\n", line);
			continue;
		}
		memcpy(name, line, start-line);
		name[start-line] = 0;

		// XXX normalize name;

		// ts = ts / 1000;  // XXX

		if (strcmp(name, lastname) == 0) {
			// same name, skip hash lookup, use old i
		} else {
			if (shgeti(metrics, name) < 0)
				shput(metrics, name, 0);
			i = shgeti(metrics, name);
		}

		arrput(metrics[i].value, ev);

		strcpy(lastname, name);
	}

	printf("metrics: %ld\n", shlen(metrics));
	for (int i = 0; i < shlen(metrics); i++) {
		printf("%s %ld\n",
		    metrics[i].key,
		    arrlen(metrics[i].value));
	}

        zip_t *zip = zip_open(argv[1], ZIP_CREATE | ZIP_TRUNCATE, 0);
        if (!zip)
                exit(-1);

	for (int m = 0; m < shlen(metrics); m++) {
		// ensure events are ordered by size
		qsort(metrics[m].value, arrlen(metrics[m].value),
		    sizeof (struct event), timecmp);

		struct bitfile ts = { 0 };
		struct bitfile vs = { 0 };
		ts.output = open_memstream(&ts.mem, &ts.memlen);
		vs.output = open_memstream(&vs.mem, &vs.memlen);

		int64_t t, pt = 0, ppt = 0;
		double v, pv = 0.0;

		put1(&ts, arrlen(metrics[m].value));
		put1(&vs, arrlen(metrics[m].value));

		for (ssize_t i = 0; i < arrlen(metrics[m].value); i++) {
			t = metrics[m].value[i].t;
			v = metrics[m].value[i].v;

			if (t == pt) {
				fprintf(stderr, "duplicate timestamp: %ld", t);
				continue;
			}

			put1(&ts, (t - pt) - (pt - ppt));
			put1(&vs, (int64_t)((v - pv) * 10000));

			ppt = pt;
			pt = t;
			pv = v;
		}

		putbits1_flush(&ts);
		putbits1_flush(&vs);

		printf("%s: %ld + %ld\n", metrics[m].key,
		    ts.memlen, vs.memlen);

		char prefix[255];
		char path[255];
		strcpy(prefix, metrics[m].key);
		char *s = strchr(prefix, '{');
		if (!s) {
			abort();
		}
		*s = '/';
		s = strrchr(prefix, '}');
		if (s)
			*s = 0;

		time_t mtime = metrics[m].value[0].t / 1000;
		snprintf(path, sizeof path, "%s/time.dd", prefix);

		zip_source_t *buft = zip_source_buffer(zip,
		    ts.mem, ts.memlen, 0);
		int jj= zip_file_add(zip, path, buft, ZIP_FL_ENC_UTF_8);
		printf("index of %s = %d\n", path, jj);
		if (jj < 0) {
			printf("error adding file: %s\n", zip_strerror(zip));
		}
		zip_file_set_mtime(zip, jj, mtime, 0);

		snprintf(path, sizeof path, "%s/data.d", prefix);

		zip_source_t *bufv = zip_source_buffer(zip,
		    vs.mem, vs.memlen, 0);
		jj= zip_file_add(zip, path, bufv, ZIP_FL_ENC_UTF_8);
		printf("index of %s = %d\n", path, jj);
		if (jj < 0) {
			printf("error adding file: %s\n", zip_strerror(zip));
		}
		zip_file_set_mtime(zip, jj, mtime, 0);

		// note that data must be kept in memory until zip_close
		// actually writes the archive.

		// close and reopen the zip to force write and free all buffers
		zip_close(zip);
		zip = zip_open(argv[1], 0, 0);
		if (!zip)
			exit(-1);

		arrfree(metrics[m].value);
		free(ts.mem);
		free(vs.mem);
	}

	zip_close(zip);
}