2064487356

Build # 2064487356

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/tremor-rs/tremor-runtime/commit/<a class=hub.com/tremor-rs/tremor-runtime/commit/c9211bc248f2ca778483da6bbfd7916c3925d095">c9211bc24<a href="https://github.com/tremor-rs/tremor-runtime/commit/c9211bc248f2ca778483da6bbfd7916c3925d095">">Merge </a><a class="double-link" href="https://github.com/tremor-rs/tremor-runtime/commit/<a class="double-link" href="https://github.com/tremor-rs/tremor-runtime/commit/3e05221c4210561623bfdb515f1006ec57c5b9e2">3e05221c4</a>">3e05221c4</a><a href="https://github.com/tremor-rs/tremor-runtime/commit/c9211bc248f2ca778483da6bbfd7916c3925d095"> into 85f4dfb1e">85f4dfb1e</a>

Pull Request Pull Request #1077: Connectors Implementation

Run Details

13867 of 13867 new or added lines in 166 files covered. (100.0%)

23960 of 27932 relevant lines covered (85.78%)

4.3 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

37.21

/src/connectors/utils/metrics.rs

// Copyright 2021, The Tremor Team
//
// 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.

use beef::Cow;
use halfbrown::HashMap;
use tremor_common::ports::{ERR, IN, OUT};
use tremor_pipeline::metrics::{value, value_count};
use tremor_pipeline::MetricsSender;
use tremor_script::EventPayload;
use tremor_value::prelude::*;

/// metrics reporter for connector sources
pub struct SourceReporter {
    alias: String,
    metrics_out: u64,
    metrics_err: u64,
    tx: MetricsSender,
    flush_interval_ns: Option<u64>,
    last_flush_ns: u64,
}

impl SourceReporter {
    pub(crate) fn new(alias: String, tx: MetricsSender, flush_interval_s: Option<u64>) -> Self {
        Self {
            alias,
            metrics_out: 0,
            metrics_err: 0,
            tx,
            flush_interval_ns: flush_interval_s.map(|s| s * 1_000_000_000),
            last_flush_ns: 0,
        }
    }

    pub(crate) fn increment_out(&mut self) {
        self.metrics_out += 1;
    }

    pub(crate) fn increment_err(&mut self) {
        self.metrics_err += 1;
    }

    /// Flush the metrics and send them out if the flush interval is set and the time has come
    /// returns `Some(timestamp)` if it did flush the system metrics
    pub(crate) fn periodic_flush(&mut self, timestamp: u64) -> Option<u64> {
        if let Some(interval) = self.flush_interval_ns {
            if timestamp >= self.last_flush_ns + interval {
                let payload_out =
                    make_event_count_metrics_payload(timestamp, OUT, self.metrics_out, &self.alias);
                let payload_err =
                    make_event_count_metrics_payload(timestamp, ERR, self.metrics_err, &self.alias);
                send(&self.tx, payload_out, &self.alias);
                send(&self.tx, payload_err, &self.alias);
                self.last_flush_ns = timestamp;
                return Some(timestamp);
            }
        }
        None
    }

    /// simply send source metrics
    pub(crate) fn send_source_metrics(&self, metrics: Vec<EventPayload>) {
        for metric in metrics {
            send(&self.tx, metric, &self.alias);
        }
    }
}

/// metrics reporter for connector sinks
pub(crate) struct SinkReporter {
    alias: String,
    metrics_in: u64,
    tx: MetricsSender,
    flush_interval_ns: Option<u64>,
    last_flush_ns: u64,
}

impl SinkReporter {
    pub(crate) fn new(alias: String, tx: MetricsSender, flush_interval_s: Option<u64>) -> Self {
        Self {
            alias,
            metrics_in: 0,
            tx,
            flush_interval_ns: flush_interval_s.map(|s| s * 1_000_000_000),
            last_flush_ns: 0,
        }
    }

    pub(crate) fn increment_in(&mut self) {
        self.metrics_in += 1;
    }

