Fix visibility of pipeline metrics along with reserved/duplicate node name detection

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## Unreleased
 
+* Terminate pipeline creation when a node already exists with the given name [#650](https://github.com/tremor-rs/tremor-runtime/issues/650)
+* Fix visibility of pipeline metrics [#648](https://github.com/tremor-rs/tremor-runtime/pull/648)
 * Fix panic upon usage of postgres ramps due to incompatbility with tokio and async-std runtime. [#641](https://github.com/tremor-rs/tremor-runtime/pull/641)
 
 ## 0.9.3

tests/query_error.rs

@@ -140,4 +140,8 @@ test_cases!(
     //INSERT
     bad_into,
     bad_from,
+    node_duplicate_name_operator,
+    node_duplicate_name_script,
+    node_reserved_name_operator,
+    node_reserved_name_script,
 );

tests/query_errors/node_duplicate_name_operator/error.txt

@@ -0,0 +1,7 @@
+Error: 
+    2 | 
+    3 | create operator c1 from counter;
+    4 | create operator c1 from counter;
+      | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Name `c1` is already in use for another node, please use another name.
+    5 | 
+    6 | select event from in into c1;

tests/query_errors/node_duplicate_name_operator/query.trickle

@@ -0,0 +1,7 @@
+define generic::counter operator counter;
+
+create operator c1 from counter;
+create operator c1 from counter;
+
+select event from in into c1;
+select event from c1 into out;

tests/query_errors/node_duplicate_name_script/error.txt

@@ -0,0 +1,7 @@
+Error: 
+    6 | 
+    7 | create script process from pass;
+    8 | create script process from pass;
+      | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Name `process` is already in use for another node, please use another name.
+    9 | 
+   10 | select event from in into process;

tests/query_errors/node_duplicate_name_script/query.trickle

@@ -0,0 +1,11 @@
+define script pass
+script
+  event
+end;
+create script pass;
+
+create script process from pass;
+create script process from pass;
+
+select event from in into process;
+select event from process into out;

tests/query_errors/node_reserved_name_operator/error.txt

@@ -0,0 +1,6 @@
+Error: 
+    1 | define generic::counter operator counter;
+    2 | create operator metrics from counter;
+      | ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ Name `metrics` is reserved for built-in nodes, please use another name.
+    3 | 
+    4 | select event from in into metrics;

tests/query_errors/node_reserved_name_operator/query.trickle

@@ -0,0 +1,5 @@
+define generic::counter operator counter;
+create operator metrics from counter;
+
+select event from in into metrics;
+select event from metrics into out;

tests/query_errors/node_reserved_name_script/error.txt

@@ -0,0 +1,7 @@
+Error: 
+    3 |   event
+    4 | end;
+    5 | create script metrics;
+      | ^^^^^^^^^^^^^^^^^^^^^ Name `metrics` is reserved for built-in nodes, please use another name.
+    6 | 
+    7 | select event from in into metrics;

tests/query_errors/node_reserved_name_script/query.trickle

@@ -0,0 +1,8 @@
+define script metrics
+script
+  event
+end;
+create script metrics;
+
+select event from in into metrics;
+select event from metrics into out;

tremor-cli/tests/integration/system_metrics/assert.yaml

@@ -0,0 +1,11 @@
+status: 0
+name: Output of tremor's system::metrics pipeline
+asserts:
+  - source: metrics_pipeline.log
+    equals_file: "expected_metrics_pipeline.json"
+  - source: metrics_onramp.log
+    equals_file: "expected_metrics_onramp.json"
+  - source: metrics_offramp.log
+    equals_file: "expected_metrics_offramp.json"
+  - source: events.log
+    equals_file: "expected_events.json"

tremor-cli/tests/integration/system_metrics/config.yaml

@@ -0,0 +1,52 @@
+onramp:
+  - id: in
+    metrics_interval_s: 3
+    type: metronome
+    config:
+      interval: 1000
+
+offramp:
+  - id: out
+    metrics_interval_s: 3
+    type: file
+    config:
+      file: "events.log"
+
+  - id: metrics_pipeline
+    type: file
+    config:
+      file: "metrics_pipeline.log"
+
+  - id: metrics_onramp
+    type: file
+    config:
+      file: "metrics_onramp.log"
+
+  - id: metrics_offramp
+    type: file
+    config:
+      file: "metrics_offramp.log"
+
+  - id: exit
+    type: exit
+
+binding:
+  - id: test
+    links:
+      "/onramp/in/{instance}/out": ["/pipeline/main/{instance}/in"]
+      "/pipeline/main/{instance}/out": ["/offramp/out/{instance}/in"]
+      "/pipeline/main/{instance}/err": ["/offramp/out/{instance}/in"]
+
+      "/pipeline/system::metrics/system/out": ["/pipeline/metrics/{instance}/in"]
+      "/pipeline/metrics/{instance}/pipeline": ["/offramp/metrics_pipeline/{instance}/in"]
+      "/pipeline/metrics/{instance}/onramp": ["/offramp/metrics_onramp/{instance}/in"]
+      "/pipeline/metrics/{instance}/offramp": ["/offramp/metrics_offramp/{instance}/in"]
+      # catching other outputs, just in case
+      "/pipeline/metrics/{instance}/out": ["/offramp/system::stdout/system/in"]
+      "/pipeline/metrics/{instance}/err": ["/offramp/system::stderr/system/in"]
+
+      "/pipeline/main/{instance}/done": ["/offramp/exit/{instance}/in"]
+
+mapping:
+  /binding/test/1:
+    instance: "1"

tremor-cli/tests/integration/system_metrics/expected_events.json

@@ -0,0 +1,6 @@
+{"onramp":"metronome","ingest_ns":null,"id":0}
+{"onramp":"metronome","ingest_ns":null,"id":1}
+{"onramp":"metronome","ingest_ns":null,"id":2}
+{"onramp":"metronome","ingest_ns":null,"id":3}
+{"onramp":"metronome","ingest_ns":null,"id":4}
+{"onramp":"metronome","ingest_ns":null,"id":5}

tremor-cli/tests/integration/system_metrics/expected_metrics_offramp.json

@@ -0,0 +1,6 @@
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/offramp/out/1/in","port":"in"},"fields":{"count":0},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/offramp/out/1/in","port":"out"},"fields":{"count":0},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/offramp/out/1/in","port":"error"},"fields":{"count":0},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/offramp/out/1/in","port":"in"},"fields":{"count":3},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/offramp/out/1/in","port":"out"},"fields":{"count":3},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/offramp/out/1/in","port":"error"},"fields":{"count":0},"timestamp":null}

tremor-cli/tests/integration/system_metrics/expected_metrics_onramp.json

@@ -0,0 +1,9 @@
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/onramp/in/1/out","port":"in"},"fields":{"count":0},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/onramp/in/1/out","port":"out"},"fields":{"count":0},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/onramp/in/1/out","port":"error"},"fields":{"count":0},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/onramp/in/1/out","port":"in"},"fields":{"count":0},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/onramp/in/1/out","port":"out"},"fields":{"count":3},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/onramp/in/1/out","port":"error"},"fields":{"count":0},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/onramp/in/1/out","port":"in"},"fields":{"count":0},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/onramp/in/1/out","port":"out"},"fields":{"count":6},"timestamp":null}
+{"measurement":"ramp_events","tags":{"ramp":"tremor://localhost/onramp/in/1/out","port":"error"},"fields":{"count":0},"timestamp":null}

tremor-cli/tests/integration/system_metrics/expected_metrics_pipeline.json

@@ -0,0 +1,4 @@
+{"measurement":"events","tags":{"pipeline":"tremor://localhost/pipeline/main/1","node":"out","direction":"input","port":"in"},"fields":{"count":3},"timestamp":null}
+{"measurement":"events","tags":{"pipeline":"tremor://localhost/pipeline/main/1","node":"process","direction":"output","port":"out"},"fields":{"count":3},"timestamp":null}
+{"measurement":"events","tags":{"pipeline":"tremor://localhost/pipeline/main/1","node":"out","direction":"input","port":"in"},"fields":{"count":6},"timestamp":null}
+{"measurement":"events","tags":{"pipeline":"tremor://localhost/pipeline/main/1","node":"process","direction":"output","port":"out"},"fields":{"count":6},"timestamp":null}

tremor-cli/tests/integration/system_metrics/main.trickle

@@ -0,0 +1,19 @@
+#!config metrics_interval_s = 3
+
+define script process
+script
+  let event.ingest_ns = null; # for testing
+  match event.id of
+    # 7th event from metronome
+    case 6 => emit {"exit": 0, "delay": 100} => "done"
+    default => event
+  end
+end;
+
+create script process;
+
+select event from in into process;
+select event from process into out;
+select event from process/err into err;
+
+select event from process/done into out/done;

tremor-cli/tests/integration/system_metrics/metrics.trickle

@@ -0,0 +1,27 @@
+define script process
+script
+  let event.timestamp = null; # for testing
+
+  let port = match event.measurement of
+    case "events" => "pipeline"
+    case "ramp_events" =>
+      match event.tags.ramp of
+        case "tremor://localhost/onramp/in/1/out" => "onramp"
+        case "tremor://localhost/offramp/out/1/in" => "offramp"
+        default => "out"
+      end
+    default => "out"
+  end;
+
+  emit event => "{port}";
+end;
+
+create script process;
+
+select event from in into process;
+
+select event from process/pipeline into out/pipeline;
+select event from process/onramp into out/onramp;
+select event from process/offramp into out/offramp;
+
+select event from process/err into err;

tremor-cli/tests/integration/system_metrics/tags.json

@@ -0,0 +1,4 @@
+[
+    "system",
+    "metrics"
+]

tremor-pipeline/src/op/prelude.rs

@@ -24,3 +24,4 @@ pub use value_trait::Value as ValueTrait;
 pub const OUT: Cow<'static, str> = Cow::Borrowed("out");
 pub const IN: Cow<'static, str> = Cow::Borrowed("in");
 pub const ERR: Cow<'static, str> = Cow::Borrowed("err");
+pub const METRICS: Cow<'static, str> = Cow::Borrowed("metrics");