Add OTel log support to Carnot plan protos and implement

exec node support

Signed-off-by: Dom Del Nano <ddelnano@gmail.com>
(cherry picked from commit ac0d014)

Author: ddelnano

bazel/external/opentelemetry.BUILD

@@ -123,3 +123,42 @@ cc_grpc_library(
     grpc_only = True,
     deps = [":metrics_service_proto_cc"],
 )
+
+proto_library(
+    name = "logs_proto",
+    srcs = [
+        "opentelemetry/proto/logs/v1/logs.proto",
+    ],
+    deps = [
+        ":common_proto",
+        ":resource_proto",
+    ],
+)
+
+cc_proto_library(
+    name = "logs_proto_cc",
+    deps = [":logs_proto"],
+)
+
+proto_library(
+    name = "logs_service_proto",
+    srcs = [
+        "opentelemetry/proto/collector/logs/v1/logs_service.proto",
+    ],
+    deps = [
+        ":logs_proto",
+    ],
+)
+
+cc_proto_library(
+    name = "logs_service_proto_cc",
+    deps = [":logs_service_proto"],
+)
+
+cc_grpc_library(
+    name = "logs_service_grpc_cc",
+    srcs = [":logs_service_proto"],
+    generate_mocks = True,
+    grpc_only = True,
+    deps = [":logs_service_proto_cc"],
+)

src/carnot/exec/BUILD.bazel

