servers/http/prom_store/
exemplar.rs

1// Copyright 2023 Greptime 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
15use std::collections::HashSet;
16use std::sync::Arc;
17
18use api::v1::RowInsertRequests;
19use common_error::ext::BoxedError;
20use futures::future::ready;
21use operator::insert::PreparedMetricInserts;
22use session::context::QueryContextRef;
23use snafu::ResultExt;
24
25use crate::batcher::logical_table::LogicalTablePendingRowsBatcher;
26use crate::error::{self, Result};
27use crate::exemplar::{ExemplarRequests, into_rows};
28use crate::http::prom_store::PromWriteBatch;
29use crate::query_handler::PromStoreProtocolHandlerRef;
30
31#[derive(Default)]
32pub(super) struct Outcome {
33    pub cost: usize,
34    pub accepted: bool,
35    /// Accepted rows on success; confirmed storage writes on error.
36    pub written: [u64; 3],
37    pub error: Option<error::Error>,
38}
39
40enum PreparedWrite {
41    Metric(PreparedMetricInserts),
42    Rows(RowInsertRequests),
43}
44
45/// Preflights all destinations together, retaining per-kind write progress.
46pub(super) async fn write(
47    handler: PromStoreProtocolHandlerRef,
48    batcher: Option<Arc<LogicalTablePendingRowsBatcher>>,
49    metric_engine: bool,
50    samples: Vec<PromWriteBatch>,
51    histograms: Vec<PromWriteBatch>,
52    exemplars: ExemplarRequests,
53    ctx: QueryContextRef,
54) -> Outcome {
55    match prepare(&handler, samples, histograms, exemplars, ctx).await {
56        Ok((writes, targets)) => {
57            let mut sources = Vec::new();
58            let mut prepared = Vec::new();
59            let mut source_preparation_failed = false;
60            for (kind, ctx, request) in writes {
61                if kind != 2 && metric_engine {
62                    let (pinned, other): (Vec<_>, Vec<_>) =
63                        request.inserts.into_iter().partition(|request| {
64                            targets.contains(&(
65                                ctx.current_catalog().to_string(),
66                                ctx.current_schema(),
67                                request.table_name.clone(),
68                            ))
69                        });
70                    if !other.is_empty() {
71                        prepared.push((
72                            kind,
73                            ctx.clone(),
74                            Ok(PreparedWrite::Rows(RowInsertRequests { inserts: other })),
75                        ));
76                    }
77                    if !pinned.is_empty() {
78                        let result = handler
79                            .prepare_metric_write(
80                                RowInsertRequests { inserts: pinned },
81                                ctx.clone(),
82                            )
83                            .await;
84                        if let Ok(request) = &result {
85                            sources.extend(request.source_tables().cloned());
86                        } else {
87                            source_preparation_failed = true;
88                        }
89                        prepared.push((kind, ctx, result.map(PreparedWrite::Metric)));
90                    }
91                } else {
92                    prepared.push((kind, ctx, Ok(PreparedWrite::Rows(request))));
93                }
94            }
95            // A failed source preparation must not fall back to a newer table with the same name.
96            if source_preparation_failed {
97                prepared.retain(|(kind, _, _)| *kind != 2);
98            }
99            let sample_batches = prepared.iter().filter_map(|(kind, ctx, request)| {
100                if *kind == 0
101                    && let Ok(PreparedWrite::Rows(request)) = request
102                {
103                    Some((ctx, request))
104                } else {
105                    None
106                }
107            });
108            let batcher = if metric_engine
109                && let Some(batcher) = batcher.as_ref()
110                && batcher.accepts_bulk_destinations(sample_batches).await
111            {
112                Some(batcher)
113            } else {
114                None
115            };
116            let mut outputs = Vec::with_capacity(prepared.len());
117            // Serialize schema preparation across batches that resolve to the same table.
118            for (kind, ctx, request) in prepared {
119                let mut deferred = false;
120                let output = match request {
121                    Err(error) => Err(error),
122                    Ok(PreparedWrite::Metric(request)) => {
123                        handler.write_prepared_metric(request, ctx).await
124                    }
125                    Ok(PreparedWrite::Rows(request)) => {
126                        if kind == 2 && !metric_engine {
127                            error::NotSupportedSnafu {
128                                feat: "exemplars require metric-engine ingestion",
129                            }
130                            .fail()
131                        } else if kind == 2 {
132                            handler
133                                .write_exemplars_with_sources(request, ctx, &sources)
134                                .await
135                        } else if kind == 0
136                            && let Some(batcher) = batcher
137                        {
138                            deferred = !batcher.is_synchronous();
139                            batcher
140                                .submit_output_with(request, ctx, |_| ready(Ok(())))
141                                .await
142                                .map(|(output, ())| output)
143                        } else {
144                            handler.write_prepared(request, ctx, metric_engine).await
145                        }
146                    }
147                };
148                outputs.push((kind, output, deferred));
149            }
150            let mut outcome = Outcome::default();
151            let mut pending_samples = Vec::new();
152            for (kind, output, deferred) in outputs {
153                match output {
154                    Ok(output) => {
155                        outcome.cost += output.meta.cost;
156                        if deferred && !output.meta.write_completions.is_empty() {
157                            pending_samples.push(output);
158                            continue;
159                        }
160                        let progress = output.write_progress().await;
161                        outcome.written[kind] += progress.affected_rows as u64;
162                        if let Err(error) =
163                            progress.into_result().context(error::ExecuteGrpcQuerySnafu)
164                        {
165                            outcome.error.get_or_insert(error);
166                        }
167                    }
168                    Err(error) => {
169                        outcome.error.get_or_insert(error);
170                    }
171                }
172            }
173            if outcome.error.is_none() {
174                outcome.accepted = !pending_samples.is_empty();
175                outcome.written[0] += pending_samples
176                    .iter()
177                    .map(|output| output.extract_rows_and_cost().0 as u64)
178                    .sum::<u64>();
179            } else {
180                // Split submissions may share a counter; each completion owns its row count.
181                for output in pending_samples {
182                    outcome.written[0] += output.write_progress().await.affected_rows as u64;
183                }
184            }
185            outcome
186        }
187        Err(error) => Outcome {
188            error: Some(error),
189            ..Default::default()
190        },
191    }
192}
193
194async fn prepare(
195    handler: &PromStoreProtocolHandlerRef,
196    samples: Vec<PromWriteBatch>,
197    histograms: Vec<PromWriteBatch>,
198    exemplars: ExemplarRequests,
199    ctx: QueryContextRef,
200) -> Result<(
201    Vec<(usize, QueryContextRef, RowInsertRequests)>,
202    HashSet<(String, String, String)>,
203)> {
204    let mut kinds = Vec::new();
205    let mut prepared = Vec::new();
206    let mut targets = HashSet::new();
207    for (kind, batches) in [samples, histograms].into_iter().enumerate() {
208        for (ctx, request) in batches {
209            handler.pre_write(&request, ctx.clone()).await?;
210            kinds.push(kind);
211            // Keep per-batch routing options when admitting database-wide totals.
212            prepared.push((Arc::new(ctx.with_confirmed_writes()), request));
213        }
214    }
215    for (schema, exemplars) in exemplars {
216        let target = ctx.fork();
217        if let Some(schema) = schema {
218            target.set_current_schema(&schema);
219        }
220        let (request, sources) = into_rows(exemplars)?;
221        let target = Arc::new(target);
222        handler.pre_write(&request, target.clone()).await?;
223        let target = Arc::new(target.with_confirmed_writes());
224        if !sources.is_empty() {
225            handler
226                .check_exemplar_sources(&sources, target.clone())
227                .await?;
228        }
229        targets.extend(sources.into_iter().map(|name| {
230            (
231                target.current_catalog().to_string(),
232                target.current_schema(),
233                name,
234            )
235        }));
236        kinds.push(2);
237        prepared.push((target, request));
238    }
239    operator::insert::admit_row_insert_batches(&mut prepared)
240        .await
241        .map_err(BoxedError::new)
242        .context(error::ExecuteGrpcQuerySnafu)?;
243    let writes = kinds
244        .into_iter()
245        .zip(prepared)
246        .map(|(kind, (ctx, request))| (kind, ctx, request))
247        .collect();
248    Ok((writes, targets))
249}
250
251#[cfg(test)]
252mod tests {
253    use std::collections::{BTreeMap, BTreeSet};
254    use std::sync::Mutex;
255    use std::sync::atomic::{AtomicBool, Ordering};
256
257    use api::prom_store::remote::ReadRequest;
258    use api::v1::{Row, RowInsertRequest, Rows};
259    use async_trait::async_trait;
260    use common_query::{Output, OutputData, OutputMeta, WriteCompletion, WriteProgress};
261    use session::context::QueryContext;
262    use tokio::sync::Notify;
263
264    use super::*;
265    use crate::exemplar::Exemplar;
266    use crate::query_handler::{PromStoreProtocolHandler, PromStoreResponse};
267
268    #[derive(Default)]
269    struct Handler {
270        events: Mutex<Vec<String>>,
271        exemplar_done: Arc<Notify>,
272        deny_schema: bool,
273        fail_preparation: bool,
274        successful_writes: bool,
275        check_serial: bool,
276        writing: AtomicBool,
277    }
278
279    #[async_trait]
280    impl PromStoreProtocolHandler for Handler {
281        async fn prepare_metric_write(
282            &self,
283            _: RowInsertRequests,
284            _: QueryContextRef,
285        ) -> Result<PreparedMetricInserts> {
286            if self.fail_preparation {
287                return error::InvalidPromRemoteRequestSnafu {
288                    msg: "source preparation failed",
289                }
290                .fail();
291            }
292            Ok(PreparedMetricInserts::default())
293        }
294
295        async fn write_prepared_metric(
296            &self,
297            _: PreparedMetricInserts,
298            ctx: QueryContextRef,
299        ) -> Result<Output> {
300            self.write_prepared(RowInsertRequests::default(), ctx, true)
301                .await
302        }
303
304        async fn pre_write(&self, request: &RowInsertRequests, ctx: QueryContextRef) -> Result<()> {
305            self.events
306                .lock()
307                .unwrap()
308                .push(format!("pre:{}", ctx.current_schema()));
309            assert!(!request.inserts.is_empty());
310            if self.deny_schema && ctx.current_schema() == "other" {
311                return error::InvalidPromRemoteRequestSnafu {
312                    msg: "denied destination",
313                }
314                .fail();
315            }
316            if let Some(schema) = ctx.extension("resolved_schema") {
317                ctx.set_current_schema(schema);
318            }
319            Ok(())
320        }
321
322        async fn check_exemplar_sources(
323            &self,
324            sources: &BTreeSet<String>,
325            ctx: QueryContextRef,
326        ) -> Result<()> {
327            assert_eq!(sources, &BTreeSet::from(["metric".into()]));
328            self.events
329                .lock()
330                .unwrap()
331                .push(format!("source:{}", ctx.current_schema()));
332            Ok(())
333        }
334
335        async fn write_prepared(
336            &self,
337            _request: RowInsertRequests,
338            ctx: QueryContextRef,
339            _: bool,
340        ) -> Result<Output> {
341            if self.check_serial {
342                assert!(!self.writing.swap(true, Ordering::SeqCst));
343                tokio::task::yield_now().await;
344                self.writing.store(false, Ordering::SeqCst);
345            }
346            assert!(ctx.confirmed_writes());
347            assert_eq!(
348                ctx.write_rows_to_admit(ctx.current_catalog(), &ctx.current_schema(), 3),
349                0
350            );
351            self.events.lock().unwrap().push("metric".into());
352            if self.successful_writes {
353                if let Some(counter) = ctx.write_progress() {
354                    counter.add_written(2);
355                }
356                return Ok(Output::new_with_affected_rows(2));
357            }
358            let done = self.exemplar_done.clone();
359            let counter = ctx.write_progress().cloned();
360            let completion = WriteCompletion::with_progress(async move {
361                done.notified().await;
362                if let Some(counter) = counter {
363                    counter.add_written(1);
364                }
365                WriteProgress {
366                    affected_rows: 1,
367                    error: Some(Arc::new(BoxedError::new(
368                        error::InternalSnafu {
369                            err_msg: "late metric failure",
370                        }
371                        .build(),
372                    ))),
373                }
374            });
375            Ok(Output::new(
376                OutputData::AffectedRows(2),
377                OutputMeta {
378                    cost: 2,
379                    write_completions: vec![completion],
380                    ..Default::default()
381                },
382            ))
383        }
384
385        async fn write_exemplars_prepared(
386            &self,
387            _: RowInsertRequests,
388            ctx: QueryContextRef,
389        ) -> Result<Output> {
390            assert!(ctx.confirmed_writes());
391            assert_eq!(
392                ctx.write_rows_to_admit(ctx.current_catalog(), &ctx.current_schema(), 1),
393                0
394            );
395            self.events.lock().unwrap().push("exemplar".into());
396            if let Some(counter) = ctx.write_progress() {
397                counter.add_written(1);
398            }
399            self.exemplar_done.notify_one();
400            Ok(Output::new(
401                OutputData::AffectedRows(1),
402                OutputMeta::new_with_cost(1),
403            ))
404        }
405
406        async fn write(&self, _: RowInsertRequests, _: QueryContextRef, _: bool) -> Result<Output> {
407            unreachable!()
408        }
409        async fn write_all(
410            &self,
411            _: Vec<(QueryContextRef, RowInsertRequests)>,
412            _: bool,
413        ) -> Result<Vec<Result<Output>>> {
414            unreachable!()
415        }
416        async fn read(&self, _: ReadRequest, _: QueryContextRef) -> Result<PromStoreResponse> {
417            unreachable!()
418        }
419    }
420
421    #[tokio::test]
422    async fn exemplar_requires_successful_metric_engine_source_preparation() {
423        for metric_engine in [false, true] {
424            let handler = Arc::new(Handler {
425                fail_preparation: true,
426                ..Default::default()
427            });
428            let ctx = QueryContext::arc();
429            let samples = if metric_engine {
430                vec![(
431                    ctx.clone(),
432                    RowInsertRequests {
433                        inserts: vec![RowInsertRequest {
434                            table_name: "metric".into(),
435                            rows: Some(Rows {
436                                schema: vec![],
437                                rows: vec![Row { values: vec![] }],
438                            }),
439                        }],
440                    },
441                )]
442            } else {
443                vec![]
444            };
445            let exemplars = ExemplarRequests::from([(
446                None,
447                vec![Exemplar {
448                    metric: "metric".into(),
449                    series_labels: BTreeMap::new().into(),
450                    labels: BTreeMap::new(),
451                    timestamp: 1,
452                    value: 1.0,
453                }],
454            )]);
455            let outcome = write(
456                handler.clone(),
457                None,
458                metric_engine,
459                samples,
460                vec![],
461                exemplars,
462                ctx,
463            )
464            .await;
465            assert!(outcome.error.is_some());
466            assert_eq!(outcome.written, [0, 0, 0]);
467            assert!(
468                !handler
469                    .events
470                    .lock()
471                    .unwrap()
472                    .iter()
473                    .any(|e| e == "metric" || e == "exemplar")
474            );
475        }
476    }
477
478    #[tokio::test]
479    async fn exemplar_preserves_batch_routing_after_preflight() {
480        use crate::http::prom_store::PHYSICAL_TABLE_PARAM;
481
482        let handler: PromStoreProtocolHandlerRef = Arc::new(Handler::default());
483        let mut ctx = QueryContext::with("greptime", "public");
484        ctx.set_extension("resolved_schema", "tenant");
485        let ctx = Arc::new(ctx);
486        let batch = |physical: Option<&str>| {
487            let mut target = ctx.fork();
488            if let Some(physical) = physical {
489                target.set_extension(PHYSICAL_TABLE_PARAM, physical);
490            }
491            (
492                Arc::new(target),
493                RowInsertRequests {
494                    inserts: vec![RowInsertRequest {
495                        table_name: "metric".into(),
496                        rows: Some(Rows {
497                            schema: vec![],
498                            rows: vec![Row { values: vec![] }],
499                        }),
500                    }],
501                },
502            )
503        };
504        let exemplars = ExemplarRequests::from([(
505            None,
506            vec![Exemplar {
507                metric: "metric".into(),
508                series_labels: BTreeMap::new().into(),
509                labels: BTreeMap::new(),
510                timestamp: 1,
511                value: 1.0,
512            }],
513        )]);
514        let (writes, targets) = prepare(
515            &handler,
516            vec![
517                batch(Some("physical_a")),
518                batch(Some("physical_b")),
519                batch(None),
520            ],
521            vec![batch(Some("histogram_physical"))],
522            exemplars,
523            ctx,
524        )
525        .await
526        .unwrap();
527        assert_eq!(
528            targets,
529            HashSet::from([("greptime".into(), "tenant".into(), "metric".into())])
530        );
531        let expected = [
532            (0, Some("physical_a")),
533            (0, Some("physical_b")),
534            (0, None),
535            (1, Some("histogram_physical")),
536            (2, None),
537        ];
538        assert_eq!(writes.len(), expected.len());
539        for ((kind, ctx, request), (expected_kind, physical)) in writes.into_iter().zip(expected) {
540            assert_eq!(kind, expected_kind);
541            assert_eq!(ctx.extension(PHYSICAL_TABLE_PARAM), physical);
542            assert_eq!(ctx.current_schema(), "tenant");
543            assert!(ctx.confirmed_writes());
544            assert_eq!(ctx.write_rows_to_admit("greptime", "tenant", 1), 0);
545            assert_eq!(request.inserts.len(), 1);
546        }
547    }
548
549    #[tokio::test]
550    async fn exemplar_submits_metric_batches_sequentially() {
551        let handler = Arc::new(Handler {
552            successful_writes: true,
553            check_serial: true,
554            ..Default::default()
555        });
556        let ctx = QueryContext::arc();
557        let samples = ["metric", "other", "other"]
558            .into_iter()
559            .map(|name| {
560                (
561                    Arc::new(ctx.fork()),
562                    RowInsertRequests {
563                        inserts: vec![RowInsertRequest {
564                            table_name: name.into(),
565                            rows: Some(Rows {
566                                schema: vec![],
567                                rows: vec![Row { values: vec![] }; 2],
568                            }),
569                        }],
570                    },
571                )
572            })
573            .collect();
574        let exemplars = ExemplarRequests::from([(
575            None,
576            vec![Exemplar {
577                metric: "metric".into(),
578                series_labels: BTreeMap::new().into(),
579                labels: BTreeMap::new(),
580                timestamp: 1,
581                value: 1.0,
582            }],
583        )]);
584        let outcome = write(handler, None, true, samples, vec![], exemplars, ctx).await;
585        assert!(outcome.error.is_none());
586        assert_eq!(outcome.written, [6, 0, 1]);
587    }
588
589    #[tokio::test]
590    async fn exemplar_v2_batching_acknowledgement_and_partial_counts() {
591        use api::greptime_proto::io::prometheus::write::v2;
592        use common_memory_manager::OnExhaustedPolicy;
593        use prost::Message;
594
595        use crate::batcher::logical_table::test_util::{WorkerCommand, submission_batcher};
596        use crate::http::extractor::PipelineInfo;
597        use crate::http::prom_store::{PromStoreState, RemoteWriteQuery, remote_write_v2};
598        use crate::prom_remote_write::REMOTE_WRITE_V2_VERSION;
599        use crate::prom_remote_write::validation::PromValidationMode;
600        use crate::request_memory_limiter::ServerMemoryLimiter;
601
602        for (sync, successful_writes) in
603            [(false, false), (false, true), (true, false), (true, true)]
604        {
605            let (batcher, mut receiver) = submission_batcher(sync).await;
606            let limiter = ServerMemoryLimiter::new(1024 * 1024, OnExhaustedPolicy::Fail);
607            let db = "exemplar_confirmed_counts";
608            let counters = [
609                &*crate::metrics::PROM_STORE_REMOTE_WRITE_SAMPLES,
610                &*crate::metrics::PROM_STORE_REMOTE_WRITE_EXEMPLARS,
611            ]
612            .map(|counter| counter.with_label_values(&[db, REMOTE_WRITE_V2_VERSION]));
613            let before = counters.each_ref().map(|counter| counter.get());
614            let request = v2::Request {
615                symbols: vec![
616                    String::new(),
617                    "__name__".into(),
618                    "metric".into(),
619                    "cpu".into(),
620                ],
621                timeseries: vec![
622                    v2::TimeSeries {
623                        labels_refs: vec![1, 2],
624                        samples: vec![
625                            v2::Sample {
626                                timestamp: 1,
627                                value: 1.0,
628                                ..Default::default()
629                            };
630                            2
631                        ],
632                        exemplars: vec![v2::Exemplar {
633                            labels_refs: vec![],
634                            value: 0.25,
635                            timestamp: 1,
636                        }],
637                        ..Default::default()
638                    },
639                    v2::TimeSeries {
640                        labels_refs: vec![1, 3],
641                        samples: vec![
642                            v2::Sample {
643                                timestamp: 1,
644                                value: 1.0,
645                                ..Default::default()
646                            };
647                            2
648                        ],
649                        ..Default::default()
650                    },
651                ],
652            };
653            let response = remote_write_v2(
654                PromStoreState {
655                    prom_store_handler: Arc::new(Handler {
656                        successful_writes,
657                        ..Default::default()
658                    }),
659                    pipeline_handler: None,
660                    prom_store_with_metric_engine: true,
661                    prom_validation_mode: PromValidationMode::Strict,
662                    pending_rows_batcher: Some(batcher),
663                    memory_limiter: limiter.clone(),
664                },
665                RemoteWriteQuery {
666                    db: Some(db.into()),
667                    ..Default::default()
668                },
669                QueryContext::with("greptime", "public"),
670                PipelineInfo {
671                    pipeline_name: None,
672                    pipeline_version: None,
673                    pipeline_params: Default::default(),
674                },
675                false,
676                crate::prom_store::snappy_compress(&request.encode_to_vec())
677                    .unwrap()
678                    .into(),
679                0,
680            );
681            tokio::pin!(response);
682            let mut early_response = None;
683            let command = tokio::select! {
684                biased;
685                result = &mut response => {
686                    early_response = Some(result.unwrap());
687                    receiver.try_recv().unwrap()
688                }
689                command = receiver.recv() => command.unwrap(),
690            };
691            assert_eq!(early_response.is_some(), !sync && successful_writes);
692            assert!(limiter.used_bytes() > 0);
693            let WorkerCommand::Submit {
694                ctx,
695                response_tx,
696                _completion,
697                ..
698            } = command
699            else {
700                panic!("expected sample submission")
701            };
702            let affected_rows = if successful_writes { 2 } else { 1 };
703            ctx.write_progress().unwrap().add_written(affected_rows);
704            let progress = WriteProgress {
705                affected_rows,
706                error: (!successful_writes).then(|| {
707                    Arc::new(BoxedError::new(
708                        error::InternalSnafu {
709                            err_msg: "partial queued write",
710                        }
711                        .build(),
712                    ))
713                }),
714            };
715            assert_eq!(
716                response_tx.send(progress).is_ok(),
717                sync || !successful_writes
718            );
719            drop(_completion);
720            let response = match early_response {
721                Some(response) => response,
722                None => response.await.unwrap(),
723            };
724            if successful_writes {
725                assert_eq!(response.status(), if sync { 204 } else { 202 });
726            } else {
727                assert!(response.status().is_server_error());
728            }
729            let samples = if successful_writes { "4" } else { "2" };
730            for (kind, count) in [
731                ("samples", samples),
732                ("histograms", "0"),
733                ("exemplars", "1"),
734            ] {
735                assert_eq!(
736                    response.headers()[format!("x-prometheus-remote-write-{kind}-written")],
737                    count
738                );
739            }
740            for ((counter, before), count) in counters
741                .into_iter()
742                .zip(before)
743                .zip([if successful_writes { 4 } else { 2 }, 1])
744            {
745                assert_eq!(counter.get() - before, count);
746            }
747            drop(ctx);
748            assert_eq!(limiter.used_bytes(), 0);
749        }
750    }
751
752    #[tokio::test]
753    async fn exemplar_preflights_all_databases_and_retains_delayed_partial_progress() {
754        for denied in [false, true] {
755            let handler = Arc::new(Handler {
756                deny_schema: denied,
757                ..Default::default()
758            });
759            let mut exemplars = ExemplarRequests::new();
760            for schema in if denied {
761                vec![None, Some("other".into())]
762            } else {
763                vec![None]
764            } {
765                exemplars.insert(
766                    schema,
767                    vec![Exemplar {
768                        metric: "metric".into(),
769                        series_labels: BTreeMap::new().into(),
770                        labels: BTreeMap::new(),
771                        timestamp: 1,
772                        value: 1.0,
773                    }],
774                );
775            }
776            let ctx = QueryContext::arc();
777            let samples = vec![(
778                ctx.clone(),
779                RowInsertRequests {
780                    inserts: vec![RowInsertRequest {
781                        table_name: "metric".into(),
782                        rows: Some(Rows {
783                            schema: vec![],
784                            rows: vec![Row { values: vec![] }; 2],
785                        }),
786                    }],
787                },
788            )];
789            let outcome = write(handler.clone(), None, true, samples, vec![], exemplars, ctx).await;
790            assert!(outcome.error.is_some());
791            let events = handler.events.lock().unwrap();
792            if denied {
793                assert_eq!(outcome.written, [0, 0, 0]);
794                assert!(!events.iter().any(|e| e == "metric" || e == "exemplar"));
795            } else {
796                assert_eq!(outcome.written, [1, 0, 1]);
797                assert_eq!(outcome.cost, 3);
798                assert_eq!(
799                    *events,
800                    [
801                        "pre:public",
802                        "pre:public",
803                        "source:public",
804                        "metric",
805                        "exemplar"
806                    ]
807                );
808            }
809        }
810    }
811}
servers/http/prom_store/exemplar.rs

servers/http/prom_store/
exemplar.rs
