Skip to content

cilium: defer response logging until stream completion #1879

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
nvibert wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

cilium: defer response logging until stream completion #1879

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 32 additions & 3 deletions cilium/l7policy.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@
#include <string>
#include <utility>

#include "envoy/buffer/buffer.h"
#include "envoy/common/time.h"
#include "envoy/http/codes.h"
#include "envoy/http/filter.h"
Expand Down Expand Up @@ -303,6 +304,20 @@ void AccessFilter::onStreamComplete() {
if (log_entry_ && !log_entry_->request_logged_) {
config_->log(*log_entry_, ::cilium::EntryType::Request);
}

if (!response_logged_ && response_headers_.has_value()) {
logResponse();
}
}

void AccessFilter::logResponse() {
if (response_logged_ || log_entry_ == nullptr || !response_headers_.has_value()) {
return;
}

log_entry_->updateFromResponse(*response_headers_, config_->time_source_);
config_->log(*log_entry_, ::cilium::EntryType::Response);
response_logged_ = true;
}

Http::FilterHeadersStatus AccessFilter::encodeHeaders(Http::ResponseHeaderMap& headers, bool) {
Expand Down Expand Up @@ -374,11 +389,25 @@ Http::FilterHeadersStatus AccessFilter::encodeHeaders(Http::ResponseHeaderMap& h
config_->log(*log_entry_, log_type);
}

// Log the response
log_entry_->updateFromResponse(headers, config_->time_source_);
config_->log(*log_entry_, ::cilium::EntryType::Response);
response_headers_ = headers;

// Note: We intentionally do NOT call logResponse() here even when end_stream is true.
// Other encode filters (e.g. ext_proc with BUFFERED body mode) may still mutate
// response headers after we return. We defer logging to onStreamComplete() so that
// the access log captures the final, post-mutation header state regardless of where
// cilium.l7policy sits in the HCM filter chain.
return Http::FilterHeadersStatus::Continue;
}

Http::FilterDataStatus AccessFilter::encodeData(Buffer::Instance&, bool) {
// See encodeHeaders(): logging is deferred to onStreamComplete().
return Http::FilterDataStatus::Continue;
}

Http::FilterTrailersStatus AccessFilter::encodeTrailers(Http::ResponseTrailerMap&) {
// See encodeHeaders(): logging is deferred to onStreamComplete().
return Http::FilterTrailersStatus::Continue;
}

} // namespace Cilium
} // namespace Envoy
11 changes: 5 additions & 6 deletions cilium/l7policy.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,18 +94,15 @@ class AccessFilter : public Http::StreamFilter,
}
Http::FilterHeadersStatus encodeHeaders(Http::ResponseHeaderMap& headers,
bool end_stream) override;
Http::FilterDataStatus encodeData(Buffer::Instance&, bool) override {
return Http::FilterDataStatus::Continue;
}
Http::FilterTrailersStatus encodeTrailers(Http::ResponseTrailerMap&) override {
return Http::FilterTrailersStatus::Continue;
}
Http::FilterDataStatus encodeData(Buffer::Instance&, bool end_stream) override;
Http::FilterTrailersStatus encodeTrailers(Http::ResponseTrailerMap&) override;
void setEncoderFilterCallbacks(Http::StreamEncoderFilterCallbacks&) override {}
Http::FilterMetadataStatus encodeMetadata(Http::MetadataMap&) override {
return Http::FilterMetadataStatus::Continue;
}

private:
void logResponse();
void sendLocalError(absl::string_view details);

ConfigSharedPtr config_;
Expand All @@ -114,7 +111,9 @@ class AccessFilter : public Http::StreamFilter,
AccessLog::Entry* log_entry_ = nullptr;

OptRef<Http::RequestHeaderMap> latched_headers_;
OptRef<Http::ResponseHeaderMap> response_headers_;
absl::optional<bool> latched_end_stream_;
bool response_logged_ = false;
};

} // namespace Cilium
Expand Down
42 changes: 42 additions & 0 deletions tests/accesslog_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,5 +86,47 @@ TEST_F(CiliumTest, AccessLog) {
EXPECT_STREQ(log.entry_.http().headers(1).value().c_str(), "response");
}

