Skip to content

Commit 5b04aa6

Browse files
ddelnanoddelnano
committed
Add log support to OTelExportSinkIR
Signed-off-by: Dom Del Nano <ddelnano@gmail.com> (cherry picked from commit 9d5e616)
1 parent 301198f commit 5b04aa6

File tree

3 files changed

+205
-0
lines changed

3 files changed

+205
-0
lines changed

‎src/carnot/planner/ir/otel_export_sink_ir.cc

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -160,6 +160,29 @@ Status OTelExportSinkIR::ProcessConfig(const OTelData& data) {
160160
new_span.span_kind = span.span_kind;
161161
data_.spans.push_back(std::move(new_span));
162162
}
163+
for (const auto& log : data.logs) {
164+
OTelLog new_log;
165+
166+
PX_ASSIGN_OR_RETURN(new_log.time_column, AddColumn(log.time_column));
167+
PX_ASSIGN_OR_RETURN(new_log.body_column, AddColumn(log.body_column));
168+
if (log.observed_time_column != nullptr) {
169+
PX_ASSIGN_OR_RETURN(new_log.observed_time_column, AddColumn(log.observed_time_column));
170+
}
171+
172+
new_log.severity_text = log.severity_text;
173+
new_log.severity_number = log.severity_number;
174+
175+
for (const auto& attr : log.attributes) {
176+
if (attr.column_reference == nullptr) {
177+
new_log.attributes.push_back({attr.name, nullptr, attr.string_value});
178+
continue;
179+
}
180+
PX_ASSIGN_OR_RETURN(auto column, AddColumn(attr.column_reference));
181+
new_log.attributes.push_back({attr.name, column, ""});
182+
}
183+
184+
data_.logs.push_back(std::move(new_log));
185+
}
163186
return Status::OK();
164187
}
165188

@@ -330,6 +353,46 @@ Status OTelExportSinkIR::ToProto(planpb::Operator* op) const {
330353
}
331354
span_pb->set_kind_value(span.span_kind);
332355
}
356+
for (const auto& log : data_.logs) {
357+
auto log_pb = otel_op->add_logs();
358+
359+
if (log.time_column->EvaluatedDataType() != types::TIME64NS) {
360+
return log.time_column->CreateIRNodeError(
361+
"Expected time column '$0' to be TIME64NS, received $1", log.time_column->col_name(),
362+
types::ToString(log.time_column->EvaluatedDataType()));
363+
}
364+
PX_ASSIGN_OR_RETURN(auto time_column_index, log.time_column->GetColumnIndex());
365+
log_pb->set_time_column_index(time_column_index);
366+
367+
if (log.observed_time_column != nullptr) {
368+
if (log.observed_time_column->EvaluatedDataType() != types::TIME64NS) {
369+
return log.observed_time_column->CreateIRNodeError(
370+
"Expected observed_time column '$0' to be TIME64NS, received $1",
371+
log.observed_time_column->col_name(),
372+
types::ToString(log.observed_time_column->EvaluatedDataType()));
373+
}
374+
PX_ASSIGN_OR_RETURN(auto observed_time_column_index,
375+
log.observed_time_column->GetColumnIndex());
376+
log_pb->set_observed_time_column_index(observed_time_column_index);
377+
} else {
378+
log_pb->set_observed_time_column_index(-1);
379+
}
380+
381+
log_pb->set_severity_text(log.severity_text);
382+
log_pb->set_severity_number(log.severity_number);
383+
384+
if (log.body_column->EvaluatedDataType() != types::STRING) {
385+
return log.body_column->CreateIRNodeError(
386+
"Expected body column '$0' to be STRING, received $1", log.body_column->col_name(),
387+
types::ToString(log.body_column->EvaluatedDataType()));
388+
}
389+
PX_ASSIGN_OR_RETURN(auto body_column_index, log.body_column->GetColumnIndex());
390+
log_pb->set_body_column_index(body_column_index);
391+
392+
for (const auto& attribute : log.attributes) {
393+
PX_RETURN_IF_ERROR(attribute.ToProto(log_pb->add_attributes()));
394+
}
395+
}
333396
return Status::OK();
334397
}
335398

