diff --git a/src/postgres/src/backend/utils/misc/pg_yb_utils.c b/src/postgres/src/backend/utils/misc/pg_yb_utils.c index b233ee289eb2..d8bd6e20601f 100644 --- a/src/postgres/src/backend/utils/misc/pg_yb_utils.c +++ b/src/postgres/src/backend/utils/misc/pg_yb_utils.c @@ -1206,7 +1206,7 @@ YBInitPostgresBackend(const char *program_name, const YbcPgInitPostgresInfo *ini hex_encode((const char *) YbGetLocalTServerUuid(), UUID_LEN, hex_uuid); hex_uuid[2 * UUID_LEN] = '\0'; - YBCInitDistTrace(MyProcPid, hex_uuid); + YBCInitDistTrace(hex_uuid); } } } @@ -1217,7 +1217,7 @@ YBOnPostgresBackendShutdown() YBCDestroyPgGate(); if (YBCIsDistTraceEnabled()) - YBCCleanupDistTrace(); + YBCShutdownDistTrace(); } void diff --git a/src/yb/rpc/outbound_call.cc b/src/yb/rpc/outbound_call.cc index 216cbfd66618..1714cca24cde 100644 --- a/src/yb/rpc/outbound_call.cc +++ b/src/yb/rpc/outbound_call.cc @@ -151,13 +151,11 @@ bool FinishedState(RpcCallState state) { return false; } -void SetSpanStatus(opentelemetry::trace::Span& span, RpcCallState state) { +void SetSpanStatus(opentelemetry::trace::Span& span, RpcCallState state, const Status& status) { switch (state) { case TIMED_OUT: - span.SetStatus(opentelemetry::trace::StatusCode::kError, "Call TimedOut"); - return; case FINISHED_ERROR: - span.SetStatus(opentelemetry::trace::StatusCode::kError, "Call ErroredOut"); + span.SetStatus(opentelemetry::trace::StatusCode::kError, status.ToUserMessage()); return; case FINISHED_SUCCESS: span.SetStatus(opentelemetry::trace::StatusCode::kOk); @@ -279,9 +277,16 @@ OutboundCall::OutboundCall(const RemoteMethod& remote_method, IncrementCounter(rpc_metrics_->outbound_calls_created); IncrementGauge(rpc_metrics_->outbound_calls_alive); - if (dist_trace::HasActiveContext()) { - otel_span_ = dist_trace::StartSpan( - Format("rpc $0", remote_method_.ToString()), dist_trace::GetPendingRpcAttrPairs()); + // Capture this call's parent before StartClientSpanWithScope makes otel_span_ current; + // InvokeCallbackSync restores it around the callback so its follow-on work nests as a sibling. + trace_parent_ = dist_trace::GetActiveSpanContext(); + + otel_span_ = dist_trace::StartClientSpanWithScope(Format("rpc $0", remote_method_.ToString())); + if (otel_span_) { + otel_span_->SetAttribute("rpc.system", "outbound_rpc"); + otel_span_->SetAttribute("rpc.service", remote_method_.service_name()); + otel_span_->SetAttribute("rpc.method", remote_method_.method_name()); + otel_span_->SetAttribute("rpc.call_id", call_id_); } } @@ -372,6 +377,33 @@ Status OutboundCall::SetRequestParam( metadata_size += 1; // add tag size of RequestHeader::kMetadataFieldNumber } + // Distributed-trace context: extract the outbound span's SpanContext (if any) and pre-compute the + // serialized size of the TraceContextPB submessage so it can be folded into the header length. + size_t trace_context_size = 0; + size_t trace_context_message_size = 0; + uint32_t version_and_flags = 0; + opentelemetry::trace::TraceId trace_id; + opentelemetry::trace::SpanId span_id; + if (otel_span_) { + auto span_context = otel_span_->GetContext(); + if (span_context.IsValid()) { + trace_id = span_context.trace_id(); + span_id = span_context.span_id(); + auto trace_flags = span_context.trace_flags(); + // TraceContextPB submessage layout: + // - trace_id_hi: 1 byte tag + 8 bytes fixed64 + // - trace_id_lo: 1 byte tag + 8 bytes fixed64 + // - span_id: 1 byte tag + 8 bytes fixed64 + // - version_and_flags: 1 byte tag + varint (size depends on the combined value) + constexpr uint32_t kVersion = 0; + version_and_flags = (kVersion << 8) | trace_flags.flags(); + size_t version_and_flags_varint_size = Output::VarintSize32(version_and_flags); + trace_context_message_size = 3 * 9 + 1 + version_and_flags_varint_size; + trace_context_size = + 1 + Output::VarintSize64(trace_context_message_size) + trace_context_message_size; + } + } + auto use_crc = FLAGS_rpc_enable_crc; size_t header_pb_len = 1 + call_id_size + // int32 call_id = 1 serialized_remote_method.size() + // RemoteMethodPB remote_method = 2 @@ -380,6 +412,7 @@ Status OutboundCall::SetRequestParam( if (pool_tag) { header_pb_len += 1 + Output::VarintSize64(pool_tag); // uint64 pool_tag = 7 } + header_pb_len += trace_context_size; // TraceContext trace_context = 8 if (use_crc) { header_pb_len += 1 + sizeof(uint32_t); // fixed32 crc = 15 } @@ -431,6 +464,31 @@ Status OutboundCall::SetRequestParam( dst = Output::WriteVarint64ToArray(pool_tag, dst); } + if (trace_context_size > 0) { + // Write the TraceContextPB submessage. Field numbers match yb.TraceContextPB in common.proto. + // The 16-byte trace id splits into two big-endian 64-bit halves; the span id is one big-endian + // 64-bit. + dst = Output::WriteTagToArray( + (RequestHeader::kTraceContextFieldNumber << 3) | WireFormatLite::WIRETYPE_LENGTH_DELIMITED, + dst); + dst = Output::WriteVarint32ToArray(narrow_cast(trace_context_message_size), dst); + + dst = Output::WriteTagToArray( + (TraceContextPB::kTraceIdHiFieldNumber << 3) | WireFormatLite::WIRETYPE_FIXED64, dst); + dst = Output::WriteLittleEndian64ToArray(BigEndian::Load64(trace_id.Id().data()), dst); + + dst = Output::WriteTagToArray( + (TraceContextPB::kTraceIdLoFieldNumber << 3) | WireFormatLite::WIRETYPE_FIXED64, dst); + dst = Output::WriteLittleEndian64ToArray(BigEndian::Load64(trace_id.Id().data() + 8), dst); + + dst = Output::WriteTagToArray( + (TraceContextPB::kSpanIdFieldNumber << 3) | WireFormatLite::WIRETYPE_FIXED64, dst); + dst = Output::WriteLittleEndian64ToArray(BigEndian::Load64(span_id.Id().data()), dst); + + dst = Output::WriteTagToArray(TraceContextPB::kVersionAndFlagsFieldNumber << 3, dst); + dst = Output::WriteVarint32ToArray(version_and_flags, dst); + } + // CRC should be at the end of header, otherwise adjust CRC filling logic below. if (use_crc) { dst = Output::WriteTagToArray( @@ -472,7 +530,7 @@ OutboundCall::State OutboundCall::state() const { return state_.load(std::memory_order_acquire); } -bool OutboundCall::SetState(State new_state) { +bool OutboundCall::SetState(State new_state, const Status& status) { auto old_state = state(); // Sanity check state transitions. DVLOG(3) << "OutboundCall " << this << " (" << ToString() << ") switching from " @@ -487,7 +545,8 @@ bool OutboundCall::SetState(State new_state) { } if (state_.compare_exchange_weak(old_state, new_state, std::memory_order_acq_rel)) { if (otel_span_ && FinishedState(new_state)) { - SetSpanStatus(*otel_span_, new_state); + DCHECK(otel_span_->span); + SetSpanStatus(*otel_span_->span, new_state, status); otel_span_->End(); } return true; @@ -557,7 +616,13 @@ void OutboundCall::InvokeCallbackSync(std::optional now_optiona // TODO: consider removing the cycle-based mechanism of reporting slow callbacks below. int64_t start_cycles = CycleClock::Now(); - callback_(); + // Re-activate the call's parent context so RPCs the callback issues nest as siblings, not + // parentless roots. No-op when trace_parent_ is invalid; parent_scope drops at block end so it + // can't leak. + { + auto parent_scope = dist_trace::ActivateParentScope(trace_parent_); + callback_(); + } // Clear the callback, since it may be holding onto reference counts // via bound parameters. We do this inside the timer because it's possible // the user has naughty destructors that block, and we want to account for that @@ -692,7 +757,7 @@ void OutboundCall::SetFailed(const Status &status, std::unique_ptrSetAttribute("network.peer.name", *hostname); + } + otel_span_->SetAttribute("network.peer.address", yb::ToString(value.remote())); + // Drop the OTEL span's scope here, on the calling thread, before the call is queued to the + // reactor. The span ends later, but the scope must be released on the thread that installed + // it. + otel_span_->DropScope(); + } } void SetThreadPoolFailure(const Status& status) EXCLUDES(mtx_) { @@ -453,6 +462,10 @@ class OutboundCall : public RpcCall { virtual size_t GetSidecarsCount() const; virtual size_t TransferSidecars(Sidecars* dest); + // Distributed-trace span for this call; LocalOutboundCall reads its context to parent the local + // inbound span. + const dist_trace::SpanWithScopePtr& otel_span() const { return otel_span_; } + // ---------------------------------------------------------------------------------------------- // Protected fields set in constructor or during initialization // ---------------------------------------------------------------------------------------------- @@ -481,7 +494,7 @@ class OutboundCall : public RpcCall { void NotifyTransferred(const Status& status, const ConnectionPtr& conn) override; - MUST_USE_RESULT bool SetState(State new_state); + MUST_USE_RESULT bool SetState(State new_state, const Status& status = Status::OK()); State state() const; // return current status @@ -595,9 +608,13 @@ class OutboundCall : public RpcCall { std::unique_ptr metadata_serializer_; - // OpenTelemetry span for distributed tracing. Created when the call starts if there is an - // active trace context, ended when the call completes (success, failure, or timeout). - opentelemetry::nostd::shared_ptr otel_span_; + // OpenTelemetry span for this call, created at start (if a trace context is active) and ended at + // completion. + dist_trace::SpanWithScopePtr otel_span_; + + // The trace context active when this call was constructed -- its PARENT, re-activated around the + // completion callback so follow-on RPCs nest as SIBLINGS of this call. + dist_trace::trace::SpanContext trace_parent_ = dist_trace::trace::SpanContext::GetInvalid(); // InvokeCallbackTask should be able to call InvokeCallbackSync and we don't want other that // method to be public. diff --git a/src/yb/util/dist_trace.cc b/src/yb/util/dist_trace.cc index e369ef66bd8a..1dec941c319b 100644 --- a/src/yb/util/dist_trace.cc +++ b/src/yb/util/dist_trace.cc @@ -11,43 +11,35 @@ // under the License. // -#include "yb/util/dist_trace.h" - -#include -#include - #include "opentelemetry/context/propagation/global_propagator.h" -#include "opentelemetry/context/propagation/text_map_propagator.h" -#include "opentelemetry/context/runtime_context.h" #include "opentelemetry/exporters/otlp/otlp_http_exporter_factory.h" #include "opentelemetry/exporters/otlp/otlp_http_exporter_options.h" #include "opentelemetry/sdk/common/global_log_handler.h" #include "opentelemetry/sdk/trace/batch_span_processor_factory.h" -#include "opentelemetry/sdk/trace/batch_span_processor_options.h" -#include "opentelemetry/sdk/trace/tracer_provider.h" #include "opentelemetry/sdk/trace/tracer_provider_factory.h" #include "opentelemetry/sdk/trace/provider.h" -#include "opentelemetry/trace/context.h" #include "opentelemetry/trace/propagation/http_trace_context.h" #include "opentelemetry/trace/provider.h" -#include "opentelemetry/trace/span.h" -#include "opentelemetry/trace/span_metadata.h" -#include "opentelemetry/trace/tracer.h" +#include "yb/util/dist_trace.h" #include "yb/util/flag_validators.h" -#include "yb/util/flags.h" #include "yb/util/signal_util.h" DEFINE_NON_RUNTIME_PREVIEW_string(otel_collector_traces_endpoint, "", "OTLP HTTP endpoint for the OpenTelemetry collector. When set, distributed tracing is " "enabled and spans are exported to this endpoint on each query execution."); -DEFINE_NON_RUNTIME_uint32(otel_batch_max_queue_size, 2048, +DEFINE_NON_RUNTIME_uint32(otel_batch_max_queue_size, 16384, "Maximum number of spans that can be buffered in the batch span processor queue. " "Spans arriving after this limit are dropped. Must be greater than 0 and at least as " "large as otel_batch_max_export_batch_size."); -DEFINE_NON_RUNTIME_uint32(otel_batch_schedule_delay_ms, 5000, +DEFINE_NON_RUNTIME_uint32(otel_ysql_batch_max_queue_size, 2048, + "Like otel_batch_max_queue_size, but used only by the ysql (postgres backend) process, which " + "runs one batch span processor per connection. Must be greater than 0 and at least as large as " + "otel_batch_max_export_batch_size."); + +DEFINE_NON_RUNTIME_uint32(otel_batch_schedule_delay_ms, 500, "Time interval in milliseconds between two consecutive batch exports of spans to the " "OpenTelemetry collector. Lower values reduce latency but increase export frequency."); @@ -62,6 +54,9 @@ DEFINE_NON_RUNTIME_string(otel_internal_log_level, "info", DEFINE_validator(otel_batch_max_queue_size, FLAG_GE_FLAG_VALIDATOR(otel_batch_max_export_batch_size)); +DEFINE_validator(otel_ysql_batch_max_queue_size, + FLAG_GE_FLAG_VALIDATOR(otel_batch_max_export_batch_size)); + DEFINE_validator(otel_internal_log_level, FLAG_IN_SET_VALIDATOR("debug", "info", "warning", "error", "none")); @@ -75,38 +70,39 @@ namespace context = opentelemetry::context; namespace { -const nostd::string_view ysql_resource_name = "ysql"; +// Service name for the tracing resource and tracer (e.g. "ysql", "Master", "TabletServer"). +static std::string g_service_name; -// Owns string attribute data and maintains a parallel vector of string_view/AttributeValue pairs -// that can be passed directly to the OTel Tracer::StartSpan API. Uses std::deque for pointer -// stability -- unlike std::vector, deque does not relocate existing elements on insertion, so -// string_views into earlier entries remain valid. -class RpcSpanAttrs { - public: - void AddStringAttr(std::string key, std::string value) { - auto& owned_key = owned_keys_.emplace_back(std::move(key)); - auto& owned_val = owned_values_.emplace_back(std::move(value)); - attrs_.emplace_back(owned_key, owned_val); - } +// The ysql process gets its own queue-size flag; tserver/master share otel_batch_max_queue_size. +static uint32_t EffectiveBatchMaxQueueSize() { + return g_service_name == kYsqlServiceName ? FLAGS_otel_ysql_batch_max_queue_size + : FLAGS_otel_batch_max_queue_size; +} - const std::vector>& attrs() - const { - return attrs_; - } +// A batch of pending RPC span attributes, owned as plain (key, value) strings. +using PendingRpcSpanAttrs = std::vector>; - void clear() { - attrs_.clear(); - owned_keys_.clear(); - owned_values_.clear(); - } +thread_local PendingRpcSpanAttrs pending_rpc_attrs; - private: - std::deque owned_keys_; - std::deque owned_values_; - std::vector> attrs_; -}; +// Moves the pending attributes out of the thread-local buffer, leaving it empty. The returned batch +// owns the strings. +static PendingRpcSpanAttrs ConsumePendingRpcAttrs() { + PendingRpcSpanAttrs consumed = std::move(pending_rpc_attrs); + // std::move leaves the source unspecified; force it empty. + pending_rpc_attrs.clear(); + return consumed; +} -thread_local RpcSpanAttrs pending_rpc_attrs; +// Builds the OTel-API attribute vector (string_view/AttributeValue pairs) viewing into `attrs`. +std::vector> SpanAttrsView( + const PendingRpcSpanAttrs& attrs) { + std::vector> view; + view.reserve(attrs.size()); + for (const auto& [key, value] : attrs) { + view.emplace_back(key, value); + } + return view; +} internal_log::LogLevel GetOtelInternalLogLevel() { const auto& flag_value = FLAGS_otel_internal_log_level; @@ -158,10 +154,11 @@ class YbOtelLogHandler : public internal_log::LogHandler { } }; -resource_sdk::Resource CreateResource(int64_t process_pid, nostd::string_view node_uuid) { +resource_sdk::Resource CreateResource( + nostd::string_view service_name, nostd::string_view node_uuid) { resource_sdk::ResourceAttributes attrs; - attrs.SetAttribute("service.name", ysql_resource_name); - attrs.SetAttribute("process.pid", process_pid); + attrs.SetAttribute("service.name", service_name); + attrs.SetAttribute("process.pid", static_cast(getpid())); attrs.SetAttribute("service.instance.id", node_uuid); return resource_sdk::Resource::Create(attrs); @@ -177,7 +174,7 @@ auto CreateExporter() { trace_sdk::BatchSpanProcessorOptions MakeBatchProcessorOptions() { trace_sdk::BatchSpanProcessorOptions batching_opts; - batching_opts.max_queue_size = static_cast(FLAGS_otel_batch_max_queue_size); + batching_opts.max_queue_size = static_cast(EffectiveBatchMaxQueueSize()); batching_opts.schedule_delay_millis = std::chrono::milliseconds(FLAGS_otel_batch_schedule_delay_ms); batching_opts.max_export_batch_size = static_cast(FLAGS_otel_batch_max_export_batch_size); @@ -207,8 +204,8 @@ Status InitDistTraceProvider(const resource_sdk::Resource& resource_attrs) { // supplied (through GUC or comment) traceparent header. class TraceparentCarrier : public context::propagation::TextMapCarrier { public: - explicit TraceparentCarrier(nostd::string_view traceparent) - : traceparent_(traceparent) {} + explicit TraceparentCarrier(nostd::string_view traceparent = {}) + : traceparent_(traceparent.data(), traceparent.size()) {} nostd::string_view Get(nostd::string_view key) const noexcept override { if (key == trace::propagation::kTraceParent) { @@ -217,10 +214,16 @@ class TraceparentCarrier : public context::propagation::TextMapCarrier { return {}; } - void Set(nostd::string_view, nostd::string_view) noexcept override {} + void Set(nostd::string_view key, nostd::string_view value) noexcept override { + if (key == trace::propagation::kTraceParent) { + traceparent_.assign(value.data(), value.size()); + } + } + + const std::string& traceparent() const { return traceparent_; } private: - nostd::string_view traceparent_; + std::string traceparent_; }; } // namespace @@ -229,7 +232,7 @@ bool IsDistTraceEnabled() { return !FLAGS_otel_collector_traces_endpoint.empty(); } -void InitDistTrace(int64_t process_pid, nostd::string_view node_uuid) { +void InitDistTrace(nostd::string_view service_name, nostd::string_view node_uuid) { DCHECK(IsDistTraceEnabled()); internal_log::GlobalLogHandler::SetLogHandler( @@ -239,7 +242,8 @@ void InitDistTrace(int64_t process_pid, nostd::string_view node_uuid) { // are mapped to LOG(...), where YB logging applies its own routing. internal_log::GlobalLogHandler::SetLogLevel(GetOtelInternalLogLevel()); - auto resource_attrs = CreateResource(process_pid, node_uuid); + g_service_name = std::string(service_name); + auto resource_attrs = CreateResource(service_name, node_uuid); const auto status = InitDistTraceProvider(resource_attrs); if (!status.ok()) { LOG(DFATAL) << "Failed to initialize OpenTelemetry tracing: " << status; @@ -250,13 +254,13 @@ void InitDistTrace(int64_t process_pid, nostd::string_view node_uuid) { nostd::shared_ptr( new trace::propagation::HttpTraceContext())); - LOG(INFO) << "OTEL: Initialized tracing for service: " << ysql_resource_name - << "\nBatchSpanProcessor config: max_queue_size=" << FLAGS_otel_batch_max_queue_size + LOG(INFO) << "OTEL: Initialized tracing for service: " << g_service_name + << "\nBatchSpanProcessor config: max_queue_size=" << EffectiveBatchMaxQueueSize() << ", schedule_delay_ms=" << FLAGS_otel_batch_schedule_delay_ms << ", max_export_batch_size=" << FLAGS_otel_batch_max_export_batch_size; } -void CleanupDistTrace() { +void ShutdownDistTrace() { DCHECK(IsDistTraceEnabled()); std::shared_ptr none; @@ -267,7 +271,7 @@ void CleanupDistTrace() { nostd::shared_ptr GetDistTracer() { DCHECK(IsDistTraceEnabled()); - return DCHECK_NOTNULL(trace::Provider::GetTracerProvider()->GetTracer(ysql_resource_name)); + return DCHECK_NOTNULL(trace::Provider::GetTracerProvider()->GetTracer(g_service_name)); } // A SpanContext is not valid when either its trace ID or span ID is all zeros. @@ -292,6 +296,17 @@ trace::SpanContext GetTraceparentSpanContext(const char* traceparent) { return trace::GetSpan(parent_context)->GetContext(); } +std::string GetActiveTraceparent() { + if (!HasActiveContext()) { + return {}; + } + TraceparentCarrier carrier; + static const auto propagator = + context::propagation::GlobalTextMapPropagator::GetGlobalPropagator(); + propagator->Inject(carrier, context::RuntimeContext::GetCurrent()); + return carrier.traceparent(); +} + bool HasActiveContext() { if (!IsDistTraceEnabled()) { return false; @@ -300,6 +315,13 @@ bool HasActiveContext() { return current_span && current_span->GetContext().IsValid(); } +trace::SpanContext GetActiveSpanContext() { + if (!HasActiveContext()) { + return trace::SpanContext::GetInvalid(); + } + return trace::Tracer::GetCurrentSpan()->GetContext(); +} + nostd::shared_ptr StartSpan( std::string_view op_name, const std::vector>& attrs, @@ -321,17 +343,68 @@ nostd::shared_ptr StartSpan(std::string_view op_name) { return StartSpan(op_name, {}); } -void AddPendingRpcStringAttr(std::string key, std::string value) { - pending_rpc_attrs.AddStringAttr(std::move(key), std::move(value)); +SpanWithScopePtr StartSpanWithScope( + std::string_view op_name, + const std::vector>& attrs, + trace::SpanKind kind) { + if (!HasActiveContext()) { + return nullptr; + } + trace::StartSpanOptions options; + options.kind = kind; + return std::make_shared(StartSpan(op_name, attrs, options)); } -const std::vector>& GetPendingRpcAttrPairs() { - return pending_rpc_attrs.attrs(); +SpanWithScopePtr StartSpanWithScope(std::string_view op_name, trace::SpanKind kind) { + return StartSpanWithScope(op_name, {}, kind); } -void ClearPendingRpcAttrs() { - pending_rpc_attrs.clear(); +SpanWithScopePtr StartClientSpanWithScope(std::string_view op_name) { + const auto pending = ConsumePendingRpcAttrs(); + + if (!IsDistTraceEnabled()) { + return nullptr; + } + + auto current_span = trace::Tracer::GetCurrentSpan(); + if (current_span && current_span->GetContext().IsValid()) { + return StartSpanWithScope(op_name, SpanAttrsView(pending), trace::SpanKind::kClient); + } + + return nullptr; +} + +SpanWithScopePtr StartServerSpanWithScope( + std::string_view op_name, + const trace::SpanContext& parent_context, + const std::vector>& + attrs) { + if (!IsDistTraceEnabled()) { + return nullptr; + } + trace::StartSpanOptions options; + options.kind = trace::SpanKind::kServer; + options.parent = parent_context; + return std::make_shared(GetDistTracer()->StartSpan( + nostd::string_view(op_name.data(), op_name.size()), attrs, options)); +} + +SpanWithScopePtr StartServerSpanWithScope( + std::string_view op_name, const trace::SpanContext& parent_context) { + return StartServerSpanWithScope(op_name, parent_context, {}); +} + +SpanWithScopePtr ActivateParentScope(const trace::SpanContext& parent_context) { + if (!IsDistTraceEnabled() || !parent_context.IsValid()) { + return nullptr; + } + // A non-recording span that merely carries parent_context. + return std::make_shared( + nostd::shared_ptr(new trace::DefaultSpan(parent_context))); +} + +void AddPendingRpcStringAttr(std::string key, std::string value) { + pending_rpc_attrs.emplace_back(std::move(key), std::move(value)); } } // namespace yb::dist_trace diff --git a/src/yb/util/dist_trace.h b/src/yb/util/dist_trace.h index 4371b492c2e7..5343686153ef 100644 --- a/src/yb/util/dist_trace.h +++ b/src/yb/util/dist_trace.h @@ -13,28 +13,89 @@ #pragma once -#include -#include -#include -#include -#include - -#include "opentelemetry/common/attribute_value.h" -#include "opentelemetry/trace/span_metadata.h" +#include + +#include "opentelemetry/trace/scope.h" #include "opentelemetry/trace/span_startoptions.h" #include "yb/util/dist_trace_fwd.h" +#include "yb/util/logging.h" namespace yb::dist_trace { namespace nostd = opentelemetry::nostd; namespace trace = opentelemetry::trace; -void InitDistTrace(int64_t process_pid, opentelemetry::nostd::string_view node_uuid); -void CleanupDistTrace(); +// Bundles a span with an activated (thread-local) scope so work started after it inherits it as +// parent; DropScope on the constructing thread before hopping threads. End() is safe from any +// thread. +struct SpanWithScope { + explicit SpanWithScope(nostd::shared_ptr s) + : span(std::move(s)), scope(span) {} + + ~SpanWithScope() { End(); } + + SpanWithScope(SpanWithScope&&) = default; + SpanWithScope& operator=(SpanWithScope&&) = default; + SpanWithScope(const SpanWithScope&) = delete; + SpanWithScope& operator=(const SpanWithScope&) = delete; + + void SetAttribute(nostd::string_view key, const opentelemetry::common::AttributeValue& value) { + if (span) { + span->SetAttribute(key, value); + } + } + + void SetStatus(trace::StatusCode code, nostd::string_view description = "") { + if (span) { + span->SetStatus(code, description); + } + } + + trace::SpanContext GetContext() const { + return span ? span->GetContext() : trace::SpanContext::GetInvalid(); + } + + // Releases the thread-local scope. Must be called on the thread that constructed this object. + void DropScope() { + scope.reset(); + owner_thread = {}; + } + + void End() { + if (span && span->IsRecording()) { + // The scope must be dropped on its creating thread; catch an unintended thread hop. + DCHECK(owner_thread == std::thread::id() || std::this_thread::get_id() == owner_thread) + << "SpanWithScope scope released off its creating thread"; + scope.reset(); + span->End(); + } + } + + nostd::shared_ptr span; + std::optional scope; + std::thread::id owner_thread = std::this_thread::get_id(); +}; + +using SpanWithScopePtr = std::shared_ptr; + +// OTel service.name for the ysql (postgres backend) process, passed to InitDistTrace at startup. +inline const std::string kYsqlServiceName = "ysql"; + +void InitDistTrace( + opentelemetry::nostd::string_view service_name, opentelemetry::nostd::string_view node_uuid); +void ShutdownDistTrace(); nostd::shared_ptr GetDistTracer(); bool IsDistTraceEnabled(); trace::SpanContext GetTraceparentSpanContext(const char* traceparent); + +// Serializes the active span into a W3C traceparent string via the global propagator (inverse of +// GetTraceparentSpanContext); hands the context to another process. Empty when disabled/no context. +std::string GetActiveTraceparent(); + +// Get SpanContext of the active span +trace::SpanContext GetActiveSpanContext(); + bool IsSpanContextValidAndRemote(const trace::SpanContext& span_context); // Returns true if distributed tracing is enabled and there is an active span in the OTEL context. @@ -48,11 +109,34 @@ nostd::shared_ptr StartSpan( const std::vector>& attrs); nostd::shared_ptr StartSpan(std::string_view op_name); +// Starts a child span of the active context, bundled with an activated scope so it +// becomes current; nullptr when no active context. +SpanWithScopePtr StartSpanWithScope( + std::string_view op_name, + const std::vector>& attrs, + trace::SpanKind kind = trace::SpanKind::kInternal); +SpanWithScopePtr StartSpanWithScope( + std::string_view op_name, trace::SpanKind kind = trace::SpanKind::kInternal); + +// Client span for an outbound RPC; drains pending thread-local attrs onto it. +// Child of active context, else an optional root gated by otel_rpc_sampling_ratio. +SpanWithScopePtr StartClientSpanWithScope(std::string_view op_name); + +// Span as a remote child of parent_context (from an inbound request) + activated scope -- +// the server end of a propagated trace; needs no local active context. +SpanWithScopePtr StartServerSpanWithScope( + std::string_view op_name, + const trace::SpanContext& parent_context, + const std::vector>& attrs); +SpanWithScopePtr StartServerSpanWithScope( + std::string_view op_name, const trace::SpanContext& parent_context); + +// Re-establishes parent_context as this thread's active context WITHOUT a new span, so RPCs built +// here nest under it -- for RPCs issued off the origin's thread. +SpanWithScopePtr ActivateParentScope(const trace::SpanContext& parent_context); + // Thread-local attribute buffer for the next RPC span. Producers (e.g. PgSession) add -// attributes here; the OutboundCall constructor consumes them when starting a span. +// attributes here; the OutboundCall Span consumes them when started. void AddPendingRpcStringAttr(std::string key, std::string value); -const std::vector>& - GetPendingRpcAttrPairs(); -void ClearPendingRpcAttrs(); } // namespace yb::dist_trace diff --git a/src/yb/yql/pggate/pg_client.cc b/src/yb/yql/pggate/pg_client.cc index b8900c8ccab7..e9ee212120c9 100644 --- a/src/yb/yql/pggate/pg_client.cc +++ b/src/yb/yql/pggate/pg_client.cc @@ -282,9 +282,23 @@ struct ResponseReadyTraits; std::string_view GetSharedMemSpanName(tserver::PgSharedExchangeReqType req_type) { switch (req_type) { case tserver::PgSharedExchangeReqType::PERFORM: - return "shmem req yb.tserver.PgClientService.Perform"; + return "shmem yb.tserver.PgClientService.Perform"; case tserver::PgSharedExchangeReqType::ACQUIRE_OBJECT_LOCK: - return "shmem req yb.tserver.PgClientService.AcquireObjectLock"; + return "shmem yb.tserver.PgClientService.AcquireObjectLock"; + case tserver::PgSharedExchangeReqType_INT_MIN_SENTINEL_DO_NOT_USE_: [[fallthrough]]; + case tserver::PgSharedExchangeReqType_INT_MAX_SENTINEL_DO_NOT_USE_: break; + } + FATAL_INVALID_ENUM_VALUE(tserver::PgSharedExchangeReqType, req_type); +} + +// Method name attribute for the outbound shared-memory span, mirroring rpc.method on RPC spans (the +// service is always PgClientService). Paired with the tserver's SharedMemMethodName. +const char* GetSharedMemMethodName(tserver::PgSharedExchangeReqType req_type) { + switch (req_type) { + case tserver::PgSharedExchangeReqType::PERFORM: + return "Perform"; + case tserver::PgSharedExchangeReqType::ACQUIRE_OBJECT_LOCK: + return "AcquireObjectLock"; case tserver::PgSharedExchangeReqType_INT_MIN_SENTINEL_DO_NOT_USE_: [[fallthrough]]; case tserver::PgSharedExchangeReqType_INT_MAX_SENTINEL_DO_NOT_USE_: break; } @@ -428,14 +442,19 @@ struct PgClientData : public FetchBigDataCallback { rpc::CallData big_call_data GUARDED_BY(exchange_mutex); // Only the owning future accesses this span while starting or finishing the shared-memory // request. Exchange callbacks do not touch it, so it does not need exchange_mutex protection. - opentelemetry::nostd::shared_ptr otel_span; + dist_trace::SpanWithScopePtr otel_span; PgClientData(const LWReqPB& req_, ThreadSafeArena* arena_) : req(req_), resp(arena_) {} void StartSharedMemorySpan() { - if (dist_trace::HasActiveContext()) { - otel_span = dist_trace::StartSpan( - GetSharedMemSpanName(kSharedExchangeRequestType), dist_trace::GetPendingRpcAttrPairs()); + otel_span = dist_trace::StartClientSpanWithScope( + GetSharedMemSpanName(kSharedExchangeRequestType)); + if (otel_span) { + // Mirror the attributes the RPC outbound span carries (outbound_call.cc). + otel_span->SetAttribute("rpc.system", "outbound_shmem"); + otel_span->SetAttribute("rpc.service", "yb.tserver.PgClientService"); + otel_span->SetAttribute("rpc.method", GetSharedMemMethodName(kSharedExchangeRequestType)); + otel_span->DropScope(); } } @@ -1197,8 +1216,30 @@ class PgClient::Impl : public BigDataFetcher { if (tablespace_oid) { lock_oid.set_tablespace_oid(*tablespace_oid); } - req.set_lock_type(static_cast(mode)); + const auto lock_type = static_cast(mode); + req.set_lock_type(lock_type); req.set_is_session_lock(is_session_lock); + + // Publish the details of AcquireObjectLock. + if (dist_trace::HasActiveContext()) { + dist_trace::AddPendingRpcStringAttr( + "rpc.object_lock.database_oid", std::to_string(lock_id.db_oid)); + dist_trace::AddPendingRpcStringAttr( + "rpc.object_lock.relation_oid", std::to_string(lock_id.relation_oid)); + dist_trace::AddPendingRpcStringAttr( + "rpc.object_lock.object_oid", std::to_string(lock_id.object_oid)); + dist_trace::AddPendingRpcStringAttr( + "rpc.object_lock.object_sub_oid", std::to_string(lock_id.object_sub_oid)); + dist_trace::AddPendingRpcStringAttr( + "rpc.object_lock.lock_mode", tserver::ObjectLockMode_Name(lock_type)); + dist_trace::AddPendingRpcStringAttr( + "rpc.object_lock.is_session_lock", is_session_lock ? "true" : "false"); + if (tablespace_oid) { + dist_trace::AddPendingRpcStringAttr( + "rpc.object_lock.tablespace_oid", std::to_string(*tablespace_oid)); + } + } + auto method = [](auto* proxy, const auto& req, auto* resp, auto* controller, auto callback) { proxy->AcquireObjectLockAsync(req, resp, controller, std::move(callback)); }; diff --git a/src/yb/yql/pggate/pg_session.cc b/src/yb/yql/pggate/pg_session.cc index 57078451cc0b..083c3b817a45 100644 --- a/src/yb/yql/pggate/pg_session.cc +++ b/src/yb/yql/pggate/pg_session.cc @@ -109,12 +109,23 @@ void Erase(Container* container, const Key& key) { } void PublishPendingRpcTableInfo( + const PgsqlOps& operations, const std::vector& relations, const std::unordered_map& table_cache) { if (!dist_trace::HasActiveContext() || relations.empty()) { return; } - dist_trace::ClearPendingRpcAttrs(); + + // Publish the details of Perform RPC. + size_t reads = 0; + size_t writes = 0; + for (const auto& op : operations) { + (op->is_read() ? reads : writes)++; + } + dist_trace::AddPendingRpcStringAttr("rpc.perform.op_count", std::to_string(operations.size())); + dist_trace::AddPendingRpcStringAttr("rpc.perform.reads", std::to_string(reads)); + dist_trace::AddPendingRpcStringAttr("rpc.perform.writes", std::to_string(writes)); + std::set unique_relations(relations.begin(), relations.end()); std::string joined_names; for (const auto& relation : unique_relations) { @@ -993,7 +1004,7 @@ Result PgSession::Perform(BufferableOperations&& ops, PerformOpti std::move(ops).MoveTo(operations, relations); // Must run before `relations` is moved into PerformFuture below; otherwise the vector is // empty and no table info gets published for the upcoming RPC client span. - PublishPendingRpcTableInfo(relations, table_cache_); + PublishPendingRpcTableInfo(operations, relations, table_cache_); return PerformFuture( pg_client_.PerformAsync(&options, std::move(operations), metrics_), std::move(relations)); diff --git a/src/yb/yql/pggate/ybc_dist_trace.cc b/src/yb/yql/pggate/ybc_dist_trace.cc index 8e806557000e..160e6cccbb4f 100644 --- a/src/yb/yql/pggate/ybc_dist_trace.cc +++ b/src/yb/yql/pggate/ybc_dist_trace.cc @@ -21,6 +21,7 @@ #include "opentelemetry/trace/tracer.h" #include "yb/util/dist_trace.h" +#include "yb/util/flags.h" #include "yb/util/logging.h" #include "yb/yql/pggate/pg_memctx.h" @@ -96,14 +97,12 @@ void YBCDestroySpanContext(YbcOtelSpanContext span_ctx) { PgMemctx::Destroy(span_ctx); } -void YBCInitDistTrace(int64_t process_pid, const char* node_uuid) { - DCHECK_GT(process_pid, 0); - - dist_trace::InitDistTrace(process_pid, DCHECK_NOTNULL(node_uuid)); +void YBCInitDistTrace(const char* node_uuid) { + dist_trace::InitDistTrace(dist_trace::kYsqlServiceName, DCHECK_NOTNULL(node_uuid)); } -void YBCCleanupDistTrace() { - dist_trace::CleanupDistTrace(); +void YBCShutdownDistTrace() { + dist_trace::ShutdownDistTrace(); } void YBCDistTraceClearStack() { diff --git a/src/yb/yql/pggate/ybc_dist_trace.h b/src/yb/yql/pggate/ybc_dist_trace.h index 092bb6a2583f..569d77d232ea 100644 --- a/src/yb/yql/pggate/ybc_dist_trace.h +++ b/src/yb/yql/pggate/ybc_dist_trace.h @@ -60,8 +60,8 @@ extern "C" { } while (0) bool YBCIsOtelScopeStackEmpty(); -void YBCInitDistTrace(int64_t process_pid, const char* node_uuid); -void YBCCleanupDistTrace(); +void YBCInitDistTrace(const char* node_uuid); +void YBCShutdownDistTrace(); bool YBCIsDistTraceEnabled(); bool YBCIsDistTraceActive(); bool YBCIsTraceParentValidAndRemote(const char* traceparent); diff --git a/src/yb/yql/pgwrapper/dist_trace-test.cc b/src/yb/yql/pgwrapper/dist_trace-test.cc index dfd3c57c5d0f..5ed12c865073 100644 --- a/src/yb/yql/pgwrapper/dist_trace-test.cc +++ b/src/yb/yql/pgwrapper/dist_trace-test.cc @@ -1677,9 +1677,9 @@ TEST_F(DistTraceRpcTest, TestOtelInternalMessagesAreLogged) { RegexWaiterLogSink info_waiter(Format("I.*$0.*", kInfo)); RegexWaiterLogSink debug_waiter(Format("I.*$0.*", kDebug)); - dist_trace::InitDistTrace(0 /* process_pid */, "dist-trace-otel-log-test"); + dist_trace::InitDistTrace("ysql" /* service_name */, "dist-trace-otel-log-test"); auto cleanup = ScopeExit([] { - dist_trace::CleanupDistTrace(); + dist_trace::ShutdownDistTrace(); }); OTEL_INTERNAL_LOG_ERROR(kError); @@ -1705,9 +1705,9 @@ TEST_F(DistTraceRpcTest, TestOtelInternalLogLevelDefaultsToInfo) { RegexWaiterLogSink info_waiter(Format("I.*$0.*", kInfo)); RegexWaiterLogSink debug_waiter(Format("I.*$0.*", kDebug)); - dist_trace::InitDistTrace(0 /* process_pid */, "dist-trace-otel-default-log-level-test"); + dist_trace::InitDistTrace("ysql" /* service_name */, "dist-trace-otel-default-log-level-test"); auto cleanup = ScopeExit([] { - dist_trace::CleanupDistTrace(); + dist_trace::ShutdownDistTrace(); }); OTEL_INTERNAL_LOG_ERROR(kError); @@ -1734,9 +1734,9 @@ TEST_F(DistTraceRpcTest, TestOtelInternalLogLevelGFlagControlsSdkFiltering) { RegexWaiterLogSink info_waiter(Format("I.*$0.*", kInfo)); RegexWaiterLogSink debug_waiter(Format("I.*$0.*", kDebug)); - dist_trace::InitDistTrace(0 /* process_pid */, "dist-trace-otel-error-log-level-test"); + dist_trace::InitDistTrace("ysql" /* service_name */, "dist-trace-otel-error-log-level-test"); auto cleanup = ScopeExit([] { - dist_trace::CleanupDistTrace(); + dist_trace::ShutdownDistTrace(); }); OTEL_INTERNAL_LOG_ERROR(kError); @@ -1765,9 +1765,9 @@ TEST_F(DistTraceRpcTest, TestOtelInternalLogLevelNoneSuppressesAllMessages) { RegexWaiterLogSink info_waiter(Format("I.*$0.*", kInfo)); RegexWaiterLogSink debug_waiter(Format("I.*$0.*", kDebug)); - dist_trace::InitDistTrace(0 /* process_pid */, "dist-trace-otel-none-log-level-test"); + dist_trace::InitDistTrace("ysql" /* service_name */, "dist-trace-otel-none-log-level-test"); auto cleanup = ScopeExit([] { - dist_trace::CleanupDistTrace(); + dist_trace::ShutdownDistTrace(); }); OTEL_INTERNAL_LOG_ERROR(kError); @@ -1789,9 +1789,9 @@ TEST_F(DistTraceRpcTest, TestErroredRpcSpanStatus) { kOtelBatchMaxExportBatchSize; ANNOTATE_UNPROTECTED_WRITE(FLAGS_otel_batch_max_queue_size) = kOtelBatchMaxQueueSize; - dist_trace::InitDistTrace(0 /* process_pid */, "dist-trace-rpc-error-test"); + dist_trace::InitDistTrace("ysql" /* service_name */, "dist-trace-rpc-error-test"); auto cleanup = ScopeExit([] { - dist_trace::CleanupDistTrace(); + dist_trace::ShutdownDistTrace(); }); auto root_span = dist_trace::GetDistTracer()->StartSpan("rpc-error-test");