fluentd_access_logger: check cluster existance (envoyproxy#35138)

Additional Description: If a cluster does not exist at the time of
logger creation, Envoy would crash. When running in validation mode, the
cluster manager returns ``nullptr`` by design, which causes the process
to crash. Fixes envoyproxy#35098
Risk Level: low
Testing: unit tests
Docs Changes: none
Release Notes: none
Platform Specific Features: none

Signed-off-by: Ohad Vano <ohadvano@gmail.com>

source/extensions/access_loggers/fluentd/fluentd_access_log_impl.cc

@@ -167,6 +167,10 @@ FluentdAccessLoggerCacheImpl::getOrCreateLogger(const FluentdAccessLogConfigShar
   }
 
   auto* cluster = cluster_manager_.getThreadLocalCluster(config->cluster());
+  if (!cluster) {
+    return nullptr;
+  }
+
   auto client =
       cluster->tcpAsyncClient(nullptr, std::make_shared<const Tcp::AsyncTcpClientOptions>(false));
 

test/extensions/access_loggers/fluentd/fluentd_access_log_impl_test.cc

@@ -325,6 +325,21 @@ class FluentdAccessLoggerCacheImplTest : public testing::Test {
   std::unique_ptr<FluentdAccessLoggerCacheImpl> logger_cache_;
 };
 
+TEST_F(FluentdAccessLoggerCacheImplTest, CreateLoggerWhenClusterNotFound) {
+  tls_.setDispatcher(&dispatcher_);
+  logger_cache_ = std::make_unique<FluentdAccessLoggerCacheImpl>(cluster_manager_, scope_, tls_);
+  EXPECT_CALL(cluster_manager_, getThreadLocalCluster(cluster_name_)).WillOnce(Return(nullptr));
+
+  envoy::extensions::access_loggers::fluentd::v3::FluentdAccessLogConfig config;
+  config.set_cluster(cluster_name_);
+  config.set_tag("test.tag");
+  config.mutable_buffer_size_bytes()->set_value(123);
+  auto logger =
+      logger_cache_->getOrCreateLogger(std::make_shared<FluentdAccessLogConfig>(config), random_);
+
+  EXPECT_TRUE(logger == nullptr);
+}
+
 TEST_F(FluentdAccessLoggerCacheImplTest, CreateNonExistingLogger) {
   init();
   EXPECT_CALL(cluster_manager_, getThreadLocalCluster(cluster_name_)).WillOnce(Return(&cluster_));