Feat/implement tetragon connector

BUILD.bazel

@@ -60,6 +60,7 @@ gazelle(
 # gazelle:resolve go px.dev/pixie/src/carnot/planner/compilerpb //src/carnot/planner/compilerpb:compiler_status_pl_go_proto
 # gazelle:resolve go px.dev/pixie/src/carnot/planner/distributedpb //src/carnot/planner/distributedpb:distributed_plan_pl_go_proto
 # gazelle:resolve go px.dev/pixie/src/carnot/planner/file_source/ir //src/carnot/planner/file_source/ir:logical_pl_go_proto
+# gazelle:resolve go px.dev/pixie/src/carnot/planner/tetragon/ir //src/carnot/planner/tetragon/ir:logical_pl_go_proto
 # gazelle:resolve go px.dev/pixie/src/carnot/planner/dynamic_tracing/ir/logicalpb //src/carnot/planner/dynamic_tracing/ir/logicalpb:logical_pl_go_proto
 # gazelle:resolve go px.dev/pixie/src/carnot/planner/plannerpb //src/carnot/planner/plannerpb:service_pl_go_proto
 # gazelle:resolve go px.dev/pixie/src/carnot/planpb //src/carnot/planpb:plan_pl_go_proto

src/carnot/planner/compiler/BUILD.bazel

@@ -41,6 +41,7 @@ pl_cc_library(
         "//src/carnot/planner/compiler_error_context:cc_library",
         "//src/carnot/planner/compiler_state:cc_library",
         "//src/carnot/planner/file_source:cc_library",
+        "//src/carnot/planner/tetragon:cc_library",
         "//src/carnot/planner/ir:cc_library",
         "//src/carnot/planner/metadata:cc_library",
         "//src/carnot/planner/objects:cc_library",

src/carnot/planner/compiler/ast_visitor.cc

@@ -106,6 +106,8 @@ Status ASTVisitorImpl::SetupModules(
                       TraceModule::Create(mutations_, this));
   PX_ASSIGN_OR_RETURN((*module_handler_)[LogModule::kLogModuleObjName],
                       LogModule::Create(mutations_, this));
+  PX_ASSIGN_OR_RETURN((*module_handler_)[TetragonModule::kTetragonModuleObjName],
+                      TetragonModule::Create(mutations_, this));
   PX_ASSIGN_OR_RETURN((*module_handler_)[ConfigModule::kConfigModuleObjName],
                       ConfigModule::Create(mutations_, this));
   for (const auto& [module_name, module_text] : module_name_to_pxl_map) {

src/carnot/planner/compiler/ast_visitor.h

@@ -34,6 +34,7 @@
 #include "src/carnot/planner/ast/ast_visitor.h"
 #include "src/carnot/planner/compiler_state/compiler_state.h"
 #include "src/carnot/planner/file_source/log_module.h"
+#include "src/carnot/planner/tetragon/tetragon_module.h"
 #include "src/carnot/planner/ir/ast_utils.h"
 #include "src/carnot/planner/ir/ir.h"
 #include "src/carnot/planner/objects/dataframe.h"

src/carnot/planner/plannerpb/BUILD.bazel

@@ -29,6 +29,7 @@ pl_proto_library(
         "//src/carnot/planner/distributedpb:distributed_plan_pl_proto",
         "//src/carnot/planner/dynamic_tracing/ir/logicalpb:logical_pl_proto",
         "//src/carnot/planner/file_source/ir:logical_pl_proto",
+        "//src/carnot/planner/tetragon/ir:logical_pl_proto",
         "//src/carnot/planpb:plan_pl_proto",
         "//src/common/base/statuspb:status_pl_proto",
         "//src/shared/types/typespb:types_pl_proto",
@@ -44,6 +45,7 @@ pl_cc_proto_library(
         "//src/carnot/planner/distributedpb:distributed_plan_pl_cc_proto",
         "//src/carnot/planner/dynamic_tracing/ir/logicalpb:logical_pl_cc_proto",
         "//src/carnot/planner/file_source/ir:logical_pl_cc_proto",
+        "//src/carnot/planner/tetragon/ir:logical_pl_cc_proto",
         "//src/carnot/planpb:plan_pl_cc_proto",
         "//src/common/base/statuspb:status_pl_cc_proto",
         "//src/shared/types/typespb/wrapper:cc_library",
@@ -59,6 +61,7 @@ pl_go_proto_library(
         "//src/carnot/planner/distributedpb:distributed_plan_pl_go_proto",
         "//src/carnot/planner/dynamic_tracing/ir/logicalpb:logical_pl_go_proto",
         "//src/carnot/planner/file_source/ir:logical_pl_go_proto",
+        "//src/carnot/planner/tetragon/ir:logical_pl_go_proto",
         "//src/carnot/planpb:plan_pl_go_proto",
         "//src/common/base/statuspb:status_pl_go_proto",
         "//src/shared/types/typespb:types_pl_go_proto",

src/carnot/planner/plannerpb/service.proto

@@ -24,6 +24,7 @@ option go_package = "plannerpb";
 
 import "src/carnot/planner/dynamic_tracing/ir/logicalpb/logical.proto";
 import "src/carnot/planner/file_source/ir/logical.proto";
+import "src/carnot/planner/tetragon/ir/logical.proto";
 import "src/common/base/statuspb/status.proto";
 import "gogoproto/gogo.proto";
 import "src/carnot/planner/distributedpb/distributed_plan.proto";
@@ -135,6 +136,11 @@ message DeleteFileSource {
   string glob_pattern = 1;
 }
 
+message DeleteTetragon {
+  // The glob pattern to use to find files to read. Also doubles as the name of the tetragon log file.
+  string glob_pattern = 1;
+}
+
 // The definition of a mutation to perfom on Vizier. Mutations include operations
 // that add and delete tables to the database.
 message CompileMutation {
@@ -150,6 +156,9 @@ message CompileMutation {
     carnot.planner.file_source.ir.FileSourceDeployment file_source = 5;
     // Mutation that deletes a file source/poller
     DeleteFileSource delete_file_source = 6;
+    carnot.planner.tetragon.ir.TetragonDeployment tetragon = 7;
+    // Mutation that deletes a tetragon/poller
+    DeleteTetragon delete_tetragon = 8;
   }
 }
 

src/carnot/planner/probes/BUILD.bazel

@@ -38,6 +38,7 @@ pl_cc_library(
     deps = [
         "//src/carnot/planner/dynamic_tracing/ir/logicalpb:logical_pl_cc_proto",
         "//src/carnot/planner/file_source/ir:logical_pl_cc_proto",
+        "//src/carnot/planner/tetragon/ir:logical_pl_cc_proto",
         "//src/carnot/planner/objects:cc_library",
         "//src/common/uuid:cc_library",
     ],

src/carnot/planner/probes/probes.cc

@@ -312,6 +312,14 @@ Status MutationsIR::ToProto(plannerpb::CompileMutationsResponse* pb) {
     *(pb->add_mutations()->mutable_file_source()) = file_source;
   }
 
+  for (const auto& tetragon_to_delete : TetragonsToDelete()) {
+    pb->add_mutations()->mutable_delete_tetragon()->set_glob_pattern(tetragon_to_delete);
+  }
+
+  for (const auto& tetragon : tetragon_deployments_) {
+    *(pb->add_mutations()->mutable_tetragon()) = tetragon;
+  }
+
   return Status::OK();
 }
 
@@ -329,6 +337,26 @@ void MutationsIR::CreateFileSourceDeployment(const std::string& glob_pattern,
   file_source_deployments_.push_back(file_source);
 }
 
+std::vector<tetragon::ir::TetragonDeployment> MutationsIR::TetragonDeployments() {
+  std::vector<tetragon::ir::TetragonDeployment> tetragon_deployments;
+  for (size_t i = 0; i < tetragon_deployments_.size(); i++) {
+    tetragon_deployments.push_back(tetragon_deployments_[i]);
+  }
+  return tetragon_deployments;
+}
+
+void MutationsIR::CreateTetragonDeployment(const std::string& glob_pattern,
+                                             const std::string& table_name, int64_t ttl_ns) {
+  tetragon::ir::TetragonDeployment tetragon;
+  tetragon.set_name(glob_pattern);
+  tetragon.set_glob_pattern(glob_pattern);
+  tetragon.set_table_name(table_name);
+  auto one_sec = std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::seconds(1));
+  tetragon.mutable_ttl()->set_seconds(ttl_ns / one_sec.count());
+  tetragon.mutable_ttl()->set_nanos(ttl_ns % one_sec.count());
+  tetragon_deployments_.push_back(tetragon);
+}
+
 }  // namespace compiler
 }  // namespace planner
 }  // namespace carnot

src/carnot/planner/probes/probes.h

@@ -24,6 +24,7 @@
 
 #include "src/carnot/planner/dynamic_tracing/ir/logicalpb/logical.pb.h"
 #include "src/carnot/planner/file_source/ir/logical.pb.h"
+#include "src/carnot/planner/tetragon/ir/logical.pb.h"
 #include "src/carnot/planner/objects/funcobject.h"
 #include "src/carnot/planner/plannerpb/service.pb.h"
 #include "src/carnot/planner/probes/label_selector_target.h"
@@ -181,6 +182,20 @@ class FileSourceDeployment {
   int64_t ttl_ns_;
 };
 
+class TetragonDeployment {
+ public:
+  TetragonDeployment(const std::string& glob_pattern, const std::string& table_name,
+                       int64_t ttl_ns)
+      : glob_pattern_(glob_pattern), table_name_(table_name), ttl_ns_(ttl_ns) {}
+
+  Status ToProto(tetragon::ir::TetragonDeployment pb) const;
+
+ private:
+  std::string glob_pattern_;
+  std::string table_name_;
+  int64_t ttl_ns_;
+};
+
 class TracepointDeployment {
  public:
   TracepointDeployment(const std::string& trace_name, int64_t ttl_ns)
@@ -244,6 +259,12 @@ class MutationsIR {
                                   int64_t ttl_ns);
 
   void CreateDeleteFileSource(const std::string& glob_pattern);
+
+  void CreateTetragonDeployment(const std::string& glob_pattern, const std::string& table_name,
+                                  int64_t ttl_ns);
+
+  void CreateDeleteTetragon(const std::string& glob_pattern);
+
   /**
    * @brief Create a TraceProgram for the MutationsIR w/ the specified UPID.
    *
@@ -363,7 +384,20 @@ class MutationsIR {
 
   const std::vector<std::string>& FileSourcesToDelete() { return file_sources_to_delete_; }
 
- private:
+  std::vector<tetragon::ir::TetragonDeployment> TetragonDeployments();
+
+  /**
+   * @brief Deletes the tetragon log passed in.
+   *
+   * @param tetragon_to_delete
+   */
+  void DeleteTetragon(const std::string& tetragon_to_delete) {
+    tetragons_to_delete_.push_back(tetragon_to_delete);
+  }
+
+  const std::vector<std::string>& TetragonsToDelete() { return tetragons_to_delete_; }
+
+  private:
   // All the new tracepoints added as part of this mutation. DeploymentSpecs are protobufs because
   // we only modify these upon inserting the new tracepoint, while the Tracepoint definition is
   // still modified aftered adding the tracepoint.
@@ -383,6 +417,9 @@ class MutationsIR {
 
   std::vector<file_source::ir::FileSourceDeployment> file_source_deployments_;
   std::vector<std::string> file_sources_to_delete_;
+
+  std::vector<tetragon::ir::TetragonDeployment> tetragon_deployments_;
+  std::vector<std::string> tetragons_to_delete_;
 };
 
 }  // namespace compiler

src/carnot/planner/probes/tracepoint_generator.cc

@@ -29,6 +29,7 @@
 
 #include "src/carnot/planner/dynamic_tracing/ir/logicalpb/logical.pb.h"
 #include "src/carnot/planner/file_source/ir/logical.pb.h"
+#include "src/carnot/planner/tetragon/ir/logical.pb.h"
 
 namespace px {
 namespace carnot {
@@ -83,6 +84,11 @@ StatusOr<file_source::ir::FileSourceDeployment> CompileFileSource(std::string_vi
   return pb.mutations()[0].file_source();
 }
 
+StatusOr<tetragon::ir::TetragonDeployment> CompileTetragon(std::string_view query) {
+  PX_ASSIGN_OR_RETURN(auto pb, CompileMutations(query));
+  return pb.mutations()[0].tetragon();
+}
+
 }  // namespace compiler
 }  // namespace planner
 }  // namespace carnot

src/carnot/planner/probes/tracepoint_generator.h

@@ -39,6 +39,12 @@ StatusOr<carnot::planner::dynamic_tracing::ir::logical::TracepointDeployment> Co
  */
 StatusOr<file_source::ir::FileSourceDeployment> CompileFileSource(std::string_view query);
 
+/**
+ * Take a tetragon log specification in PXL format, and compiles it to a logical tetragon log
+ * deployment.
+ */
+StatusOr<tetragon::ir::TetragonDeployment> CompileTetragon(std::string_view query);
+
 }  // namespace compiler
 }  // namespace planner
 }  // namespace carnot

