Skip to content

Entlein/adaptive write #37

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
entlein wants to merge 15 commits into
base: main
Choose a base branch
from
Open

Entlein/adaptive write #37

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
86397a8
Changes needed to get clickhouse e2e test working with external click…
ddelnano Apr 13, 2026
f590005
Implement parquet export format
ddelnano Apr 16, 2026
3510794
Allow prometheus recorders to specifiy different kubeconfig or kubeco…
ddelnano Apr 16, 2026
5a8fb65
Fix parquet file overflow bug
ddelnano Apr 16, 2026
17188d5
Add duck db wasm visualization file
ddelnano Apr 16, 2026
63f7d5f
Temporary changes to make load testing easier
ddelnano Apr 17, 2026
839af02
Add clickhouse perf_tool suite, ability to query cross kubeconfig/kub…
ddelnano Apr 20, 2026
06a8d3a
Ensure px delete works with external k8s ApiService
ddelnano Apr 20, 2026
af340a5
experiment with the adaptive feature
entlein Apr 20, 2026
de1e2f2
Merge remote-tracking branch 'origin/main' into demo/adaptive-export-…
Apr 30, 2026
14d2fc9
redesigning the adaptive write
entlein May 7, 2026
57183a5
addressing the rabbit;
entlein May 8, 2026
95b6b36
adaptive_export/trigger: dedupe at watermark boundary
May 8, 2026
1370479
adaptive_export/trigger: validate identifiers + cover dedup with a test
May 8, 2026
3f59c6d
adaptive_export/trigger: stricter Endpoint validation, streaming pars…
May 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,7 @@ require (
github.com/ory/dockertest/v3 v3.8.1
github.com/ory/hydra-client-go v1.9.2
github.com/ory/kratos-client-go v0.10.1
github.com/parquet-go/parquet-go v0.25.1
github.com/phayes/freeport v0.0.0-20171002181615-b8543db493a5
github.com/prometheus/client_golang v1.14.0
github.com/prometheus/client_model v0.3.0
Expand Down Expand Up @@ -115,6 +116,7 @@ require (
github.com/VividCortex/ewma v1.1.1 // indirect
github.com/a8m/envsubst v1.3.0 // indirect
github.com/alecthomas/participle/v2 v2.0.0-beta.5 // indirect
github.com/andybalholm/brotli v1.1.0 // indirect
github.com/andybalholm/cascadia v1.1.0 // indirect
github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 // indirect
github.com/beorn7/perks v1.0.1 // indirect
Expand Down Expand Up @@ -171,7 +173,7 @@ require (
github.com/google/go-querystring v1.1.0 // indirect
github.com/google/gofuzz v1.2.0 // indirect
github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
github.com/google/uuid v1.3.0 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/googleapis/gax-go/v2 v2.7.0 // indirect
github.com/gorilla/securecookie v1.1.1 // indirect
github.com/gorilla/websocket v1.5.0 // indirect
Expand All @@ -191,7 +193,7 @@ require (
github.com/json-iterator/go v1.1.12 // indirect
github.com/jstemmer/go-junit-report v0.9.1 // indirect
github.com/kevinburke/ssh_config v0.0.0-20190725054713-01f96b0aa0cd // indirect
github.com/klauspost/compress v1.17.2 // indirect
github.com/klauspost/compress v1.17.9 // indirect
github.com/kr/pretty v0.2.1 // indirect
github.com/kr/text v0.2.0 // indirect
github.com/kylelemons/godebug v1.1.0 // indirect
Expand Down Expand Up @@ -232,6 +234,7 @@ require (
github.com/patrickmn/go-cache v2.1.0+incompatible // indirect
github.com/pelletier/go-toml v1.9.3 // indirect
github.com/peterbourgon/diskv v2.0.1+incompatible // indirect
github.com/pierrec/lz4/v4 v4.1.21 // indirect
github.com/pkg/errors v0.9.1 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/prometheus/procfs v0.9.0 // indirect
Expand Down Expand Up @@ -276,7 +279,7 @@ require (
golang.org/x/xerrors v0.0.0-20220907171357-04be3eba64a2 // indirect
gomodules.xyz/jsonpatch/v2 v2.2.0 // indirect
google.golang.org/appengine v1.6.7 // indirect
google.golang.org/protobuf v1.29.1 // indirect
google.golang.org/protobuf v1.34.2 // indirect
gopkg.in/inf.v0 v0.9.1 // indirect
gopkg.in/ini.v1 v1.67.0 // indirect
gopkg.in/launchdarkly/go-jsonstream.v1 v1.0.1 // indirect
Expand Down Expand Up @@ -317,3 +320,5 @@ replace (
google.golang.org/grpc => google.golang.org/grpc v1.43.0
gopkg.in/yaml.v2 => gopkg.in/yaml.v2 v2.4.0
)

replace google.golang.org/protobuf => google.golang.org/protobuf v1.29.1
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot May 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

🧩 Analysis chain

🌐 Web query:

What is the first patched (non-vulnerable) version of google.golang.org/protobuf for GHSA-8r3f-844c-mc37 / GO-2024-2611?

💡 Result:

The first patched (non-vulnerable) version of google.golang.org/protobuf for GHSA-8r3f-844c-mc37 (also known as GO-2024-2611 and CVE-2024-24786) is v1.33.0.[1][2][3] This version fixes the infinite loop in protojson.Unmarshal when handling certain invalid JSON inputs, particularly with google.protobuf.Any or UnmarshalOptions.DiscardUnknown set.[4] Official advisories confirm affected versions are before v1.33.0, with v1.33.0 as the patched version.[1][5][2]

Citations:

🏁 Script executed:

# Read the go.mod file around line 324 to see context and any comments
head -n 330 go.mod | tail -n 20

Repository: k8sstormcenter/pixie

Length of output: 1339

🏁 Script executed:

# Search for all protobuf-related entries in go.mod
rg "protobuf" go.mod -n -A 1 -B 1

Repository: k8sstormcenter/pixie

Length of output: 1000

🏁 Script executed:

# Check the require section to see what version of protobuf would be used without the replace
rg "google.golang.org/protobuf" go.mod -n

Repository: k8sstormcenter/pixie

Length of output: 193

Remove the forced protobuf downgrade to a vulnerable version.

Line 324 pins google.golang.org/protobuf to v1.29.1, which contains the high-severity infinite-loop vulnerability (GHSA-8r3f-844c-mc37 / GO-2024-2611). This replace directive downgrades from the safe v1.34.2 already declared in the require section. Removing this replace allows the patched version to be used.

Proposed fix 
-replace google.golang.org/protobuf => google.golang.org/protobuf v1.29.1
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
replace google.golang.org/protobuf => google.golang.org/protobuf v1.29.1
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@go.mod` at line 324, The go.mod file contains a replace directive forcing
google.golang.org/protobuf to v1.29.1 (the vulnerable downgrade) while the
require already specifies a safe v1.34.2; remove the replace directive
referencing google.golang.org/protobuf v1.29.1 so the module can use the patched
version and then run module cleanup (e.g., go mod tidy) to update the dependency
graph and ensure go.mod/go.sum reflect the removal.

18 changes: 10 additions & 8 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,8 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy
github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
github.com/andybalholm/brotli v1.1.0 h1:eLKJA0d02Lf0mVpIDgYnqXcUn0GqVmEFny3VuID1U3M=
github.com/andybalholm/brotli v1.1.0/go.mod h1:sms7XGricyQI9K10gOSf56VKKWS4oLer58Q+mhRPtnY=
github.com/andybalholm/cascadia v1.1.0 h1:BuuO6sSfQNFRu1LppgbD25Hr2vLYW25JvxHs5zzsLTo=
github.com/andybalholm/cascadia v1.1.0/go.mod h1:GsXiBklL0woXo1j/WYWtSYYC4ouU9PqHO0sqidkEA4Y=
github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239 h1:kFOfPq6dUM1hTo4JG6LR5AXSUEsOjtdm0kw0FtQtMJA=
Expand Down Expand Up @@ -447,8 +449,8 @@ github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaU
github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ=
github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
github.com/googleapis/gax-go/v2 v2.7.0 h1:IcsPKeInNvYi7eqSaDjiZqDDKu5rsmunY0Y1YupQSSQ=
github.com/googleapis/gax-go/v2 v2.7.0/go.mod h1:TEop28CZZQ2y+c0VxMUmu1lV+fQx57QpBWsYpwqHJx8=
Expand Down Expand Up @@ -579,8 +581,8 @@ github.com/klauspost/compress v1.8.2/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0
github.com/klauspost/compress v1.9.0/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
github.com/klauspost/compress v1.9.5/go.mod h1:RyIbtBH6LamlWaDj8nUwkbUhJ87Yi3uG0guNDohfE1A=
github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk=
github.com/klauspost/compress v1.17.2 h1:RlWWUY/Dr4fL8qk9YG7DTZ7PDgME2V4csBXA8L/ixi4=
github.com/klauspost/compress v1.17.2/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
github.com/klauspost/cpuid v1.2.1/go.mod h1:Pj4uuM528wm8OyEC2QMXAi2YiTZ96dNQPGgoMS4s3ek=
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
Expand Down Expand Up @@ -775,6 +777,8 @@ github.com/ory/hydra-client-go v1.9.2 h1:sbp+8zwEJvhqSxcY8HiOkXeY2FspsfSOJ5ajJ07
github.com/ory/hydra-client-go v1.9.2/go.mod h1:TTg4Gt0SDC8+XoGtj5qzdtqxapfFW+Vmm41PFuC6n/E=
github.com/ory/kratos-client-go v0.10.1 h1:kSRk+0leCJ1nPMS+FPho8b9WMzrKNpgszvta0Xo32QU=
github.com/ory/kratos-client-go v0.10.1/go.mod h1:dOQIsar76K07wMPJD/6aMhrWyY+sFGEagLDLso1CpsA=
github.com/parquet-go/parquet-go v0.25.1 h1:l7jJwNM0xrk0cnIIptWMtnSnuxRkwq53S+Po3KG8Xgo=
github.com/parquet-go/parquet-go v0.25.1/go.mod h1:AXBuotO1XiBtcqJb/FKFyjBG4aqa3aQAAWF3ZPzCanY=
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
Expand All @@ -788,6 +792,8 @@ github.com/peterbourgon/diskv v2.0.1+incompatible h1:UBdAOUP5p4RWqPBg048CAvpKN+v
github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR/tNboyR3/BZd58JJSHlUSCU=
github.com/phayes/freeport v0.0.0-20171002181615-b8543db493a5 h1:rZQtoozkfsiNs36c7Tdv/gyGNzD1X1XWKO8rptVNZuM=
github.com/phayes/freeport v0.0.0-20171002181615-b8543db493a5/go.mod h1:iIss55rKnNBTvrwdmkUpLnDpZoAHvWaiq5+iMmen4AE=
github.com/pierrec/lz4/v4 v4.1.21 h1:yOVMLb6qSIDP67pl/5F7RepeKYu/VmTyEXvuMI5d9mQ=
github.com/pierrec/lz4/v4 v4.1.21/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4=
github.com/pingcap/errors v0.11.4/go.mod h1:Oi8TUi2kEtXXLMJk9l1cGmz20kV3TaQ0usTwv5KuLY8=
github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e h1:aoZm08cpOy4WuID//EZDgcC4zIxODThtZNPirFr42+A=
Expand Down Expand Up @@ -1327,10 +1333,6 @@ google.golang.org/genproto v0.0.0-20211208223120-3a66f561d7aa h1:I0YcKz0I7OAhddo
google.golang.org/genproto v0.0.0-20211208223120-3a66f561d7aa/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
google.golang.org/grpc v1.43.0 h1:Eeu7bZtDZ2DpRCsLhUlcrLnvYaMK1Gz86a+hMVvELmM=
google.golang.org/grpc v1.43.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU=
google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
google.golang.org/protobuf v1.25.0/go.mod h1:9JNX74DMeImyA3h4bdi1ymwjUzf21/xIlbajtzgsN7c=
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.29.1 h1:7QBf+IK2gx70Ap/hDsOmam3GE0v9HicjfEdAxE62UoM=
google.golang.org/protobuf v1.29.1/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I=
gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
Expand Down
27 changes: 21 additions & 6 deletions go_deps.bzl
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

37 changes: 34 additions & 3 deletions src/carnot/exec/clickhouse_export_sink_node.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,9 @@ namespace px {
namespace carnot {
namespace exec {

// TODO(ddelnano): Defend against columns that don't exist. These should be
// ignored by the Node.
Comment on lines +38 to +39
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot May 8, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major | ⚡ Quick win

Implement the missing-column guard before dereferencing mapped indices.

Line 38 states missing columns should be ignored, but Line 107 still unconditionally calls rb.ColumnAt(mapping.input_column_index()). A stale or partial mapping can crash the sink instead of skipping the column.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/carnot/exec/clickhouse_export_sink_node.cc` around lines 38 - 39, The
sink currently dereferences mapped indices without guarding for missing columns
(see mapping.input_column_index() being passed to rb.ColumnAt), so update the
ClickhouseExportSinkNode code that iterates mappings to check that the mapped
input column actually exists before calling
rb.ColumnAt(mapping.input_column_index()); e.g., verify
mapping.input_column_index() is within rb.NumColumns() (or use rb.HasColumn /
equivalent) and skip/continue for mappings that are out-of-range or otherwise
invalid so the node ignores missing columns instead of crashing.


using table_store::schema::RowBatch;
using table_store::schema::RowDescriptor;

Expand Down Expand Up @@ -148,12 +151,12 @@ Status ClickHouseExportSinkNode::ConsumeNextImpl(ExecState* /*exec_state*/, cons
break;
}
case types::UINT128: {
// UINT128 is exported as STRING (UUID format)
// UINT128 is exported as STRING in "high:low" format to match
// the ClickHouseSourceNode's parsing in clickhouse_source_node.cc
auto col = std::make_shared<clickhouse::ColumnString>();
for (int64_t i = 0; i < num_rows; ++i) {
auto val = types::GetValueFromArrowArray<types::UINT128>(arrow_col.get(), i);
std::string uuid_str = sole::rebuild(absl::Uint128High64(val), absl::Uint128Low64(val)).str();
col->Append(uuid_str);
col->Append(absl::Substitute("$0:$1", absl::Uint128High64(val), absl::Uint128Low64(val)));
}
block.AppendColumn(mapping.clickhouse_column_name(), col);
break;
Expand All @@ -164,6 +167,34 @@ Status ClickHouseExportSinkNode::ConsumeNextImpl(ExecState* /*exec_state*/, cons
}
}

// Auto-derive event_time from time_ if time_ is present but event_time is not.
// The ClickHouse table schema uses event_time (DateTime64(3), milliseconds) for
// partitioning and ordering, but the Pixie table has time_ (TIME64NS, nanoseconds).
bool has_time_ = false;
bool has_event_time = false;
int time_col_index = -1;
for (const auto& mapping : plan_node_->column_mappings()) {
if (mapping.clickhouse_column_name() == "time_") {
has_time_ = true;
time_col_index = mapping.input_column_index();
}
if (mapping.clickhouse_column_name() == "event_time") {
has_event_time = true;
}
}

if (has_time_ && !has_event_time && time_col_index >= 0) {
auto arrow_col = rb.ColumnAt(time_col_index);
int64_t num_rows = arrow_col->length();
auto event_time_col = std::make_shared<clickhouse::ColumnDateTime64>(3);
for (int64_t i = 0; i < num_rows; ++i) {
int64_t ns_val = types::GetValueFromArrowArray<types::TIME64NS>(arrow_col.get(), i);
// Convert nanoseconds to milliseconds for DateTime64(3)
event_time_col->Append(ns_val / 1000000LL);
}
block.AppendColumn("event_time", event_time_col);
}

// Insert the block into ClickHouse
clickhouse_client_->Insert(plan_node_->table_name(), block);

Expand Down
1 change: 1 addition & 0 deletions src/carnot/funcs/metadata/metadata_ops.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,6 +127,7 @@ void RegisterMetadataOpsOrDie(px::carnot::udf::Registry* registry) {
registry->RegisterOrDie<UPIDToDeploymentIDUDF>("upid_to_deployment_id");
registry->RegisterOrDie<UPIDToStringUDF>("upid_to_string");
registry->RegisterOrDie<HostnameUDF>("_exec_hostname");
registry->RegisterOrDie<PEMHostnameUDF>("_pem_hostname");
registry->RegisterOrDie<HostNumCPUsUDF>("_exec_host_num_cpus");
registry->RegisterOrDie<VizierIDUDF>("vizier_id");
registry->RegisterOrDie<VizierNameUDF>("vizier_name");
Expand Down
27 changes: 27 additions & 0 deletions src/carnot/funcs/metadata/metadata_ops.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2926,6 +2926,33 @@ class HostnameUDF : public ScalarUDF {
}
};

class PEMHostnameUDF : public ScalarUDF {
public:
/**
* @brief Gets the hostname of the PEM agent's machine.
* Unlike _exec_hostname (UDF_ALL), this is restricted to UDF_PEM so the
* distributed planner is forced to execute it on the PEM before data is
* shipped to Kelvin. Use this when the hostname must reflect the agent
* that collected the data rather than the agent that exports it.
*/
StringValue Exec(FunctionContext* ctx) {
auto md = GetMetadataState(ctx);
return md->hostname();
}

static udf::ScalarUDFDocBuilder Doc() {
return udf::ScalarUDFDocBuilder("Get the hostname of the PEM agent.")
.Details(
"Get the hostname of the PEM agent that collected the data. "
"This UDF is restricted to PEM execution, so the distributed planner "
"will always run it on the PEM even when the downstream sink is on Kelvin.")
.Example("df.hostname = px._pem_hostname()")
.Returns("The hostname of the PEM agent.");
}

static udfspb::UDFSourceExecutor Executor() { return udfspb::UDFSourceExecutor::UDF_PEM; }
};

class HostNumCPUsUDF : public ScalarUDF {
public:
/**
Expand Down
1 change: 1 addition & 0 deletions src/e2e_test/perf_tool/cmd/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ go_library(
"//src/e2e_test/perf_tool/pkg/cluster",
"//src/e2e_test/perf_tool/pkg/cluster/gke",
"//src/e2e_test/perf_tool/pkg/cluster/local",
"//src/e2e_test/perf_tool/pkg/exporter",
"//src/e2e_test/perf_tool/pkg/pixie",
"//src/e2e_test/perf_tool/pkg/run",
"//src/e2e_test/perf_tool/pkg/suites",
Expand Down
Loading
Loading