sort of dnstap query log module

src/dnstap/writer.c

@@ -59,10 +59,12 @@ dt_writer_t* dt_writer_create(const char *file_path, const char *version)
 	if (writer->fw == NULL) {
 		goto fail;
 	}
+	/*
 	res = fstrm_writer_open(writer->fw);
 	if (res != fstrm_res_success) {
 		goto fail;
 	}
+	*/
 
 	return writer;
 fail:

src/knot/modules/dnstap.c

@@ -20,30 +20,50 @@
 #include "dnstap/dnstap.pb-c.h"
 #include "dnstap/writer.h"
 #include "dnstap/message.h"
+#include "dnstap/dnstap.h"
 #include "common/descriptor.h"
 
 /* Defines. */
 #define MODULE_ERR(msg...) log_zone_error("Module 'dnstap': " msg)
 
-static int dnstap_query_log(int state, knot_pkt_t *pkt, struct query_data *qdata, void *ctx)
+static int log_message(int state, const knot_pkt_t *pkt, struct query_data *qdata, void *ctx, const Dnstap__Message__Type msgtype)
 {
 	if (pkt == NULL || qdata == NULL || ctx == NULL) {
 		return NS_PROC_FAIL;
 	}
 
 	char *qname = knot_dname_to_str(knot_pkt_qname(pkt));
-	MODULE_ERR("query_log: %s (%u qdcount)\n", qname, knot_wire_get_qdcount(pkt->wire));
+	MODULE_ERR("answer_log: %s (%u ancount)\n", qname, knot_wire_get_ancount(pkt->wire));
 	free(qname);
 
+	struct fstrm_iothr* iothread = ctx;
+	struct fstrm_iothr_queue *ioq = fstrm_iothr_get_input_queue_idx(iothread, 0);
+
 	struct timeval tv;
 	gettimeofday(&tv, NULL);
 	Dnstap__Message msg;
-	memset(&msg, 0, sizeof(Dnstap__Message));
-	dt_message_fill(&msg, DNSTAP__MESSAGE__TYPE__AUTH_QUERY,
+	int ret = dt_message_fill(&msg, msgtype,
 					(const struct sockaddr *)qdata->param->query_source, IPPROTO_UDP,
-					pkt->wire, pkt->size, &tv, NULL);
+					pkt->wire, pkt->size, NULL, &tv);
+	assert(ret == KNOT_EOK);
+
 	/* deal with it later (ret) */
-	dt_writer_write(ctx, (const ProtobufCMessage *)&msg);
+	Dnstap__Dnstap dnstap = DNSTAP__DNSTAP__INIT;
+	dnstap.type = DNSTAP__DNSTAP__TYPE__MESSAGE;
+	dnstap.message = (Dnstap__Message *)&msg;
+	uint8_t *frame = NULL;
+	size_t size = 0;
+	dt_pack(&dnstap, &frame, &size);
+	assert(size > 0);
+	assert(frame);
+	/* return value */
+	fstrm_res res = fstrm_iothr_submit(iothread, ioq, frame, size, fstrm_free_wrapper, NULL);
+	if (res != fstrm_res_success) {
+		free(frame);
+		assert(0);
+	}
+
+	/* return value */
 
 	return state;
 }
@@ -53,21 +73,8 @@ static int dnstap_answer_log(int state, knot_pkt_t *pkt, struct query_data *qdat
 	if (pkt == NULL || qdata == NULL || ctx == NULL) {
 		return NS_PROC_FAIL;
 	}
-
-	char *qname = knot_dname_to_str(knot_pkt_qname(pkt));
-	MODULE_ERR("answer_log: %s (%u ancount)\n", qname, knot_wire_get_ancount(pkt->wire));
-	free(qname);
-
-	struct timeval tv;
-	gettimeofday(&tv, NULL);
-	Dnstap__Message msg;
-	memset(&msg, 0, sizeof(Dnstap__Message));
-	dt_message_fill(&msg, DNSTAP__MESSAGE__TYPE__AUTH_RESPONSE,
-					(const struct sockaddr *)qdata->param->query_source, IPPROTO_UDP,
-					pkt->wire, pkt->size, NULL, &tv);
-	/* deal with it later (ret) */
-	dt_writer_write(ctx, (const ProtobufCMessage *)&msg);
-
+	log_message(state, qdata->query, qdata, ctx, DNSTAP__MESSAGE__TYPE__AUTH_QUERY);
+	log_message(state, pkt, qdata, ctx, DNSTAP__MESSAGE__TYPE__AUTH_RESPONSE);
 	return state;
 }
 
@@ -75,9 +82,15 @@ int dnstap_load(struct query_plan *plan, struct query_module *self)
 {
 
 	/* Save in query module, it takes ownership from now on. */
-	self->ctx = dt_writer_create(self->param, "something");
+	dt_writer_t *writer = dt_writer_create(self->param, "something");
+	assert(writer);
+	struct fstrm_iothr_options* opt = fstrm_iothr_options_init();
+	assert(opt);
+	fstrm_iothr_options_set_queue_model(opt, FSTRM_IOTHR_QUEUE_MODEL_MPSC);
+	struct fstrm_iothr* iothread = fstrm_iothr_init(opt, &writer->fw);
+	self->ctx = iothread;
+	assert(iothread);
 
-	query_plan_step(plan, QPLAN_BEGIN, dnstap_query_log, self->ctx);
 	query_plan_step(plan, QPLAN_END, dnstap_answer_log, self->ctx);
 
 	return KNOT_EOK;
@@ -85,7 +98,8 @@ int dnstap_load(struct query_plan *plan, struct query_module *self)
 
 int dnstap_unload(struct query_module *self)
 {
-	dt_writer_free(self->ctx);
+	struct fstrm_iothr* iothread = self->ctx;
+	fstrm_iothr_destroy(&iothread);
 	self->ctx = NULL;
 	return KNOT_EOK;
 }