feat: Meta-service cluster is gated by Enterprise Edition

By default meta-service disallows clustering.
Meta-service cluster is only enabled when raft-config
`databend_enterprise_license` is set and is valid.
No feature in the jwt claim is examined.

The EE gate check when a meta node initiate raft-protocol network
instance. Thus without a valid EE token, all raft-protocol are disabled,
including `RequestVote`, `AppendEntries`, `InstallSnapshot` and internal
request forward.

If EE token is not set, an error will be outputed to log file.

- New config `databend_enterprise_license`:
  ```
  [raft_config]
  databend_enterprise_license = "<token>"
  ```

  This token is same as the one used by databend-query.

- When testing, a temp key pair and jwt claim is created to pass
  integration tests, this is enabled by `fake_ee_license` config entry.

- Other changes: Add `DisplaySlice` and `DisplayUnixTimeStampExt` to
  display slice of `Display` instance and unix timestamp.

Author: drmingdrmer

Cargo.lock

@@ -0,0 +1,95 @@
+// Copyright 2021 Datafuse Labs
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt;
+
+/// Implement `Display` for `&[T]` if T is `Display`.
+///
+/// It outputs at most `MAX` elements, excluding those from the 5th to the second-to-last one:
+/// - `DisplaySlice(&[1,2,3,4,5,6])` outputs: `"[1,2,3,4,...,6]"`.
+pub struct DisplaySlice<'a, T: fmt::Display, const MAX: usize = 5>(pub &'a [T]);
+
+impl<'a, T: fmt::Display, const MAX: usize> fmt::Display for DisplaySlice<'a, T, MAX> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let slice = self.0;
+        let len = slice.len();
+
+        write!(f, "[")?;
+
+        if len > MAX {
+            for (i, t) in slice[..(MAX - 1)].iter().enumerate() {
+                if i > 0 {
+                    write!(f, ",")?;
+                }
+
+                write!(f, "{}", t)?;
+            }
+
+            write!(f, ",..,")?;
+            write!(f, "{}", slice.last().unwrap())?;
+        } else {
+            for (i, t) in slice.iter().enumerate() {
+                if i > 0 {
+                    write!(f, ",")?;
+                }
+
+                write!(f, "{}", t)?;
+            }
+        }
+
+        write!(f, "]")
+    }
+}
+
+pub trait DisplaySliceExt<'a, T: fmt::Display> {
+    fn display(&'a self) -> DisplaySlice<'a, T>;
+
+    /// Display at most `MAX` elements.
+    fn display_n<const MAX: usize>(&'a self) -> DisplaySlice<'a, T, MAX>;
+}
+
+impl<T> DisplaySliceExt<'_, T> for [T]
+where T: fmt::Display
+{
+    fn display(&self) -> DisplaySlice<T> {
+        DisplaySlice(self)
+    }
+
+    fn display_n<const MAX: usize>(&'_ self) -> DisplaySlice<'_, T, MAX> {
+        DisplaySlice(self)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::DisplaySlice;
+
+    #[test]
+    fn test_display_slice() {
+        let a = vec![1, 2, 3, 4];
+        assert_eq!("[1,2,3,4]", DisplaySlice::<_>(&a).to_string());
+
+        let a = vec![1, 2, 3, 4, 5];
+        assert_eq!("[1,2,3,4,5]", DisplaySlice::<_>(&a).to_string());
+
+        let a = vec![1, 2, 3, 4, 5, 6];
+        assert_eq!("[1,2,3,4,..,6]", DisplaySlice::<_>(&a).to_string());
+
+        let a = vec![1, 2, 3, 4, 5, 6, 7];
+        assert_eq!("[1,2,3,4,..,7]", DisplaySlice::<_>(&a).to_string());
+
+        let a = vec![1, 2, 3, 4, 5, 6, 7];
+        assert_eq!("[1,..,7]", DisplaySlice::<_, 2>(&a).to_string());
+    }
+}

src/common/base/src/display/display_slice.rs

@@ -13,4 +13,5 @@
 // limitations under the License.
 
 pub mod display_option;
+pub mod display_slice;
 pub mod display_unix_epoch;

src/common/base/src/display/mod.rs

@@ -12,6 +12,11 @@ doctest = false
 test = true
 
 [dependencies]
+anyhow = { workspace = true }
+databend-common-base = { workspace = true }
+databend-common-license = { workspace = true }
+jwt-simple = { workspace = true }
+log = { workspace = true }
 
 [build-dependencies]
 databend-common-building = { workspace = true }

src/meta/ee/Cargo.toml

@@ -11,3 +11,111 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
+
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::time::Duration;
+
+use databend_common_base::display::display_unix_epoch::DisplayUnixTimeStampExt;
+use databend_common_license::license::LicenseInfo;
+use jwt_simple::algorithms::ECDSAP256KeyPairLike;
+use jwt_simple::algorithms::ECDSAP256PublicKeyLike;
+use jwt_simple::algorithms::ES256KeyPair;
+use jwt_simple::algorithms::ES256PublicKey;
+use jwt_simple::claims::Claims;
+use jwt_simple::claims::JWTClaims;
+use jwt_simple::prelude::Clock;
+
+#[derive(Clone, Default)]
+pub struct MetaServiceEnterpriseGate {
+    /// License ES256 signed jwt claim token.
+    license_token: Arc<Mutex<Option<String>>>,
+
+    /// The public key to verify the license token.
+    public_key: String,
+}
+
+impl MetaServiceEnterpriseGate {
+    const LICENSE_PUBLIC_KEY: &'static str = r#"-----BEGIN PUBLIC KEY-----
+MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEGsKCbhXU7j56VKZ7piDlLXGhud0a
+pWjW3wxSdeARerxs/BeoWK7FspDtfLaAT8iJe4YEmR0JpkRQ8foWs0ve3w==
+-----END PUBLIC KEY-----"#;
+
+    pub fn new(token: Option<String>) -> Self {
+        Self {
+            license_token: Arc::new(Mutex::new(token)),
+            public_key: Self::LICENSE_PUBLIC_KEY.to_string(),
+        }
+    }
+
+    /// Create with a temp license for testing.
+    pub fn new_testing() -> Self {
+        let lic = LicenseInfo {
+            r#type: Some("trial".to_string()),
+            org: Some("databend".to_string()),
+            tenants: Some(vec!["test".to_string()]),
+            features: None,
+        };
+
+        let key_pair = ES256KeyPair::generate();
+        let claims = Claims::with_custom_claims(lic, jwt_simple::prelude::Duration::from_hours(2));
+        let token = key_pair.sign(claims).unwrap();
+
+        let public_key = key_pair.public_key().to_pem().unwrap();
+
+        Self {
+            license_token: Arc::new(Mutex::new(Some(token))),
+            public_key,
+        }
+    }
+
+    /// Parse the JWT token and restore the claims.
+    fn parse_jwt_token(&self, raw: &str) -> Result<JWTClaims<LicenseInfo>, anyhow::Error> {
+        let public_key = ES256PublicKey::from_pem(&self.public_key)?;
+
+        let claim = public_key.verify_token::<LicenseInfo>(raw, None)?;
+
+        Ok(claim)
+    }
+
+    fn check_license(&self, raw: &str) -> Result<(), anyhow::Error> {
+        let claim = self.parse_jwt_token(raw)?;
+
+        let now = Clock::now_since_epoch();
+
+        if Some(now) > claim.expires_at {
+            let expires_at = claim.expires_at.unwrap_or_default();
+            let unix_timestamp = Duration::from_millis(expires_at.as_millis());
+
+            return Err(anyhow::anyhow!(format!(
+                "License is expired at: {}",
+                unix_timestamp.display_unix_timestamp()
+            )));
+        }
+
+        Ok(())
+    }
+
+    pub fn assert_cluster_enabled(&self) -> Result<(), anyhow::Error> {
+        let token = {
+            let x = self.license_token.lock().unwrap();
+            x.clone()
+        };
+
+        let Some(token) = token.as_ref() else {
+            log::error!(
+                "No license set in config `databend_enterprise_license`, clustering is disabled",
+            );
+            return Err(anyhow::anyhow!(
+                "No license set in config `databend_enterprise_license`, clustering is disabled"
+            ));
+        };
+
+        if let Err(e) = self.check_license(token) {
+            log::error!("Check license failed: {}", e);
+            return Err(e);
+        }
+
+        Ok(())
+    }
+}

src/meta/ee/src/lib.rs

@@ -131,6 +131,12 @@ pub struct RaftConfig {
 
     /// Max timeout(in milli seconds) when waiting a cluster leader.
     pub wait_leader_timeout: u64,
+
+    /// License token in string to enable enterprise features(including: `cluster`)
+    pub databend_enterprise_license: Option<String>,
+
+    /// For test only: whether to fake an enterprise license.
+    pub fake_ee_license: bool,
 }
 
 pub fn get_default_raft_advertise_host() -> String {
@@ -172,6 +178,8 @@ impl Default for RaftConfig {
             sled_max_cache_size_mb: 10 * 1024,
             cluster_name: "foo_cluster".to_string(),
             wait_leader_timeout: 70000,
+            databend_enterprise_license: None,
+            fake_ee_license: false,
         }
     }
 }

src/meta/raft-store/src/config.rs

@@ -26,11 +26,13 @@ anyerror = { workspace = true }
 anyhow = { workspace = true }
 async-trait = { workspace = true }
 backon = "0.4"
+chrono = { workspace = true }
 clap = { workspace = true }
 databend-common-arrow = { workspace = true }
 databend-common-base = { workspace = true }
 databend-common-grpc = { workspace = true }
 databend-common-http = { workspace = true }
+databend-common-license = { workspace = true }
 databend-common-meta-api = { workspace = true }
 databend-common-meta-client = { workspace = true }
 databend-common-meta-kvapi = { workspace = true }
@@ -40,6 +42,7 @@ databend-common-meta-stoerr = { workspace = true }
 databend-common-meta-types = { workspace = true }
 databend-common-metrics = { workspace = true }
 databend-common-tracing = { workspace = true }
+databend-enterprise-meta = { workspace = true }
 deepsize = { workspace = true }
 derive_more = { workspace = true }
 fastrace = { workspace = true }
@@ -48,6 +51,7 @@ futures = { workspace = true }
 futures-async-stream = { workspace = true }
 http = { workspace = true }
 itertools = { workspace = true }
+jwt-simple = { workspace = true }
 log = { workspace = true }
 logcall = { workspace = true }
 maplit = "1.0.2"

src/meta/service/Cargo.toml

@@ -303,6 +303,7 @@ pub struct ConfigViaEnv {
     pub sled_tree_prefix: String,
     pub sled_max_cache_size_mb: u64,
     pub cluster_name: String,
+    pub databend_enterprise_license: Option<String>,
 }
 
 impl Default for ConfigViaEnv {
@@ -356,6 +357,7 @@ impl From<Config> for ConfigViaEnv {
             sled_tree_prefix: cfg.raft_config.sled_tree_prefix,
             sled_max_cache_size_mb: cfg.raft_config.sled_max_cache_size_mb,
             cluster_name: cfg.raft_config.cluster_name,
+            databend_enterprise_license: cfg.raft_config.databend_enterprise_license,
         }
     }
 }
@@ -393,6 +395,7 @@ impl Into<Config> for ConfigViaEnv {
             sled_tree_prefix: self.sled_tree_prefix,
             sled_max_cache_size_mb: self.sled_max_cache_size_mb,
             cluster_name: self.cluster_name,
+            databend_enterprise_license: self.databend_enterprise_license,
         };
         let log_config = LogConfig {
             file: FileLogConfig {
@@ -564,6 +567,10 @@ pub struct RaftConfig {
     /// Max timeout(in milli seconds) when waiting a cluster leader.
     #[clap(long, default_value = "180000")]
     pub wait_leader_timeout: u64,
+
+    /// License token in string to enable enterprise features(including: `cluster`)
+    #[clap(long, value_name = "VALUE")]
+    pub databend_enterprise_license: Option<String>,
 }
 
 // TODO(rotbl): should not be used.
@@ -602,6 +609,8 @@ impl From<RaftConfig> for InnerRaftConfig {
             sled_max_cache_size_mb: x.sled_max_cache_size_mb,
             cluster_name: x.cluster_name,
             wait_leader_timeout: x.wait_leader_timeout,
+            databend_enterprise_license: x.databend_enterprise_license,
+            fake_ee_license: false,
         }
     }
 }
@@ -635,6 +644,7 @@ impl From<InnerRaftConfig> for RaftConfig {
             sled_max_cache_size_mb: inner.sled_max_cache_size_mb,
             cluster_name: inner.cluster_name,
             wait_leader_timeout: inner.wait_leader_timeout,
+            databend_enterprise_license: inner.databend_enterprise_license,
         }
     }
 }