    pub(crate) fn periodic_flush(&mut self, timestamp: u64) -> Option<u64> {
        if let Some(interval) = self.flush_interval_ns {
            if timestamp >= self.last_flush_ns + interval {
                let payload =
                    make_event_count_metrics_payload(timestamp, IN, self.metrics_in, &self.alias);
                send(&self.tx, payload, &self.alias);
                self.last_flush_ns = timestamp;
                return Some(timestamp);
            }
        }
        None
    }

    /// simply send source metrics
    pub(crate) fn send_sink_metrics(&self, metrics: Vec<EventPayload>) {
        for metric in metrics {
            send(&self.tx, metric, &self.alias);
        }
    }
}

// this is simple forwarding
#[cfg(not(tarpaulin_include))]
pub(crate) fn send(tx: &MetricsSender, metric: EventPayload, alias: &str) {
    use tremor_pipeline::MetricsMsg;

    if let Err(_e) = tx.try_broadcast(MetricsMsg::new(metric, None)) {
        error!(
            "[Connector::{}] Error sending to system metrics connector.",
            &alias
        );
    }
}

#[must_use]
pub(crate) fn make_event_count_metrics_payload(
    timestamp: u64,
    port: Cow<'static, str>,
    count: u64,
    artefact_id: &str,
) -> EventPayload {
    let mut tags: HashMap<Cow<'static, str>, Value<'static>> = HashMap::with_capacity(2);
    tags.insert_nocheck(Cow::const_str("connector"), artefact_id.to_string().into());
    tags.insert_nocheck(Cow::const_str("port"), port.into());

    let value = value_count(Cow::from("connector_events"), tags, count, timestamp);
    // full metrics payload
    (value, Value::object()).into()
}

// TODO: add convenience functions for creating custom metrics payloads
#[must_use]
pub(crate) fn make_metrics_payload(
    name: &'static str,
    fields: HashMap<Cow<'static, str>, Value<'static>>,
    tags: HashMap<Cow<'static, str>, Value<'static>>,
    timestamp: u64,
) -> EventPayload {
    let value = value(Cow::const_str(name), tags, fields, timestamp);
    (value, Value::object()).into()
}

