Skip to content

Commit

Permalink
Solve comments.
Browse files Browse the repository at this point in the history 
* Avoid sending tags for submitted_payloads, active_sessions and
  memory_usage.
* Comment code managing trace_api_bytes as there is no suppor it.
* Add status code to HTTP error logs.
* Define constants for for literals.
* Replace 'trace-chunk-*' for 'trace-chunks-*.
  • Loading branch information
@hoolioh
hoolioh committed May 27, 2024
1 parent 512cdc5 commit 711e6e2
Show file tree
Hide file tree
Showing 2 changed files with 49 additions and 33 deletions.
55 changes: 29 additions & 26 deletions sidecar/src/self_telemetry.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,9 +27,10 @@ struct MetricData<'a> {
trace_api_requests: ContextKey,
trace_api_responses: ContextKey,
trace_api_errors: ContextKey,
trace_api_bytes: ContextKey,
trace_chunk_sent: ContextKey,
trace_chunk_dropped: ContextKey,
// TODO: APMSP-1157 - Enable this metric when support is enabled.
// trace_api_bytes: ContextKey,
trace_chunks_sent: ContextKey,
trace_chunks_dropped: ContextKey,
}
impl<'a> MetricData<'a> {
async fn send(&self, key: ContextKey, value: f64, tags: Vec<Tag>) {
Expand All @@ -46,19 +47,19 @@ impl<'a> MetricData<'a> {
self.send(
self.submitted_payloads,
self.server.submitted_payloads.swap(0, Ordering::Relaxed) as f64,
vec![Tag::new("src_library", "libdatadog").unwrap()],
vec![],
),
self.send(
self.active_sessions,
self.server.active_session_count() as f64,
vec![Tag::new("src_library", "libdatadog").unwrap()],
vec![],
),
self.send(
self.memory_usage,
self.sidecar_watchdog
.mem_usage_bytes
.load(Ordering::Relaxed) as f64,
vec![Tag::new("src_library", "libdatadog").unwrap()],
vec![],
),
];
for (level, count) in log::MULTI_LOG_FILTER
Expand Down Expand Up @@ -111,23 +112,24 @@ impl<'a> MetricData<'a> {
],
));
}
if trace_metrics.bytes_sent > 0 {
futures.push(self.send(
self.trace_api_bytes,
trace_metrics.bytes_sent as f64,
vec![Tag::new("src_library", "libdatadog").unwrap()],
));
}
// TODO: APMSP-1157 - Enable this metric when support is enabled.
// if trace_metrics.bytes_sent > 0 {
// futures.push(self.send(
// self.trace_api_bytes,
// trace_metrics.bytes_sent as f64,
// vec![Tag::new("src_library", "libdatadog").unwrap()],
// ));
// }
if trace_metrics.chunks_sent > 0 {
futures.push(self.send(
self.trace_chunk_sent,
self.trace_chunks_sent,
trace_metrics.chunks_sent as f64,
vec![Tag::new("src_library", "libdatadog").unwrap()],
));
}
if trace_metrics.chunks_dropped > 0 {
futures.push(self.send(
self.trace_chunk_dropped,
self.trace_chunks_dropped,
trace_metrics.chunks_dropped as f64,
vec![Tag::new("src_library", "libdatadog").unwrap()],
));
Expand Down Expand Up @@ -256,22 +258,23 @@ impl SelfTelemetry {
true,
MetricNamespace::Tracers,
),
trace_api_bytes: worker.register_metric_context(
"trace_api_bytes".to_string(),
vec![],
MetricType::Count,
true,
MetricNamespace::Tracers,
),
trace_chunk_sent: worker.register_metric_context(
"trace_chunk_sent".to_string(),
// TODO: APMSP-1157 - Enable this metric when support is enabled.
// trace_api_bytes: worker.register_metric_context(
// "trace_api_bytes".to_string(),
// vec![],
// MetricType::Distribution,
// true,
// MetricNamespace::Tracers,
// ),
trace_chunks_sent: worker.register_metric_context(
"trace_chunks_sent".to_string(),
vec![],
MetricType::Count,
true,
MetricNamespace::Tracers,
),
trace_chunk_dropped: worker.register_metric_context(
"trace_chunk_dropped".to_string(),
trace_chunks_dropped: worker.register_metric_context(
"trace_chunks_dropped".to_string(),
vec![],
MetricType::Count,
true,
Expand Down
27 changes: 20 additions & 7 deletions trace-utils/src/send_data.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -11,15 +11,26 @@ use crate::tracer_header_tags::TracerHeaderTags;
use datadog_trace_protobuf::pb::{self, TracerPayload};
use ddcommon::{connector, Endpoint, HttpRequestBuilder};

const DD_API_KEY: &str = "DD-API-KEY";
const HEADER_DD_TRACE_COUNT: &str = "X-Datadog-Trace-Count";
const HEADER_HTTP_CTYPE: &str = "Content-Type";
const HEADER_CTYPE_MSGPACK: &str = "application/msgpack";
const HEADER_CTYPE_PROTOBUF: &str = "application/x-protobuf";

type BytesSent = u64;
type ChunksSent = u64;
type ChunksDropped = u64;

pub enum RequestResult {
// Holds information from a succesful request.
Success((Response<Body>, BytesSent, ChunksSent)),
// Treats HTTP errors.
Error((Response<Body>, ChunksDropped)),
// Treats timeout errors originated in the transport layer.
TimeoutError(ChunksDropped),
// Treats errors coming from networking.
NetworkError(ChunksDropped),
// Treats errors coming from building the request
BuildError(ChunksDropped),
}

Expand Down Expand Up @@ -93,18 +104,21 @@ impl SendDataResult {
self.last_result = Ok(response);
}
RequestResult::Error((response, chunks)) => {
let status_code = response.status().as_u16();
self.errors_status_code += 1;
*self
.responses_count_per_code
.entry(response.status().as_u16())
.entry(status_code)
.or_default() += 1;
self.chunks_dropped += chunks;

let body_bytes = hyper::body::to_bytes(response.into_body()).await;
let response_body =
String::from_utf8(body_bytes.unwrap_or_default().to_vec()).unwrap_or_default();
self.last_result = Err(anyhow::format_err!(
"Server did not accept traces: {response_body}"
"{} - Server did not accept traces: {}",
status_code,
response_body,
));
}
RequestResult::TimeoutError(chunks) => {
Expand Down Expand Up @@ -172,7 +186,7 @@ impl SendData {
target: &Endpoint,
) -> SendData {
let headers = if let Some(api_key) = &target.api_key {
HashMap::from([("DD-API-KEY", api_key.as_ref().to_string())])
HashMap::from([(DD_API_KEY, api_key.as_ref().to_string())])
} else {
tracer_header_tags.into()
};
Expand Down Expand Up @@ -249,11 +263,10 @@ impl SendData {
}
}

//let mut result = SendDataResult::new();
let mut result = SendDataResult::default();

if target.api_key.is_some() {
req = req.header("Content-type", "application/x-protobuf");
req = req.header(HEADER_HTTP_CTYPE, HEADER_CTYPE_PROTOBUF);

let mut chunks: u64 = 0;
for tracer_payload in &self.tracer_payloads {
Expand All @@ -272,7 +285,7 @@ impl SendData {
result.update(res).await;
result
} else {
req = req.header("Content-type", "application/msgpack");
req = req.header(HEADER_HTTP_CTYPE, HEADER_CTYPE_MSGPACK);

let (template, _) = req.body(()).unwrap().into_parts();

Expand All @@ -283,7 +296,7 @@ impl SendData {
.method(template.method.clone())
.uri(template.uri.clone())
.version(template.version)
.header("X-Datadog-Trace-Count", chunks.to_string());
.header(HEADER_DD_TRACE_COUNT, chunks.to_string());
builder
.headers_mut()
.unwrap()
Expand Down

0 comments on commit 711e6e2

Please sign in to comment.