feat: Add API to update license to meta-service in flight

Update EE license token with the following API:

```
curl -qs '127.0.0.1:28101/v1/ctrl/update_license?license=<license_token>'
```

On success, i.e., the token is valid and not expired, meta-service
respond with the 200 OK with token info, such as:

```
{"Success":"JWTClaims{issuer: databend, issued_at: 2024-05-13T05:57:24.000000Z+0000, expires_at: 2025-05-13T05:57:24.000000Z+0000, custom: LicenseInfo{ type: enterprise, org: databend-interval-test, tenants: None, features: [Unlimited] }}"}
```

If the token is invalid, or expired, it responds with 400 Bad Request
and a reason that cause the failure, such as:

```
Invalid license: JWT compact encoding error
```

src/meta/ee/src/lib.rs

@@ -70,7 +70,7 @@ pWjW3wxSdeARerxs/BeoWK7FspDtfLaAT8iJe4YEmR0JpkRQ8foWs0ve3w==
     }
 
     /// Parse the JWT token and restore the claims.
-    fn parse_jwt_token(&self, raw: &str) -> Result<JWTClaims<LicenseInfo>, anyhow::Error> {
+    pub 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)?;
@@ -118,4 +118,13 @@ pWjW3wxSdeARerxs/BeoWK7FspDtfLaAT8iJe4YEmR0JpkRQ8foWs0ve3w==
 
         Ok(())
     }
+
+    pub fn update_license(&self, license: String) -> Result<(), anyhow::Error> {
+        self.check_license(&license)?;
+
+        let mut x = self.license_token.lock().unwrap();
+        *x = Some(license);
+
+        Ok(())
+    }
 }

src/meta/service/Cargo.toml

@@ -31,6 +31,7 @@ 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 }

src/meta/service/src/api/http/v1/ctrl.rs

@@ -12,8 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::BTreeMap;
 use std::sync::Arc;
 
+use databend_common_license::display_jwt_claims::DisplayJWTClaimsExt;
 use databend_common_meta_sled_store::openraft::async_runtime::watch::WatchReceiver;
 use databend_common_meta_types::NodeId;
 use http::StatusCode;
@@ -127,3 +129,52 @@ pub async fn trigger_transfer_leader(
         voter_ids,
     }))
 }
+
+#[derive(serde::Serialize, serde::Deserialize, Debug)]
+pub struct UpdateLicenseQuery {
+    pub(crate) license: String,
+}
+
+#[derive(serde::Serialize, serde::Deserialize, Debug)]
+pub struct UpdateLicenseResponse {
+    pub from: NodeId,
+    pub to: NodeId,
+    pub voter_ids: Vec<NodeId>,
+}
+
+/// Update the `databend_enterprise_license` of this meta node.
+#[poem::handler]
+pub async fn update_license(
+    meta_node: Data<&Arc<MetaNode>>,
+    query: Option<Query<UpdateLicenseQuery>>,
+) -> poem::Result<impl IntoResponse> {
+    let Some(query) = query else {
+        return Err(poem::Error::from_string(
+            "Invalid license",
+            StatusCode::BAD_REQUEST,
+        ));
+    };
+
+    let metrics = meta_node.raft.metrics().borrow_watched().clone();
+    let id = metrics.id;
+
+    let saved = meta_node
+        .ee_gate
+        .parse_jwt_token(query.license.as_str())
+        .map_err(|e| {
+            poem::Error::from_string(format!("Invalid license: {}", e), StatusCode::BAD_REQUEST)
+        })?;
+
+    meta_node
+        .ee_gate
+        .update_license(query.license.clone())
+        .map_err(|e| poem::Error::from_string(e.to_string(), StatusCode::BAD_REQUEST))?;
+
+    let claim_str = saved.display_jwt_claims().to_string();
+    info!("id={} Updated license: {}", id, claim_str);
+
+    let mut resp = BTreeMap::new();
+    resp.insert("Success", claim_str);
+
+    Ok(Json(resp))
+}

src/meta/service/src/api/http_service.rs

@@ -65,6 +65,10 @@ impl HttpService {
                 "/v1/ctrl/trigger_transfer_leader",
                 get(super::http::v1::ctrl::trigger_transfer_leader),
             )
+            .at(
+                "/v1/ctrl/update_license",
+                get(super::http::v1::ctrl::update_license),
+            )
             .at(
                 "/v1/cluster/nodes",
                 get(super::http::v1::cluster_state::nodes_handler),

src/meta/service/src/meta_service/meta_node.rs

@@ -108,6 +108,7 @@ pub struct MetaNode {
     pub sto: RaftStore,
     pub dispatcher_handle: EventDispatcherHandle,
     pub raft: MetaRaft,
+    pub(crate) ee_gate: MetaServiceEnterpriseGate,
     pub running_tx: watch::Sender<()>,
     pub running_rx: watch::Receiver<()>,
     pub join_handles: Mutex<Vec<JoinHandle<Result<(), AnyError>>>>,
@@ -146,7 +147,7 @@ impl MetaNodeBuilder {
 
         let ee_gate = self.ee_gate.clone();
 
-        let net = NetworkFactory::new(sto.clone(), ee_gate);
+        let net = NetworkFactory::new(sto.clone(), ee_gate.clone());
 
         let log_store = sto.clone();
         let sm_store = sto.clone();
@@ -167,6 +168,7 @@ impl MetaNodeBuilder {
             sto: sto.clone(),
             dispatcher_handle: EventDispatcherHandle::new(dispatcher_tx),
             raft: raft.clone(),
+            ee_gate,
             running_tx: tx,
             running_rx: rx,
             join_handles: Mutex::new(Vec::new()),