Merge pipeline-support-wip from ddelnano

.arclint

@@ -23,6 +23,7 @@
     "(^src/stirling/bpf_tools/bcc_bpf/system-headers)",
     "(^src/stirling/mysql/testing/.*\\.json$)",
     "(^src/stirling/obj_tools/testdata/go/test_go_binary.go)",
+    "(^src/stirling/source_connectors/file_source/testdata/test.json$)",
     "(^src/stirling/source_connectors/socket_tracer/protocols/http2/testing/go_grpc_client/main.go$)",
     "(^src/stirling/source_connectors/socket_tracer/protocols/http2/testing/go_grpc_server/main.go$)",
     "(^src/stirling/utils/testdata/config$)",

.bazelrc

@@ -3,6 +3,9 @@
 # Use strict action env to prevent leaks of env vars.
 build --incompatible_strict_action_env
 
+# Use cache
+# build --disk_cache=/tmp/bazel/cache # must not be merged dev only settng
+
 # Only pass through GH_API_KEY for stamped builds.
 # This is still not ideal as it still busts the cache of stamped builds.
 build:stamp --stamp

BUILD.bazel

@@ -1,3 +1,6 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@px//bazel:pl_build_system.bzl", "pl_go_binary")
+
 # Copyright 2018- The Pixie Authors.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