‎src/carnot/planner/ir/otel_export_sink_ir.h

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,11 +127,23 @@ struct OTelSpan {
127127
int64_t span_kind;
128128
};
129129

130+
struct OTelLog {
131+
std::vector<OTelAttribute> attributes;
132+
133+
ColumnIR* time_column;
134+
ColumnIR* observed_time_column = nullptr;
135+
ColumnIR* body_column;
136+
137+
int64_t severity_number;
138+
std::string severity_text;
139+
};
140+
130141
struct OTelData {
131142
planpb::OTelEndpointConfig endpoint_config;
132143
std::vector<OTelAttribute> resource_attributes;
133144
std::vector<OTelMetric> metrics;
134145
std::vector<OTelSpan> spans;
146+
std::vector<OTelLog> logs;
135147
};
136148

137149
/**

‎src/carnot/planner/ir/otel_export_sink_ir_test.cc

Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -443,6 +443,84 @@ INSTANTIATE_TEST_SUITE_P(
443443
.ConsumeValueOrDie();
444444
},
445445
},
446+
{
447+
"logs_basic",
448+
table_store::schema::Relation{{types::TIME64NS, types::STRING, types::STRING},
449+
{"start_time", "attribute_str", "log_message"},
450+
{types::ST_NONE, types::ST_NONE, types::ST_NONE}},
451+
R"pb(
452+
endpoint_config {}
453+
resource {}
454+
logs {
455+
attributes {
456+
name: "service.name"
457+
column {
458+
column_type: STRING
459+
column_index: 1
460+
}
461+
}
462+
time_column_index: 0
463+
observed_time_column_index: -1
464+
severity_number: 4
465+
severity_text: "INFO"
466+
body_column_index: 2
467+
}
468+
)pb",
469+
[](IR* graph, OperatorIR* parent, table_store::schema::Relation* relation) {
470+
OTelData data;
471+
472+
auto& log = data.logs.emplace_back();
473+
log.time_column = CreateTypedColumn(graph, "start_time", relation);
474+
log.attributes.push_back(
475+
{"service.name", CreateTypedColumn(graph, "attribute_str", relation), ""});
476+
log.severity_number = 4;
477+
log.severity_text = "INFO";
478+
log.body_column = CreateTypedColumn(graph, "log_message", relation);
479+
480+
return graph->CreateNode<OTelExportSinkIR>(parent->ast(), parent, data)
481+
.ConsumeValueOrDie();
482+
},
483+
},
484+
{
485+
"logs_with_observed_time_col",
486+
table_store::schema::Relation{
487+
{types::TIME64NS, types::TIME64NS, types::STRING, types::STRING},
488+
{"start_time", "observed_time", "attribute_str", "log_message"},
489+
{types::ST_NONE, types::ST_NONE, types::ST_NONE, types::ST_NONE}},
490+
R"pb(
491+
endpoint_config {}
492+
resource {}
493+
logs {
494+
attributes {
495+
name: "service.name"
496+
column {
497+
column_type: STRING
498+
column_index: 2
499+
}
500+
}
501+
time_column_index: 0
502+
observed_time_column_index: 1
503+
severity_number: 4
504+
severity_text: "INFO"
505+
body_column_index: 3
506+
}
507+
)pb",
508+
[](IR* graph, OperatorIR* parent, table_store::schema::Relation* relation) {
509+
OTelData data;
510+
511+
auto& log = data.logs.emplace_back();
512+
log.time_column = CreateTypedColumn(graph, "start_time", relation);
513+
log.observed_time_column = CreateTypedColumn(graph, "observed_time", relation);
514+
log.attributes.push_back(
515+
{"service.name", CreateTypedColumn(graph, "attribute_str", relation), ""});
516+
log.severity_number = 4;
517+
log.severity_text = "INFO";
518+
log.body_column = CreateTypedColumn(graph, "log_message", relation);
519+
520+
return graph->CreateNode<OTelExportSinkIR>(parent->ast(), parent, data)
521+
.ConsumeValueOrDie();
522+
},
523+
},
446524
{
447525
"string_value_attributes",
448526
table_store::schema::Relation{{types::TIME64NS, types::INT64},
@@ -557,6 +635,33 @@ OTelExportSinkIR* CreateSpanWithNameString(IR* graph, OperatorIR* parent,
557635
return graph->CreateNode<OTelExportSinkIR>(parent->ast(), parent, data).ConsumeValueOrDie();
558636
}
559637

638+
OTelExportSinkIR* CreateLog(IR* graph, OperatorIR* parent,
639+
table_store::schema::Relation* relation) {
640+
OTelData data;
641+
642+
auto& log = data.logs.emplace_back();
643+
log.time_column = CreateTypedColumn(graph, "start_time", relation);
644+
log.body_column = CreateTypedColumn(graph, "log_message", relation);
645+
log.severity_number = 4;
646+
log.severity_text = "INFO";
647+
648+
return graph->CreateNode<OTelExportSinkIR>(parent->ast(), parent, data).ConsumeValueOrDie();
649+
}
650+
651+
OTelExportSinkIR* CreateLogWithObservedTime(IR* graph, OperatorIR* parent,
652+
table_store::schema::Relation* relation) {
653+
OTelData data;
654+
655+
auto& log = data.logs.emplace_back();
656+
log.time_column = CreateTypedColumn(graph, "start_time", relation);
657+
log.observed_time_column = CreateTypedColumn(graph, "observed_time", relation);
658+
log.body_column = CreateTypedColumn(graph, "log_message", relation);
659+
log.severity_number = 4;
660+
log.severity_text = "INFO";
661+
662+
return graph->CreateNode<OTelExportSinkIR>(parent->ast(), parent, data).ConsumeValueOrDie();
663+
}
664+
560665
INSTANTIATE_TEST_SUITE_P(
561666
ErrorTests, WrongColumnTypesTest,
562667
::testing::ValuesIn(std::vector<WrongColumnTypesTestCase>{
@@ -723,6 +828,31 @@ INSTANTIATE_TEST_SUITE_P(
723828
.ConsumeValueOrDie();
724829
},
725830
},
831+
{
832+
"log_time_column_wrong",
833+
table_store::schema::Relation{{types::INT64, types::STRING, types::STRING},
834+
{"start_time", "attribute_str", "log_message"},
835+
{types::ST_NONE, types::ST_NONE, types::ST_NONE}},
836+
"Expected time column 'start_time' to be TIME64NS, received INT64",
837+
&CreateLog,
838+
},
839+
{
840+
"log_body_column_wrong",
841+
table_store::schema::Relation{{types::TIME64NS, types::STRING, types::TIME64NS},
842+
{"start_time", "attribute_str", "log_message"},
843+
{types::ST_NONE, types::ST_NONE, types::ST_NONE}},
844+
"Expected body column 'log_message' to be STRING, received TIME64NS",
845+
&CreateLog,
846+
},
847+
{
848+
"log_observed_time_column_wrong",
849+
table_store::schema::Relation{
850+
{types::TIME64NS, types::INT64, types::STRING, types::STRING},
851+
{"start_time", "observed_time", "attribute_str", "log_message"},
852+
{types::ST_NONE, types::ST_NONE, types::ST_NONE, types::ST_NONE}},
853+
"Expected observed_time column 'observed_time' to be TIME64NS, received INT64",
854+
&CreateLogWithObservedTime,
855+
},
726856
}),
727857
[](const ::testing::TestParamInfo<WrongColumnTypesTestCase>& info) { return info.param.name; });
728858
} // namespace planner

0 commit comments

Comments
 (0)