@@ -47,6 +47,7 @@ pl_cc_library(
         "//src/table_store/table:cc_library",
         "@com_github_apache_arrow//:arrow",
         "@com_github_grpc_grpc//:grpc++",
+        "@com_github_opentelemetry_proto//:logs_service_grpc_cc",
         "@com_github_opentelemetry_proto//:metrics_service_grpc_cc",
         "@com_github_opentelemetry_proto//:trace_service_grpc_cc",
         "@com_github_rlyeh_sole//:sole",

src/carnot/exec/exec_state.h

@@ -37,6 +37,7 @@
 #include "src/shared/metadata/metadata_state.h"
 #include "src/table_store/table/table_store.h"
 
+#include "opentelemetry/proto/collector/logs/v1/logs_service.grpc.pb.h"
 #include "opentelemetry/proto/collector/metrics/v1/metrics_service.grpc.pb.h"
 #include "opentelemetry/proto/collector/trace/v1/trace_service.grpc.pb.h"
 #include "src/carnot/carnotpb/carnot.grpc.pb.h"
@@ -54,6 +55,9 @@ using MetricsStubGenerator = std::function<
 using TraceStubGenerator = std::function<
     std::unique_ptr<opentelemetry::proto::collector::trace::v1::TraceService::StubInterface>(
         const std::string& address, bool insecure)>;
+using LogsStubGenerator = std::function<
+    std::unique_ptr<opentelemetry::proto::collector::logs::v1::LogsService::StubInterface>(
+        const std::string& address, bool insecure)>;
 
 /**
  * ExecState manages the execution state for a single query. A new one will
@@ -69,15 +73,16 @@ class ExecState {
       udf::Registry* func_registry, std::shared_ptr<table_store::TableStore> table_store,
       const ResultSinkStubGenerator& stub_generator,
       const MetricsStubGenerator& metrics_stub_generator,
-      const TraceStubGenerator& trace_stub_generator, const sole::uuid& query_id,
-      udf::ModelPool* model_pool, GRPCRouter* grpc_router = nullptr,
+      const TraceStubGenerator& trace_stub_generator, const LogsStubGenerator& logs_stub_generator,
+      const sole::uuid& query_id, udf::ModelPool* model_pool, GRPCRouter* grpc_router = nullptr,
       std::function<void(grpc::ClientContext*)> add_auth_func = [](grpc::ClientContext*) {},
       ExecMetrics* exec_metrics = nullptr)
       : func_registry_(func_registry),
         table_store_(std::move(table_store)),
         stub_generator_(stub_generator),
         metrics_stub_generator_(metrics_stub_generator),
         trace_stub_generator_(trace_stub_generator),
+        logs_stub_generator_(logs_stub_generator),
         query_id_(query_id),
         model_pool_(model_pool),
         grpc_router_(grpc_router),
@@ -157,6 +162,19 @@ class ExecState {
     trace_service_stubs_pool_.push_back(std::move(stub_));
     return raw;
   }
+  opentelemetry::proto::collector::logs::v1::LogsService::StubInterface* LogsServiceStub(
+      const std::string& remote_address, bool insecure) {
+    if (logs_service_stub_map_.contains(remote_address)) {
+      return logs_service_stub_map_[remote_address];
+    }
+    std::unique_ptr<opentelemetry::proto::collector::logs::v1::LogsService::StubInterface> stub_ =
+        logs_stub_generator_(remote_address, insecure);
+    opentelemetry::proto::collector::logs::v1::LogsService::StubInterface* raw = stub_.get();
+    logs_service_stub_map_[remote_address] = raw;
+    // Push to the pool.
+    logs_service_stubs_pool_.push_back(std::move(stub_));
+    return raw;
+  }
 
   udf::ScalarUDFDefinition* GetScalarUDFDefinition(int64_t id) { return id_to_scalar_udf_map_[id]; }
 
@@ -209,6 +227,7 @@ class ExecState {
   const ResultSinkStubGenerator stub_generator_;
   const MetricsStubGenerator metrics_stub_generator_;
   const TraceStubGenerator trace_stub_generator_;
+  const LogsStubGenerator logs_stub_generator_;
   std::map<int64_t, udf::ScalarUDFDefinition*> id_to_scalar_udf_map_;
   std::map<int64_t, udf::UDADefinition*> id_to_uda_map_;
   const sole::uuid query_id_;
@@ -239,6 +258,15 @@ class ExecState {
   absl::flat_hash_map<std::string,
                       opentelemetry::proto::collector::trace::v1::TraceService::StubInterface*>
       trace_service_stub_map_;
+
+  std::vector<
+      std::unique_ptr<opentelemetry::proto::collector::logs::v1::LogsService::StubInterface>>
+      logs_service_stubs_pool_;
+  absl::flat_hash_map<std::string,
+                      opentelemetry::proto::collector::logs::v1::LogsService::StubInterface*>
+      logs_service_stub_map_;
+
+  types::Time64NSValue time_now_;
 };
 
 }  // namespace exec

src/carnot/exec/otel_export_sink_node.cc

@@ -77,6 +77,9 @@ Status OTelExportSinkNode::OpenImpl(ExecState* exec_state) {
   if (plan_node_->spans().size()) {
     trace_service_stub_ = exec_state->TraceServiceStub(plan_node_->url(), plan_node_->insecure());
   }
+  if (plan_node_->logs().size()) {
+    logs_service_stub_ = exec_state->LogsServiceStub(plan_node_->url(), plan_node_->insecure());
+  }
   return Status::OK();
 }
 
@@ -438,13 +441,79 @@ Status OTelExportSinkNode::ConsumeSpans(ExecState* exec_state, const RowBatch& r
   return Status::OK();
 }
 
+using ::opentelemetry::proto::logs::v1::ResourceLogs;
+Status OTelExportSinkNode::ConsumeLogs(ExecState* exec_state, const RowBatch& rb) {
+  grpc::ClientContext context;
+  for (const auto& header : plan_node_->endpoint_headers()) {
+    context.AddMetadata(header.first, header.second);
+  }
+  context.set_compression_algorithm(GRPC_COMPRESS_GZIP);
+
+  logs_response_.Clear();
+  opentelemetry::proto::collector::logs::v1::ExportLogsServiceRequest request;
+
+  for (int64_t row_idx = 0; row_idx < rb.ColumnAt(0)->length(); ++row_idx) {
+    // TODO(ddelnano) aggregate spans by resource.
+    ::opentelemetry::proto::logs::v1::ResourceLogs resource_logs;
+    auto resource = resource_logs.mutable_resource();
+    AddAttributes(resource->mutable_attributes(), plan_node_->resource_attributes_normal_encoding(),
+                  rb, row_idx);
+    auto scope_logs = resource_logs.add_scope_logs();
+    for (const auto& log_pb : plan_node_->logs()) {
+      auto log = scope_logs->add_log_records();
+
+      AddAttributes(log->mutable_attributes(), log_pb.attributes(), rb, row_idx);
+
+      auto time_col = rb.ColumnAt(log_pb.time_column_index()).get();
+      log->set_time_unix_nano(types::GetValueFromArrowArray<types::TIME64NS>(time_col, row_idx));
+      if (log_pb.observed_time_column_index() >= 0) {
+        auto observed_time_col = rb.ColumnAt(log_pb.observed_time_column_index()).get();
+        log->set_observed_time_unix_nano(
+            types::GetValueFromArrowArray<types::TIME64NS>(observed_time_col, row_idx));
+      } else {
+        log->set_observed_time_unix_nano(
+            types::GetValueFromArrowArray<types::TIME64NS>(time_col, row_idx));
+      }
+      log->set_severity_number(
+          static_cast<::opentelemetry::proto::logs::v1::SeverityNumber>(log_pb.severity_number()));
+      log->set_severity_text(log_pb.severity_text());
+      log->mutable_body()->set_string_value(types::GetValueFromArrowArray<types::STRING>(
+          rb.ColumnAt(log_pb.body_column_index()).get(), row_idx));
+    }
+
+    ReplicateData<ResourceLogs>(
+        plan_node_->resource_attributes_optional_json_encoded(),
+        [&request](ResourceLogs log) { *request.add_resource_logs() = std::move(log); },
+        std::move(resource_logs), rb, row_idx);
+  }
+  // Set timeout, to avoid blocking on query.
+  if (plan_node_->timeout() > 0) {
+    std::chrono::system_clock::time_point deadline =
+        std::chrono::system_clock::now() + std::chrono::seconds{plan_node_->timeout()};
+    context.set_deadline(deadline);
+  }
+
+  grpc::Status status = logs_service_stub_->Export(&context, request, &logs_response_);
+  if (!status.ok()) {
+    if (status.error_code() == grpc::StatusCode::DEADLINE_EXCEEDED) {
+      exec_state->exec_metrics()->otlp_spans_timeout_counter.Increment();
+    }
+
+    return FormatOTelStatus(plan_node_->id(), status);
+  }
+  return Status::OK();
+}
+
 Status OTelExportSinkNode::ConsumeNextImpl(ExecState* exec_state, const RowBatch& rb, size_t) {
   if (plan_node_->metrics().size()) {
     PX_RETURN_IF_ERROR(ConsumeMetrics(exec_state, rb));
   }
   if (plan_node_->spans().size()) {
     PX_RETURN_IF_ERROR(ConsumeSpans(exec_state, rb));
   }
+  if (plan_node_->logs().size()) {
+    PX_RETURN_IF_ERROR(ConsumeLogs(exec_state, rb));
+  }
   if (rb.eos()) {
     sent_eos_ = true;
   }

src/carnot/exec/otel_export_sink_node.h

@@ -21,9 +21,6 @@
 #include <memory>
 #include <string>
 
-#include "opentelemetry/proto/collector/metrics/v1/metrics_service.grpc.pb.h"
-#include "opentelemetry/proto/collector/metrics/v1/metrics_service.pb.h"
-
 #include "src/carnot/exec/exec_node.h"
 #include "src/carnot/planpb/plan.pb.h"
 #include "src/common/base/base.h"
@@ -51,6 +48,7 @@ class OTelExportSinkNode : public SinkNode {
                          size_t parent_index) override;
 
  private:
+  Status ConsumeLogs(ExecState* exec_state, const table_store::schema::RowBatch& rb);
   Status ConsumeMetrics(ExecState* exec_state, const table_store::schema::RowBatch& rb);
   Status ConsumeSpans(ExecState* exec_state, const table_store::schema::RowBatch& rb);
 
@@ -60,6 +58,8 @@ class OTelExportSinkNode : public SinkNode {
       metrics_service_stub_;
   opentelemetry::proto::collector::trace::v1::ExportTraceServiceResponse trace_response_;
   opentelemetry::proto::collector::trace::v1::TraceService::StubInterface* trace_service_stub_;
+  opentelemetry::proto::collector::logs::v1::ExportLogsServiceResponse logs_response_;
+  opentelemetry::proto::collector::logs::v1::LogsService::StubInterface* logs_service_stub_;
   std::unique_ptr<plan::OTelExportSinkOperator> plan_node_;
 
   std::unique_ptr<SpanConfig> span_config_;