Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Simplify logging #68

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Simplify logging #68

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 13 additions & 22 deletions src/exports.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -158,8 +158,7 @@ _json_to_pqtimestamp(json_object *found, Needles current)
goto error;

if (len < 21 || len > 31) {
logger_log("%s %d: Datestring %s not supported",
__FILE__, __LINE__, ts);
log("Datestring %s not supported", ts);
goto error;
}

Expand All @@ -168,8 +167,7 @@ _json_to_pqtimestamp(json_object *found, Needles current)
if (ts[4] != '-' || ts[7] != '-' || ts[10] != 'T'
|| ts[13] != ':' || ts[16] != ':' || ts[19] != '.'
|| ts[len-1] != 'Z') {
logger_log("%s %d: Datestring %s not supported",
__FILE__, __LINE__, ts);
log("Datestring %s not supported", ts);
goto error;
}

Expand All @@ -184,8 +182,7 @@ _json_to_pqtimestamp(json_object *found, Needles current)
tm.micro = strtoull(ts+20,NULL,10);

if(errno) {
logger_log("%s %d: Error %s in date conversion",
__FILE__, __LINE__, strerror(errno));
log("Error %s in date conversion", strerror(errno));
goto error;
}

Expand All @@ -198,19 +195,16 @@ _json_to_pqtimestamp(json_object *found, Needles current)
}

if(tm.year < 2000 || tm.year > 4027) {
logger_log("%s %d: Date %s out of range",
__FILE__, __LINE__, ts);
log("Date %s out of range", ts);
goto error;
}
if(tm.month < 1 || tm.month > 12 || tm.day >31
|| tm.hour > 23 || tm.minute > 59 || tm.second > 59) {
logger_log("%s %d: Datestring %s not a date",
__FILE__, __LINE__, ts);
log("Datestring %s not a date", ts);
goto error;
}
if(tm.month == 2 && tm.day > 29) {
logger_log("%s %d: Datestring %s not a date",
__FILE__, __LINE__, ts);
log("Datestring %s not a date", ts);
goto error;
}

Expand Down Expand Up @@ -320,12 +314,12 @@ exports_meta_init(const char *host, const char *topic, config_setting_t *needles
m->conn_master = PQconnectdb(m->conninfo);
if (PQstatus(m->conn_master) != CONNECTION_OK)
{
logger_log("%s %d: %s", __FILE__, __LINE__, PQerrorMessage(m->conn_master));
log("%s", PQerrorMessage(m->conn_master));
abort();
}

if (pthread_mutex_init(&m->commit_mutex, NULL) != 0) {
logger_log("%s %d: unable to create mutex", __FILE__, __LINE__ );
log("unable to create mutex");
abort();
}

Expand Down Expand Up @@ -374,7 +368,7 @@ exports_producer_init(config_setting_t *config)
NULL,
commit_worker,
(void *)&(exports->meta))) {
logger_log("%s %d: Failed to create commit worker!", __FILE__, __LINE__);
log("Failed to create commit worker!");
abort();
}

Expand All @@ -395,8 +389,7 @@ _deref(json_object *haystack, Internal internal)
needles[i]->length = (uint32_t)~0;
} else {
if(!needles[i]->format(found,needles[i])) {
logger_log("%s %d: Failed jpointer deref %s",
__FILE__, __LINE__, needles[i]->jpointer);
log("Failed jpointer deref %s", needles[i]->jpointer);
return false;
}
}
Expand Down Expand Up @@ -431,8 +424,7 @@ exports_producer_produce(Producer p, Message msg)
m->res = PQexec(m->conn_master, m->cpycmd);
if (PQresultStatus(m->res) != PGRES_COPY_IN)
{
logger_log("%s %d: %s", __FILE__, __LINE__,
PQerrorMessage(m->conn_master));
log(PQerrorMessage(m->conn_master));
abort();
}
PQclear(m->res);
Expand All @@ -446,13 +438,12 @@ exports_producer_produce(Producer p, Message msg)

haystack = json_tokener_parse(data);
if(!haystack) {
logger_log("%s %d: Failed to tokenize json!", __FILE__, __LINE__);
log("Failed to tokenize json!");
goto error;
}