TEST_F(CiliumTest, AccessLogPreservesAIResponseHeaders) {
Http::TestRequestHeaderMapImpl headers{{":method", "POST"},
{":path", "/v1/chat/completions"},
{":authority", "model-server"},
{"x-forwarded-proto", "http"},
{"x-request-id", "req-123"}};
Network::MockConnection connection;
auto source_address = std::make_shared<Network::Address::Ipv4Instance>("5.6.7.8", 45678);
auto destination_address = std::make_shared<Network::Address::Ipv4Instance>("1.2.3.4", 80);
connection.stream_info_.protocol_ = Http::Protocol::Http11;
connection.stream_info_.start_time_ = time_system_.systemTime();
connection.stream_info_.downstream_connection_info_provider_->setRemoteAddress(source_address);
connection.stream_info_.downstream_connection_info_provider_->setLocalAddress(
destination_address);

AccessLog::Entry log;
log.initFromRequest("1.2.3.4", 42, true, 1, source_address, 173, destination_address,
connection.stream_info_, headers);

Http::TestResponseHeaderMapImpl response_headers{{":status", "200"},
{"x-ai-model", "llama3.1-8b-instruct"},
{"x-ai-input-tokens", "11"},
{"x-ai-output-tokens", "6"},
{"x-ai-total-tokens", "17"}};

NiceMock<Event::SimulatedTimeSystem> time_source;
log.updateFromResponse(response_headers, time_source);

EXPECT_EQ(log.entry_.http().status(), 200);
EXPECT_EQ(log.entry_.http().headers_size(), 5);
EXPECT_STREQ(log.entry_.http().headers(0).key().c_str(), "x-request-id");
EXPECT_STREQ(log.entry_.http().headers(0).value().c_str(), "req-123");
EXPECT_STREQ(log.entry_.http().headers(1).key().c_str(), "x-ai-model");
EXPECT_STREQ(log.entry_.http().headers(1).value().c_str(), "llama3.1-8b-instruct");
EXPECT_STREQ(log.entry_.http().headers(2).key().c_str(), "x-ai-input-tokens");
EXPECT_STREQ(log.entry_.http().headers(2).value().c_str(), "11");
EXPECT_STREQ(log.entry_.http().headers(3).key().c_str(), "x-ai-output-tokens");
EXPECT_STREQ(log.entry_.http().headers(3).value().c_str(), "6");
EXPECT_STREQ(log.entry_.http().headers(4).key().c_str(), "x-ai-total-tokens");
EXPECT_STREQ(log.entry_.http().headers(4).value().c_str(), "17");
}

} // namespace Cilium
} // namespace Envoy
42 changes: 42 additions & 0 deletions tests/cilium_http_integration_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -903,6 +903,48 @@ TEST_P(CiliumIntegrationTest, AcceptedMethod) {
}));
}

TEST_P(CiliumIntegrationTest, AcceptedResponsePreservesArbitraryHeaders) {
initialize();
codec_client_ = makeHttpConnection(lookupPort("http"));

Http::TestResponseHeaderMapImpl response_headers{
{":status", "200"}, {"x-ai-model", "llama3.1-8b-instruct"}, {"x-ai-total-tokens", "17"}};
auto response = sendRequestAndWaitForResponse(
{{":method", "PUT"}, {":path", "/public/opinions"}, {":authority", "host"}}, 0,
response_headers, 0);

absl::optional<std::string> maybe_x_request_id;
EXPECT_TRUE(expectAccessLogRequestTo([&maybe_x_request_id](const ::cilium::LogEntry& entry) {
maybe_x_request_id = getHeader(entry.http().headers(), "x-request-id");
return entry.http().status() == 0;
}));
ASSERT_TRUE(maybe_x_request_id.has_value());

absl::optional<std::string> maybe_x_request_id_resp;
absl::optional<std::string> maybe_model;
absl::optional<std::string> maybe_total_tokens;
EXPECT_TRUE(expectAccessLogResponseTo([&maybe_x_request_id_resp, &maybe_model,
&maybe_total_tokens](const ::cilium::LogEntry& entry) {
maybe_x_request_id_resp = getHeader(entry.http().headers(), "x-request-id");
maybe_model = getHeader(entry.http().headers(), "x-ai-model");
maybe_total_tokens = getHeader(entry.http().headers(), "x-ai-total-tokens");
return entry.http().status() == 200;
}));

ASSERT_TRUE(maybe_x_request_id_resp.has_value());
EXPECT_EQ(maybe_x_request_id.value(), maybe_x_request_id_resp.value());
ASSERT_TRUE(maybe_model.has_value());
EXPECT_EQ("llama3.1-8b-instruct", maybe_model.value());
ASSERT_TRUE(maybe_total_tokens.has_value());
EXPECT_EQ("17", maybe_total_tokens.value());

EXPECT_TRUE(response->complete());
EXPECT_EQ("200", response->headers().getStatusValue());
EXPECT_TRUE(upstream_request_->complete());
EXPECT_EQ(0, upstream_request_->bodyLength());
cleanupUpstreamAndDownstream();
}

TEST_P(CiliumIntegrationTest, L3DeniedPath) {
denied({{":method", "GET"}, {":path", "/only-2-allowed"}, {":authority", "host"}});

Expand Down
Loading