diff --git a/CMakeLists.txt b/CMakeLists.txt index 06df886bf..d89ab009c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -114,7 +114,7 @@ pkg_check_modules(LIB60870 IMPORTED_TARGET lib60870>=2.3.1) pkg_check_modules(LIBCONFIG IMPORTED_TARGET libconfig>=1.4.9) pkg_check_modules(MOSQUITTO IMPORTED_TARGET libmosquitto>=1.6.9) pkg_check_modules(MODBUS IMPORTED_TARGET libmodbus>=3.1.0) -pkg_check_modules(RDKAFKA IMPORTED_TARGET rdkafka>=1.5.0) +pkg_check_modules(RDKAFKAPP IMPORTED_TARGET rdkafka++>=1.5.0) pkg_check_modules(HIREDIS IMPORTED_TARGET hiredis>=1.0.0) pkg_check_modules(REDISPP IMPORTED_TARGET redis++>=1.2.0) pkg_check_modules(RABBITMQ_C IMPORTED_TARGET librabbitmq>=0.8.0) @@ -205,7 +205,7 @@ cmake_dependent_option(WITH_NODE_IEC60870 "Build with iec60870 node-types" cmake_dependent_option(WITH_NODE_IEC61850 "Build with iec61850 node-types" "${WITH_DEFAULTS}" "LIBIEC61850_FOUND; NOT WITHOUT_GPL" OFF) cmake_dependent_option(WITH_NODE_INFINIBAND "Build with infiniband node-type" "${WITH_DEFAULTS}" "IBVerbs_FOUND; RDMACM_FOUND" OFF) # Infiniband node-type is currenly broken cmake_dependent_option(WITH_NODE_INFLUXDB "Build with influxdb node-type" "${WITH_DEFAULTS}" "" OFF) -cmake_dependent_option(WITH_NODE_KAFKA "Build with kafka node-type" "${WITH_DEFAULTS}" "RDKAFKA_FOUND" OFF) +cmake_dependent_option(WITH_NODE_KAFKA "Build with kafka node-type" "${WITH_DEFAULTS}" "RDKAFKAPP_FOUND" OFF) cmake_dependent_option(WITH_NODE_LOOPBACK "Build with loopback node-type" "${WITH_DEFAULTS}" "" OFF) cmake_dependent_option(WITH_NODE_MODBUS "Build with modbus node-type" "${WITH_DEFAULTS}" "MODBUS_FOUND" OFF) cmake_dependent_option(WITH_NODE_MQTT "Build with mqtt node-type" "${WITH_DEFAULTS}" "MOSQUITTO_FOUND" OFF) @@ -321,7 +321,7 @@ add_feature_info(NODE_MODBUS WITH_NODE_MODBUS "Build with add_feature_info(NODE_MQTT WITH_NODE_MQTT "Build with mqtt node-type") add_feature_info(NODE_NANOMSG WITH_NODE_NANOMSG "Build with nanomsg node-type") add_feature_info(NODE_NGSI WITH_NODE_NGSI "Build with ngsi node-type") -add_feature_info(NODE_OPAL_AYSNC WITH_NODE_OPAL_ASYNC "Build with opal.async node-type") +add_feature_info(NODE_OPAL_ASYNC WITH_NODE_OPAL_ASYNC "Build with opal.async node-type") add_feature_info(NODE_OPAL_ORCHESTRA WITH_NODE_OPAL_ORCHESTRA "Build with opal.orchestra node-type") add_feature_info(NODE_OPENDSS WITH_NODE_OPENDSS "Build with opendss node-type") add_feature_info(NODE_REDIS WITH_NODE_REDIS "Build with redis node-type") diff --git a/doc/openapi/components/schemas/config/nodes/kafka.yaml b/doc/openapi/components/schemas/config/nodes/kafka.yaml index a86024f8f..56624e6e6 100644 --- a/doc/openapi/components/schemas/config/nodes/kafka.yaml +++ b/doc/openapi/components/schemas/config/nodes/kafka.yaml @@ -56,7 +56,7 @@ allOf: in: type: object properties: - consume: + topic: type: string description: The Kafka topic to which this node-type will subscribe for receiving messages. @@ -67,7 +67,7 @@ allOf: out: type: object properties: - produce: + topic: type: string description: The Kafka topic to which this node-type will publish messages. diff --git a/etc/examples/nodes/kafka.conf b/etc/examples/nodes/kafka.conf index 4f74d5be6..e17f2ba84 100644 --- a/etc/examples/nodes/kafka.conf +++ b/etc/examples/nodes/kafka.conf @@ -7,27 +7,39 @@ nodes = { format = "json.kafka" - server = "localhost:9094" - protocol = "SASL_SSL" + server = "localhost:9092" + protocol = "PLAINTEXT" client_id = "villas-node" in = { - consume = "test-topic" + topic = "test-topic" group_id = "villas-node" } out = { - produce = "test-topic" + topic = "test-topic" } + } - ssl = { - ca = "/etc/ssl/certs/ca.pem" - } + siggen = { + type = "signal" - sasl = { - mechanisms = "SCRAM-SHA-512" - username = "scram-sha-512-usr" - password = "scram-sha-512-pwd" - } + rate = 20 + values = 5 + signal = "mixed" } } + +paths = ( + { + in = "siggen" + out = "kafka_node" + }, + { + in = "kafka_node" + + hooks = [ + "print" + ] + } +) diff --git a/include/villas/nodes/kafka.hpp b/include/villas/nodes/kafka.hpp deleted file mode 100644 index 4ca6a765a..000000000 --- a/include/villas/nodes/kafka.hpp +++ /dev/null @@ -1,83 +0,0 @@ -/* Node type: kafka. - * - * Author: Juan Pablo Noreña - * SPDX-FileCopyrightText: 2021 Universidad Nacional de Colombia - * SPDX-License-Identifier: Apache-2.0 - */ - -#pragma once - -#include - -#include -#include -#include - -namespace villas { -namespace node { - -// Forward declarations -class NodeCompat; - -struct kafka { - struct CQueueSignalled queue; - struct Pool pool; - - double timeout; // Timeout in seconds. - char *server; // Hostname/IP:Port address of the bootstrap server. - char *protocol; // Security protocol. - char *produce; // Producer topic. - char *consume; // Consumer topic. - char *client_id; // Client ID. - - struct { - rd_kafka_t *client; - rd_kafka_topic_t *topic; - } producer; - - struct { - rd_kafka_t *client; - char *group_id; // Group id. - } consumer; - - struct { - char *ca; // SSL CA file. - } ssl; - - struct { - char *mechanisms; // SASL mechanisms. - char *username; // SSL CA path. - char *password; // SSL certificate. - } sasl; - - Format *formatter; -}; - -int kafka_reverse(NodeCompat *n); - -char *kafka_print(NodeCompat *n); - -int kafka_init(NodeCompat *n); - -int kafka_prepare(NodeCompat *n); - -int kafka_parse(NodeCompat *n, json_t *json); - -int kafka_start(NodeCompat *n); - -int kafka_destroy(NodeCompat *n); - -int kafka_stop(NodeCompat *n); - -int kafka_type_start(SuperNode *sn); - -int kafka_type_stop(); - -int kafka_poll_fds(NodeCompat *n, int fds[]); - -int kafka_read(NodeCompat *n, struct Sample *const smps[], unsigned cnt); - -int kafka_write(NodeCompat *n, struct Sample *const smps[], unsigned cnt); - -} // namespace node -} // namespace villas diff --git a/lib/nodes/CMakeLists.txt b/lib/nodes/CMakeLists.txt index 540154c96..5ae13d1db 100644 --- a/lib/nodes/CMakeLists.txt +++ b/lib/nodes/CMakeLists.txt @@ -120,7 +120,7 @@ endif() # Enable Kafka support if(WITH_NODE_KAFKA) list(APPEND NODE_SRC kafka.cpp) - list(APPEND LIBRARIES PkgConfig::RDKAFKA) + list(APPEND LIBRARIES PkgConfig::RDKAFKAPP) endif() # Enable Comedi support diff --git a/lib/nodes/kafka.cpp b/lib/nodes/kafka.cpp index 035fd6e74..6aaccdf2e 100644 --- a/lib/nodes/kafka.cpp +++ b/lib/nodes/kafka.cpp @@ -1,600 +1,412 @@ /* Node type: kafka. * * Author: Juan Pablo Noreña + * Author: Steffen Vogel * SPDX-FileCopyrightText: 2021 Universidad Nacional de Colombia + * SPDX-FileCopyrightText: 2025 OPAL-RT Germany GmbH * SPDX-License-Identifier: Apache-2.0 */ -#include +#include #include -#include +#include +#include #include +#include +#include +#include #include -#include +#include +#include #include using namespace villas; -using namespace villas::node; using namespace villas::utils; +using namespace villas::node; -// Each process has a list of clients for which a thread invokes the kafka loop -static struct List clients; -static pthread_t thread; static Logger logger; -static void kafka_logger_cb(const rd_kafka_t *rk, int level, const char *fac, - const char *buf) { - - switch (level) { - case LOG_EMERG: - case LOG_CRIT: - case LOG_ERR: - logger->error("{}: {}", fac, buf); - break; - - case LOG_ALERT: - case LOG_WARNING: - logger->warn("{}: {}", fac, buf); - break; - - case LOG_DEBUG: - logger->debug("{}: {}", fac, buf); - break; - - case LOG_NOTICE: - case LOG_INFO: - default: - logger->info("{}: {}", fac, buf); - break; - } -} - -static void kafka_message_cb(void *ctx, const rd_kafka_message_t *msg) { - int ret; - auto *n = (NodeCompat *)ctx; - auto *k = n->getData(); - struct Sample *smps[n->in.vectorize]; - - n->logger->debug("Received a message of {} bytes from broker {}", msg->len, - k->server); - - ret = sample_alloc_many(&k->pool, smps, n->in.vectorize); - if (ret <= 0) { - n->logger->warn("Pool underrun in consumer"); - return; - } - - ret = k->formatter->sscan((char *)msg->payload, msg->len, nullptr, smps, - n->in.vectorize); - if (ret < 0) { - n->logger->warn("Received an invalid message"); - n->logger->warn(" Payload: {}", (char *)msg->payload); - return; - } - - if (ret == 0) { - n->logger->debug("Skip empty message"); - sample_decref_many(smps, n->in.vectorize); - return; - } - - ret = queue_signalled_push_many(&k->queue, (void **)smps, n->in.vectorize); - if (ret < (int)n->in.vectorize) - n->logger->warn("Failed to enqueue samples"); -} - -static void *kafka_loop_thread(void *ctx) { - int ret; - - // Set the cancel type of this thread to async - ret = pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, nullptr); - if (ret != 0) - throw RuntimeError("Unable to set cancel type of Kafka communication " - "thread to asynchronous."); - - while (true) { - for (unsigned i = 0; i < list_length(&clients); i++) { - auto *n = (NodeCompat *)list_at(&clients, i); - auto *k = n->getData(); - - // Execute kafka loop for this client - if (k->consumer.client) { - rd_kafka_message_t *msg = - rd_kafka_consumer_poll(k->consumer.client, k->timeout * 1000); - if (msg) { - kafka_message_cb((void *)n, msg); - rd_kafka_message_destroy(msg); - } - } +class KafkaNode : public Node, public RdKafka::EventCb { + +protected: + // Settings. + std::chrono::milliseconds timeout; + std::string server; // Hostname/IP:Port address of the bootstrap server. + std::string protocol; // Security protocol. + std::string produce; // Producer topic. + std::string consume; // Consumer topic. + std::string client_id; // Client ID. + std::string group_id; // Group ID. + std::string ssl_ca; // SSL CA file. + + struct { + std::unique_ptr client; + std::unique_ptr topic; + } producer; + + struct { + std::unique_ptr client; + std::unique_ptr topic; + std::unique_ptr queue; + int eventFd; + } consumer; + + struct { + std::string mechanisms; // SASL mechanisms. + std::string username; // SSL CA path. + std::string password; // SSL certificate. + } sasl; + + std::unique_ptr formatter; + + int _read(struct Sample *smps[], unsigned cnt) override { + assert(consumer.client != nullptr); + + auto msg = consumer.client->consume(consumer.queue.get(), timeout.count()); + + auto ret = formatter->sscan((char *)msg->payload(), msg->len(), nullptr, + smps, cnt); + if (ret < 0) { + logger->warn("Received an invalid message"); + logger->warn(" Payload: {}", (char *)msg->payload()); + return -1; } - } - - return nullptr; -} - -int villas::node::kafka_reverse(NodeCompat *n) { - auto *k = n->getData(); - - std::swap(k->produce, k->consume); - - return 0; -} - -int villas::node::kafka_init(NodeCompat *n) { - auto *k = n->getData(); - - // Default values - k->server = nullptr; - k->protocol = nullptr; - k->produce = nullptr; - k->consume = nullptr; - k->client_id = nullptr; - k->timeout = 1.0; - - k->consumer.client = nullptr; - k->consumer.group_id = nullptr; - k->producer.client = nullptr; - k->producer.topic = nullptr; - - k->sasl.mechanisms = nullptr; - k->sasl.username = nullptr; - k->sasl.password = nullptr; - - k->ssl.ca = nullptr; - - k->formatter = nullptr; - - return 0; -} - -int villas::node::kafka_parse(NodeCompat *n, json_t *json) { - int ret; - auto *k = n->getData(); - - const char *server; - const char *produce = nullptr; - const char *consume = nullptr; - const char *protocol; - const char *client_id = "villas-node"; - const char *group_id = nullptr; - - json_error_t err; - json_t *json_ssl = nullptr; - json_t *json_sasl = nullptr; - json_t *json_format = nullptr; - - ret = json_unpack_ex(json, &err, 0, - "{ s?: { s?: s }, s?: { s?: s, s?: s }, s?: o, s: s, " - "s?: F, s: s, s?: s, s?: o, s?: o }", - "out", "produce", &produce, "in", "consume", &consume, - "group_id", &group_id, "format", &json_format, "server", - &server, "timeout", &k->timeout, "protocol", &protocol, - "client_id", &client_id, "ssl", &json_ssl, "sasl", - &json_sasl); - if (ret) - throw ConfigError(json, err, "node-config-node-kafka"); - - k->server = strdup(server); - k->produce = produce ? strdup(produce) : nullptr; - k->consume = consume ? strdup(consume) : nullptr; - k->protocol = strdup(protocol); - k->client_id = strdup(client_id); - k->consumer.group_id = group_id ? strdup(group_id) : nullptr; - - if (strcmp(protocol, "SSL") && strcmp(protocol, "PLAINTEXT") && - strcmp(protocol, "SASL_SSL") && strcmp(protocol, "SASL_PLAINTEXT")) - throw ConfigError(json, "node-config-node-kafka-protocol", - "Invalid security protocol: {}", protocol); - - if (!k->produce && !k->consume) - throw ConfigError(json, "node-config-node-kafka", - "At least one topic has to be specified for node {}", - n->getName()); - - if (json_ssl) { - const char *ca; - - ret = json_unpack_ex(json_ssl, &err, 0, "{ s: s }", "ca", &ca); - if (ret) - throw ConfigError(json_ssl, err, "node-config-node-kafka-ssl", - "Failed to parse SSL configuration of node {}", - n->getName()); - - k->ssl.ca = strdup(ca); - } - - if (json_sasl) { - const char *mechanisms; - const char *username; - const char *password; - - ret = json_unpack_ex(json_sasl, &err, 0, "{ s: s, s: s, s: s }", - "mechanisms", &mechanisms, "username", &username, - "password", &password); - if (ret) - throw ConfigError(json_sasl, err, "node-config-node-kafka-sasl", - "Failed to parse SASL configuration"); - - k->sasl.mechanisms = strdup(mechanisms); - k->sasl.username = strdup(username); - k->sasl.password = strdup(password); - } - - // Format - if (k->formatter) - delete k->formatter; - k->formatter = json_format ? FormatFactory::make(json_format) - : FormatFactory::make("villas.binary"); - if (!k->formatter) - throw ConfigError(json_format, "node-config-node-kafka-format", - "Invalid format configuration"); - - return 0; -} - -int villas::node::kafka_prepare(NodeCompat *n) { - int ret; - auto *k = n->getData(); - - k->formatter->start(n->getInputSignals(false), ~(int)SampleFlags::HAS_OFFSET); - - ret = pool_init(&k->pool, 1024, - SAMPLE_LENGTH(n->getInputSignals(false)->size())); - if (ret) - return ret; - - ret = queue_signalled_init(&k->queue, 1024); - if (ret) - return ret; - - return 0; -} - -char *villas::node::kafka_print(NodeCompat *n) { - auto *k = n->getData(); - - char *buf = nullptr; - - strcatf(&buf, "bootstrap.server=%s, client.id=%s, security.protocol=%s", - k->server, k->client_id, k->protocol); - - // Only show if not default - if (k->produce) - strcatf(&buf, ", out.produce=%s", k->produce); - - if (k->consume) - strcatf(&buf, ", in.consume=%s", k->consume); - - return buf; -} - -int villas::node::kafka_destroy(NodeCompat *n) { - int ret; - auto *k = n->getData(); - - if (k->producer.client) - rd_kafka_destroy(k->producer.client); - - if (k->consumer.client) - rd_kafka_destroy(k->consumer.client); - - if (k->formatter) - delete k->formatter; - ret = pool_destroy(&k->pool); - if (ret) return ret; + } - ret = queue_signalled_destroy(&k->queue); - if (ret) - return ret; + int _write(struct Sample *smps[], unsigned cnt) override { + assert(producer.client != nullptr); - if (k->produce) - free(k->produce); + size_t wbytes; - if (k->consume) - free(k->consume); + char data[DEFAULT_FORMAT_BUFFER_LENGTH]; - if (k->protocol) - free(k->protocol); + auto ret = formatter->sprint(data, sizeof(data), &wbytes, smps, cnt); + if (ret < 0) + return ret; - if (k->client_id) - free(k->client_id); + if (!produce.empty()) { + auto ret = producer.client->produce( + producer.topic.get(), RdKafka::Topic::PARTITION_UA, + RdKafka::Producer::RK_MSG_COPY, data, wbytes, NULL, 0, NULL); + if (ret != RdKafka::ErrorCode::ERR_NO_ERROR) { + logger->warn("Publish failed"); + return -abs(ret); + } + } else + logger->warn( + "No produce possible because no produce topic is configured"); - free(k->server); + return cnt; + } - return 0; -} + int startProducer() { + std::string errstr; -int villas::node::kafka_start(NodeCompat *n) { - int ret; - char errstr[1024]; - auto *k = n->getData(); + auto conf_prod = createCommonConf(); + if (!conf_prod) + throw MemoryAllocationError(); - rd_kafka_conf_t *rdkconf = rd_kafka_conf_new(); - if (!rdkconf) - throw MemoryAllocationError(); + producer.client = std::unique_ptr( + RdKafka::Producer::create(conf_prod.get(), errstr)); + if (!producer.client) + throw RuntimeError("{}", errstr); - rd_kafka_conf_set_log_cb(rdkconf, kafka_logger_cb); + auto topic_conf = std::unique_ptr( + RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC)); + if (!topic_conf) + throw MemoryAllocationError(); - ret = rd_kafka_conf_set(rdkconf, "client.id", k->client_id, errstr, - sizeof(errstr)); - if (ret != RD_KAFKA_CONF_OK) - goto kafka_config_error; + auto cr = topic_conf->set("acks", "all", errstr); + if (cr != RdKafka::Conf::CONF_OK) + throw RuntimeError("{}", errstr); - ret = rd_kafka_conf_set(rdkconf, "bootstrap.servers", k->server, errstr, - sizeof(errstr)); - if (ret != RD_KAFKA_CONF_OK) - goto kafka_config_error; + producer.topic = std::unique_ptr(RdKafka::Topic::create( + producer.client.get(), produce, topic_conf.get(), errstr)); + if (!producer.topic) + throw MemoryAllocationError(); - ret = rd_kafka_conf_set(rdkconf, "security.protocol", k->protocol, errstr, - sizeof(errstr)); - if (ret != RD_KAFKA_CONF_OK) - goto kafka_config_error; + logger->info("Connected producer to bootstrap server {}", server); - if (!strcmp(k->protocol, "SASL_SSL") || !strcmp(k->protocol, "SSL")) { - ret = rd_kafka_conf_set(rdkconf, "ssl.ca.location", k->ssl.ca, errstr, - sizeof(errstr)); - if (ret != RD_KAFKA_CONF_OK) - goto kafka_config_error; + return 0; } - if (!strcmp(k->protocol, "SASL_PLAINTEXT") || - !strcmp(k->protocol, "SASL_SSL")) { - ret = rd_kafka_conf_set(rdkconf, "sasl.mechanisms", k->sasl.mechanisms, - errstr, sizeof(errstr)); - if (ret != RD_KAFKA_CONF_OK) - goto kafka_config_error; - - ret = rd_kafka_conf_set(rdkconf, "sasl.username", k->sasl.username, errstr, - sizeof(errstr)); - if (ret != RD_KAFKA_CONF_OK) - goto kafka_config_error; - - ret = rd_kafka_conf_set(rdkconf, "sasl.password", k->sasl.password, errstr, - sizeof(errstr)); - if (ret != RD_KAFKA_CONF_OK) - goto kafka_config_error; - } + int startConsumer() { + std::string errstr; - if (k->produce) { - // rd_kafka_new() will take ownership and free the passed rd_kafka_conf_t object, - // so we will need to create a copy first - rd_kafka_conf_t *rdkconf_prod = rd_kafka_conf_dup(rdkconf); - if (!rdkconf_prod) + auto conf_cons = createCommonConf(); + if (!conf_cons) throw MemoryAllocationError(); - k->producer.client = - rd_kafka_new(RD_KAFKA_PRODUCER, rdkconf_prod, errstr, sizeof(errstr)); - if (!k->producer.client) - goto kafka_config_error; + auto cr = conf_cons->set("group.id", group_id, errstr); + if (cr != RdKafka::Conf::CONF_OK) + throw RuntimeError("{}", errstr); - rd_kafka_topic_conf_t *topic_conf = rd_kafka_topic_conf_new(); - if (!topic_conf) + consumer.client = std::unique_ptr( + RdKafka::Consumer::create(conf_cons.get(), errstr)); + if (!consumer.client) throw MemoryAllocationError(); - ret = rd_kafka_topic_conf_set(topic_conf, "acks", "all", errstr, - sizeof(errstr)); - if (ret != RD_KAFKA_CONF_OK) - goto kafka_config_error; - - k->producer.topic = - rd_kafka_topic_new(k->producer.client, k->produce, topic_conf); - if (!k->producer.topic) + consumer.topic = std::unique_ptr(RdKafka::Topic::create( + consumer.client.get(), consume, nullptr, errstr)); + if (!consumer.topic) throw MemoryAllocationError(); - n->logger->info("Connected producer to bootstrap server {}", k->server); - } - - if (k->consume) { - // rd_kafka_new() will take ownership and free the passed rd_kafka_conf_t object, - // so we will need to create a copy first - rd_kafka_conf_t *rdkconf_cons = rd_kafka_conf_dup(rdkconf); - if (!rdkconf_cons) + consumer.queue = std::unique_ptr( + RdKafka::Queue::create(consumer.client.get())); + if (!consumer.queue) throw MemoryAllocationError(); - rd_kafka_topic_partition_list_t *partitions = - rd_kafka_topic_partition_list_new(1); - if (!partitions) - throw MemoryAllocationError(); + consumer.eventFd = eventfd(0, 0); - rd_kafka_topic_partition_t *partition = - rd_kafka_topic_partition_list_add(partitions, k->consume, 0); - if (!partition) - throw RuntimeError("Failed to add new partition"); + uint64_t incr = 1; + consumer.queue->io_event_enable(consumer.eventFd, &incr, sizeof(incr)); - ret = rd_kafka_conf_set(rdkconf_cons, "group.id", k->consumer.group_id, - errstr, sizeof(errstr)); - if (ret != RD_KAFKA_CONF_OK) - goto kafka_config_error; - - k->consumer.client = - rd_kafka_new(RD_KAFKA_CONSUMER, rdkconf_cons, errstr, sizeof(errstr)); - if (!k->consumer.client) - throw MemoryAllocationError(); + auto ec = consumer.client->start(consumer.topic.get(), 0, 0, + consumer.queue.get()); + if (ec != RdKafka::ErrorCode::ERR_NO_ERROR) + throw RuntimeError("Error subscribing to {} at {}: {}", consume, server, + RdKafka::err2str(ec)); - ret = rd_kafka_subscribe(k->consumer.client, partitions); - if (ret != RD_KAFKA_RESP_ERR_NO_ERROR) - throw RuntimeError("Error subscribing to {} at {}: {}", k->consume, - k->server, rd_kafka_err2str((rd_kafka_resp_err_t)ret)); + logger->info("Subscribed consumer from bootstrap server {}", server); - n->logger->info("Subscribed consumer from bootstrap server {}", k->server); + return 0; } - // Add client to global list of kafka clients - // so that thread can call kafka loop for this client - list_push(&clients, n); + std::unique_ptr createCommonConf() { + std::string errstr; - rd_kafka_conf_destroy(rdkconf); + auto conf = std::unique_ptr( + RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL)); + if (!conf) + throw MemoryAllocationError(); + + auto ret = conf->set("event_cb", this, errstr); + if (ret != RdKafka::Conf::CONF_OK) + throw RuntimeError("{}", errstr); - return 0; + ret = conf->set("client.id", client_id, errstr); + if (ret != RdKafka::Conf::CONF_OK) + throw RuntimeError("{}", errstr); -kafka_config_error: - rd_kafka_conf_destroy(rdkconf); + ret = conf->set("bootstrap.servers", server, errstr); + if (ret != RdKafka::Conf::CONF_OK) + throw RuntimeError("{}", errstr); - throw RuntimeError("{}", errstr); + ret = conf->set("security.protocol", protocol, errstr); + if (ret != RdKafka::Conf::CONF_OK) + throw RuntimeError("{}", errstr); - return -1; -} + if (protocol == "SASL_SSL" || protocol == "SSL") { + ret = conf->set("ssl.ca.location", ssl_ca, errstr); + if (ret != RdKafka::Conf::CONF_OK) + throw RuntimeError("{}", errstr); + } -int villas::node::kafka_stop(NodeCompat *n) { - int ret; - auto *k = n->getData(); + if (protocol == "SASL_PLAINTEXT" || protocol == "SASL_SSL") { + ret = conf->set("sasl.mechanisms", sasl.mechanisms, errstr); + if (ret != RdKafka::Conf::CONF_OK) + throw RuntimeError("{}", errstr); - if (k->producer.client) { - ret = rd_kafka_flush(k->producer.client, k->timeout * 1000); - if (ret != RD_KAFKA_RESP_ERR_NO_ERROR) - n->logger->error("Failed to flush messages: {}", - rd_kafka_err2str((rd_kafka_resp_err_t)ret)); + ret = conf->set("sasl.username", sasl.username, errstr); + if (ret != RdKafka::Conf::CONF_OK) + throw RuntimeError("{}", errstr); + + ret = conf->set("sasl.password", sasl.password, errstr); + if (ret != RdKafka::Conf::CONF_OK) + throw RuntimeError("{}", errstr); + } - /* If the output queue is still not empty there is an issue - * with producing messages to the clusters. */ - if (rd_kafka_outq_len(k->producer.client) > 0) - n->logger->warn("{} message(s) were not delivered", - rd_kafka_outq_len(k->producer.client)); + return conf; } - // Unregister client from global kafka client list - // so that kafka loop is no longer invoked for this client - // important to do that before disconnecting from broker, otherwise, kafka thread will attempt to reconnect - list_remove_all(&clients, n); +public: + KafkaNode(const uuid_t &id = {}, const std::string &name = "") + : Node(id, name), timeout(1000), client_id("villas-node"), producer({}), + consumer({.eventFd = -1}) {} - ret = queue_signalled_close(&k->queue); - if (ret) - return ret; + virtual ~KafkaNode() {} - return 0; -} + int prepare() override { + formatter->start(getInputSignals(false), ~(int)SampleFlags::HAS_OFFSET); -int villas::node::kafka_type_start(villas::node::SuperNode *sn) { - int ret; + return Node::prepare(); + } - logger = Log::get("node:kafka"); + int parse(json_t *json) override { + int ret = Node::parse(json); + if (ret) + return ret; - ret = list_init(&clients); - if (ret) - goto kafka_error; + Json config = json; - // Start thread here to run kafka loop for registered clients - ret = pthread_create(&thread, nullptr, kafka_loop_thread, nullptr); - if (ret) - goto kafka_error; + try { + server = config.at("server").get(); + protocol = config.at("protocol").get(); + client_id = config.value("client_id", client_id); - return 0; + if (auto out = config.find("out"); out != config.end()) + produce = out->value("topic", produce); -kafka_error: - logger->warn("Error initialazing node type kafka"); + if (auto in = config.find("in"); in != config.end()) { + consume = in->value("topic", consume); + group_id = in->value("group_id", group_id); + } - return ret; -} + if (auto to = config.value("timeout", -1.0); to >= 0) + timeout = std::chrono::milliseconds(static_cast(to * 1000)); -int villas::node::kafka_type_stop() { - int ret; + if (auto ssl = config.find("ssl"); ssl != config.end()) + ssl_ca = ssl->value("ca", ssl_ca); - // Stop thread here that executes kafka loop - ret = pthread_cancel(thread); - if (ret) - return ret; + if (auto sasl_cfg = config.find("sasl"); sasl_cfg != config.end()) { + sasl.mechanisms = sasl_cfg->at("mechanisms").get(); + sasl.username = sasl_cfg->at("username").get(); + sasl.password = sasl_cfg->at("password").get(); + } + } catch (Json::exception &e) { + throw ConfigError(json, "node-config-node-kafka", + "Failed to parse configuration: {}", e.what()); + } - logger->debug( - "Called pthread_cancel() on kafka communication management thread."); + if (protocol != "SSL" && protocol != "PLAINTEXT" && + protocol != "SASL_SSL" && protocol != "SASL_PLAINTEXT") + throw ConfigError(json, "node-config-node-kafka-protocol", + "Invalid security protocol: {}", protocol); + + if (produce.empty() && consume.empty()) + throw ConfigError(json, "node-config-node-kafka", + "At least one topic has to be specified for node {}", + getName()); + + // Format + json_t *json_format = json_object_get(json, "format"); + formatter = std::unique_ptr( + json_format ? FormatFactory::make(json_format) + : FormatFactory::make("villas.binary")); + if (!formatter) + throw ConfigError(json_format, "node-config-node-kafka-format", + "Invalid format configuration"); + + return 0; + } - ret = pthread_join(thread, nullptr); - if (ret) - goto kafka_error; + const std::string &getDetails() override { + details = fmt::format("server={}, client_id={}, protocol={}", server, + client_id, protocol); - // When this is called the list of clients should be empty - if (list_length(&clients) > 0) - throw RuntimeError( - "List of kafka clients contains elements at time of destruction. Call " - "node_stop for each kafka node before stopping node type!"); + if (!produce.empty()) + details += fmt::format(", produce={}", produce); - ret = list_destroy(&clients, nullptr, false); - if (ret) - goto kafka_error; + if (!consume.empty()) + details += fmt::format(", consume={}", consume); - return 0; + return details; + } -kafka_error: - logger->warn("Error stoping node type kafka"); + int start() override { + if (!produce.empty()) { + auto ret = startProducer(); + if (ret) + return ret; + } - return ret; -} + if (!consume.empty()) { + auto ret = startConsumer(); + if (ret) + return ret; + } -int villas::node::kafka_read(NodeCompat *n, struct Sample *const smps[], - unsigned cnt) { - int pulled; - auto *k = n->getData(); - struct Sample *smpt[cnt]; + int ret = Node::start(); + if (!ret) + state = State::STARTED; - pulled = queue_signalled_pull_many(&k->queue, (void **)smpt, cnt); + return 0; + } - sample_copy_many(smps, smpt, pulled); - sample_decref_many(smpt, pulled); + int stop() override { + int ret = Node::stop(); + if (ret) + return ret; + + if (producer.client) { + auto ret = producer.client->flush(timeout.count()); + if (ret != RdKafka::ErrorCode::ERR_NO_ERROR) + logger->error("Failed to flush messages: {}", RdKafka::err2str(ret)); + + // If the output queue is still not empty there is an issue + // with producing messages to the clusters. + if (producer.client->outq_len() > 0) + logger->warn("{} message(s) were not delivered", + producer.client->outq_len()); + } + + return 0; + } - return pulled; -} + int reverse() override { + std::swap(produce, consume); -int villas::node::kafka_write(NodeCompat *n, struct Sample *const smps[], - unsigned cnt) { - int ret; - auto *k = n->getData(); + return 0; + } - size_t wbytes; + std::vector getPollFDs() override { + if (consumer.eventFd >= 0) + return {consumer.eventFd}; - char data[DEFAULT_FORMAT_BUFFER_LENGTH]; + return {}; + } - ret = k->formatter->sprint(data, sizeof(data), &wbytes, smps, cnt); - if (ret < 0) - return ret; + void event_cb(RdKafka::Event &event) override { + switch (event.type()) { + case RdKafka::Event::EVENT_ERROR: + logger->error("Kafka error: {}", event.str()); + break; + + case RdKafka::Event::EVENT_STATS: + logger->info("Kafka stats: {}", event.str()); + break; + + case RdKafka::Event::EVENT_LOG: + switch (event.severity()) { + case RdKafka::Event::EVENT_SEVERITY_DEBUG: + logger->debug("{}", event.str()); + break; + + case RdKafka::Event::EVENT_SEVERITY_NOTICE: + case RdKafka::Event::EVENT_SEVERITY_INFO: + logger->info("{}", event.str()); + break; + + case RdKafka::Event::EVENT_SEVERITY_ALERT: + case RdKafka::Event::EVENT_SEVERITY_WARNING: + logger->warn("{}", event.str()); + break; + + case RdKafka::Event::EVENT_SEVERITY_ERROR: + case RdKafka::Event::EVENT_SEVERITY_CRITICAL: + case RdKafka::Event::EVENT_SEVERITY_EMERG: + logger->error("{}", event.str()); + break; + } - if (k->produce) { - ret = rd_kafka_produce(k->producer.topic, RD_KAFKA_PARTITION_UA, - RD_KAFKA_MSG_F_COPY, data, wbytes, NULL, 0, NULL); + break; - if (ret != RD_KAFKA_RESP_ERR_NO_ERROR) { - n->logger->warn("Publish failed"); - return -abs(ret); + default: + logger->info("Kafka event {}: {}", (int)event.type(), event.str()); + break; } - } else - n->logger->warn( - "No produce possible because no produce topic is configured"); - - return cnt; -} - -int villas::node::kafka_poll_fds(NodeCompat *n, int fds[]) { - auto *k = n->getData(); - - fds[0] = queue_signalled_fd(&k->queue); - - return 1; -} - -static NodeCompatType p; - -__attribute__((constructor(110))) static void register_plugin() { - p.name = "kafka"; - p.description = "Kafka event message streaming (rdkafka)"; - p.vectorize = 0; - p.size = sizeof(struct kafka); - p.type.start = kafka_type_start; - p.type.stop = kafka_type_stop; - p.destroy = kafka_destroy; - p.prepare = kafka_prepare; - p.parse = kafka_parse; - p.prepare = kafka_prepare; - p.print = kafka_print; - p.init = kafka_init; - p.destroy = kafka_destroy; - p.start = kafka_start; - p.stop = kafka_stop; - p.read = kafka_read; - p.write = kafka_write; - p.reverse = kafka_reverse; - p.poll_fds = kafka_poll_fds; - - static NodeCompatFactory ncp(&p); -} + } +}; + +// Register node +static char n[] = "kafka"; +static char d[] = "Kafka event message streaming (rdkafka)"; +static NodePlugin + p; diff --git a/lib/nodes/kafka_old.cpp b/lib/nodes/kafka_old.cpp new file mode 100644 index 000000000..1c56e1e5f --- /dev/null +++ b/lib/nodes/kafka_old.cpp @@ -0,0 +1,600 @@ +/* Node type: kafka. + * + * Author: Juan Pablo Noreña + * SPDX-FileCopyrightText: 2021 Universidad Nacional de Colombia + * SPDX-License-Identifier: Apache-2.0 + */ + +#include + +#include +#include + +#include +#include +#include +#include + +using namespace villas; +using namespace villas::node; +using namespace villas::utils; + +// Each process has a list of clients for which a thread invokes the kafka loop +static struct List clients; +static pthread_t thread; +static Logger logger; + +static void kafka_logger_cb(const rd_kafka_t *rk, int level, const char *fac, + const char *buf) { + + switch (level) { + case LOG_EMERG: + case LOG_CRIT: + case LOG_ERR: + logger->error("{}: {}", fac, buf); + break; + + case LOG_ALERT: + case LOG_WARNING: + logger->warn("{}: {}", fac, buf); + break; + + case LOG_DEBUG: + logger->debug("{}: {}", fac, buf); + break; + + case LOG_NOTICE: + case LOG_INFO: + default: + logger->info("{}: {}", fac, buf); + break; + } +} + +static void kafka_message_cb(void *ctx, const rd_kafka_message_t *msg) { + int ret; + auto *n = (NodeCompat *)ctx; + auto *k = n->getData(); + struct Sample *smps[n->in.vectorize]; + + n->logger->debug("Received a message of {} bytes from broker {}", msg->len, + k->server); + + ret = sample_alloc_many(&k->pool, smps, n->in.vectorize); + if (ret <= 0) { + n->logger->warn("Pool underrun in consumer"); + return; + } + + ret = k->formatter->sscan((char *)msg->payload, msg->len, nullptr, smps, + n->in.vectorize); + if (ret < 0) { + n->logger->warn("Received an invalid message"); + n->logger->warn(" Payload: {}", (char *)msg->payload); + return; + } + + if (ret == 0) { + n->logger->debug("Skip empty message"); + sample_decref_many(smps, n->in.vectorize); + return; + } + + ret = queue_signalled_push_many(&k->queue, (void **)smps, n->in.vectorize); + if (ret < (int)n->in.vectorize) + n->logger->warn("Failed to enqueue samples"); +} + +static void *kafka_loop_thread(void *ctx) { + int ret; + + // Set the cancel type of this thread to async + ret = pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, nullptr); + if (ret != 0) + throw RuntimeError("Unable to set cancel type of Kafka communication " + "thread to asynchronous."); + + while (true) { + for (unsigned i = 0; i < list_length(&clients); i++) { + auto *n = (NodeCompat *)list_at(&clients, i); + auto *k = n->getData(); + + // Execute kafka loop for this client + if (k->consumer.client) { + rd_kafka_message_t *msg = + rd_kafka_consumer_poll(k->consumer.client, k->timeout * 1000); + if (msg) { + kafka_message_cb((void *)n, msg); + rd_kafka_message_destroy(msg); + } + } + } + } + + return nullptr; +} + +int villas::node::kafka_reverse(NodeCompat *n) { + auto *k = n->getData(); + + SWAP(k->produce, k->consume); + + return 0; +} + +int villas::node::kafka_init(NodeCompat *n) { + auto *k = n->getData(); + + // Default values + k->server = nullptr; + k->protocol = nullptr; + k->produce = nullptr; + k->consume = nullptr; + k->client_id = nullptr; + k->timeout = 1.0; + + k->consumer.client = nullptr; + k->consumer.group_id = nullptr; + k->producer.client = nullptr; + k->producer.topic = nullptr; + + k->sasl.mechanisms = nullptr; + k->sasl.username = nullptr; + k->sasl.password = nullptr; + + k->ssl.ca = nullptr; + + k->formatter = nullptr; + + return 0; +} + +int villas::node::kafka_parse(NodeCompat *n, json_t *json) { + int ret; + auto *k = n->getData(); + + const char *server; + const char *produce = nullptr; + const char *consume = nullptr; + const char *protocol; + const char *client_id = "villas-node"; + const char *group_id = nullptr; + + json_error_t err; + json_t *json_ssl = nullptr; + json_t *json_sasl = nullptr; + json_t *json_format = nullptr; + + ret = json_unpack_ex(json, &err, 0, + "{ s?: { s?: s }, s?: { s?: s, s?: s }, s?: o, s: s, " + "s?: F, s: s, s?: s, s?: o, s?: o }", + "out", "produce", &produce, "in", "consume", &consume, + "group_id", &group_id, "format", &json_format, "server", + &server, "timeout", &k->timeout, "protocol", &protocol, + "client_id", &client_id, "ssl", &json_ssl, "sasl", + &json_sasl); + if (ret) + throw ConfigError(json, err, "node-config-node-kafka"); + + k->server = strdup(server); + k->produce = produce ? strdup(produce) : nullptr; + k->consume = consume ? strdup(consume) : nullptr; + k->protocol = strdup(protocol); + k->client_id = strdup(client_id); + k->consumer.group_id = group_id ? strdup(group_id) : nullptr; + + if (strcmp(protocol, "SSL") && strcmp(protocol, "PLAINTEXT") && + strcmp(protocol, "SASL_SSL") && strcmp(protocol, "SASL_PLAINTEXT")) + throw ConfigError(json, "node-config-node-kafka-protocol", + "Invalid security protocol: {}", protocol); + + if (!k->produce && !k->consume) + throw ConfigError(json, "node-config-node-kafka", + "At least one topic has to be specified for node {}", + n->getName()); + + if (json_ssl) { + const char *ca; + + ret = json_unpack_ex(json_ssl, &err, 0, "{ s: s }", "ca", &ca); + if (ret) + throw ConfigError(json_ssl, err, "node-config-node-kafka-ssl", + "Failed to parse SSL configuration of node {}", + n->getName()); + + k->ssl.ca = strdup(ca); + } + + if (json_sasl) { + const char *mechanisms; + const char *username; + const char *password; + + ret = json_unpack_ex(json_sasl, &err, 0, "{ s: s, s: s, s: s }", + "mechanisms", &mechanisms, "username", &username, + "password", &password); + if (ret) + throw ConfigError(json_sasl, err, "node-config-node-kafka-sasl", + "Failed to parse SASL configuration"); + + k->sasl.mechanisms = strdup(mechanisms); + k->sasl.username = strdup(username); + k->sasl.password = strdup(password); + } + + // Format + if (k->formatter) + delete k->formatter; + k->formatter = json_format ? FormatFactory::make(json_format) + : FormatFactory::make("villas.binary"); + if (!k->formatter) + throw ConfigError(json_format, "node-config-node-kafka-format", + "Invalid format configuration"); + + return 0; +} + +int villas::node::kafka_prepare(NodeCompat *n) { + int ret; + auto *k = n->getData(); + + k->formatter->start(n->getInputSignals(false), ~(int)SampleFlags::HAS_OFFSET); + + ret = pool_init(&k->pool, 1024, + SAMPLE_LENGTH(n->getInputSignals(false)->size())); + if (ret) + return ret; + + ret = queue_signalled_init(&k->queue, 1024); + if (ret) + return ret; + + return 0; +} + +char *villas::node::kafka_print(NodeCompat *n) { + auto *k = n->getData(); + + char *buf = nullptr; + + strcatf(&buf, "bootstrap.server=%s, client.id=%s, security.protocol=%s", + k->server, k->client_id, k->protocol); + + // Only show if not default + if (k->produce) + strcatf(&buf, ", out.produce=%s", k->produce); + + if (k->consume) + strcatf(&buf, ", in.consume=%s", k->consume); + + return buf; +} + +int villas::node::kafka_destroy(NodeCompat *n) { + int ret; + auto *k = n->getData(); + + if (k->producer.client) + rd_kafka_destroy(k->producer.client); + + if (k->consumer.client) + rd_kafka_destroy(k->consumer.client); + + if (k->formatter) + delete k->formatter; + + ret = pool_destroy(&k->pool); + if (ret) + return ret; + + ret = queue_signalled_destroy(&k->queue); + if (ret) + return ret; + + if (k->produce) + free(k->produce); + + if (k->consume) + free(k->consume); + + if (k->protocol) + free(k->protocol); + + if (k->client_id) + free(k->client_id); + + free(k->server); + + return 0; +} + +int villas::node::kafka_start(NodeCompat *n) { + int ret; + char errstr[1024]; + auto *k = n->getData(); + + rd_kafka_conf_t *rdkconf = rd_kafka_conf_new(); + if (!rdkconf) + throw MemoryAllocationError(); + + rd_kafka_conf_set_log_cb(rdkconf, kafka_logger_cb); + + ret = rd_kafka_conf_set(rdkconf, "client.id", k->client_id, errstr, + sizeof(errstr)); + if (ret != RD_KAFKA_CONF_OK) + goto kafka_config_error; + + ret = rd_kafka_conf_set(rdkconf, "bootstrap.servers", k->server, errstr, + sizeof(errstr)); + if (ret != RD_KAFKA_CONF_OK) + goto kafka_config_error; + + ret = rd_kafka_conf_set(rdkconf, "security.protocol", k->protocol, errstr, + sizeof(errstr)); + if (ret != RD_KAFKA_CONF_OK) + goto kafka_config_error; + + if (!strcmp(k->protocol, "SASL_SSL") || !strcmp(k->protocol, "SSL")) { + ret = rd_kafka_conf_set(rdkconf, "ssl.ca.location", k->ssl.ca, errstr, + sizeof(errstr)); + if (ret != RD_KAFKA_CONF_OK) + goto kafka_config_error; + } + + if (!strcmp(k->protocol, "SASL_PLAINTEXT") || + !strcmp(k->protocol, "SASL_SSL")) { + ret = rd_kafka_conf_set(rdkconf, "sasl.mechanisms", k->sasl.mechanisms, + errstr, sizeof(errstr)); + if (ret != RD_KAFKA_CONF_OK) + goto kafka_config_error; + + ret = rd_kafka_conf_set(rdkconf, "sasl.username", k->sasl.username, errstr, + sizeof(errstr)); + if (ret != RD_KAFKA_CONF_OK) + goto kafka_config_error; + + ret = rd_kafka_conf_set(rdkconf, "sasl.password", k->sasl.password, errstr, + sizeof(errstr)); + if (ret != RD_KAFKA_CONF_OK) + goto kafka_config_error; + } + + if (k->produce) { + // rd_kafka_new() will take ownership and free the passed rd_kafka_conf_t object, + // so we will need to create a copy first + rd_kafka_conf_t *rdkconf_prod = rd_kafka_conf_dup(rdkconf); + if (!rdkconf_prod) + throw MemoryAllocationError(); + + k->producer.client = + rd_kafka_new(RD_KAFKA_PRODUCER, rdkconf_prod, errstr, sizeof(errstr)); + if (!k->producer.client) + goto kafka_config_error; + + rd_kafka_topic_conf_t *topic_conf = rd_kafka_topic_conf_new(); + if (!topic_conf) + throw MemoryAllocationError(); + + ret = rd_kafka_topic_conf_set(topic_conf, "acks", "all", errstr, + sizeof(errstr)); + if (ret != RD_KAFKA_CONF_OK) + goto kafka_config_error; + + k->producer.topic = + rd_kafka_topic_new(k->producer.client, k->produce, topic_conf); + if (!k->producer.topic) + throw MemoryAllocationError(); + + n->logger->info("Connected producer to bootstrap server {}", k->server); + } + + if (k->consume) { + // rd_kafka_new() will take ownership and free the passed rd_kafka_conf_t object, + // so we will need to create a copy first + rd_kafka_conf_t *rdkconf_cons = rd_kafka_conf_dup(rdkconf); + if (!rdkconf_cons) + throw MemoryAllocationError(); + + rd_kafka_topic_partition_list_t *partitions = + rd_kafka_topic_partition_list_new(1); + if (!partitions) + throw MemoryAllocationError(); + + rd_kafka_topic_partition_t *partition = + rd_kafka_topic_partition_list_add(partitions, k->consume, 0); + if (!partition) + throw RuntimeError("Failed to add new partition"); + + ret = rd_kafka_conf_set(rdkconf_cons, "group.id", k->consumer.group_id, + errstr, sizeof(errstr)); + if (ret != RD_KAFKA_CONF_OK) + goto kafka_config_error; + + k->consumer.client = + rd_kafka_new(RD_KAFKA_CONSUMER, rdkconf_cons, errstr, sizeof(errstr)); + if (!k->consumer.client) + throw MemoryAllocationError(); + + ret = rd_kafka_subscribe(k->consumer.client, partitions); + if (ret != RD_KAFKA_RESP_ERR_NO_ERROR) + throw RuntimeError("Error subscribing to {} at {}: {}", k->consume, + k->server, rd_kafka_err2str((rd_kafka_resp_err_t)ret)); + + n->logger->info("Subscribed consumer from bootstrap server {}", k->server); + } + + // Add client to global list of kafka clients + // so that thread can call kafka loop for this client + list_push(&clients, n); + + rd_kafka_conf_destroy(rdkconf); + + return 0; + +kafka_config_error: + rd_kafka_conf_destroy(rdkconf); + + throw RuntimeError("{}", errstr); + + return -1; +} + +int villas::node::kafka_stop(NodeCompat *n) { + int ret; + auto *k = n->getData(); + + if (k->producer.client) { + ret = rd_kafka_flush(k->producer.client, k->timeout * 1000); + if (ret != RD_KAFKA_RESP_ERR_NO_ERROR) + n->logger->error("Failed to flush messages: {}", + rd_kafka_err2str((rd_kafka_resp_err_t)ret)); + + /* If the output queue is still not empty there is an issue + * with producing messages to the clusters. */ + if (rd_kafka_outq_len(k->producer.client) > 0) + n->logger->warn("{} message(s) were not delivered", + rd_kafka_outq_len(k->producer.client)); + } + + // Unregister client from global kafka client list + // so that kafka loop is no longer invoked for this client + // important to do that before disconnecting from broker, otherwise, kafka thread will attempt to reconnect + list_remove_all(&clients, n); + + ret = queue_signalled_close(&k->queue); + if (ret) + return ret; + + return 0; +} + +int villas::node::kafka_type_start(villas::node::SuperNode *sn) { + int ret; + + logger = Log::get("node:kafka"); + + ret = list_init(&clients); + if (ret) + goto kafka_error; + + // Start thread here to run kafka loop for registered clients + ret = pthread_create(&thread, nullptr, kafka_loop_thread, nullptr); + if (ret) + goto kafka_error; + + return 0; + +kafka_error: + logger->warn("Error initialazing node type kafka"); + + return ret; +} + +int villas::node::kafka_type_stop() { + int ret; + + // Stop thread here that executes kafka loop + ret = pthread_cancel(thread); + if (ret) + return ret; + + logger->debug( + "Called pthread_cancel() on kafka communication management thread."); + + ret = pthread_join(thread, nullptr); + if (ret) + goto kafka_error; + + // When this is called the list of clients should be empty + if (list_length(&clients) > 0) + throw RuntimeError( + "List of kafka clients contains elements at time of destruction. Call " + "node_stop for each kafka node before stopping node type!"); + + ret = list_destroy(&clients, nullptr, false); + if (ret) + goto kafka_error; + + return 0; + +kafka_error: + logger->warn("Error stoping node type kafka"); + + return ret; +} + +int villas::node::kafka_read(NodeCompat *n, struct Sample *const smps[], + unsigned cnt) { + int pulled; + auto *k = n->getData(); + struct Sample *smpt[cnt]; + + pulled = queue_signalled_pull_many(&k->queue, (void **)smpt, cnt); + + sample_copy_many(smps, smpt, pulled); + sample_decref_many(smpt, pulled); + + return pulled; +} + +int villas::node::kafka_write(NodeCompat *n, struct Sample *const smps[], + unsigned cnt) { + int ret; + auto *k = n->getData(); + + size_t wbytes; + + char data[DEFAULT_FORMAT_BUFFER_LENGTH]; + + ret = k->formatter->sprint(data, sizeof(data), &wbytes, smps, cnt); + if (ret < 0) + return ret; + + if (k->produce) { + ret = rd_kafka_produce(k->producer.topic, RD_KAFKA_PARTITION_UA, + RD_KAFKA_MSG_F_COPY, data, wbytes, NULL, 0, NULL); + + if (ret != RD_KAFKA_RESP_ERR_NO_ERROR) { + n->logger->warn("Publish failed"); + return -abs(ret); + } + } else + n->logger->warn( + "No produce possible because no produce topic is configured"); + + return cnt; +} + +int villas::node::kafka_poll_fds(NodeCompat *n, int fds[]) { + auto *k = n->getData(); + + fds[0] = queue_signalled_fd(&k->queue); + + return 1; +} + +static NodeCompatType p; + +__attribute__((constructor(110))) static void register_plugin() { + p.name = "kafka"; + p.description = "Kafka event message streaming (rdkafka)"; + p.vectorize = 0; + p.size = sizeof(struct kafka); + p.type.start = kafka_type_start; + p.type.stop = kafka_type_stop; + p.destroy = kafka_destroy; + p.prepare = kafka_prepare; + p.parse = kafka_parse; + p.prepare = kafka_prepare; + p.print = kafka_print; + p.init = kafka_init; + p.destroy = kafka_destroy; + p.start = kafka_start; + p.stop = kafka_stop; + p.read = kafka_read; + p.write = kafka_write; + p.reverse = kafka_reverse; + p.poll_fds = kafka_poll_fds; + + static NodeCompatFactory ncp(&p); +}