Add OTel log support to Carnot plan protos and implement exec node support (#2161)

Summary: Add OTel log support to Carnot plan protos and implement exec
node support

My upcoming Kubecon talk will be demoing functionality that would
benefit from OTel export log support. This is the first step in
supporting this. The next set of PRs will include the planner and pxl
frontend changes to leverage these changes.

Relevant Issues: #705

Type of change: /kind feature

Test Plan: New tests verify added functionality

---------

Signed-off-by: Dom Del Nano <ddelnano@gmail.com>

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/engine_state.h

@@ -87,6 +87,9 @@ class EngineState : public NotCopyable {
         [this](const std::string& remote_addr, bool insecure) {
           return TraceStubGenerator(remote_addr, insecure);
         },
+        [this](const std::string& remote_addr, bool insecure) {
+          return LogsStubGenerator(remote_addr, insecure);
+        },
         query_id, model_pool_.get(), grpc_router_, add_auth_to_grpc_context_func_, metrics_.get());
   }
   std::shared_ptr<grpc::Channel> CreateChannel(const std::string& remote_addr, bool insecure) {
@@ -115,6 +118,12 @@ class EngineState : public NotCopyable {
         CreateChannel(remote_addr, insecure));
   }
 
+  std::unique_ptr<opentelemetry::proto::collector::logs::v1::LogsService::StubInterface>
+  LogsStubGenerator(const std::string& remote_addr, bool insecure) {
+    return opentelemetry::proto::collector::logs::v1::LogsService::NewStub(
+        CreateChannel(remote_addr, insecure));
+  }
+
   std::unique_ptr<plan::PlanState> CreatePlanState() {
     return std::make_unique<plan::PlanState>(func_registry_.get());
   }

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/agg_node_test.cc

