meta_srv/event/
region_migration.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::any::Any;
16use std::time::Duration;
17
18use api::v1::value::ValueData;
19use api::v1::{ColumnSchema, Row};
20use common_event_recorder::error::{Result, SerializeEventSnafu};
21use common_event_recorder::event_table::{
22    REGION_ID_COLUMN, REGION_MIGRATION_DST_NODE_ID_COLUMN, REGION_MIGRATION_DST_PEER_ADDR_COLUMN,
23    REGION_MIGRATION_SRC_NODE_ID_COLUMN, REGION_MIGRATION_SRC_PEER_ADDR_COLUMN,
24    REGION_MIGRATION_TRIGGER_REASON_COLUMN, REGION_NUMBER_COLUMN, TABLE_ID_COLUMN, column_schemas,
25};
26use common_event_recorder::{Event, TriggerReason};
27use serde::Serialize;
28use snafu::ResultExt;
29use store_api::storage::RegionId;
30
31use crate::procedure::region_migration::{PersistentContext, RegionMigrationTriggerReason};
32
33pub const REGION_MIGRATION_EVENT_TYPE: &str = "region_migration";
34
35/// RegionMigrationEvent is the event of region migration.
36#[derive(Debug)]
37pub(crate) struct RegionMigrationEvent {
38    // The region ids of the region migration.
39    region_ids: Vec<RegionId>,
40    // The trigger reason of the region migration.
41    trigger_reason: RegionMigrationTriggerReason,
42    // The source node id of the region migration.
43    src_node_id: u64,
44    // The source peer address of the region migration.
45    src_peer_addr: String,
46    // The destination node id of the region migration.
47    dst_node_id: u64,
48    // The destination peer address of the region migration.
49    dst_peer_addr: String,
50    // The timeout of the region migration.
51    timeout: Duration,
52}
53
54#[derive(Debug, Serialize)]
55struct Payload {
56    #[serde(with = "humantime_serde")]
57    timeout: Duration,
58}
59
60impl RegionMigrationEvent {
61    pub fn from_persistent_ctx(ctx: &PersistentContext, trigger_reason: TriggerReason) -> Self {
62        Self {
63            region_ids: ctx.region_ids.clone(),
64            trigger_reason: RegionMigrationTriggerReason::from_trigger_reason(trigger_reason),
65            src_node_id: ctx.from_peer.id,
66            src_peer_addr: ctx.from_peer.addr.clone(),
67            dst_node_id: ctx.to_peer.id,
68            dst_peer_addr: ctx.to_peer.addr.clone(),
69            timeout: ctx.timeout,
70        }
71    }
72}
73
74impl Event for RegionMigrationEvent {
75    fn event_type(&self) -> &str {
76        REGION_MIGRATION_EVENT_TYPE
77    }
78
79    fn extra_schema(&self) -> Vec<ColumnSchema> {
80        column_schemas([
81            &REGION_ID_COLUMN,
82            &TABLE_ID_COLUMN,
83            &REGION_NUMBER_COLUMN,
84            &REGION_MIGRATION_TRIGGER_REASON_COLUMN,
85            &REGION_MIGRATION_SRC_NODE_ID_COLUMN,
86            &REGION_MIGRATION_SRC_PEER_ADDR_COLUMN,
87            &REGION_MIGRATION_DST_NODE_ID_COLUMN,
88            &REGION_MIGRATION_DST_PEER_ADDR_COLUMN,
89        ])
90    }
91
92    fn extra_rows(&self) -> Result<Vec<Row>> {
93        let mut extra_rows = Vec::with_capacity(self.region_ids.len());
94        for region_id in &self.region_ids {
95            extra_rows.push(Row {
96                values: vec![
97                    ValueData::U64Value(region_id.as_u64()).into(),
98                    ValueData::U32Value(region_id.table_id()).into(),
99                    ValueData::U32Value(region_id.region_number()).into(),
100                    ValueData::StringValue(self.trigger_reason.to_string()).into(),
101                    ValueData::U64Value(self.src_node_id).into(),
102                    ValueData::StringValue(self.src_peer_addr.clone()).into(),
103                    ValueData::U64Value(self.dst_node_id).into(),
104                    ValueData::StringValue(self.dst_peer_addr.clone()).into(),
105                ],
106            });
107        }
108
109        Ok(extra_rows)
110    }
111
112    fn json_payload(&self) -> Result<serde_json::Value> {
113        serde_json::to_value(Payload {
114            timeout: self.timeout,
115        })
116        .context(SerializeEventSnafu)
117    }
118
119    fn as_any(&self) -> &dyn Any {
120        self
121    }
122}
meta_srv/event/region_migration.rs

meta_srv/event/
region_migration.rs
