: cluster_manager_(cluster_manager), http_service_(http_service),

      headers_applicator_(std::move(headers_applicator)),

      buffer_flush_interval_(getBufferFlushInterval(config)),

      max_buffer_size_bytes_(getBufferSizeBytes(config)),

      stats_({ALL_OTLP_ACCESS_LOG_STATS(

          POOL_COUNTER_PREFIX(server_context.serverScope(),

                              absl::StrCat(OtlpAccessLogStatsPrefix, config.stat_prefix())))}) {

  root_ = initOtlpMessageRoot(message_, config, server_context.localInfo());

  flush_timer_ = dispatcher.createTimer([this]() {

    flush();

    flush_timer_->enableTimer(buffer_flush_interval_);

  });

  flush_timer_->enableTimer(buffer_flush_interval_);

}

void HttpAccessLoggerImpl::log(opentelemetry::proto::logs::v1::LogRecord&& entry) {

  approximate_message_size_bytes_ += entry.ByteSizeLong();

  batched_log_entries_++;

  root_->mutable_log_records()->Add(std::move(entry));

  if (approximate_message_size_bytes_ >= max_buffer_size_bytes_) {

    flush();

  }

}

void HttpAccessLoggerImpl::flush() {

  if (root_->log_records().empty()) {

    return;

  }

  std::string request_body;

  const auto ok = message_.SerializeToString(&request_body);

  if (!ok) {

  const auto thread_local_cluster =

      cluster_manager_.getThreadLocalCluster(http_service_.http_uri().cluster());

  if (thread_local_cluster == nullptr) {

    ENVOY_LOG(error, "OTLP HTTP access log exporter failed: [cluster = {}] is not configured",

              http_service_.http_uri().cluster());

    root_->clear_log_records();

    approximate_message_size_bytes_ = 0;

    return;

  }

  Http::RequestMessagePtr message = Http::Utility::prepareHeaders(http_service_.http_uri());

  message->headers().setReferenceMethod(Http::Headers::get().MethodValues.Post);

  message->headers().setReferenceContentType(Http::Headers::get().ContentTypeValues.Protobuf);

  message->headers().setReferenceUserAgent(getOtlpUserAgentHeader());

  headers_applicator_->apply(message->headers());

  message->body().add(request_body);

  const auto options =

      Http::AsyncClient::RequestOptions()

          .setTimeout(std::chrono::milliseconds(

              DurationUtil::durationToMilliseconds(http_service_.http_uri().timeout())))

          .setDiscardResponseBody(true);

  Http::AsyncClient::Request* in_flight_request =

      thread_local_cluster->httpAsyncClient().send(std::move(message), *this, options);

  if (in_flight_request != nullptr) {

    active_requests_.add(*in_flight_request);

    in_flight_log_entries_ = batched_log_entries_;

  } else {

    stats_.logs_dropped_.add(batched_log_entries_);

  }

  root_->clear_log_records();

  approximate_message_size_bytes_ = 0;

  batched_log_entries_ = 0;

}

                                     Http::ResponseMessagePtr&& http_response) {

  active_requests_.remove(request);

  const auto response_code = Http::Utility::getResponseStatus(http_response->headers());

  if (response_code == enumToInt(Http::Code::OK)) {

    stats_.logs_written_.add(in_flight_log_entries_);

  } else {

    ENVOY_LOG(error,

              "OTLP HTTP access log exporter received a non-success status code: {} while "

              "exporting the OTLP message",

              response_code);

    stats_.logs_dropped_.add(in_flight_log_entries_);

  }

  in_flight_log_entries_ = 0;

}

                                     Http::AsyncClient::FailureReason reason) {

  active_requests_.remove(request);

  ENVOY_LOG(warn, "OTLP HTTP access log export request failed. Failure reason: {}",

            enumToInt(reason));

  stats_.logs_dropped_.add(in_flight_log_entries_);

  in_flight_log_entries_ = 0;

}

    : tls_slot_(server_context.threadLocal().allocateSlot()), server_context_(server_context) {

  tls_slot_->set(

      [](Event::Dispatcher& dispatcher) { return std::make_shared<ThreadLocalCache>(dispatcher); });

}

    std::shared_ptr<const Http::HttpServiceHeadersApplicator> headers_applicator) {

  auto& cache = tls_slot_->getTyped<ThreadLocalCache>();

  const std::size_t config_hash = MessageUtil::hash(config) ^ MessageUtil::hash(http_service);

  const auto it = cache.access_loggers_.find(config_hash);

  if (it != cache.access_loggers_.end()) {

    return it->second;

  }

  auto logger = std::make_shared<HttpAccessLoggerImpl>(server_context_.clusterManager(),

                                                       http_service, std::move(headers_applicator),

                                                       config, cache.dispatcher_, server_context_);

  cache.access_loggers_.emplace(config_hash, logger);

  return logger;

}

    Server::Configuration::ServerFactoryContext& server_context) {

  ASSERT_IS_MAIN_OR_TEST_THREAD();

  const std::size_t config_hash = MessageUtil::hash(http_service);

  absl::MutexLock lock(&applicator_mutex_);

  const auto it = applicators_.find(config_hash);

  if (it != applicators_.end()) {

    auto existing = it->second.lock();

    if (existing) {

      return existing;

    }

  }

  absl::Status creation_status = absl::OkStatus();

  std::shared_ptr<HttpAccessLoggerCacheImpl> self = shared_from_this();

  std::shared_ptr<const Http::HttpServiceHeadersApplicator> applicator(

      new Http::HttpServiceHeadersApplicator(http_service, server_context, creation_status),

      [self, config_hash](const Http::HttpServiceHeadersApplicator* ptr) {

        {

          absl::MutexLock lock(&self->applicator_mutex_);

          const auto it = self->applicators_.find(config_hash);

          if (it != self->applicators_.end() && it->second.expired()) {

            self->applicators_.erase(it);

          }

        }

        delete ptr;

      });

  THROW_IF_NOT_OK_REF(creation_status);

  applicators_.insert_or_assign(config_hash, applicator);

  return applicator;

}

    : logger_(std::move(logger)) {}

    : Common::ImplBase(std::move(filter)), tls_slot_(server_context.threadLocal().allocateSlot()),

      access_logger_cache_(std::move(access_logger_cache)), http_service_(config.http_service()),

      filter_state_objects_to_log_(getFilterStateObjectsToLog(config)),

      custom_tags_(getCustomTags(config)) {

  std::shared_ptr<const Http::HttpServiceHeadersApplicator> headers_applicator =

      access_logger_cache_->getOrCreateApplicator(http_service_, server_context);

  tls_slot_->set([this, config, headers_applicator](Event::Dispatcher&) {

    return std::make_shared<ThreadLocalLogger>(

        access_logger_cache_->getOrCreateLogger(config, http_service_, headers_applicator));

  });

  if (config.body().value_case() != ::opentelemetry::proto::common::v1::AnyValue::VALUE_NOT_SET) {

    body_formatter_ = std::make_unique<OpenTelemetryFormatter>(packBody(config.body()), commands);

  }

  attributes_formatter_ = std::make_unique<OpenTelemetryFormatter>(config.attributes(), commands);

}

                            const StreamInfo::StreamInfo& stream_info) {

  opentelemetry::proto::logs::v1::LogRecord log_entry;

  log_entry.set_time_unix_nano(std::chrono::duration_cast<std::chrono::nanoseconds>(

                                   stream_info.startTime().time_since_epoch())

                                   .count());

  if (body_formatter_) {

    const auto formatted_body = unpackBody(body_formatter_->format(log_context, stream_info));

    *log_entry.mutable_body() = formatted_body;

  }

  const auto formatted_attributes = attributes_formatter_->format(log_context, stream_info);

  *log_entry.mutable_attributes() = formatted_attributes.values();

  const std::string trace_id_hex =

      log_context.activeSpan().has_value() ? log_context.activeSpan()->getTraceId() : "";

  const std::string span_id_hex =

      log_context.activeSpan().has_value() ? log_context.activeSpan()->getSpanId() : "";

  populateTraceContext(log_entry, trace_id_hex, span_id_hex);

  addFilterStateToAttributes(stream_info, filter_state_objects_to_log_, log_entry);

  addCustomTagsToAttributes(custom_tags_, log_context, stream_info, log_entry);

  tls_slot_->getTyped<ThreadLocalLogger>().logger_->log(std::move(log_entry));

}