1	// Copyright 2021, The Tremor Team
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	use beef::Cow;
16	use halfbrown::HashMap;
17	use tremor_common::ports::{ERR, IN, OUT};
18	use tremor_pipeline::metrics::{value, value_count};
19	use tremor_pipeline::MetricsSender;
20	use tremor_script::EventPayload;
21	use tremor_value::prelude::*;
22
23	/// metrics reporter for connector sources
24	pub struct SourceReporter {
25	alias: String,
26	metrics_out: u64,
27	metrics_err: u64,
28	tx: MetricsSender,
29	flush_interval_ns: Option<u64>,
30	last_flush_ns: u64,
31	}
32
33	impl SourceReporter {
34	pub(crate) fn new(alias: String, tx: MetricsSender, flush_interval_s: Option<u64>) -> Self {	3✔
35	Self {
36	alias,
37	metrics_out: 0,
38	metrics_err: 0,
39	tx,
40	flush_interval_ns: flush_interval_s.map(\|s\| s * 1_000_000_000),	3✔
41	last_flush_ns: 0,
42	}
43	}
44
45	pub(crate) fn increment_out(&mut self) {	2✔
46	self.metrics_out += 1;	2✔
47	}
48
49	pub(crate) fn increment_err(&mut self) {	1✔
50	self.metrics_err += 1;	1✔
51	}
52
53	/// Flush the metrics and send them out if the flush interval is set and the time has come
54	/// returns `Some(timestamp)` if it did flush the system metrics
55	pub(crate) fn periodic_flush(&mut self, timestamp: u64) -> Option<u64> {	2✔
56	if let Some(interval) = self.flush_interval_ns {	2✔
57	if timestamp >= self.last_flush_ns + interval {	×
58	let payload_out =	×
59	make_event_count_metrics_payload(timestamp, OUT, self.metrics_out, &self.alias);
60	let payload_err =	×
61	make_event_count_metrics_payload(timestamp, ERR, self.metrics_err, &self.alias);
62	send(&self.tx, payload_out, &self.alias);	×
63	send(&self.tx, payload_err, &self.alias);	×
64	self.last_flush_ns = timestamp;	×
65	return Some(timestamp);	×
66	}
67	}
68	None	2✔
69	}
70
71	/// simply send source metrics
72	pub(crate) fn send_source_metrics(&self, metrics: Vec<EventPayload>) {	×
73	for metric in metrics {	×
74	send(&self.tx, metric, &self.alias);	×
75	}
76	}
77	}
78
79	/// metrics reporter for connector sinks
80	pub(crate) struct SinkReporter {
81	alias: String,
82	metrics_in: u64,
83	tx: MetricsSender,
84	flush_interval_ns: Option<u64>,
85	last_flush_ns: u64,
86	}
87
88	impl SinkReporter {
89	pub(crate) fn new(alias: String, tx: MetricsSender, flush_interval_s: Option<u64>) -> Self {	3✔
90	Self {
91	alias,
92	metrics_in: 0,
93	tx,
94	flush_interval_ns: flush_interval_s.map(\|s\| s * 1_000_000_000),	3✔
95	last_flush_ns: 0,
96	}
97	}
98
99	pub(crate) fn increment_in(&mut self) {	2✔
100	self.metrics_in += 1;	2✔
101	}
102
103	pub(crate) fn periodic_flush(&mut self, timestamp: u64) -> Option<u64> {	2✔
104	if let Some(interval) = self.flush_interval_ns {	2✔
105	if timestamp >= self.last_flush_ns + interval {	×
106	let payload =	×
107	make_event_count_metrics_payload(timestamp, IN, self.metrics_in, &self.alias);
108	send(&self.tx, payload, &self.alias);	×
109	self.last_flush_ns = timestamp;	×
110	return Some(timestamp);	×
111	}
112	}
113	None	2✔
114	}
115
116	/// simply send source metrics
117	pub(crate) fn send_sink_metrics(&self, metrics: Vec<EventPayload>) {	×
118	for metric in metrics {	×
119	send(&self.tx, metric, &self.alias);	×
120	}
121	}
122	}
123
124	// this is simple forwarding
125	#[cfg(not(tarpaulin_include))]
126	pub(crate) fn send(tx: &MetricsSender, metric: EventPayload, alias: &str) {
127	use tremor_pipeline::MetricsMsg;
128
129	if let Err(_e) = tx.try_broadcast(MetricsMsg::new(metric, None)) {
130	error!(
131	"[Connector::{}] Error sending to system metrics connector.",
132	&alias
133	);
134	}
135	}
136
137	#[must_use]
138	pub(crate) fn make_event_count_metrics_payload(	×
139	timestamp: u64,
140	port: Cow<'static, str>,
141	count: u64,
142	artefact_id: &str,
143	) -> EventPayload {
144	let mut tags: HashMap<Cow<'static, str>, Value<'static>> = HashMap::with_capacity(2);	×
145	tags.insert_nocheck(Cow::const_str("connector"), artefact_id.to_string().into());	×
146	tags.insert_nocheck(Cow::const_str("port"), port.into());	×
147
148	let value = value_count(Cow::from("connector_events"), tags, count, timestamp);	×
149	// full metrics payload
150	(value, Value::object()).into()	×
151	}
152
153	// TODO: add convenience functions for creating custom metrics payloads
154	#[must_use]
155	pub(crate) fn make_metrics_payload(	×
156	name: &'static str,
157	fields: HashMap<Cow<'static, str>, Value<'static>>,
158	tags: HashMap<Cow<'static, str>, Value<'static>>,
159	timestamp: u64,
160	) -> EventPayload {
161	let value = value(Cow::const_str(name), tags, fields, timestamp);	×
162	(value, Value::object()).into()	×
163	}

tremor-rs / tremor-runtime / 2064487356

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous