frontend/service_config/
prom_store.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::num::NonZeroUsize;
16use std::time::Duration;
17
18use serde::{Deserialize, Serialize};
19use servers::prom_remote_write::validation::PromValidationMode;
20
21#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
22pub struct PromStoreOptions {
23    pub enable: bool,
24    pub with_metric_engine: bool,
25    /// Validation mode while decoding Prometheus remote write requests.
26    #[serde(default)]
27    pub prom_validation_mode: PromValidationMode,
28    /// Enables experimental Prometheus remote write v2 native histogram ingestion.
29    #[serde(default)]
30    pub experimental_enable_prometheus_native_histogram: bool,
31    #[serde(default, with = "humantime_serde")]
32    pub pending_rows_flush_interval: Duration,
33    #[serde(default = "default_max_batch_rows")]
34    pub max_batch_rows: usize,
35    #[serde(default = "default_max_concurrent_flushes")]
36    pub max_concurrent_flushes: usize,
37    #[serde(default = "default_worker_channel_capacity")]
38    pub worker_channel_capacity: usize,
39    #[serde(default = "default_max_inflight_requests")]
40    pub max_inflight_requests: usize,
41    /// Maximum number of logical-table flow notifications waiting in the shared queue.
42    #[serde(default = "default_flow_notification_queue_capacity")]
43    pub flow_notification_queue_capacity: NonZeroUsize,
44}
45
46fn default_max_batch_rows() -> usize {
47    100_000
48}
49
50fn default_max_concurrent_flushes() -> usize {
51    256
52}
53
54fn default_worker_channel_capacity() -> usize {
55    65526
56}
57
58fn default_max_inflight_requests() -> usize {
59    3000
60}
61
62fn default_flow_notification_queue_capacity() -> NonZeroUsize {
63    NonZeroUsize::new(1024).unwrap_or(NonZeroUsize::MIN)
64}
65
66impl PromStoreOptions {
67    /// Returns whether the pending rows batcher can be enabled with these
68    /// options. Mirrors the enablement conditions of
69    /// `PendingRowsBatcher::try_new` in the servers crate, which returns
70    /// `None` when any of these knobs is zero.
71    pub fn pending_rows_batching_enabled(&self) -> bool {
72        self.enable
73            && self.with_metric_engine
74            && !self.pending_rows_flush_interval.is_zero()
75            && self.max_batch_rows > 0
76            && self.max_concurrent_flushes > 0
77            && self.worker_channel_capacity > 0
78            && self.max_inflight_requests > 0
79    }
80}
81
82impl Default for PromStoreOptions {
83    fn default() -> Self {
84        Self {
85            enable: true,
86            with_metric_engine: true,
87            prom_validation_mode: PromValidationMode::Strict,
88            experimental_enable_prometheus_native_histogram: false,
89            pending_rows_flush_interval: Duration::ZERO,
90            max_batch_rows: default_max_batch_rows(),
91            max_concurrent_flushes: default_max_concurrent_flushes(),
92            worker_channel_capacity: default_worker_channel_capacity(),
93            max_inflight_requests: default_max_inflight_requests(),
94            flow_notification_queue_capacity: default_flow_notification_queue_capacity(),
95        }
96    }
97}
98
99#[cfg(test)]
100mod tests {
101    use std::time::Duration;
102
103    use super::{PromStoreOptions, PromValidationMode};
104    use crate::service_config::prom_store::{
105        default_flow_notification_queue_capacity, default_max_batch_rows,
106        default_max_concurrent_flushes, default_max_inflight_requests,
107        default_worker_channel_capacity,
108    };
109
110    #[test]
111    fn test_prom_store_options() {
112        let default = PromStoreOptions::default();
113        assert!(default.enable);
114        assert!(default.with_metric_engine);
115        assert_eq!(default.prom_validation_mode, PromValidationMode::Strict);
116        assert!(!default.experimental_enable_prometheus_native_histogram);
117        assert_eq!(default.pending_rows_flush_interval, Duration::ZERO);
118        assert_eq!(default.max_batch_rows, default_max_batch_rows());
119        assert_eq!(
120            default.max_concurrent_flushes,
121            default_max_concurrent_flushes()
122        );
123        assert_eq!(
124            default.worker_channel_capacity,
125            default_worker_channel_capacity()
126        );
127        assert_eq!(
128            default.max_inflight_requests,
129            default_max_inflight_requests()
130        );
131        assert_eq!(
132            default.flow_notification_queue_capacity,
133            default_flow_notification_queue_capacity()
134        );
135    }
136}
frontend/service_config/prom_store.rs

frontend/service_config/
prom_store.rs
