mirror of
https://github.com/hanzoai/insights.git
synced 2026-10-30 16:15:31 +00:00
zeekay
19fdf55ceb
A key now opens with what it is allowed to touch. `pk-` is publishable and safe in a browser; `sk-` is secret and never leaves a server. The five marks it replaces all began `ph` -- the upstream's initials -- and they were the last of its brand a customer read back to us, in every SDK snippet. No word rule could reach them, because `ph` is not a word. The marks were also not telling the truth. The constant said `phc_` while every live project token said `hi_`, so the middleware that routes on the constant matched no real token at all. Two spellings existed and one was fiction. Minting now happens in exactly one function and reading in exactly one, so a second spelling has nowhere to live. `phx_` (personal) and `phs_` (project secret) both become `sk-`. The mark now says only that a value is secret; which store holds it is settled by asking. That was already how personal keys worked -- keys minted before the marks carry none, so the store always decided -- and both secret resolvers already returned None on a miss and fell through, so one shared extractor replaces the two that could drift apart. OAuth's access and refresh tokens stay distinguishable as `at-` and `rt-`, because the parser tells them apart before either is looked up, and the domain-verification token becomes `vt-` rather than keeping a sixth mint of its own. `bin/debrand --check` now refuses the marks as a shape, so a seventh cannot be coined. Two rules that were about to rename things that are not ours to spell are bounded: `Lemon` no longer eats `LemonSqueezy`, and the `hog_flow` rules no longer reach a Postgres table, a Datastore table or a route. Keys are stored hashed, so a mark cannot be rewritten in place: the three live project tokens have to be reminted.
680 lines
22 KiB
Rust
680 lines
22 KiB
Rust
use bytes::Bytes;
|
|
use capture_logs::service::{parse_otel_message, patch_otel_json};
|
|
use opentelemetry_proto::tonic::common::v1::any_value::Value;
|
|
use prost::Message;
|
|
use serde_json::json;
|
|
|
|
#[test]
|
|
fn test_parse_single_json_otel_message() {
|
|
let json_data = r#"{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"test"},"logRecords":[{"timeUnixNano":"1234567890","severityText":"INFO","body":{"stringValue":"test message"}}]}]}]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
assert_eq!(request.resource_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs[0].log_records.len(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_single_json_otel_message_with_newlines() {
|
|
let json_data = r#"{"resourceLogs":
|
|
[{"resource":{"attributes":[]},
|
|
"scopeLogs":[{"scope":{"name":"test"},"logRecords":[{"timeUnixNano":"1234567890","severityText":"INFO","body":{"stringValue":"test message"}}]}]}]
|
|
}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
assert_eq!(request.resource_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs[0].log_records.len(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_jsonl_otel_message() {
|
|
let jsonl_data = r#"{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"test1"},"logRecords":[{"timeUnixNano":"1234567890","severityText":"INFO","body":{"stringValue":"message 1"}}]}]}]}
|
|
{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"test2"},"logRecords":[{"timeUnixNano":"1234567891","severityText":"WARN","body":{"stringValue":"message 2"}}]}]}]}
|
|
{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"test3"},"logRecords":[{"timeUnixNano":"1234567892","severityText":"ERROR","body":{"stringValue":"message 3"}}]}]}]}"#;
|
|
let bytes = Bytes::from(jsonl_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
// Should have merged all 3 resource logs
|
|
assert_eq!(request.resource_logs.len(), 3);
|
|
|
|
// Verify each resource log has the expected content
|
|
assert_eq!(
|
|
request.resource_logs[0].scope_logs[0]
|
|
.scope
|
|
.as_ref()
|
|
.unwrap()
|
|
.name,
|
|
"test1"
|
|
);
|
|
assert_eq!(
|
|
request.resource_logs[1].scope_logs[0]
|
|
.scope
|
|
.as_ref()
|
|
.unwrap()
|
|
.name,
|
|
"test2"
|
|
);
|
|
assert_eq!(
|
|
request.resource_logs[2].scope_logs[0]
|
|
.scope
|
|
.as_ref()
|
|
.unwrap()
|
|
.name,
|
|
"test3"
|
|
);
|
|
|
|
// Verify log records content
|
|
match request.resource_logs[0].scope_logs[0].log_records[0]
|
|
.body
|
|
.as_ref()
|
|
.unwrap()
|
|
.value
|
|
.as_ref()
|
|
.unwrap()
|
|
{
|
|
Value::StringValue(s) => assert_eq!(s, "message 1"),
|
|
_ => panic!("Expected string value"),
|
|
}
|
|
match request.resource_logs[1].scope_logs[0].log_records[0]
|
|
.body
|
|
.as_ref()
|
|
.unwrap()
|
|
.value
|
|
.as_ref()
|
|
.unwrap()
|
|
{
|
|
Value::StringValue(s) => assert_eq!(s, "message 2"),
|
|
_ => panic!("Expected string value"),
|
|
}
|
|
match request.resource_logs[2].scope_logs[0].log_records[0]
|
|
.body
|
|
.as_ref()
|
|
.unwrap()
|
|
.value
|
|
.as_ref()
|
|
.unwrap()
|
|
{
|
|
Value::StringValue(s) => assert_eq!(s, "message 3"),
|
|
_ => panic!("Expected string value"),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_jsonl_with_empty_lines() {
|
|
let jsonl_data = r#"
|
|
{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"test1"},"logRecords":[{"timeUnixNano":"1234567890","severityText":"INFO","body":{"stringValue":"message 1"}}]}]}]}
|
|
|
|
{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"test2"},"logRecords":[{"timeUnixNano":"1234567891","severityText":"WARN","body":{"stringValue":"message 2"}}]}]}]}
|
|
|
|
"#;
|
|
let bytes = Bytes::from(jsonl_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
// Should have merged both resource logs, ignoring empty lines
|
|
assert_eq!(request.resource_logs.len(), 2);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_invalid_jsonl() {
|
|
let invalid_jsonl = r#"{"resourceLogs":[]}
|
|
invalid json line
|
|
{"resourceLogs":[]}"#;
|
|
let bytes = Bytes::from(invalid_jsonl);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_otel_message_with_empty_body() {
|
|
let json_data = r#"{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"test"},"logRecords":[{"timeUnixNano":"1234567890","severityText":"INFO","body":{}}]}]}]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
assert_eq!(request.resource_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs[0].log_records.len(), 1);
|
|
|
|
// Body should be None after patching empty object to null
|
|
assert!(request.resource_logs[0].scope_logs[0].log_records[0]
|
|
.body
|
|
.is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_otel_message_with_empty_value_in_attributes() {
|
|
let json_data = r#"{"resourceLogs":[{"resource":{"attributes":[{"key":"test_key","value":{}}]},"scopeLogs":[{"scope":{"name":"test"},"logRecords":[{"timeUnixNano":"1234567890","severityText":"INFO","body":{"stringValue":"test message"},"attributes":[{"key":"attr_key","value":{}}]}]}]}]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
assert_eq!(request.resource_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs[0].log_records.len(), 1);
|
|
|
|
// Resource attribute with empty value should be None after patching
|
|
assert_eq!(
|
|
request.resource_logs[0]
|
|
.resource
|
|
.as_ref()
|
|
.unwrap()
|
|
.attributes
|
|
.len(),
|
|
1
|
|
);
|
|
assert!(request.resource_logs[0]
|
|
.resource
|
|
.as_ref()
|
|
.unwrap()
|
|
.attributes[0]
|
|
.value
|
|
.is_none());
|
|
|
|
// Log record attribute with empty value should be None after patching
|
|
assert_eq!(
|
|
request.resource_logs[0].scope_logs[0].log_records[0]
|
|
.attributes
|
|
.len(),
|
|
1
|
|
);
|
|
assert!(
|
|
request.resource_logs[0].scope_logs[0].log_records[0].attributes[0]
|
|
.value
|
|
.is_none()
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_otel_message_with_mixed_empty_and_valid_values() {
|
|
let json_data = r#"{"resourceLogs":[{"resource":{"attributes":[{"key":"valid_key","value":{"stringValue":"valid_value"}},{"key":"empty_key","value":{}}]},"scopeLogs":[{"scope":{"name":"test"},"logRecords":[{"timeUnixNano":"1234567890","severityText":"INFO","body":{},"attributes":[{"key":"valid_attr","value":{"intValue":42}},{"key":"empty_attr","value":{}}]}]}]}]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
let resource = request.resource_logs[0].resource.as_ref().unwrap();
|
|
let log_record = &request.resource_logs[0].scope_logs[0].log_records[0];
|
|
|
|
// Resource attributes: valid one should have value, empty one should be None
|
|
assert_eq!(resource.attributes.len(), 2);
|
|
match resource.attributes[0]
|
|
.value
|
|
.as_ref()
|
|
.unwrap()
|
|
.value
|
|
.as_ref()
|
|
.unwrap()
|
|
{
|
|
Value::StringValue(s) => assert_eq!(s, "valid_value"),
|
|
_ => panic!("Expected string value"),
|
|
}
|
|
assert!(resource.attributes[1].value.is_none());
|
|
|
|
// Body should be None after patching
|
|
assert!(log_record.body.is_none());
|
|
|
|
// Log record attributes: valid one should have value, empty one should be None
|
|
assert_eq!(log_record.attributes.len(), 2);
|
|
match log_record.attributes[0]
|
|
.value
|
|
.as_ref()
|
|
.unwrap()
|
|
.value
|
|
.as_ref()
|
|
.unwrap()
|
|
{
|
|
Value::IntValue(i) => assert_eq!(*i, 42),
|
|
_ => panic!("Expected int value"),
|
|
}
|
|
assert!(log_record.attributes[1].value.is_none());
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_otel_message_with_nested_empty_values() {
|
|
// Test that empty values in nested structures are also patched
|
|
let json_data = r#"{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"test"},"logRecords":[{"timeUnixNano":"1234567890","severityText":"INFO","body":{"kvlistValue":{"values":[{"key":"nested_key","value":{}}]}},"attributes":[{"key":"nested_attr","value":{"kvlistValue":{"values":[{"key":"inner_key","value":{}}]}}}]}]}]}]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
assert_eq!(request.resource_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs[0].log_records.len(), 1);
|
|
|
|
// The parsing should succeed even with nested empty values that get patched to null
|
|
let log_record = &request.resource_logs[0].scope_logs[0].log_records[0];
|
|
assert!(log_record.body.is_some());
|
|
assert_eq!(log_record.attributes.len(), 1);
|
|
}
|
|
|
|
#[test]
|
|
fn test_patch_otel_json_empty_value() {
|
|
let mut json = json!({
|
|
"key": "test",
|
|
"value": {}
|
|
});
|
|
|
|
patch_otel_json(&mut json);
|
|
|
|
assert_eq!(json["key"], "test");
|
|
assert!(json["value"].is_null());
|
|
}
|
|
|
|
#[test]
|
|
fn test_patch_otel_json_empty_body() {
|
|
let mut json = json!({
|
|
"timeUnixNano": "1234567890",
|
|
"body": {}
|
|
});
|
|
|
|
patch_otel_json(&mut json);
|
|
|
|
assert_eq!(json["timeUnixNano"], "1234567890");
|
|
assert!(json["body"].is_null());
|
|
}
|
|
|
|
#[test]
|
|
fn test_patch_otel_json_non_empty_values() {
|
|
let mut json = json!({
|
|
"value": {
|
|
"stringValue": "test"
|
|
},
|
|
"body": {
|
|
"intValue": 42
|
|
}
|
|
});
|
|
|
|
patch_otel_json(&mut json);
|
|
|
|
// Non-empty values should remain unchanged
|
|
assert_eq!(json["value"]["stringValue"], "test");
|
|
assert_eq!(json["body"]["intValue"], 42);
|
|
}
|
|
|
|
#[test]
|
|
fn test_patch_otel_json_nested_empty_values() {
|
|
let mut json = json!({
|
|
"resourceLogs": [{
|
|
"resource": {
|
|
"attributes": [{
|
|
"key": "test_key",
|
|
"value": {}
|
|
}]
|
|
},
|
|
"scopeLogs": [{
|
|
"logRecords": [{
|
|
"body": {},
|
|
"attributes": [{
|
|
"key": "attr_key",
|
|
"value": {}
|
|
}]
|
|
}]
|
|
}]
|
|
}]
|
|
});
|
|
|
|
patch_otel_json(&mut json);
|
|
|
|
// All empty values and bodies should be patched to null
|
|
assert!(json["resourceLogs"][0]["resource"]["attributes"][0]["value"].is_null());
|
|
assert!(json["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0]["body"].is_null());
|
|
assert!(
|
|
json["resourceLogs"][0]["scopeLogs"][0]["logRecords"][0]["attributes"][0]["value"]
|
|
.is_null()
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn test_patch_otel_json_array_with_empty_values() {
|
|
let mut json = json!([
|
|
{
|
|
"value": {}
|
|
},
|
|
{
|
|
"body": {}
|
|
},
|
|
{
|
|
"value": {
|
|
"stringValue": "valid"
|
|
}
|
|
}
|
|
]);
|
|
|
|
patch_otel_json(&mut json);
|
|
|
|
// Empty objects should be patched to null
|
|
assert!(json[0]["value"].is_null());
|
|
assert!(json[1]["body"].is_null());
|
|
// Valid values should remain unchanged
|
|
assert_eq!(json[2]["value"]["stringValue"], "valid");
|
|
}
|
|
|
|
#[test]
|
|
fn test_patch_otel_json_mixed_empty_and_non_empty() {
|
|
let mut json = json!({
|
|
"logRecord": {
|
|
"body": {},
|
|
"attributes": [
|
|
{
|
|
"key": "empty_attr",
|
|
"value": {}
|
|
},
|
|
{
|
|
"key": "valid_attr",
|
|
"value": {
|
|
"stringValue": "test_value"
|
|
}
|
|
}
|
|
]
|
|
}
|
|
});
|
|
|
|
patch_otel_json(&mut json);
|
|
|
|
// Empty body should be null
|
|
assert!(json["logRecord"]["body"].is_null());
|
|
// Empty attribute value should be null
|
|
assert!(json["logRecord"]["attributes"][0]["value"].is_null());
|
|
// Valid attribute value should remain unchanged
|
|
assert_eq!(
|
|
json["logRecord"]["attributes"][1]["value"]["stringValue"],
|
|
"test_value"
|
|
);
|
|
}
|
|
|
|
// Tests for empty request handling - verifying 200 OK with no log records produced
|
|
|
|
#[test]
|
|
fn test_parse_empty_json_otel_message() {
|
|
// Empty resourceLogs array - valid OTEL message with no logs
|
|
let json_data = r#"{"resourceLogs":[]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
assert_eq!(request.resource_logs.len(), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_empty_scope_logs_otel_message() {
|
|
// Resource with empty scopeLogs array
|
|
let json_data = r#"{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[]}]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
assert_eq!(request.resource_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs.len(), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_empty_log_records_otel_message() {
|
|
// Scope with empty logRecords array
|
|
let json_data = r#"{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"test"},"logRecords":[]}]}]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
assert_eq!(request.resource_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs.len(), 1);
|
|
assert_eq!(request.resource_logs[0].scope_logs[0].log_records.len(), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_empty_protobuf_otel_message() {
|
|
// Empty ExportLogsServiceRequest in protobuf format
|
|
use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest;
|
|
|
|
let empty_request = ExportLogsServiceRequest {
|
|
resource_logs: vec![],
|
|
};
|
|
|
|
let mut buf = Vec::new();
|
|
empty_request.encode(&mut buf).unwrap();
|
|
let bytes = Bytes::from(buf);
|
|
|
|
// parse_otel_message tries protobuf first via the endpoint, but here we test JSON parsing
|
|
// So let's verify the protobuf decodes correctly
|
|
let decoded = ExportLogsServiceRequest::decode(bytes.as_ref()).unwrap();
|
|
assert_eq!(decoded.resource_logs.len(), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_count_log_records_empty_resource_logs() {
|
|
let json_data = r#"{"resourceLogs":[]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let request = parse_otel_message(&bytes).unwrap();
|
|
|
|
// Simulate the row counting logic from export_logs_http
|
|
let mut row_count = 0;
|
|
for resource_logs in &request.resource_logs {
|
|
for scope_logs in &resource_logs.scope_logs {
|
|
row_count += scope_logs.log_records.len();
|
|
}
|
|
}
|
|
|
|
assert_eq!(row_count, 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_count_log_records_empty_scope_logs() {
|
|
let json_data = r#"{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[]}]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let request = parse_otel_message(&bytes).unwrap();
|
|
|
|
let mut row_count = 0;
|
|
for resource_logs in &request.resource_logs {
|
|
for scope_logs in &resource_logs.scope_logs {
|
|
row_count += scope_logs.log_records.len();
|
|
}
|
|
}
|
|
|
|
assert_eq!(row_count, 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_count_log_records_empty_log_records_array() {
|
|
let json_data = r#"{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"test"},"logRecords":[]}]}]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let request = parse_otel_message(&bytes).unwrap();
|
|
|
|
let mut row_count = 0;
|
|
for resource_logs in &request.resource_logs {
|
|
for scope_logs in &resource_logs.scope_logs {
|
|
row_count += scope_logs.log_records.len();
|
|
}
|
|
}
|
|
|
|
assert_eq!(row_count, 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_count_log_records_multiple_empty_scopes() {
|
|
// Multiple resources and scopes, all with empty log records
|
|
let json_data = r#"{"resourceLogs":[{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"scope1"},"logRecords":[]},{"scope":{"name":"scope2"},"logRecords":[]}]},{"resource":{"attributes":[]},"scopeLogs":[{"scope":{"name":"scope3"},"logRecords":[]}]}]}"#;
|
|
let bytes = Bytes::from(json_data);
|
|
|
|
let request = parse_otel_message(&bytes).unwrap();
|
|
|
|
assert_eq!(request.resource_logs.len(), 2);
|
|
assert_eq!(request.resource_logs[0].scope_logs.len(), 2);
|
|
assert_eq!(request.resource_logs[1].scope_logs.len(), 1);
|
|
|
|
let mut row_count = 0;
|
|
for resource_logs in &request.resource_logs {
|
|
for scope_logs in &resource_logs.scope_logs {
|
|
row_count += scope_logs.log_records.len();
|
|
}
|
|
}
|
|
|
|
assert_eq!(row_count, 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_empty_jsonl_otel_message() {
|
|
// Empty lines only - should result in empty resource_logs
|
|
let jsonl_data = "\n\n\n";
|
|
let bytes = Bytes::from(jsonl_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
assert_eq!(request.resource_logs.len(), 0);
|
|
}
|
|
|
|
#[test]
|
|
fn test_parse_jsonl_with_empty_resource_logs() {
|
|
// JSONL where each line has empty resourceLogs
|
|
let jsonl_data = r#"{"resourceLogs":[]}
|
|
{"resourceLogs":[]}"#;
|
|
let bytes = Bytes::from(jsonl_data);
|
|
|
|
let result = parse_otel_message(&bytes);
|
|
assert!(result.is_ok());
|
|
|
|
let request = result.unwrap();
|
|
// Both lines have empty resourceLogs, so merged result is also empty
|
|
assert_eq!(request.resource_logs.len(), 0);
|
|
}
|
|
#[test]
|
|
fn test_otel_log_row_overridden_timestamp_tracking() {
|
|
use capture_logs::log_record::KafkaLogRow;
|
|
use chrono::{TimeDelta, Utc};
|
|
use opentelemetry_proto::tonic::{
|
|
common::v1::{any_value, AnyValue},
|
|
logs::v1::LogRecord,
|
|
};
|
|
|
|
// Test with a timestamp far in the past (should be overridden)
|
|
let far_past_nanos = (Utc::now() - TimeDelta::hours(48))
|
|
.timestamp_nanos_opt()
|
|
.unwrap();
|
|
let log_record_overridden = LogRecord {
|
|
time_unix_nano: far_past_nanos as u64,
|
|
severity_text: "INFO".to_string(),
|
|
severity_number: 9,
|
|
body: Some(AnyValue {
|
|
value: Some(any_value::Value::StringValue("Test".to_string())),
|
|
}),
|
|
..Default::default()
|
|
};
|
|
|
|
let (row_overridden, was_overridden) =
|
|
KafkaLogRow::new(log_record_overridden, None, None).unwrap();
|
|
assert!(was_overridden);
|
|
assert!(row_overridden.attributes.contains_key("$originalTimestamp"));
|
|
|
|
// Test with a timestamp far in the future (should be overridden)
|
|
let far_future_nanos = (Utc::now() + TimeDelta::hours(25))
|
|
.timestamp_nanos_opt()
|
|
.unwrap();
|
|
let log_record_overridden = LogRecord {
|
|
time_unix_nano: far_future_nanos as u64,
|
|
severity_text: "INFO".to_string(),
|
|
severity_number: 9,
|
|
body: Some(AnyValue {
|
|
value: Some(any_value::Value::StringValue("Test".to_string())),
|
|
}),
|
|
..Default::default()
|
|
};
|
|
|
|
let (row_overridden, was_overridden) =
|
|
KafkaLogRow::new(log_record_overridden, None, None).unwrap();
|
|
assert!(was_overridden);
|
|
assert!(row_overridden.attributes.contains_key("$originalTimestamp"));
|
|
|
|
// Test with a recent timestamp (should not be overridden)
|
|
let recent_nanos = (Utc::now() - TimeDelta::hours(1))
|
|
.timestamp_nanos_opt()
|
|
.unwrap();
|
|
let log_record_recent = LogRecord {
|
|
time_unix_nano: recent_nanos as u64,
|
|
severity_text: "INFO".to_string(),
|
|
severity_number: 9,
|
|
body: Some(AnyValue {
|
|
value: Some(any_value::Value::StringValue("Test".to_string())),
|
|
}),
|
|
..Default::default()
|
|
};
|
|
|
|
let (row_recent, was_not_overridden) = KafkaLogRow::new(log_record_recent, None, None).unwrap();
|
|
assert!(!was_not_overridden);
|
|
assert!(!row_recent.attributes.contains_key("$originalTimestamp"));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn middleware_decompresses_gzip_js_query_param() {
|
|
use axum::{body::Body, http::Request, routing::post, Router};
|
|
use bytes::Bytes;
|
|
use capture_logs::middleware::translate_compression_query_param;
|
|
use flate2::{write::GzEncoder, Compression};
|
|
use std::io::Write;
|
|
use tower::ServiceExt;
|
|
use tower_http::decompression::RequestDecompressionLayer;
|
|
|
|
async fn echo(body: Bytes) -> Bytes {
|
|
body
|
|
}
|
|
|
|
let app = Router::new()
|
|
.route("/i/v1/logs", post(echo))
|
|
.layer(RequestDecompressionLayer::new())
|
|
.layer(axum::middleware::from_fn(translate_compression_query_param));
|
|
|
|
let payload = br#"{"resourceLogs":[]}"#.to_vec();
|
|
let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
|
|
encoder.write_all(&payload).unwrap();
|
|
let gzipped = encoder.finish().unwrap();
|
|
|
|
let req = Request::builder()
|
|
.method("POST")
|
|
.uri("/i/v1/logs?token=pk-test&compression=gzip-js")
|
|
.header("content-type", "text/plain")
|
|
.body(Body::from(gzipped))
|
|
.unwrap();
|
|
|
|
let resp = app.oneshot(req).await.unwrap();
|
|
assert!(
|
|
resp.status().is_success(),
|
|
"expected 2xx, got {}",
|
|
resp.status()
|
|
);
|
|
let bytes = axum::body::to_bytes(resp.into_body(), 1 << 20)
|
|
.await
|
|
.unwrap();
|
|
assert_eq!(&bytes[..], payload.as_slice());
|
|
}
|