@@ -56,6 +59,7 @@ gazelle(
 # gazelle:resolve go px.dev/pixie/src/carnot/docspb //src/carnot/docspb:docs_pl_go_proto
 # 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/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
@@ -216,3 +220,16 @@ filegroup(
     srcs = ["go.sum"],
     visibility = ["//visibility:public"],
 )
+
+go_library(
+    name = "pixie_lib",
+    srcs = ["gosym_tab_experiment.go"],
+    importpath = "px.dev/pixie",
+    visibility = ["//visibility:private"],
+)
+
+pl_go_binary(
+    name = "pixie",
+    embed = [":pixie_lib"],
+    visibility = ["//visibility:public"],
+)

WORKSPACE

@@ -204,7 +204,7 @@ bind(
 )
 
 # gazelle:repo bazel_gazelle
-# Gazelle depes need to be loaded last to make sure they don't override our dependencies.
+# Gazelle deps need to be loaded last to make sure they don't override our dependencies.
 # The first one wins when it comes to package declaration.
 load("@bazel_gazelle//:deps.bzl", "gazelle_dependencies")
 

demos/log-generator/log-generator.yaml

@@ -0,0 +1,89 @@
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: px-log-generator
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-config
+  namespace: px-log-generator
+data:
+  vector.toml: |
+    [sources.demo]
+    type = "demo_logs"
+    format = "json"
+
+    [sinks.json_output]
+    type = "file"
+    inputs = ["demo"]
+    path = "/var/log/px-log-generator.json"
+    encoding.codec = "json"
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-logrotate-config
+  namespace: px-log-generator
+data:
+  logrotate.conf: |
+    /var/log/px-log-generator.json {
+        size 30M
+        copytruncate
+        rotate 5
+        compress
+        missingok
+        notifempty
+    }
+---
+apiVersion: apps/v1
+kind: DaemonSet
+metadata:
+  name: vector
+  namespace: px-log-generator
+spec:
+  selector:
+    matchLabels:
+      app: vector
+  template:
+    metadata:
+      labels:
+        app: vector
+    spec:
+      volumes:
+        - name: log-storage
+          hostPath:
+            path: /var/log
+            type: Directory
+        - name: logrotate-config
+          configMap:
+            name: vector-logrotate-config
+        - name: config-volume
+          configMap:
+            name: vector-config
+      initContainers:
+        - name: cleanup
+          image: busybox
+          command: ["/bin/sh", "-c", "truncate -s0 /var/log/px-log-generator.json"]
+          volumeMounts:
+            - name: log-storage
+              mountPath: /var/log
+      containers:
+        - name: vector
+          image: timberio/vector@sha256:f8933ff1a3ec08df45abc6130947938d98dc85792a25592ec1aa6fe83a7f562c  # 0.44.0-debian
+          args: ["--config", "/etc/vector/vector.toml"]
+          volumeMounts:
+            - name: config-volume
+              mountPath: /etc/vector
+            - name: log-storage
+              mountPath: /var/log
+        - name: logrotate
+          image: vitess/logrotate@sha256:ba0f99827d0e2d0bda86230ff6666e75383d93babcbc6c803c4d41396214f312  # v21.0.2-bookworm
+          volumeMounts:
+            - name: logrotate-config
+              mountPath: /vt/logrotate.conf
+              subPath: logrotate.conf
+            - name: log-storage
+              mountPath: /var/log
+      terminationGracePeriodSeconds: 10
+      restartPolicy: Always

skaffold/skaffold_vizier.yaml

@@ -8,37 +8,43 @@ build:
     bazel:
       target: //src/vizier/services/agent/pem:pem_image.tar
       args:
-      - --compilation_mode=dbg
+      - --config=x86_64_sysroot
+      - --compilation_mode=opt
   - image: vizier-kelvin_image
     context: .
     bazel:
       target: //src/vizier/services/agent/kelvin:kelvin_image.tar
       args:
-      - --compilation_mode=dbg
+      - --config=x86_64_sysroot
+      - --compilation_mode=opt
   - image: vizier-metadata_server_image
     context: .
     bazel:
       target: //src/vizier/services/metadata:metadata_server_image.tar
       args:
-      - --compilation_mode=dbg
+      - --config=x86_64_sysroot
+      - --compilation_mode=opt
   - image: vizier-query_broker_server_image
     context: .
     bazel:
       target: //src/vizier/services/query_broker:query_broker_server_image.tar
       args:
-      - --compilation_mode=dbg
+      - --config=x86_64_sysroot
+      - --compilation_mode=opt
   - image: vizier-cloud_connector_server_image
     context: .
     bazel:
       target: //src/vizier/services/cloud_connector:cloud_connector_server_image.tar
       args:
-      - --compilation_mode=dbg
+        - --config=x86_64_sysroot
+        - --compilation_mode=opt
   - image: vizier-cert_provisioner_image
     context: .
     bazel:
       target: //src/utils/cert_provisioner:cert_provisioner_image.tar
       args:
-      - --compilation_mode=dbg
+      - --config=x86_64_sysroot
+      - --compilation_mode=opt
   tagPolicy:
     dateTime: {}
   local:

src/api/go/pxapi/examples/standalone_pem_example/example.go

@@ -30,18 +30,66 @@ import (
 
 // Define PxL script with one table output.
 var (
+	stream = `
+import px
+df = px.DataFrame('http_events')
+px.display(df.stream())
+`
 	pxl = `
 import px
+import pxlog
+table = 'vector.json'
+f = '/home/ddelnano/code/pixie-worktree/vector.json'
+pxlog.FileSource(f, table, '5m')
+df = px.DataFrame(table)
 
-# Look at the http_events.
-df = px.DataFrame(table='http_events')
-
-# Grab the command line from the metadata.
-df.cmdline = px.upid_to_cmdline(df.upid)
-
-# Limit to the first 10.
-df = df.head(10)
-
+px.display(df)`
+	bpftrace = `
+import pxtrace
+import px
+# Adapted from https://github.com/iovisor/bpftrace/blob/master/tools/tcpretrans.bt
+program = """
+// tcpretrans.bt Trace or count TCP retransmits
+//               For Linux, uses bpftrace and eBPF.
+//
+// Copyright (c) 2018 Dale Hamel.
+// Licensed under the Apache License, Version 2.0 (the "License")
+#include <linux/socket.h>
+#include <net/sock.h>
+kprobe:tcp_retransmit_skb
+{
+  $sk = (struct sock *)arg0;
+  $inet_family = $sk->__sk_common.skc_family;
+  $AF_INET = (uint16) 2;
+  $AF_INET6 = (uint16) 10;
+  if ($inet_family == $AF_INET || $inet_family == $AF_INET6) {
+    if ($inet_family == $AF_INET) {
+      $daddr = ntop($sk->__sk_common.skc_daddr);
+      $saddr = ntop($sk->__sk_common.skc_rcv_saddr);
+    } else {
+      $daddr = ntop($sk->__sk_common.skc_v6_daddr.in6_u.u6_addr8);
+      $saddr = ntop($sk->__sk_common.skc_v6_rcv_saddr.in6_u.u6_addr8);
+    }
+    $sport = $sk->__sk_common.skc_num;
+    $dport = $sk->__sk_common.skc_dport;
+    // Destination port is big endian, it must be flipped
+    $dport = ($dport >> 8) | (($dport << 8) & 0x00FF00);
+    printf(\"time_:%llu src_ip:%s src_port:%d dst_ip:%s dst_port:%d\",
+      nsecs,
+      $saddr,
+      $sport,
+      $daddr,
+      $dport);
+  }
+}
+"""
+table_name = 'tcp_retransmits_table'
+pxtrace.UpsertTracepoint('tcp_retranmits_probe',
+    table_name,
+    program,
+    pxtrace.kprobe(),
+    "2m")
+df = px.DataFrame(table=table_name, select=['time_', 'src_ip', 'src_port', 'dst_ip', 'dst_port'])
 px.display(df)`
 )
 

src/api/go/pxapi/vizier.go

@@ -20,6 +20,7 @@ package pxapi
 
 import (
 	"context"
+	"strings"
 
 	"px.dev/pixie/src/api/go/pxapi/errdefs"
 	"px.dev/pixie/src/api/proto/vizierpb"
@@ -40,6 +41,7 @@ func (v *VizierClient) ExecuteScript(ctx context.Context, pxl string, mux TableM
 		ClusterID:         v.vizierID,
 		QueryStr:          pxl,
 		EncryptionOptions: v.encOpts,
+		Mutation:          strings.Contains(pxl, "import pxlog") || strings.Contains(pxl, "import pxtrace"),
 	}
 	origCtx := ctx
 	ctx, cancel := context.WithCancel(ctx)

src/carnot/carnot.cc

@@ -378,9 +378,9 @@ Status CarnotImpl::ExecutePlan(const planpb::Plan& logical_plan, const sole::uui
                 int64_t total_time_ns = stats->TotalExecTime();
                 int64_t self_time_ns = stats->SelfExecTime();
                 LOG(INFO) << absl::Substitute(
-                    "self_time:$1\ttotal_time: $2\tbytes_output: $3\trows_output: $4\tnode_id:$0",
+                    "self_time:$1\ttotal_time: $2\tbytes_input: $3\tbytes_output: $4\trows_input: $5\trows_output: $6\tnode_id:$0",
                     node_name, PrettyDuration(self_time_ns), PrettyDuration(total_time_ns),
-                    stats->bytes_output, stats->rows_output);
+                    stats->bytes_input, stats->bytes_output, stats->rows_input, stats->rows_output);
 
                 queryresultspb::OperatorExecutionStats* stats_pb =
                     agent_operator_exec_stats.add_operator_execution_stats();

src/carnot/carnot_test.cc

@@ -211,7 +211,7 @@ px.display(df, 'range_output'))pxl";
   std::vector<types::Time64NSValue> col0_out1;
   std::vector<types::Float64Value> col1_out1;
   std::vector<types::Int64Value> col2_out1;
-  table_store::Table::Cursor cursor(big_table_.get());
+  table_store::Cursor cursor(big_table_.get());
   auto batch = cursor.GetNextRowBatch({0}).ConsumeValueOrDie();
   for (int64_t i = 0; i < batch->ColumnAt(0)->length(); i++) {
     if (CarnotTestUtils::big_test_col1[i].val >= 2 && CarnotTestUtils::big_test_col1[i].val < 12) {

src/carnot/exec/BUILD.bazel

@@ -226,6 +226,7 @@ pl_cc_test(
     deps = [
         ":cc_library",
         ":test_utils",
+        "//src/common/testing/event:cc_library",
         "//src/carnot/planpb:plan_testutils",
         "@com_github_apache_arrow//:arrow",
     ],
@@ -296,6 +297,7 @@ pl_cc_test(
         ":exec_node_test_helpers",
         ":test_utils",
         "//src/carnot/planpb:plan_testutils",
+        "//src/common/testing/event:cc_library",
         "@com_github_apache_arrow//:arrow",
         "@com_github_grpc_grpc//:grpc++_test",
     ],