src/carnot/planner/tetragon/BUILD.bazel

@@ -0,0 +1,52 @@
+# Copyright 2018- The Pixie Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+load("//bazel:pl_build_system.bzl", "pl_cc_binary", "pl_cc_library", "pl_cc_test")
+
+package(default_visibility = [
+    "//src/carnot:__subpackages__",
+    "//src/experimental/standalone_pem:__subpackages__",  # TODO(ddelnano): Is this needed?
+])
+
+pl_cc_library(
+    name = "cc_library",
+    srcs = glob(
+        [
+            "*.cc",
+            "*.h",
+        ],
+        exclude = [
+            "**/*_test.cc",
+            "**/*_test_utils.h",
+        ],
+    ),
+    hdrs = ["tetragon.h"],
+    deps = [
+        "//src/carnot/planner/objects:cc_library",
+        "//src/carnot/planner/probes:cc_library",
+        "//src/common/uuid:cc_library",  # TODO(ddelnano): This may not be needed
+    ],
+)
+
+pl_cc_test(
+    name = "tetragon_test",
+    srcs = ["tetragon_test.cc"],
+    deps = [
+        ":cc_library",
+        "//src/carnot/planner:test_utils",
+        "//src/carnot/planner/compiler:cc_library",
+    ],
+)

src/carnot/planner/tetragon/ir/BUILD.bazel

@@ -0,0 +1,41 @@
+# Copyright 2018- The Pixie Authors.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+load("//bazel:proto_compile.bzl", "pl_cc_proto_library", "pl_go_proto_library", "pl_proto_library")
+
+package(default_visibility = ["//src:__subpackages__"])
+
+pl_proto_library(
+    name = "logical_pl_proto",
+    srcs = ["logical.proto"],
+    deps = [
+        "@gogo_grpc_proto//gogoproto:gogo_pl_proto",
+    ],
+)
+
+pl_cc_proto_library(
+    name = "logical_pl_cc_proto",
+    proto = ":logical_pl_proto",
+    deps = [
+        "@gogo_grpc_proto//gogoproto:gogo_pl_cc_proto",
+    ],
+)
+
+pl_go_proto_library(
+    name = "logical_pl_go_proto",
+    importpath = "px.dev/pixie/src/carnot/planner/tetragon/ir",
+    proto = ":logical_pl_proto",
+)