Constellation, Spacedust, Slingshot, UFOs: atproto crates and services for microcosm
update metric names to fit prometheus requirements
use '_' instead of '.'. no real change here since the prometheus recorder thing automatically makes the same change, but now nothing should get magically changed/fixed implicitly.
+47
-47
+30
-30
link_aggregator/src/consumer/jetstream.rs
+30
-30
link_aggregator/src/consumer/jetstream.rs
···
22
22
cursor: Option<u64>,
23
23
) -> anyhow::Result<()> {
24
24
describe_counter!(
25
-
"jetstream.connnect",
25
+
"jetstream_connnect",
26
26
Unit::Count,
27
27
"attempts to connect to a jetstream server"
28
28
);
29
29
describe_counter!(
30
-
"jetstream.read",
30
+
"jetstream_read",
31
31
Unit::Count,
32
32
"attempts to read an event from jetstream"
33
33
);
34
34
describe_counter!(
35
-
"jetstream.read.fail",
35
+
"jetstream_read_fail",
36
36
Unit::Count,
37
37
"failures to read events from jetstream"
38
38
);
39
39
describe_counter!(
40
-
"jetstream.read.bytes",
40
+
"jetstream_read_bytes",
41
41
Unit::Bytes,
42
42
"total received message bytes from jetstream"
43
43
);
44
44
describe_counter!(
45
-
"jetstream.read.bytes.decompressed",
45
+
"jetstream_read_bytes_decompressed",
46
46
Unit::Bytes,
47
47
"total decompressed message bytes from jetstream"
48
48
);
49
49
describe_histogram!(
50
-
"jetstream.read.bytes.decompressed",
50
+
"jetstream_read_bytes_decompressed",
51
51
Unit::Bytes,
52
52
"decompressed size of jetstream messages"
53
53
);
54
54
describe_counter!(
55
-
"jetstream.events",
55
+
"jetstream_events",
56
56
Unit::Count,
57
57
"valid json messages received"
58
58
);
59
59
describe_histogram!(
60
-
"jetstream.events.queued",
60
+
"jetstream_events_queued",
61
61
Unit::Count,
62
62
"event messages waiting in queue"
63
63
);
64
64
describe_gauge!(
65
-
"jetstream.cursor.age",
65
+
"jetstream_cursor_age",
66
66
Unit::Microseconds,
67
67
"microseconds between our clock and the jetstream event's time_us"
68
68
);
···
81
81
})
82
82
.unwrap_or("".into())
83
83
);
84
-
counter!("jetstream.connect", "url" => url, "is_retry" => (connect_retries > 0).to_string()).increment(1);
84
+
counter!("jetstream_connect", "url" => url, "is_retry" => (connect_retries > 0).to_string()).increment(1);
85
85
println!("jetstream connecting, attempt #{connect_retries}: {stream}...");
86
86
let mut socket = match tungstenite::connect(&stream) {
87
87
Ok((socket, _)) => {
···
110
110
break;
111
111
}
112
112
113
-
counter!("jetstream.read").increment(1);
113
+
counter!("jetstream_read").increment(1);
114
114
let b = match socket.read() {
115
115
Ok(Message::Binary(b)) => b,
116
116
Ok(Message::Text(_)) => {
117
-
counter!("jetstream.read.fail", "url" => url, "reason" => "received text")
117
+
counter!("jetstream_read_fail", "url" => url, "reason" => "received text")
118
118
.increment(1);
119
119
eprintln!("jetstream: unexpected text message, should be binary for compressed (ignoring)");
120
120
continue;
121
121
}
122
122
Ok(Message::Close(f)) => {
123
-
counter!("jetstream.read.fail", "url" => url, "reason" => "server closed")
123
+
counter!("jetstream_read_fail", "url" => url, "reason" => "server closed")
124
124
.increment(1);
125
125
println!("jetstream: closing the connection: {f:?}");
126
126
continue;
127
127
}
128
128
Ok(m) => {
129
-
counter!("jetstream.read.fail", "url" => url, "reason" => "unexpected message", "message" => format!("{m:?}")).increment(1);
129
+
counter!("jetstream_read_fail", "url" => url, "reason" => "unexpected message", "message" => format!("{m:?}")).increment(1);
130
130
eprintln!("jetstream: unexpected from read (ignoring): {m:?}");
131
131
continue;
132
132
}
133
133
Err(TError::ConnectionClosed) => {
134
134
// clean exit
135
-
counter!("jetstream.read.fail", "url" => url, "reason" => "clean close")
135
+
counter!("jetstream_read_fail", "url" => url, "reason" => "clean close")
136
136
.increment(1);
137
137
println!("jetstream closed the websocket cleanly.");
138
138
break;
139
139
}
140
140
Err(TError::AlreadyClosed) => {
141
141
// programming error
142
-
counter!("jetstream.read.fail", "url" => url, "reason" => "already closed")
142
+
counter!("jetstream_read_fail", "url" => url, "reason" => "already closed")
143
143
.increment(1);
144
144
eprintln!(
145
145
"jetstream: got AlreadyClosed trying to .read() websocket. probably a bug."
···
147
147
break;
148
148
}
149
149
Err(TError::Capacity(e)) => {
150
-
counter!("jetstream.read.fail", "url" => url, "reason" => "capacity error")
150
+
counter!("jetstream_read_fail", "url" => url, "reason" => "capacity error")
151
151
.increment(1);
152
152
eprintln!("jetstream: capacity error (ignoring): {e:?}");
153
153
continue;
154
154
}
155
155
Err(TError::Utf8) => {
156
-
counter!("jetstream.read.fail", "url" => url, "reason" => "utf8 error")
156
+
counter!("jetstream_read_fail", "url" => url, "reason" => "utf8 error")
157
157
.increment(1);
158
158
eprintln!("jetstream: utf8 error (ignoring)");
159
159
continue;
···
162
162
eprintln!("jetstream: could not read message from socket. closing: {e:?}");
163
163
match socket.close(None) {
164
164
Err(TError::ConnectionClosed) => {
165
-
counter!("jetstream.read.fail", "url" => url, "reason" => "clean close").increment(1);
165
+
counter!("jetstream_read_fail", "url" => url, "reason" => "clean close").increment(1);
166
166
println!("jetstream closed the websocket cleanly.");
167
167
break;
168
168
}
169
169
r => eprintln!("jetstream: close result after error: {r:?}"),
170
170
}
171
-
counter!("jetstream.read.fail", "url" => url, "reason" => "read error")
171
+
counter!("jetstream_read_fail", "url" => url, "reason" => "read error")
172
172
.increment(1);
173
173
// if we didn't immediately get ConnectionClosed, we should keep polling read
174
174
// until we get it.
···
176
176
}
177
177
};
178
178
179
-
counter!("jetstream.read.bytes", "url" => url).increment(b.len() as u64);
179
+
counter!("jetstream_read_bytes", "url" => url).increment(b.len() as u64);
180
180
let mut cursor = Cursor::new(b);
181
181
let mut decoder =
182
182
match zstd::stream::Decoder::with_prepared_dictionary(&mut cursor, &dict) {
183
183
Ok(d) => d,
184
184
Err(e) => {
185
-
counter!("jetstream.read.fail", "url" => url, "reason" => "zstd decompress")
185
+
counter!("jetstream_read_fail", "url" => url, "reason" => "zstd decompress")
186
186
.increment(1);
187
187
eprintln!("jetstream: failed to decompress zstd message: {e:?}");
188
188
continue;
···
192
192
let mut s = String::new();
193
193
match decoder.read_to_string(&mut s) {
194
194
Ok(n) => {
195
-
counter!("jetstream.read.bytes.decompressed", "url" => url).increment(n as u64);
196
-
histogram!("jetstream.read.bytes.decompressed", "url" => url).record(n as f64);
195
+
counter!("jetstream_read_bytes_decompressed", "url" => url).increment(n as u64);
196
+
histogram!("jetstream_read_bytes_decompressed", "url" => url).record(n as f64);
197
197
}
198
198
Err(e) => {
199
-
counter!("jetstream.read.fail", "url" => url, "reason" => "zstd string decode")
199
+
counter!("jetstream_read_fail", "url" => url, "reason" => "zstd string decode")
200
200
.increment(1);
201
201
eprintln!("jetstream: failed to decode zstd: {e:?}");
202
202
continue;
···
206
206
let v = match s.parse() {
207
207
Ok(v) => v,
208
208
Err(e) => {
209
-
counter!("jetstream.read.fail", "url" => url, "reason" => "json parse")
209
+
counter!("jetstream_read_fail", "url" => url, "reason" => "json parse")
210
210
.increment(1);
211
211
eprintln!("jetstream: failed to parse message as json: {e:?}");
212
212
continue;
···
217
217
let ts = match get_event_time(&v) {
218
218
Some(ts) => ts,
219
219
None => {
220
-
counter!("jetstream.read.fail", "url" => url, "reason" => "invalid event")
220
+
counter!("jetstream_read_fail", "url" => url, "reason" => "invalid event")
221
221
.increment(1);
222
222
eprintln!("jetstream: encountered an event without a timestamp: ignoring it.");
223
223
continue;
···
225
225
};
226
226
227
227
if let Err(flume::SendError(_rejected)) = sender.send(v) {
228
-
counter!("jetstream.events", "url" => url).increment(1);
228
+
counter!("jetstream_events", "url" => url).increment(1);
229
229
if sender.is_disconnected() {
230
230
eprintln!("jetstream: send channel disconnected -- nothing to do, bye.");
231
231
break 'outer;
···
234
234
"jetstream: failed to send on channel, dropping update! (FIXME / HANDLEME)"
235
235
);
236
236
}
237
-
histogram!("jetstream.events.queued", "url" => url).record(sender.len() as f64);
237
+
histogram!("jetstream_events_queued", "url" => url).record(sender.len() as f64);
238
238
239
239
// only actually update our cursor after we've managed to queue the event
240
240
latest_cursor = Some(ts);
241
-
gauge!("jetstream.cursor.age", "url" => url).set(ts_age(ts).as_micros() as f64);
241
+
gauge!("jetstream_cursor_age", "url" => url).set(ts_age(ts).as_micros() as f64);
242
242
}
243
243
}
244
244
Err(anyhow::anyhow!("broke out of jetstream loop"))
+15
-15
link_aggregator/src/consumer/mod.rs
+15
-15
link_aggregator/src/consumer/mod.rs
···
15
15
16
16
pub fn consume(mut store: impl LinkStorage, qsize: Arc<AtomicU32>, fixture: Option<PathBuf>) {
17
17
describe_counter!(
18
-
"consumer.events.non_actionable",
18
+
"consumer_events_non_actionable",
19
19
Unit::Count,
20
20
"count of non-actionable events"
21
21
);
22
22
describe_counter!(
23
-
"consumer.events.actionable",
23
+
"consumer_events_actionable",
24
24
Unit::Count,
25
25
"count of action by type. *all* atproto record delete events are included"
26
26
);
27
27
describe_counter!(
28
-
"consumer.events.actionable.links",
28
+
"consumer_events_actionable_links",
29
29
Unit::Count,
30
30
"total links encountered"
31
31
);
32
32
describe_histogram!(
33
-
"consumer.events.actionable.links",
33
+
"consumer_events_actionable_links",
34
34
Unit::Count,
35
35
"number of links per message"
36
36
);
···
57
57
qsize.store(receiver.len().try_into().unwrap(), Ordering::Relaxed);
58
58
}
59
59
} else {
60
-
counter!("consumer.events.non_actionable").increment(1);
60
+
counter!("consumer_events_non_actionable").increment(1);
61
61
}
62
62
}
63
63
···
92
92
match commit.get("operation")? {
93
93
JsonValue::String(op) if op == "create" => {
94
94
let links = collect_links(commit.get("record")?);
95
-
counter!("consumer.events.actionable", "action_type" => "create_links", "collection" => collection.clone()).increment(1);
96
-
histogram!("consumer.events.actionable.links", "action_type" => "create_links", "collection" => collection.clone()).record(links.len() as f64);
95
+
counter!("consumer_events_actionable", "action_type" => "create_links", "collection" => collection.clone()).increment(1);
96
+
histogram!("consumer_events_actionable_links", "action_type" => "create_links", "collection" => collection.clone()).record(links.len() as f64);
97
97
for link in &links {
98
-
counter!("consumer.events.actionable.links",
98
+
counter!("consumer_events_actionable_links",
99
99
"action_type" => "create_links",
100
100
"collection" => collection.clone(),
101
101
"path" => link.path.clone(),
···
121
121
}
122
122
JsonValue::String(op) if op == "update" => {
123
123
let links = collect_links(commit.get("record")?);
124
-
counter!("consumer.events.actionable", "action_type" => "update_links", "collection" => collection.clone()).increment(1);
125
-
histogram!("consumer.events.actionable.links", "action_type" => "update_links", "collection" => collection.clone()).record(links.len() as f64);
124
+
counter!("consumer_events_actionable", "action_type" => "update_links", "collection" => collection.clone()).increment(1);
125
+
histogram!("consumer_events_actionable_links", "action_type" => "update_links", "collection" => collection.clone()).record(links.len() as f64);
126
126
for link in &links {
127
-
counter!("consumer.events.actionable.links",
127
+
counter!("consumer_events_actionable_links",
128
128
"action_type" => "update_links",
129
129
"collection" => collection.clone(),
130
130
"path" => link.path.clone(),
···
145
145
))
146
146
}
147
147
JsonValue::String(op) if op == "delete" => {
148
-
counter!("consumer.events.actionable", "action_type" => "delete_record", "collection" => collection.clone()).increment(1);
148
+
counter!("consumer_events_actionable", "action_type" => "delete_record", "collection" => collection.clone()).increment(1);
149
149
Some((
150
150
ActionableEvent::DeleteRecord(RecordId {
151
151
did: did.into(),
···
167
167
let did = account.get("did")?.get::<String>()?.clone();
168
168
match (account.get("active")?.get::<bool>()?, account.get("status")) {
169
169
(true, None) => {
170
-
counter!("consumer.events.actionable", "action_type" => "account", "action" => "activate").increment(1);
170
+
counter!("consumer_events_actionable", "action_type" => "account", "action" => "activate").increment(1);
171
171
Some((ActionableEvent::ActivateAccount(did.into()), cursor))
172
172
}
173
173
(false, Some(JsonValue::String(status))) => match status.as_ref() {
174
174
"deactivated" => {
175
-
counter!("consumer.events.actionable", "action_type" => "account", "action" => "deactivate").increment(1);
175
+
counter!("consumer_events_actionable", "action_type" => "account", "action" => "deactivate").increment(1);
176
176
Some((ActionableEvent::DeactivateAccount(did.into()), cursor))
177
177
}
178
178
"deleted" => {
179
-
counter!("consumer.events.actionable", "action_type" => "account", "action" => "delete").increment(1);
179
+
counter!("consumer_events_actionable", "action_type" => "account", "action" => "delete").increment(1);
180
180
Some((ActionableEvent::DeleteAccount(did.into()), cursor))
181
181
}
182
182
_ => None,
+2
-2
link_aggregator/src/main.rs
+2
-2
link_aggregator/src/main.rs
···
101
101
let process_collector = metrics_process::Collector::default();
102
102
process_collector.describe();
103
103
metrics::describe_gauge!(
104
-
"storage.available",
104
+
"storage_available",
105
105
metrics::Unit::Bytes,
106
106
"available to be allocated"
107
107
);
108
108
metrics::describe_gauge!(
109
-
"storage.free",
109
+
"storage_free",
110
110
metrics::Unit::Bytes,
111
111
"unused bytes in filesystem"
112
112
);