How to exit/short-circuit pipeline during transform?

[paymog]

I'm writing a transformation, extract-entity-count, which parses logs looking for a specific entry. Downstream of this there will be a postgres sink which writes the results of the transformation into a postgres table. Not all log entries will be relevant to this particular component of my pipeline. Is there a way to for a transformation to exit/short circuit such that none of the downstream components of the pipeline which use extract-entity-count as an input get triggered?

As an example, I have the following so far:

[transforms.extract-entity-count]
inputs = ["parse_graph_node_logs"]
type = "remap"
source = """
final_fields = {}
final_fields.subgraph_id = .subgraph_id
final_fields.unix_timestamp = .unix_timestamp
final_fields.cluster_name = .cluster_name

# Extract the entity count from the message
regexed, err = parse_regex(.message, r'^.*entities: (?P<entity_count>\\d+),')
if err == null {
  final_fields.entity_count = parse_int!(regexed.entity_count)
} else {
  # if we didn't extract an entity count from this message, then set all fields to null, there's nothing to write to postgres
  final_fields = {}
  . = {}
}

. |= final_fields
"""

[sinks.stdout-entity-count]
inputs = ["extract-entity-count"]
type = "console"
encoding.codec = "json"
target = "stdout"

When I run this I see many log lines from vector which simply log {}. Ideally I'd like my stdout sink to only get run when extract-entity-count has something useful. Is that possible? I ask because I'd like to reduce ingestion load on my postgres instance (and any other sinks I create) when there's nothing useful for postgres to ingest.

Here's a sample what I see when I run the above

e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {}
e2e-tests-vector-1                 | {"cluster_name":"graph-node","container_created_at":"2023-07-26T07:06:51.251591964Z","container_id":"154141814d4a109a5ec49924ab834cc74856824e37f8ec463e0f94bd7ef2091d","container_name":"e2e-tests-graph-node-1","entity_count":1652,"host":"31e19d3715f5","image":"e2e-tests-graph-node","label":{"com.docker.compose.config-hash":"9c9c097d3ce45ebaa68b7f2a18e5ef39a68c725eddad5192b3f6093d7b5a33e2","com.docker.compose.container-number":"1","com.docker.compose.depends_on":"postgres-second:service_started,ipfs:service_started,postgres:service_started","com.docker.compose.image":"sha256:9718789d195fa06a59674b20224b370f03a99de9637f57059a45fecd0fd8ac1a","com.docker.compose.oneoff":"False","com.docker.compose.project":"e2e-tests","com.docker.compose.project.config_files":"/Users/paymahn/code/goldsky/graph-tools/packages/e2e-tests/docker-compose.yml","com.docker.compose.project.working_dir":"/Users/paymahn/code/goldsky/graph-tools/packages/e2e-tests","com.docker.compose.service":"graph-node","com.docker.compose.version":"2.10.2"},"message":"Committed write batch, time_ms: 106, weight: 1686248, entities: 1652, block_count: 826, block_number: 17631324, sgd: 1, subgraph_id: QmThBaPEnXZK9w57bmnaUTe5hqRAfNnA1bfPAG9ZvxgB4H, component: SubgraphInstanceManager","severity":"INFO","source_type":"docker_logs","stream":"stderr","subgraph_id":"QmThBaPEnXZK9w57bmnaUTe5hqRAfNnA1bfPAG9ZvxgB4H","timestamp":"2023-07-26T07:28:10.541601334Z","unix_timestamp":1690356490541}

[paymog]

Oh, I think I found the answer. I think using abort and drop_on_abort = true is the key like so:

[transforms.extract-entity-count]
inputs = ["parse_graph_node_logs"]
type = "remap"
drop_on_abort = true
source = """
final_fields = {}
final_fields.subgraph_id = .subgraph_id
final_fields.unix_timestamp = .unix_timestamp
final_fields.cluster_name = .cluster_name

# Extract the entity count from the message
regexed, err = parse_regex(.message, r'^.*entities: (?P<entity_count>\\d+),')
if err == null {
  final_fields.entity_count = parse_int!(regexed.entity_count)
  . |= final_fields
} else {
  # if we didn't extract an entity count from this message abort this pipeline (note `drop_on_abort = true` for this transform)
  abort
}
"""