diff --git a/contrib/kafka/filters/network/source/kafka_types.h b/contrib/kafka/filters/network/source/kafka_types.h index d01c304984e4c..36ce3b430b83b 100644 --- a/contrib/kafka/filters/network/source/kafka_types.h +++ b/contrib/kafka/filters/network/source/kafka_types.h @@ -2,6 +2,7 @@ #include #include +#include #include #include "absl/types/optional.h" @@ -45,6 +46,11 @@ struct Uuid { bool operator==(const Uuid& rhs) const { return msb_ == rhs.msb_ && lsb_ == rhs.lsb_; }; }; +/** + * Kafka topic-partition pair. + */ +using KafkaPartition = std::pair; + } // namespace Kafka } // namespace NetworkFilters } // namespace Extensions diff --git a/contrib/kafka/filters/network/source/mesh/BUILD b/contrib/kafka/filters/network/source/mesh/BUILD index 71546a58bef01..cfc0b024924f2 100644 --- a/contrib/kafka/filters/network/source/mesh/BUILD +++ b/contrib/kafka/filters/network/source/mesh/BUILD @@ -110,6 +110,39 @@ envoy_cc_library( ], ) +envoy_cc_library( + name = "shared_consumer_manager_lib", + srcs = [ + ], + hdrs = [ + "shared_consumer_manager.h", + ], + tags = ["skip_on_windows"], + deps = [ + ":upstream_kafka_consumer_lib", + "//source/common/common:minimal_logger_lib", + ], +) + +envoy_cc_library( + name = "shared_consumer_manager_impl_lib", + srcs = [ + "shared_consumer_manager_impl.cc", + ], + hdrs = [ + "shared_consumer_manager_impl.h", + ], + tags = ["skip_on_windows"], + deps = [ + ":librdkafka_utils_lib", + ":shared_consumer_manager_lib", + ":upstream_config_lib", + ":upstream_kafka_consumer_impl_lib", + "//contrib/kafka/filters/network/source:kafka_types_lib", + "//source/common/common:minimal_logger_lib", + ], +) + envoy_cc_library( name = "inbound_record_lib", srcs = [ diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h new file mode 100644 index 0000000000000..98eac5cad8699 --- /dev/null +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h @@ -0,0 +1,32 @@ +#pragma once + +#include +#include + +#include "contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h" + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Kafka { +namespace Mesh { + +/** + * Manages (raw) Kafka consumers pointing to upstream Kafka clusters. + * It is expected to have only one instance of this object per mesh-filter type. + */ +class SharedConsumerManager { +public: + virtual ~SharedConsumerManager() = default; + + // Start the consumer (if there is none) to make sure that records can be received from the topic. + virtual void registerConsumerIfAbsent(const std::string& topic) PURE; +}; + +using SharedConsumerManagerPtr = std::unique_ptr; + +} // namespace Mesh +} // namespace Kafka +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc new file mode 100644 index 0000000000000..1550c16cb0569 --- /dev/null +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.cc @@ -0,0 +1,102 @@ +#include "contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h" + +#include "source/common/common/fmt.h" + +#include "contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer_impl.h" + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Kafka { +namespace Mesh { + +// KafkaConsumerFactoryImpl + +class KafkaConsumerFactoryImpl : public KafkaConsumerFactory { +public: + // KafkaConsumerFactory + KafkaConsumerPtr createConsumer(InboundRecordProcessor& record_processor, + Thread::ThreadFactory& thread_factory, const std::string& topic, + const int32_t partition_count, + const RawKafkaConfig& configuration) const override; + + // Default singleton accessor. + static const KafkaConsumerFactory& getDefaultInstance(); +}; + +KafkaConsumerPtr +KafkaConsumerFactoryImpl::createConsumer(InboundRecordProcessor& record_processor, + Thread::ThreadFactory& thread_factory, + const std::string& topic, const int32_t partition_count, + const RawKafkaConfig& configuration) const { + return std::make_unique(record_processor, thread_factory, topic, + partition_count, configuration); +} + +const KafkaConsumerFactory& KafkaConsumerFactoryImpl::getDefaultInstance() { + CONSTRUCT_ON_FIRST_USE(KafkaConsumerFactoryImpl); +} + +// SharedConsumerManagerImpl + +SharedConsumerManagerImpl::SharedConsumerManagerImpl( + const UpstreamKafkaConfiguration& configuration, Thread::ThreadFactory& thread_factory) + : SharedConsumerManagerImpl{configuration, thread_factory, + KafkaConsumerFactoryImpl::getDefaultInstance()} {} + +SharedConsumerManagerImpl::SharedConsumerManagerImpl( + const UpstreamKafkaConfiguration& configuration, Thread::ThreadFactory& thread_factory, + const KafkaConsumerFactory& consumer_factory) + : distributor_{std::make_unique()}, configuration_{configuration}, + thread_factory_{thread_factory}, consumer_factory_{consumer_factory} {} + +void SharedConsumerManagerImpl::registerConsumerIfAbsent(const std::string& topic) { + absl::MutexLock lock(&consumers_mutex_); + const auto it = topic_to_consumer_.find(topic); + // Return consumer already present or create new one and register it. + if (topic_to_consumer_.end() == it) { + registerNewConsumer(topic); + } +} + +void SharedConsumerManagerImpl::registerNewConsumer(const std::string& topic) { + ENVOY_LOG(debug, "Creating consumer for topic [{}]", topic); + + // Compute which upstream cluster corresponds to the topic. + const absl::optional cluster_config = + configuration_.computeClusterConfigForTopic(topic); + if (!cluster_config) { + throw EnvoyException( + fmt::format("Could not compute upstream cluster configuration for topic [{}]", topic)); + } + + // Create the consumer and register it. + KafkaConsumerPtr new_consumer = consumer_factory_.createConsumer( + *distributor_, thread_factory_, topic, cluster_config->partition_count_, + cluster_config->upstream_consumer_properties_); + ENVOY_LOG(debug, "Registering new Kafka consumer for topic [{}], consuming from cluster [{}]", + topic, cluster_config->name_); + topic_to_consumer_.emplace(topic, std::move(new_consumer)); +} + +size_t SharedConsumerManagerImpl::getConsumerCountForTest() const { + absl::MutexLock lock(&consumers_mutex_); + return topic_to_consumer_.size(); +} + +// RecordDistributor + +bool RecordDistributor::waitUntilInterest(const std::string&, const int32_t) const { + // TODO (adam.kotwasinski) To implement in future commits. + return false; +} + +void RecordDistributor::receive(InboundRecordSharedPtr) { + // TODO (adam.kotwasinski) To implement in future commits. +} + +} // namespace Mesh +} // namespace Kafka +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy diff --git a/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h new file mode 100644 index 0000000000000..849d4076442cf --- /dev/null +++ b/contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h @@ -0,0 +1,92 @@ +#pragma once + +#include + +#include "envoy/thread/thread.h" + +#include "source/common/common/logger.h" + +#include "contrib/kafka/filters/network/source/kafka_types.h" +#include "contrib/kafka/filters/network/source/mesh/librdkafka_utils.h" +#include "contrib/kafka/filters/network/source/mesh/shared_consumer_manager.h" +#include "contrib/kafka/filters/network/source/mesh/upstream_config.h" +#include "contrib/kafka/filters/network/source/mesh/upstream_kafka_consumer.h" + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Kafka { +namespace Mesh { + +/** + * Placeholder interface for now. + * In future: + * Processor implementation that stores received records (that had no interest), and callbacks + * waiting for records (that had no matching records delivered yet). + * Basically core of Fetch-handling business logic. + */ +class RecordDistributor : public InboundRecordProcessor { +public: + // InboundRecordProcessor + bool waitUntilInterest(const std::string& topic, const int32_t timeout_ms) const override; + + // InboundRecordProcessor + void receive(InboundRecordSharedPtr message) override; +}; + +using RecordDistributorPtr = std::unique_ptr; + +/** + * Injectable for tests. + */ +class KafkaConsumerFactory { +public: + virtual ~KafkaConsumerFactory() = default; + + // Create a Kafka consumer. + virtual KafkaConsumerPtr createConsumer(InboundRecordProcessor& record_processor, + Thread::ThreadFactory& thread_factory, + const std::string& topic, const int32_t partition_count, + const RawKafkaConfig& configuration) const PURE; +}; + +/** + * Maintains a collection of Kafka consumers (one per topic). + */ +class SharedConsumerManagerImpl : public SharedConsumerManager, + private Logger::Loggable { +public: + // Main constructor. + SharedConsumerManagerImpl(const UpstreamKafkaConfiguration& configuration, + Thread::ThreadFactory& thread_factory); + + // Visible for testing. + SharedConsumerManagerImpl(const UpstreamKafkaConfiguration& configuration, + Thread::ThreadFactory& thread_factory, + const KafkaConsumerFactory& consumer_factory); + + // SharedConsumerManager + void registerConsumerIfAbsent(const std::string& topic) override; + + size_t getConsumerCountForTest() const; + +private: + // Mutates 'topic_to_consumer_'. + void registerNewConsumer(const std::string& topic) + ABSL_EXCLUSIVE_LOCKS_REQUIRED(consumers_mutex_); + + RecordDistributorPtr distributor_; + + const UpstreamKafkaConfiguration& configuration_; + Thread::ThreadFactory& thread_factory_; + const KafkaConsumerFactory& consumer_factory_; + + mutable absl::Mutex consumers_mutex_; + std::map topic_to_consumer_ ABSL_GUARDED_BY(consumers_mutex_); +}; + +} // namespace Mesh +} // namespace Kafka +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy diff --git a/contrib/kafka/filters/network/source/mesh/upstream_config.cc b/contrib/kafka/filters/network/source/mesh/upstream_config.cc index 8e6917df034c8..48b1fefe31b95 100644 --- a/contrib/kafka/filters/network/source/mesh/upstream_config.cc +++ b/contrib/kafka/filters/network/source/mesh/upstream_config.cc @@ -16,6 +16,8 @@ using KafkaClusterDefinition = envoy::extensions::filters::network::kafka_mesh::v3alpha::KafkaClusterDefinition; using ForwardingRule = envoy::extensions::filters::network::kafka_mesh::v3alpha::ForwardingRule; +const std::string DEFAULT_CONSUMER_GROUP_ID = "envoy"; + UpstreamKafkaConfigurationImpl::UpstreamKafkaConfigurationImpl(const KafkaMeshProtoConfig& config) : advertised_address_{config.advertised_host(), config.advertised_port()} { @@ -43,8 +45,15 @@ UpstreamKafkaConfigurationImpl::UpstreamKafkaConfigurationImpl(const KafkaMeshPr upstream_cluster_definition.producer_config().begin(), upstream_cluster_definition.producer_config().end()}; producer_configs["bootstrap.servers"] = upstream_cluster_definition.bootstrap_servers(); + + // TODO (adam.kotwasinski) This needs to read configs just like producer does. + std::map consumer_configs = {}; + consumer_configs["bootstrap.servers"] = upstream_cluster_definition.bootstrap_servers(); + // TODO (adam.kotwasinski) When configs are read, use this only if absent. + consumer_configs["group.id"] = DEFAULT_CONSUMER_GROUP_ID; + ClusterConfig cluster_config = {cluster_name, upstream_cluster_definition.partition_count(), - producer_configs}; + producer_configs, consumer_configs}; cluster_name_to_cluster_config[cluster_name] = cluster_config; } diff --git a/contrib/kafka/filters/network/source/mesh/upstream_config.h b/contrib/kafka/filters/network/source/mesh/upstream_config.h index ad49f2f1304d6..a8b14084ce9d7 100644 --- a/contrib/kafka/filters/network/source/mesh/upstream_config.h +++ b/contrib/kafka/filters/network/source/mesh/upstream_config.h @@ -39,9 +39,14 @@ struct ClusterConfig { // producer property. std::map upstream_producer_properties_; + // This map always contains entries with keys 'bootstrap.servers' and 'group.id', as these are the + // only mandatory consumer properties. + std::map upstream_consumer_properties_; + bool operator==(const ClusterConfig& rhs) const { return name_ == rhs.name_ && partition_count_ == rhs.partition_count_ && - upstream_producer_properties_ == rhs.upstream_producer_properties_; + upstream_producer_properties_ == rhs.upstream_producer_properties_ && + upstream_consumer_properties_ == rhs.upstream_consumer_properties_; } }; diff --git a/contrib/kafka/filters/network/test/mesh/BUILD b/contrib/kafka/filters/network/test/mesh/BUILD index 47c556411c0c2..ebb23eba960bd 100644 --- a/contrib/kafka/filters/network/test/mesh/BUILD +++ b/contrib/kafka/filters/network/test/mesh/BUILD @@ -74,6 +74,15 @@ envoy_cc_test( ], ) +envoy_cc_test( + name = "shared_consumer_manager_impl_unit_test", + srcs = ["shared_consumer_manager_impl_unit_test.cc"], + tags = ["skip_on_windows"], + deps = [ + "//contrib/kafka/filters/network/source/mesh:shared_consumer_manager_impl_lib", + ], +) + envoy_cc_test( name = "upstream_kafka_consumer_impl_unit_test", srcs = ["upstream_kafka_consumer_impl_unit_test.cc"], diff --git a/contrib/kafka/filters/network/test/mesh/command_handlers/metadata_unit_test.cc b/contrib/kafka/filters/network/test/mesh/command_handlers/metadata_unit_test.cc index ba144e3ce74c2..132946cd84019 100644 --- a/contrib/kafka/filters/network/test/mesh/command_handlers/metadata_unit_test.cc +++ b/contrib/kafka/filters/network/test/mesh/command_handlers/metadata_unit_test.cc @@ -33,7 +33,7 @@ TEST(MetadataTest, shouldBeAlwaysReadyForAnswer) { const std::pair advertised_address = {"host", 1234}; EXPECT_CALL(configuration, getAdvertisedAddress()).WillOnce(Return(advertised_address)); // First topic is going to have configuration present (42 partitions for each topic). - const ClusterConfig topic1config = {"", 42, {}}; + const ClusterConfig topic1config = {"", 42, {}, {}}; EXPECT_CALL(configuration, computeClusterConfigForTopic("topic1")) .WillOnce(Return(absl::make_optional(topic1config))); // Second topic is not going to have configuration present. diff --git a/contrib/kafka/filters/network/test/mesh/shared_consumer_manager_impl_unit_test.cc b/contrib/kafka/filters/network/test/mesh/shared_consumer_manager_impl_unit_test.cc new file mode 100644 index 0000000000000..2ec7f57b5f342 --- /dev/null +++ b/contrib/kafka/filters/network/test/mesh/shared_consumer_manager_impl_unit_test.cc @@ -0,0 +1,100 @@ +#include "envoy/common/exception.h" +#include "envoy/thread/thread.h" + +#include "contrib/kafka/filters/network/source/mesh/shared_consumer_manager_impl.h" +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +namespace Envoy { +namespace Extensions { +namespace NetworkFilters { +namespace Kafka { +namespace Mesh { +namespace { + +using testing::_; +using testing::Return; + +class MockThreadFactory : public Thread::ThreadFactory { +public: + MOCK_METHOD(Thread::ThreadPtr, createThread, (std::function, Thread::OptionsOptConstRef)); + MOCK_METHOD(Thread::ThreadId, currentThreadId, ()); +}; + +class MockUpstreamKafkaConfiguration : public UpstreamKafkaConfiguration { +public: + MOCK_METHOD(absl::optional, computeClusterConfigForTopic, (const std::string&), + (const)); + MOCK_METHOD((std::pair), getAdvertisedAddress, (), (const)); +}; + +class MockKafkaConsumerFactory : public KafkaConsumerFactory { +public: + MOCK_METHOD(KafkaConsumerPtr, createConsumer, + (InboundRecordProcessor&, Thread::ThreadFactory&, const std::string&, const int32_t, + const RawKafkaConfig&), + (const)); +}; + +class SharedConsumerManagerTest : public testing::Test { +protected: + MockThreadFactory thread_factory_; + MockUpstreamKafkaConfiguration configuration_; + MockKafkaConsumerFactory consumer_factory_; + + std::unique_ptr makeTestee() { + return std::make_unique(configuration_, thread_factory_, + consumer_factory_); + } +}; + +TEST_F(SharedConsumerManagerTest, ShouldRegisterTopicOnlyOnce) { + // given + const std::string topic1 = "topic1"; + const std::string topic2 = "topic2"; + + const ClusterConfig cluster_config = {"cluster", 1, {}, {}}; + EXPECT_CALL(configuration_, computeClusterConfigForTopic(topic1)) + .WillOnce(Return(cluster_config)); + EXPECT_CALL(configuration_, computeClusterConfigForTopic(topic2)) + .WillOnce(Return(cluster_config)); + + EXPECT_CALL(consumer_factory_, createConsumer(_, _, _, _, _)).Times(2).WillRepeatedly([]() { + return nullptr; + }); + + auto testee = makeTestee(); + + // when + for (int i = 0; i < 3; ++i) { + testee->registerConsumerIfAbsent(topic1); + } + for (int i = 0; i < 3; ++i) { + testee->registerConsumerIfAbsent(topic2); + } + + // then + ASSERT_EQ(testee->getConsumerCountForTest(), 2); +} + +TEST_F(SharedConsumerManagerTest, ShouldHandleMissingConfig) { + // given + const std::string topic = "topic"; + + EXPECT_CALL(configuration_, computeClusterConfigForTopic(topic)).WillOnce(Return(absl::nullopt)); + + EXPECT_CALL(consumer_factory_, createConsumer(_, _, _, _, _)).Times(0); + + auto testee = makeTestee(); + + // when, then - construction throws and nothing gets registered. + EXPECT_THROW(testee->registerConsumerIfAbsent(topic), EnvoyException); + ASSERT_EQ(testee->getConsumerCountForTest(), 0); +} + +} // namespace +} // namespace Mesh +} // namespace Kafka +} // namespace NetworkFilters +} // namespace Extensions +} // namespace Envoy diff --git a/contrib/kafka/filters/network/test/mesh/upstream_config_unit_test.cc b/contrib/kafka/filters/network/test/mesh/upstream_config_unit_test.cc index 23bfb039b9a03..3ecb0b5fd6792 100644 --- a/contrib/kafka/filters/network/test/mesh/upstream_config_unit_test.cc +++ b/contrib/kafka/filters/network/test/mesh/upstream_config_unit_test.cc @@ -108,8 +108,14 @@ advertised_port: 42 TestUtility::loadFromYamlAndValidate(yaml, proto_config); const UpstreamKafkaConfiguration& testee = UpstreamKafkaConfigurationImpl{proto_config}; - const ClusterConfig cluster1 = {"cluster1", 1, {{"bootstrap.servers", "s1"}}}; - const ClusterConfig cluster2 = {"cluster2", 2, {{"bootstrap.servers", "s2"}}}; + const ClusterConfig cluster1 = {"cluster1", + 1, + {{"bootstrap.servers", "s1"}}, + {{"bootstrap.servers", "s1"}, {"group.id", "envoy"}}}; + const ClusterConfig cluster2 = {"cluster2", + 2, + {{"bootstrap.servers", "s2"}}, + {{"bootstrap.servers", "s2"}, {"group.id", "envoy"}}}; // when, then (advertised address is returned properly) const auto address = testee.getAdvertisedAddress(); diff --git a/contrib/kafka/filters/network/test/mesh/upstream_kafka_facade_unit_test.cc b/contrib/kafka/filters/network/test/mesh/upstream_kafka_facade_unit_test.cc index d4a0497c21476..baa6c55baf538 100644 --- a/contrib/kafka/filters/network/test/mesh/upstream_kafka_facade_unit_test.cc +++ b/contrib/kafka/filters/network/test/mesh/upstream_kafka_facade_unit_test.cc @@ -36,7 +36,8 @@ TEST(UpstreamKafkaFacadeTest, shouldCreateProducerOnlyOnceForTheSameCluster) { const std::string topic2 = "topic2"; MockUpstreamKafkaConfiguration configuration; - const ClusterConfig cluster_config = {"cluster", 1, {{"bootstrap.servers", "localhost:9092"}}}; + const ClusterConfig cluster_config = { + "cluster", 1, {{"bootstrap.servers", "localhost:9092"}}, {}}; EXPECT_CALL(configuration, computeClusterConfigForTopic(topic1)).WillOnce(Return(cluster_config)); EXPECT_CALL(configuration, computeClusterConfigForTopic(topic2)).WillOnce(Return(cluster_config)); ThreadLocal::MockInstance slot_allocator; @@ -61,10 +62,12 @@ TEST(UpstreamKafkaFacadeTest, shouldCreateDifferentProducersForDifferentClusters MockUpstreamKafkaConfiguration configuration; // Notice it's the cluster name that matters, not the producer config. - const ClusterConfig cluster_config1 = {"cluster1", 1, {{"bootstrap.servers", "localhost:9092"}}}; + const ClusterConfig cluster_config1 = { + "cluster1", 1, {{"bootstrap.servers", "localhost:9092"}}, {}}; EXPECT_CALL(configuration, computeClusterConfigForTopic(topic1)) .WillOnce(Return(cluster_config1)); - const ClusterConfig cluster_config2 = {"cluster2", 1, {{"bootstrap.servers", "localhost:9092"}}}; + const ClusterConfig cluster_config2 = { + "cluster2", 1, {{"bootstrap.servers", "localhost:9092"}}, {}}; EXPECT_CALL(configuration, computeClusterConfigForTopic(topic2)) .WillOnce(Return(cluster_config2)); ThreadLocal::MockInstance slot_allocator; @@ -87,7 +90,8 @@ TEST(UpstreamKafkaFacadeTest, shouldThrowIfThereIsNoConfigurationForGivenTopic) const std::string topic = "topic1"; MockUpstreamKafkaConfiguration configuration; - const ClusterConfig cluster_config = {"cluster", 1, {{"bootstrap.servers", "localhost:9092"}}}; + const ClusterConfig cluster_config = { + "cluster", 1, {{"bootstrap.servers", "localhost:9092"}}, {}}; EXPECT_CALL(configuration, computeClusterConfigForTopic(topic)).WillOnce(Return(absl::nullopt)); ThreadLocal::MockInstance slot_allocator; EXPECT_CALL(slot_allocator, allocateSlot())