@@ -476,7 +476,7 @@ std::unique_ptr<ExecState> MakeTestExecState(udf::Registry* registry) {
   auto table_store = std::make_shared<table_store::TableStore>();
   return std::make_unique<ExecState>(registry, table_store, MockResultSinkStubGenerator,
                                      MockMetricsStubGenerator, MockTraceStubGenerator,
-                                     sole::uuid4(), nullptr);
+                                     MockLogStubGenerator, sole::uuid4(), nullptr);
 }
 
 std::unique_ptr<plan::Operator> PlanNodeFromPbtxt(const std::string& pbtxt) {

src/carnot/exec/empty_source_node_test.cc

@@ -48,9 +48,9 @@ class EmptySourceNodeTest : public ::testing::Test {
   void SetUp() override {
     func_registry_ = std::make_unique<udf::Registry>("test_registry");
     auto table_store = std::make_shared<table_store::TableStore>();
-    exec_state_ = std::make_unique<ExecState>(func_registry_.get(), table_store,
-                                              MockResultSinkStubGenerator, MockMetricsStubGenerator,
-                                              MockTraceStubGenerator, sole::uuid4(), nullptr);
+    exec_state_ = std::make_unique<ExecState>(
+        func_registry_.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
+        MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr);
   }
 
   std::unique_ptr<ExecState> exec_state_;

src/carnot/exec/equijoin_node_test.cc

@@ -52,9 +52,9 @@ class JoinNodeTest : public ::testing::Test {
   JoinNodeTest() {
     func_registry_ = std::make_unique<udf::Registry>("test_registry");
     auto table_store = std::make_shared<table_store::TableStore>();
-    exec_state_ = std::make_unique<ExecState>(func_registry_.get(), table_store,
-                                              MockResultSinkStubGenerator, MockMetricsStubGenerator,
-                                              MockTraceStubGenerator, sole::uuid4(), nullptr);
+    exec_state_ = std::make_unique<ExecState>(
+        func_registry_.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
+        MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr);
   }
 
  protected:

src/carnot/exec/exec_graph_test.cc

@@ -74,9 +74,9 @@ class BaseExecGraphTest : public ::testing::Test {
     func_registry_->RegisterOrDie<MultiplyUDF>("multiply");
 
     auto table_store = std::make_shared<table_store::TableStore>();
-    exec_state_ = std::make_unique<ExecState>(func_registry_.get(), table_store,
-                                              MockResultSinkStubGenerator, MockMetricsStubGenerator,
-                                              MockTraceStubGenerator, sole::uuid4(), nullptr);
+    exec_state_ = std::make_unique<ExecState>(
+        func_registry_.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
+        MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr);
   }
 
   std::unique_ptr<udf::Registry> func_registry_;
@@ -174,7 +174,7 @@ TEST_P(ExecGraphExecuteTest, execute) {
   table_store->AddTable("numbers", table);
   auto exec_state_ = std::make_unique<ExecState>(
       func_registry_.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
-      MockTraceStubGenerator, sole::uuid4(), nullptr);
+      MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr);
 
   EXPECT_OK(exec_state_->AddScalarUDF(
       0, "add", std::vector<types::DataType>({types::DataType::INT64, types::DataType::FLOAT64})));
@@ -255,7 +255,7 @@ TEST_F(ExecGraphTest, execute_time) {
 
   auto exec_state_ = std::make_unique<ExecState>(
       func_registry.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
-      MockTraceStubGenerator, sole::uuid4(), nullptr);
+      MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr);
 
   EXPECT_OK(exec_state_->AddScalarUDF(
       0, "add", std::vector<types::DataType>({types::DataType::INT64, types::DataType::FLOAT64})));
@@ -322,7 +322,7 @@ TEST_F(ExecGraphTest, two_limits_dont_interfere) {
   table_store->AddTable("numbers", table);
   auto exec_state_ = std::make_unique<ExecState>(
       func_registry_.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
-      MockTraceStubGenerator, sole::uuid4(), nullptr);
+      MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr);
 
   ExecutionGraph e;
   auto s = e.Init(schema.get(), plan_state.get(), exec_state_.get(), plan_fragment_.get(),
@@ -390,7 +390,7 @@ TEST_F(ExecGraphTest, limit_w_multiple_srcs) {
   table_store->AddTable("numbers", table);
   auto exec_state_ = std::make_unique<ExecState>(
       func_registry_.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
-      MockTraceStubGenerator, sole::uuid4(), nullptr);
+      MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr);
 
   ExecutionGraph e;
   auto s = e.Init(schema.get(), plan_state.get(), exec_state_.get(), plan_fragment_.get(),
@@ -452,7 +452,7 @@ TEST_F(ExecGraphTest, two_sequential_limits) {
   table_store->AddTable("numbers", table);
   auto exec_state_ = std::make_unique<ExecState>(
       func_registry_.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
-      MockTraceStubGenerator, sole::uuid4(), nullptr);
+      MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr);
 
   ExecutionGraph e;
   auto s = e.Init(schema.get(), plan_state.get(), exec_state_.get(), plan_fragment_.get(),
@@ -515,7 +515,7 @@ TEST_F(ExecGraphTest, execute_with_two_limits) {
   table_store->AddTable("numbers", table);
   auto exec_state_ = std::make_unique<ExecState>(
       func_registry_.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
-      MockTraceStubGenerator, sole::uuid4(), nullptr);
+      MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr);
 
   ExecutionGraph e;
   auto s = e.Init(schema.get(), plan_state.get(), exec_state_.get(), plan_fragment_.get(),
@@ -702,7 +702,7 @@ class GRPCExecGraphTest : public ::testing::Test {
     auto table_store = std::make_shared<table_store::TableStore>();
     exec_state_ = std::make_unique<ExecState>(
         func_registry_.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
-        MockTraceStubGenerator, sole::uuid4(), nullptr, grpc_router_.get());
+        MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr, grpc_router_.get());
   }
 
   void SetUpPlanFragment() {

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/expression_evaluator_benchmark.cc

@@ -46,6 +46,7 @@
 using ScalarExpression = px::carnot::plan::ScalarExpression;
 using ScalarExpressionVector = std::vector<std::shared_ptr<ScalarExpression>>;
 using px::carnot::exec::ExecState;
+using px::carnot::exec::MockLogStubGenerator;
 using px::carnot::exec::MockMetricsStubGenerator;
 using px::carnot::exec::MockResultSinkStubGenerator;
 using px::carnot::exec::MockTraceStubGenerator;
@@ -85,7 +86,7 @@ void BM_ScalarExpressionTwoCols(benchmark::State& state,
   PX_CHECK_OK(func_registry->Register<AddUDF>("add"));
   auto exec_state = std::make_unique<ExecState>(
       func_registry.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
-      MockTraceStubGenerator, sole::uuid4(), nullptr);
+      MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr);
   EXPECT_OK(exec_state->AddScalarUDF(
       0, "add",
       std::vector<px::types::DataType>({px::types::DataType::INT64, px::types::DataType::INT64})));

src/carnot/exec/expression_evaluator_test.cc

@@ -116,9 +116,9 @@ class ScalarExpressionTest : public ::testing::TestWithParam<ScalarExpressionEva
 
     EXPECT_TRUE(func_registry_->Register<AddUDF>("add").ok());
     EXPECT_TRUE(func_registry_->Register<InitArgUDF>("init_arg").ok());
-    exec_state_ = std::make_unique<ExecState>(func_registry_.get(), table_store,
-                                              MockResultSinkStubGenerator, MockMetricsStubGenerator,
-                                              MockTraceStubGenerator, sole::uuid4(), nullptr);
+    exec_state_ = std::make_unique<ExecState>(
+        func_registry_.get(), table_store, MockResultSinkStubGenerator, MockMetricsStubGenerator,
+        MockTraceStubGenerator, MockLogStubGenerator, sole::uuid4(), nullptr);
     EXPECT_OK(exec_state_->AddScalarUDF(
         0, "add", std::vector<types::DataType>({types::DataType::INT64, types::DataType::INT64})));
     EXPECT_OK(