meta_client/
error.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 common_error::define_from_tonic_status;
16use common_error::ext::{ErrorExt, RetryHint};
17use common_error::status_code::StatusCode;
18use common_macro::stack_trace_debug;
19use snafu::{Location, Snafu};
20
21#[derive(Snafu)]
22#[snafu(visibility(pub))]
23#[stack_trace_debug]
24pub enum Error {
25    #[snafu(display("Illegal GRPC client state: {}", err_msg))]
26    IllegalGrpcClientState {
27        err_msg: String,
28        #[snafu(implicit)]
29        location: Location,
30    },
31
32    #[snafu(display("{}, code: {}, tonic code: {}", msg, code, tonic_code))]
33    MetaServer {
34        code: StatusCode,
35        msg: String,
36        tonic_code: tonic::Code,
37        retry_hint: RetryHint,
38        #[snafu(implicit)]
39        location: Location,
40    },
41
42    #[snafu(display("No leader, should ask leader first"))]
43    NoLeader {
44        #[snafu(implicit)]
45        location: Location,
46    },
47
48    #[snafu(display("Ask leader timeout"))]
49    AskLeaderTimeout {
50        #[snafu(implicit)]
51        location: Location,
52        #[snafu(source)]
53        error: tokio::time::error::Elapsed,
54    },
55
56    #[snafu(display("Failed to create gRPC channel"))]
57    CreateChannel {
58        #[snafu(implicit)]
59        location: Location,
60        source: common_grpc::error::Error,
61    },
62
63    #[snafu(display("{} not started", name))]
64    NotStarted {
65        name: String,
66        #[snafu(implicit)]
67        location: Location,
68    },
69
70    #[snafu(display("Procedure submission requires query context"))]
71    MissingQueryContext {
72        #[snafu(implicit)]
73        location: Location,
74    },
75
76    #[snafu(display("Failed to send heartbeat: {}", err_msg))]
77    SendHeartbeat {
78        err_msg: String,
79        #[snafu(implicit)]
80        location: Location,
81    },
82
83    #[snafu(display("Failed create heartbeat stream to server"))]
84    CreateHeartbeatStream {
85        #[snafu(implicit)]
86        location: Location,
87    },
88
89    #[snafu(display("Heartbeat handshake timeout after {timeout:?}"))]
90    HeartbeatHandshakeTimeout {
91        timeout: std::time::Duration,
92        #[snafu(implicit)]
93        location: Location,
94        #[snafu(source)]
95        error: tokio::time::error::Elapsed,
96    },
97
98    #[snafu(display("Invalid response header"))]
99    InvalidResponseHeader {
100        #[snafu(implicit)]
101        location: Location,
102        source: common_meta::error::Error,
103    },
104
105    #[snafu(display("Failed to convert Metasrv request"))]
106    ConvertMetaRequest {
107        #[snafu(implicit)]
108        location: Location,
109        source: common_meta::error::Error,
110    },
111
112    #[snafu(display("Failed to convert Metasrv response"))]
113    ConvertMetaResponse {
114        #[snafu(implicit)]
115        location: Location,
116        source: common_meta::error::Error,
117    },
118
119    #[snafu(display("Failed to get flow stat"))]
120    GetFlowStat {
121        #[snafu(implicit)]
122        location: Location,
123        source: common_meta::error::Error,
124    },
125
126    #[snafu(display("Retry exceeded max times({}), message: {}", times, msg))]
127    RetryTimesExceeded { times: usize, msg: String },
128
129    #[snafu(display("Trying to write to a read-only kv backend: {}", name))]
130    ReadOnlyKvBackend {
131        name: String,
132        #[snafu(implicit)]
133        location: Location,
134    },
135
136    #[snafu(display("Failed to convert meta config"))]
137    ConvertMetaConfig {
138        #[snafu(implicit)]
139        location: Location,
140        #[snafu(source)]
141        error: serde_json::Error,
142    },
143}
144
145#[allow(dead_code)]
146pub type Result<T> = std::result::Result<T, Error>;
147
148impl ErrorExt for Error {
149    fn as_any(&self) -> &dyn std::any::Any {
150        self
151    }
152
153    fn status_code(&self) -> StatusCode {
154        match self {
155            Error::IllegalGrpcClientState { .. }
156            | Error::NoLeader { .. }
157            | Error::AskLeaderTimeout { .. }
158            | Error::NotStarted { .. }
159            | Error::MissingQueryContext { .. }
160            | Error::SendHeartbeat { .. }
161            | Error::CreateHeartbeatStream { .. }
162            | Error::HeartbeatHandshakeTimeout { .. }
163            | Error::CreateChannel { .. }
164            | Error::RetryTimesExceeded { .. }
165            | Error::ConvertMetaConfig { .. } => StatusCode::Internal,
166
167            Error::ReadOnlyKvBackend { .. } => StatusCode::Unsupported,
168
169            Error::MetaServer { code, .. } => *code,
170
171            Error::InvalidResponseHeader { source, .. }
172            | Error::ConvertMetaRequest { source, .. }
173            | Error::ConvertMetaResponse { source, .. }
174            | Error::GetFlowStat { source, .. } => source.status_code(),
175        }
176    }
177
178    fn retry_hint(&self) -> RetryHint {
179        match self {
180            Error::MetaServer { retry_hint, .. } => *retry_hint,
181            Error::HeartbeatHandshakeTimeout { .. } => RetryHint::Retryable,
182            Error::InvalidResponseHeader { source, .. }
183            | Error::ConvertMetaRequest { source, .. }
184            | Error::ConvertMetaResponse { source, .. }
185            | Error::GetFlowStat { source, .. } => source.retry_hint(),
186            Error::CreateChannel { source, .. } => source.retry_hint(),
187            _ => RetryHint::NonRetryable,
188        }
189    }
190}
191
192impl Error {
193    pub fn is_exceeded_size_limit(&self) -> bool {
194        matches!(
195            self,
196            Error::MetaServer {
197                tonic_code: tonic::Code::OutOfRange | tonic::Code::ResourceExhausted,
198                ..
199            }
200        )
201    }
202}
203
204define_from_tonic_status!(Error, MetaServer);
205
206#[cfg(test)]
207mod tests {
208    use common_error::ext::{ErrorExt, RetryHint};
209    use common_error::{GREPTIME_DB_HEADER_ERROR_CODE, GREPTIME_DB_HEADER_ERROR_RETRY_HINT};
210    use tonic::codegen::http::{HeaderMap, HeaderValue};
211    use tonic::metadata::MetadataMap;
212
213    use super::*;
214
215    #[test]
216    fn test_from_tonic_status_fallbacks_to_status_code() {
217        let status = tonic::Status::new(tonic::Code::Internal, "blabla");
218
219        let err: Error = status.into();
220
221        assert_eq!(err.retry_hint(), RetryHint::NonRetryable);
222    }
223
224    #[test]
225    fn test_from_tonic_status_fallback_can_be_non_retryable() {
226        let mut headers = HeaderMap::new();
227        headers.insert(
228            GREPTIME_DB_HEADER_ERROR_CODE,
229            HeaderValue::from(StatusCode::InvalidArguments as u32),
230        );
231        let status = tonic::Status::with_metadata(
232            tonic::Code::Internal,
233            "blabla",
234            MetadataMap::from_headers(headers),
235        );
236
237        let err: Error = status.into();
238
239        assert_eq!(err.retry_hint(), RetryHint::NonRetryable);
240    }
241
242    #[test]
243    fn test_from_tonic_status_with_retry_hint() {
244        let mut headers = HeaderMap::new();
245        headers.insert(
246            GREPTIME_DB_HEADER_ERROR_CODE,
247            HeaderValue::from(StatusCode::Internal as u32),
248        );
249        headers.insert(
250            GREPTIME_DB_HEADER_ERROR_RETRY_HINT,
251            HeaderValue::from_static(RetryHint::Retryable.as_str()),
252        );
253        let status = tonic::Status::with_metadata(
254            tonic::Code::Internal,
255            "blabla",
256            MetadataMap::from_headers(headers),
257        );
258
259        let err: Error = status.into();
260
261        assert_eq!(err.retry_hint(), RetryHint::Retryable);
262    }
263
264    #[test]
265    fn test_is_exceeded_size_limit_for_out_of_range() {
266        let err = Error::from(tonic::Status::new(tonic::Code::OutOfRange, "any message"));
267
268        assert!(err.is_exceeded_size_limit());
269    }
270
271    #[test]
272    fn test_is_exceeded_size_limit_for_resource_exhausted() {
273        let err = Error::from(tonic::Status::new(
274            tonic::Code::ResourceExhausted,
275            "arbitrary message",
276        ));
277
278        assert!(err.is_exceeded_size_limit());
279    }
280
281    #[test]
282    fn test_is_exceeded_size_limit_for_non_size_code() {
283        let err = Error::from(tonic::Status::new(tonic::Code::Internal, "message"));
284
285        assert!(!err.is_exceeded_size_limit());
286    }
287}
meta_client/error.rs

meta_client/
error.rs