if(!_deref(haystack, internal)) {
logger_log("%s %d: Failed to dereference json!\n %s",
__FILE__, __LINE__, data);
log("Failed to dereference json!\n %s", data);
for (int i = 0; i < internal->ncount; i++) {
needles[i]->free(&needles[i]->result);
}
Expand Down
6 changes: 3 additions & 3 deletions src/file.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ file_meta_init(const char *fname, char *options)
m->fp = fopen(fname, options);
if (m->fp == NULL)
{
logger_log("%s %d: %s %s", __FILE__, __LINE__, fname, strerror(errno));
log("%s %s", fname, strerror(errno));
abort();
}
return m;
Expand All @@ -33,7 +33,7 @@ void
file_meta_free(Meta *m)
{
if ( fclose((*m)->fp) != 0)
logger_log("%s %d: %s", __FILE__, __LINE__, strerror(errno));
log(strerror(errno));
free(*m);
*m = NULL;
}
Expand Down Expand Up @@ -91,7 +91,7 @@ file_consumer_consume(Consumer c, Message msg)
ssize_t read;
if ((read = getline(&line, &bufsize, ((Meta) c->meta)->fp)) == -1)
{
logger_log("%s %d: %s", __FILE__, __LINE__, strerror(errno));
log(strerror(errno));
return -1;
}

Expand Down
43 changes: 19 additions & 24 deletions src/kafka.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ static void
dr_msg_cb (UNUSED rd_kafka_t *rk, const rd_kafka_message_t *rkmessage, UNUSED void *opaque)
{
if (rkmessage->err)
logger_log("%s %d: Message delivery failed: %s\n", __FILE__, __LINE__, rd_kafka_err2str(rkmessage->err));
log("Message delivery failed: %s\n", rd_kafka_err2str(rkmessage->err));
}

static void
Expand Down Expand Up @@ -75,13 +75,13 @@ kafka_producer_meta_init(const char *broker, const char *topic)

if (rd_kafka_conf_set(conf, "compression.codec", "lz4", errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
logger_log("%s %d: %s", __FILE__, __LINE__, errstr);
log(errstr);
abort();
}

if (rd_kafka_conf_set(conf, "queue.buffering.max.ms","1000",errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
logger_log("%s %d: %s\n", __FILE__, __LINE__, errstr);
log(errstr);
abort();
}

Expand All @@ -90,22 +90,20 @@ kafka_producer_meta_init(const char *broker, const char *topic)
rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
if (!rk)
{
logger_log("%s %d: Failed to create new producer: %s\n", __FILE__, __LINE__, errstr);
log("Failed to create new producer: %s", errstr);
abort();
}
if (rd_kafka_brokers_add(rk, broker) == 0)
{
logger_log("%s %d: Failed to add broker: %s\n", __FILE__, __LINE__, broker);
log("Failed to add broker: %s", broker);
abort();
}

rkt = rd_kafka_topic_new(rk, topic, NULL);

if (!rkt)
{
logger_log("%s %d: Failed to create topic object: %s\n",
__FILE__,
__LINE__,
log("Failed to create topic object: %s",
rd_kafka_err2str(rd_kafka_last_error()));
rd_kafka_destroy(rk);
abort();
Expand All @@ -132,29 +130,29 @@ kafka_consumer_meta_init(const char *broker, const char *topic, const char *grou

if (rd_kafka_conf_set(conf, "group.id", groupid, errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
logger_log("%s %d: %s", __FILE__, __LINE__, errstr);
log(errstr);
abort();
}

if (rd_kafka_topic_conf_set(topic_conf, "offset.store.method","broker",errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
logger_log("%s %d: %s\n", __FILE__, __LINE__, errstr);
log(errstr);
abort();
}
if (rd_kafka_topic_conf_set(topic_conf, "enable.auto.commit","true",errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
logger_log("%s %d: %s\n", __FILE__, __LINE__, errstr);
log(errstr);
abort();
}
if (rd_kafka_topic_conf_set(topic_conf, "auto.commit.interval.ms","10",errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
logger_log("%s %d: %s\n", __FILE__, __LINE__, errstr);
log(errstr);
abort();
}

if (rd_kafka_topic_conf_set(topic_conf, "auto.offset.reset","latest",errstr, sizeof(errstr)) != RD_KAFKA_CONF_OK)
{
logger_log("%s %d: %s\n", __FILE__, __LINE__, errstr);
log(errstr);
abort();
}

Expand All @@ -165,7 +163,7 @@ kafka_consumer_meta_init(const char *broker, const char *topic, const char *grou
rk = rd_kafka_new(RD_KAFKA_CONSUMER, conf, errstr, sizeof(errstr));
if (!rk)
{
logger_log("%s %d: Failed to create new consumer: %s\n", __FILE__, __LINE__, errstr);
log("Failed to create new consumer: %s", errstr);
abort();
}

Expand All @@ -178,7 +176,8 @@ kafka_consumer_meta_init(const char *broker, const char *topic, const char *grou
rkt = rd_kafka_topic_new(rk, topic, NULL);
if (!rkt)
{
logger_log("%s %d: Failed to create topic object: %s\n", __FILE__, __LINE__, rd_kafka_err2str(rd_kafka_last_error()));
log("Failed to create topic object: %s",
rd_kafka_err2str(rd_kafka_last_error()));
rd_kafka_destroy(rk);
abort();
}
Expand All @@ -190,10 +189,9 @@ kafka_consumer_meta_init(const char *broker, const char *topic, const char *grou

if ((err = rd_kafka_subscribe(rk, topics)))
{
fprintf(stderr,
"%% Failed to start consuming topics: %s\n",
rd_kafka_err2str(err));
abort();
logger_log("%% Failed to start consuming topics: %s",
rd_kafka_err2str(err));
abort();
}
m->rk = rk;
m->rkt = rkt;
Expand Down Expand Up @@ -263,12 +261,9 @@ kafka_producer_produce(Producer p, Message msg)
}
else
{
logger_log(
"%s %d Failed to produce to topic %s: %s\n",
__FILE__, __LINE__,
log("Failed to produce to topic %s: %s",
rd_kafka_topic_name(rkt),
rd_kafka_err2str(rd_kafka_last_error())
);
rd_kafka_err2str(rd_kafka_last_error()));
}
}
rd_kafka_poll(rk, 0);
Expand Down
9 changes: 5 additions & 4 deletions src/main.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
#include "utils/logger.h"
#include "utils/options.h"
#include "utils/scalloc.h"
#include "version.h"


/* Schaufel keeps track of consume and produce states.
Expand Down Expand Up @@ -114,14 +115,14 @@ consume(void *config)

if (msg == NULL)
{
logger_log("%s %d: could not init message", __FILE__, __LINE__);
log("could not init message");
return NULL;
}
Consumer c = consumer_init(*consumer_type,
(config_setting_t *) config);
if (c == NULL)
{
logger_log("%s %d: could not init consumer", __FILE__, __LINE__);
log("could not init consumer");
return NULL;
}

Expand Down Expand Up @@ -157,7 +158,7 @@ produce(void *config)
(config_setting_t *) config);
if (p == NULL)
{
logger_log("%s %d: could not init producer", __FILE__, __LINE__);
log("could not init producer");
return NULL;
}

Expand Down Expand Up @@ -336,7 +337,7 @@ main(int argc, char **argv)

q = queue_init();
if (!q) {
logger_log("%s %d: Failed to init queue\n", __FILE__, __LINE__);
log("Failed to init queue");
abort();
}

Expand Down
16 changes: 8 additions & 8 deletions src/postgres.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ postgres_meta_init(const char *host, const char *host_replica, const char *gener
m->conn_master = PQconnectdb(m->conninfo);
if (PQstatus(m->conn_master) != CONNECTION_OK)
{
logger_log("%s %d: %s", __FILE__, __LINE__, PQerrorMessage(m->conn_master));
log(PQerrorMessage(m->conn_master));
abort();
}

Expand All @@ -92,12 +92,12 @@ postgres_meta_init(const char *host, const char *host_replica, const char *gener
m->conn_replica = PQconnectdb(m->conninfo_replica);
if (PQstatus(m->conn_replica) != CONNECTION_OK)
{
logger_log("%s %d: %s", __FILE__, __LINE__, PQerrorMessage(m->conn_replica));
log(PQerrorMessage(m->conn_replica));
abort();
}

if (pthread_mutex_init(&m->commit_mutex, NULL) != 0) {
logger_log("%s %d: unable to create mutex", __FILE__, __LINE__ );
log("unable to create mutex");
abort();
}

Expand Down Expand Up @@ -137,7 +137,7 @@ postgres_producer_init(config_setting_t *config)
NULL,
commit_worker,
(void *)&(postgres->meta))) {
logger_log("%s %d: Failed to create commit worker!", __FILE__, __LINE__);
log("Failed to create commit worker!");
abort();
}

Expand All @@ -155,14 +155,14 @@ postgres_producer_produce(Producer p, Message msg)

if (buf[len] != '\0')
{
logger_log("payload doesn't end on null terminator");
log("payload doesn't end on null terminator");
return;
}

char *s = strstr(buf, "\\u0000");
if (s != NULL)
{
logger_log("found invalid unicode byte sequence: %s", buf);
log("found invalid unicode byte sequence: %s", buf);
return;
}

Expand All @@ -174,7 +174,7 @@ postgres_producer_produce(Producer p, Message msg)
m->res = PQexec(m->conn_master, m->cpycmd);
if (PQresultStatus(m->res) != PGRES_COPY_IN)
{
logger_log("%s %d: %s", __FILE__, __LINE__, PQerrorMessage(m->conn_master));
log(PQerrorMessage(m->conn_master));
abort();
}
PQclear(m->res);
Expand All @@ -184,7 +184,7 @@ postgres_producer_produce(Producer p, Message msg)
m->res = PQexec(m->conn_replica, m->cpycmd);
if (PQresultStatus(m->res) != PGRES_COPY_IN)
{
logger_log("%s %d: %s", __FILE__, __LINE__, PQerrorMessage(m->conn_replica));
log(PQerrorMessage(m->conn_replica));
abort();
}
PQclear(m->res);
Expand Down
Loading