diff --git a/CMakeLists.txt b/CMakeLists.txt index e3fabfd..11abf58 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -107,8 +107,7 @@ target_link_libraries(pg_stat_ch PRIVATE PostgreSQLServer::PostgreSQLServer opentelemetry-cpp::api opentelemetry-cpp::sdk - opentelemetry-cpp::otlp_grpc_metrics_exporter - opentelemetry-cpp::otlp_grpc_log_record_exporter + opentelemetry-cpp::otlp_http_log_record_exporter opentelemetry-cpp::metrics opentelemetry-cpp::logs Arrow::arrow_static diff --git a/docker/docker-compose.otel.yml b/docker/docker-compose.otel.yml index 4cf52c3..a1373bb 100644 --- a/docker/docker-compose.otel.yml +++ b/docker/docker-compose.otel.yml @@ -6,7 +6,7 @@ services: volumes: - ./otel/collector-config.yaml:/etc/otelcol-contrib/config.yaml ports: - - "4317:4317" # gRPC OTLP receiver (matches psch_otel_endpoint default) + - "4318:4318" # HTTP OTLP receiver (matches psch_otel_endpoint default) - "9091:9090" # Prometheus metrics exporter (host:9091 → container:9090) - "13133:13133" # Health check HTTP endpoint healthcheck: diff --git a/docker/otel/collector-config.yaml b/docker/otel/collector-config.yaml index ff0447a..cd15deb 100644 --- a/docker/otel/collector-config.yaml +++ b/docker/otel/collector-config.yaml @@ -1,8 +1,8 @@ receivers: otlp: protocols: - grpc: - endpoint: 0.0.0.0:4317 + http: + endpoint: 0.0.0.0:4318 exporters: # Expose received metrics as Prometheus for test assertions diff --git a/src/config/guc.c b/src/config/guc.c index 97b5473..a047601 100644 --- a/src/config/guc.c +++ b/src/config/guc.c @@ -27,7 +27,7 @@ int psch_batch_max = 200000; int psch_log_min_elevel = WARNING; int psch_otel_log_queue_size = 65536; int psch_otel_log_batch_size = 8192; -int psch_otel_log_max_bytes = 3 * 1024 * 1024; // 3 MiB: gRPC default max is 4 MiB +int psch_otel_log_max_bytes = 3 * 1024 * 1024; // 3 MiB: stay under collector body limits int psch_otel_log_delay_ms = 100; int psch_otel_metric_interval_ms = 5000; bool psch_debug_force_locked_overflow = false; @@ -175,10 +175,10 @@ void PschInitGuc(void) { DefineCustomStringVariable( "pg_stat_ch.otel_endpoint", - "OpenTelemetry gRPC endpoint (host:port).", + "OpenTelemetry OTLP/HTTP logs URL.", NULL, &psch_otel_endpoint, - "localhost:4317", + "http://localhost:4318/v1/logs", PGC_POSTMASTER, 0, NULL, NULL, NULL); @@ -254,7 +254,7 @@ void PschInitGuc(void) { "pg_stat_ch.otel_log_batch_size", "Maximum records per OTLP log export call.", "Caps how many log records the direct OTLP exporter puts into a single " - "gRPC ExportLogs request.", + "ExportLogs request.", &psch_otel_log_batch_size, 8192, // bootValue 1, 131072, // min, max @@ -266,7 +266,7 @@ void PschInitGuc(void) { "pg_stat_ch.otel_log_max_bytes", "Soft byte budget for a single OTLP log export call.", "The direct OTLP exporter chunks log records to stay under this per-request " - "budget. The gRPC default is 4 MiB; the default leaves a safety margin.", + "budget. Collector receivers cap request bodies; the default leaves a safety margin.", &psch_otel_log_max_bytes, 3 * 1024 * 1024, // bootValue: 3 MiB 65536, 64 * 1024 * 1024, // min: 64 KiB, max: 64 MiB @@ -277,7 +277,7 @@ void PschInitGuc(void) { DefineCustomIntVariable( "pg_stat_ch.otel_log_delay_ms", "Deadline in milliseconds for a single OTLP log export call.", - "The direct OTLP exporter uses this as the per-request gRPC timeout so " + "The direct OTLP exporter uses this as the per-request HTTP timeout so " "the bgworker cannot block indefinitely on a slow collector.", &psch_otel_log_delay_ms, 100, // bootValue @@ -361,7 +361,7 @@ void PschInitGuc(void) { "Maximum Arrow batch size in bytes per OTLP request.", "Controls the soft byte budget (estimated, pre-compression) for a single " "Arrow IPC batch before it is flushed. ZSTD compression typically shrinks " - "the payload 20-30x, so this budget can safely exceed the gRPC wire limit.", + "the payload 20-30x, so the on-wire request stays far smaller.", &psch_otel_max_block_bytes, 3 * 1024 * 1024, // bootValue: 3 MiB 65536, 16 * 1024 * 1024, // min: 64 KiB, max: 16 MiB @@ -382,7 +382,7 @@ void PschInitGuc(void) { DefineCustomStringVariable( "pg_stat_ch.debug_arrow_dump_dir", - "Directory for dumping raw Arrow IPC payloads before gRPC send.", + "Directory for dumping raw Arrow IPC payloads before OTLP send.", "When non-empty, each Arrow IPC payload is written to this directory before " "being sent through OTLP. Intended for test validation and debugging only.", &psch_debug_arrow_dump_dir, diff --git a/src/export/exporter_interface.h b/src/export/exporter_interface.h index ec86e0d..12eeb48 100644 --- a/src/export/exporter_interface.h +++ b/src/export/exporter_interface.h @@ -107,7 +107,7 @@ class StatsExporter { }; // Bridge functions for PG logging from files that cannot include postgres.h -// (e.g. otel_exporter.cc conflicts with libintl.h via gRPC headers). +// (e.g. otel_exporter.cc conflicts with libintl.h via Arrow/protobuf headers). void LogNegativeValue(const std::string& column_name, int64_t value); void LogExporterWarning(const char* context, const char* message); void RecordExporterFailure(const char* message); diff --git a/src/export/otel_exporter.cc b/src/export/otel_exporter.cc index d588ee4..463d608 100644 --- a/src/export/otel_exporter.cc +++ b/src/export/otel_exporter.cc @@ -7,21 +7,21 @@ // Arena allocation eliminates per-object malloc/free (the largest cost in the // SDK path) and makes batch destruction O(1) via Arena::Reset(). -#include -#include -#include +#include +#include #include #include #include #include +#include #include "config/guc.h" #include "export/exporter_interface.h" #include -#include #include +#include #include #include #include @@ -59,34 +59,31 @@ namespace { using string = std::string; using string_view = std::string_view; -// Create a gRPC channel with compression and keepalive settings optimized -// for high-throughput telemetry export from a long-lived bgworker. -std::shared_ptr MakeOptimizedChannel( - const otlp::OtlpGrpcLogRecordExporterOptions& opts) { - grpc::ChannelArguments args; - - // gzip — telemetry payloads with repeated attribute keys compress very well. - args.SetCompressionAlgorithm(GRPC_COMPRESS_GZIP); - - // Keepalive: the bgworker holds a persistent channel. Without keepalive - // pings, idle periods cause silent TCP drops and surprise RPC failures. - args.SetInt(GRPC_ARG_KEEPALIVE_TIME_MS, 30000); - args.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, 10000); - args.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1); - args.SetInt(GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA, 0); - - std::shared_ptr credentials; - if (opts.use_ssl_credentials) { - grpc::SslCredentialsOptions ssl_opts; - if (!opts.ssl_credentials_cacert_as_string.empty()) { - ssl_opts.pem_root_certs = opts.ssl_credentials_cacert_as_string; - } - credentials = grpc::SslCredentials(ssl_opts); - } else { - credentials = grpc::InsecureChannelCredentials(); - } +// OtlpHttpClientOptions has no default constructor, copy every field from exporter options +otlp::OtlpHttpClientOptions MakeClientOptions(const otlp::OtlpHttpLogRecordExporterOptions& o) { + return otlp::OtlpHttpClientOptions( + o.url, o.ssl_insecure_skip_verify, o.ssl_ca_cert_path, o.ssl_ca_cert_string, + o.ssl_client_key_path, o.ssl_client_key_string, o.ssl_client_cert_path, + o.ssl_client_cert_string, o.ssl_min_tls, o.ssl_max_tls, o.ssl_cipher, o.ssl_cipher_suite, + o.content_type, o.json_bytes_mapping, o.compression, o.use_json_name, o.console_debug, + o.timeout, o.http_headers, o.retry_policy_max_attempts, o.retry_policy_initial_backoff, + o.retry_policy_max_backoff, o.retry_policy_backoff_multiplier, + std::shared_ptr(nullptr)); +} - return grpc::CreateCustomChannel(opts.endpoint, credentials, args); +const char* ExportResultToString(opentelemetry::sdk::common::ExportResult result) { + using opentelemetry::sdk::common::ExportResult; + switch (result) { + case ExportResult::kSuccess: + return "success"; + case ExportResult::kFailure: + return "failure"; + case ExportResult::kFailureFull: + return "collector full"; + case ExportResult::kFailureInvalidArgument: + return "invalid argument"; + } + return "unknown"; } common_pb::KeyValue* AddAttr(logs_pb::LogRecord* rec) { @@ -108,12 +105,15 @@ void SetDouble(common_pb::KeyValue* kv, string_view key, double val) { kv->mutable_value()->set_double_value(val); } -// Conservative sizing constants to stay under the gRPC message budget. +// Conservative sizing constants to stay under collector request-body limits. constexpr size_t kMinBytesPerRecord = 1200; constexpr size_t kRequestOverheadBytes = 512; constexpr size_t kLogRecordOverheadBytes = 128; constexpr size_t kAttrOverheadBytes = 24; +// Cap concurrent chunk uploads, each in-flight body holds up to max_chunk_bytes +constexpr size_t kMaxInflightChunks = 8; + size_t EstimateStringAttrBytes(string_view key, string_view value) { return kAttrOverheadBytes + key.size() + value.size(); } @@ -130,6 +130,9 @@ class OTelExporter : public StatsExporter { void BeginBatch() final { exported_count_ = 0; batch_failed_ = false; + async_ok_records_ = 0; + async_failed_chunks_ = 0; + async_failure_result_ = -1; ResetChunk(); } @@ -137,11 +140,13 @@ class OTelExporter : public StatsExporter { if (batch_failed_) { return; } + if (async_failed_chunks_.load() > 0) { + // In-flight chunk failed, stop building; CommitBatch reports failure + batch_failed_ = true; + return; + } if (ChunkFull()) { - if (!FlushChunk()) { - batch_failed_ = true; - return; - } + FlushChunk(); ResetChunk(); } current_record_ = scope_logs_->add_log_records(); @@ -195,7 +200,7 @@ class OTelExporter : public StatsExporter { shared_ptr> DbQueryTextColumn() final { return MakeSvCol("db.query.text"); } bool EstablishNewConnection() final; - bool IsConnected() const final { return stub_ != nullptr; } + bool IsConnected() const final { return client_ != nullptr; } int NumConsecutiveFailures() const final { return consecutive_failures_; } void ResetFailures() final { consecutive_failures_ = 0; } int NumExported() const final { return exported_count_; } @@ -339,31 +344,35 @@ class OTelExporter : public StatsExporter { scope_logs_->mutable_scope()->set_version(PG_STAT_CH_VERSION); } - bool FlushChunk() { - if (stub_ == nullptr || chunk_count_ == 0) { - return true; + // Enqueue chunk upload; Export serializes into POST body before returning, + // so arena frees immediately. Blocks only when kMaxInflightChunks uploads + // already run. Result lands via callback on curl's thread; callback touches + // atomics only, Postgres logging happens in CommitBatch + void FlushChunk() { + if (client_ == nullptr || chunk_count_ == 0) { + return; } - auto context = otlp::OtlpGrpcClient::MakeClientContext(grpc_opts_); - collector_logs::ExportLogsServiceResponse response; - auto status = otlp::OtlpGrpcClient::DelegateExport( - stub_.get(), std::move(context), std::move(arena_), std::move(*request_), &response); + const size_t records = chunk_count_; + client_->Export( + *request_, + [this, records](opentelemetry::sdk::common::ExportResult result) { + if (result == opentelemetry::sdk::common::ExportResult::kSuccess) { + async_ok_records_ += records; + } else { + async_failure_result_ = static_cast(result); + ++async_failed_chunks_; + } + return true; + }, + kMaxInflightChunks); request_ = nullptr; scope_logs_ = nullptr; current_record_ = nullptr; - - if (status.ok()) { - exported_count_ += static_cast(chunk_count_); - chunk_count_ = 0; - chunk_bytes_ = 0; - return true; - } - - LogExporterWarning("gRPC export failed", status.error_message().c_str()); - RecordExporterFailure(status.error_message().c_str()); - consecutive_failures_++; - return false; + arena_.reset(); + chunk_count_ = 0; + chunk_bytes_ = 0; } static void PopulateResource(resource_pb::Resource* resource, bool arrow_ipc = false) { @@ -379,20 +388,28 @@ class OTelExporter : public StatsExporter { } void ConfigureLogExport(const string& endpoint) { - grpc_opts_ = otlp::OtlpGrpcLogRecordExporterOptions(); + // Default constructor reads OTEL_EXPORTER_OTLP_* env vars; GUC wins + http_opts_ = otlp::OtlpHttpLogRecordExporterOptions(); if (!endpoint.empty()) { - grpc_opts_.endpoint = endpoint; + http_opts_.url = endpoint; } - grpc_opts_.timeout = std::chrono::milliseconds(psch_otel_log_delay_ms); + http_opts_.timeout = std::chrono::milliseconds(psch_otel_log_delay_ms); + // bgworker owns retry backoff, disable client retry + http_opts_.retry_policy_max_attempts = 0; max_chunk_bytes_ = std::max(kMinBytesPerRecord, psch_otel_log_max_bytes); max_chunk_records_ = std::max( 1, std::min(psch_otel_log_batch_size, max_chunk_bytes_ / kMinBytesPerRecord)); } - // gRPC state - otlp::OtlpGrpcLogRecordExporterOptions grpc_opts_; - std::unique_ptr stub_; + // HTTP client state + otlp::OtlpHttpLogRecordExporterOptions http_opts_; + // Chunk upload results, written by curl-thread callbacks, drained by + // CommitBatch; declared before client_ so they outlive session callbacks + std::atomic async_ok_records_{0}; + std::atomic async_failed_chunks_{0}; + std::atomic async_failure_result_{-1}; + std::unique_ptr client_; size_t max_chunk_records_ = 1; size_t max_chunk_bytes_ = kMinBytesPerRecord; int consecutive_failures_ = 0; @@ -414,37 +431,55 @@ bool OTelExporter::EstablishNewConnection() { (psch_otel_endpoint != nullptr && *psch_otel_endpoint != '\0') ? psch_otel_endpoint : ""; ConfigureLogExport(endpoint); - auto channel = MakeOptimizedChannel(grpc_opts_); - if (channel == nullptr) { - LogExporterWarning("OTel init failed", "invalid or empty OTLP endpoint"); - stub_.reset(); + // Catch grpc-era host:port values early; OTLP/HTTP needs a full URL + if (http_opts_.url.find("://") == string::npos) { + LogExporterWarning("OTel init failed", + "otel_endpoint must be an http:// or https:// URL, e.g. " + "http://localhost:4318/v1/logs"); + client_.reset(); return false; } - stub_ = collector_logs::LogsService::NewStub(channel); + client_ = std::make_unique(MakeClientOptions(http_opts_)); return true; } catch (const std::exception& e) { LogExporterWarning("OTel init failed", e.what()); - stub_.reset(); + client_.reset(); return false; } } bool OTelExporter::CommitBatch() { - if (batch_failed_) { - return false; - } - - if (stub_ == nullptr) { + if (client_ == nullptr) { ResetFailures(); return true; } try { - bool ok = FlushChunk(); - if (ok) { + if (!batch_failed_) { + FlushChunk(); + } + // Settle all in-flight chunks so success/failure stays per batch + const bool flushed = client_->ForceFlush( + std::chrono::duration_cast(http_opts_.timeout) + + std::chrono::seconds(1)); + exported_count_ += static_cast(async_ok_records_.exchange(0)); + const int failed_chunks = async_failed_chunks_.exchange(0); + const int failure_result = async_failure_result_.exchange(-1); + + if (flushed && failed_chunks == 0) { ResetFailures(); + return true; } - return ok; + + const char* reason = + failed_chunks > 0 + ? ExportResultToString( + static_cast(failure_result)) + : "flush timeout"; + LogExporterWarning("OTLP export failed", reason); + RecordExporterFailure(reason); + consecutive_failures_++; + return false; } catch (const std::exception& e) { LogExporterWarning("export exception", e.what()); RecordExporterFailure(e.what()); @@ -454,7 +489,7 @@ bool OTelExporter::CommitBatch() { } bool OTelExporter::SendArrowBatch(const uint8_t* ipc_data, size_t ipc_len, int num_rows) { - if (stub_ == nullptr || ipc_data == nullptr || ipc_len == 0 || num_rows <= 0) { + if (client_ == nullptr || ipc_data == nullptr || ipc_len == 0 || num_rows <= 0) { return false; } @@ -483,19 +518,16 @@ bool OTelExporter::SendArrowBatch(const uint8_t* ipc_data, size_t ipc_len, int n SetString(AddAttr(record), "pg_stat_ch.block_format", "arrow_ipc"); SetInt(AddAttr(record), "pg_stat_ch.block_rows", num_rows); - auto context = otlp::OtlpGrpcClient::MakeClientContext(grpc_opts_); - collector_logs::ExportLogsServiceResponse response; - auto status = otlp::OtlpGrpcClient::DelegateExport( - stub_.get(), std::move(context), std::move(arena), std::move(*request), &response); + auto result = client_->Export(*request); - if (status.ok()) { + if (result == opentelemetry::sdk::common::ExportResult::kSuccess) { exported_count_ += num_rows; consecutive_failures_ = 0; return true; } - LogExporterWarning("Arrow batch gRPC failed", status.error_message().c_str()); - RecordExporterFailure(status.error_message().c_str()); + LogExporterWarning("Arrow batch export failed", ExportResultToString(result)); + RecordExporterFailure(ExportResultToString(result)); consecutive_failures_++; return false; } catch (const std::exception& e) { diff --git a/src/export/stats_exporter.cc b/src/export/stats_exporter.cc index f586238..cd40b67 100644 --- a/src/export/stats_exporter.cc +++ b/src/export/stats_exporter.cc @@ -528,7 +528,7 @@ int PschGetConsecutiveFailures(void) { } } -// Exception barrier: the OTel exporter's destructors (gRPC stub teardown, +// Exception barrier: the OTel exporter's destructors (HTTP client teardown, // protobuf arena release) can throw. Catching here prevents the throw from // crossing the on_proc_exit chain. void PschExporterShutdown(void) { diff --git a/t/025_otel_reconnect.pl b/t/025_otel_reconnect.pl index efded76..78f2073 100755 --- a/t/025_otel_reconnect.pl +++ b/t/025_otel_reconnect.pl @@ -84,7 +84,7 @@ } $node->safe_psql('postgres', 'SELECT pg_stat_ch_flush()'); - # Wait for export to resume; the OTel gRPC channel reconnects automatically + # Wait for export to resume; curl re-establishes the connection on the next request sleep(5); my $stats_up = psch_get_stats($node); diff --git a/t/026_arrow_dump.pl b/t/026_arrow_dump.pl index ca0083d..e4d8f93 100644 --- a/t/026_arrow_dump.pl +++ b/t/026_arrow_dump.pl @@ -27,7 +27,7 @@ # Initialize node with Arrow passthrough + dump enabled. # use_otel=on + otel_arrow_passthrough=on enables the Arrow export path. -# The otel_endpoint points at a non-existent collector — gRPC send will fail, +# The otel_endpoint points at a non-existent collector — HTTP send will fail, # but MaybeDumpArrowBatch() fires BEFORE the send, so IPC files still land. my $node = PostgreSQL::Test::Cluster->new('arrow_dump'); $node->init(); @@ -38,7 +38,7 @@ pg_stat_ch.flush_interval_ms = 100 pg_stat_ch.batch_max = 100 pg_stat_ch.use_otel = on -pg_stat_ch.otel_endpoint = 'localhost:14317' +pg_stat_ch.otel_endpoint = 'http://localhost:14318/v1/logs' pg_stat_ch.otel_arrow_passthrough = on pg_stat_ch.debug_arrow_dump_dir = '$dump_dir' pg_stat_ch.hostname = 'test-arrow-host' diff --git a/t/psch.pm b/t/psch.pm index 0fdf6cd..960115f 100644 --- a/t/psch.pm +++ b/t/psch.pm @@ -248,7 +248,7 @@ sub psch_init_node_with_otel { my $flush_interval_ms = $opts{flush_interval_ms} // 100; # Fast flush for tests my $batch_max = $opts{batch_max} // 1000; my $enabled = $opts{enabled} // 'on'; - my $otel_endpoint = $opts{otel_endpoint} // 'localhost:4317'; + my $otel_endpoint = $opts{otel_endpoint} // 'http://localhost:4318/v1/logs'; my $hostname = $opts{hostname} // 'test-host'; my $node = PostgreSQL::Test::Cluster->new($name); diff --git a/third_party/vcpkg b/third_party/vcpkg index 56bb241..cd61e1e 160000 --- a/third_party/vcpkg +++ b/third_party/vcpkg @@ -1 +1 @@ -Subproject commit 56bb2411609227288b70117ead2c47585ba07713 +Subproject commit cd61e1e26a038e82d6550a3ebbe0fbbfe7da78e3 diff --git a/vcpkg.json b/vcpkg.json index d2720ee..765583d 100644 --- a/vcpkg.json +++ b/vcpkg.json @@ -6,7 +6,7 @@ { "name": "opentelemetry-cpp", "default-features": false, - "features": ["otlp-grpc"] + "features": ["otlp-http"] }, { "name": "arrow",