feat: Add TLS support for gRPC service (#3957)

* feat: Add tls support for grpc service

* feat: add integration test

* fix: integration test

* fix: revert by suggestion

* fix: typos

* fix: optimize code

* fix: optimize code

* docs: update configs

config/config.md

@@ -20,6 +20,11 @@
 | `grpc` | -- | -- | The gRPC server options. |
 | `grpc.addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
+| `grpc.tls` | -- | -- | gRPC server TLS options, see `mysql.tls` section. |
+| `grpc.tls.mode` | String | `disable` | TLS mode. |
+| `grpc.tls.cert_path` | String | `None` | Certificate file path. |
+| `grpc.tls.key_path` | String | `None` | Private key file path. |
+| `grpc.tls.watch` | Bool | `false` | Watch for Certificate and key file change and auto reload.<br/>For now, gRPC tls config does not support auto reload. |
 | `mysql` | -- | -- | MySQL server options. |
 | `mysql.enable` | Bool | `true` | Whether to enable. |
 | `mysql.addr` | String | `127.0.0.1:4002` | The addr to bind the MySQL server. |
@@ -33,7 +38,7 @@
 | `postgres.enable` | Bool | `true` | Whether to enable |
 | `postgres.addr` | String | `127.0.0.1:4003` | The addr to bind the PostgresSQL server. |
 | `postgres.runtime_size` | Integer | `2` | The number of server worker threads. |
-| `postgres.tls` | -- | -- | PostgresSQL server TLS options, see `mysql_options.tls` section. |
+| `postgres.tls` | -- | -- | PostgresSQL server TLS options, see `mysql.tls` section. |
 | `postgres.tls.mode` | String | `disable` | TLS mode. |
 | `postgres.tls.cert_path` | String | `None` | Certificate file path. |
 | `postgres.tls.key_path` | String | `None` | Private key file path. |
@@ -159,6 +164,11 @@
 | `grpc` | -- | -- | The gRPC server options. |
 | `grpc.addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
+| `grpc.tls` | -- | -- | gRPC server TLS options, see `mysql.tls` section. |
+| `grpc.tls.mode` | String | `disable` | TLS mode. |
+| `grpc.tls.cert_path` | String | `None` | Certificate file path. |
+| `grpc.tls.key_path` | String | `None` | Private key file path. |
+| `grpc.tls.watch` | Bool | `false` | Watch for Certificate and key file change and auto reload.<br/>For now, gRPC tls config does not support auto reload. |
 | `mysql` | -- | -- | MySQL server options. |
 | `mysql.enable` | Bool | `true` | Whether to enable. |
 | `mysql.addr` | String | `127.0.0.1:4002` | The addr to bind the MySQL server. |
@@ -172,7 +182,7 @@
 | `postgres.enable` | Bool | `true` | Whether to enable |
 | `postgres.addr` | String | `127.0.0.1:4003` | The addr to bind the PostgresSQL server. |
 | `postgres.runtime_size` | Integer | `2` | The number of server worker threads. |
-| `postgres.tls` | -- | -- | PostgresSQL server TLS options, see `mysql_options.tls` section. |
+| `postgres.tls` | -- | -- | PostgresSQL server TLS options, see `mysql.tls` section. |
 | `postgres.tls.mode` | String | `disable` | TLS mode. |
 | `postgres.tls.cert_path` | String | `None` | Certificate file path. |
 | `postgres.tls.key_path` | String | `None` | Private key file path. |

config/frontend.example.toml

@@ -30,6 +30,23 @@ addr = "127.0.0.1:4001"
 ## The number of server worker threads.
 runtime_size = 8
 
+## gRPC server TLS options, see `mysql.tls` section.
+[grpc.tls]
+## TLS mode.
+mode = "disable"
+
+## Certificate file path.
+## +toml2docs:none-default
+cert_path = ""
+
+## Private key file path.
+## +toml2docs:none-default
+key_path = ""
+
+## Watch for Certificate and key file change and auto reload.
+## For now, gRPC tls config does not support auto reload.
+watch = false
+
 ## MySQL server options.
 [mysql]
 ## Whether to enable.
@@ -70,7 +87,7 @@ addr = "127.0.0.1:4003"
 ## The number of server worker threads.
 runtime_size = 2
 
-## PostgresSQL server TLS options, see `mysql_options.tls` section.
+## PostgresSQL server TLS options, see `mysql.tls` section.
 [postgres.tls]
 ## TLS mode.
 mode = "disable"

config/standalone.example.toml

@@ -25,6 +25,23 @@ addr = "127.0.0.1:4001"
 ## The number of server worker threads.
 runtime_size = 8
 
+## gRPC server TLS options, see `mysql.tls` section.
+[grpc.tls]
+## TLS mode.
+mode = "disable"
+
+## Certificate file path.
+## +toml2docs:none-default
+cert_path = ""
+
+## Private key file path.
+## +toml2docs:none-default
+key_path = ""
+
+## Watch for Certificate and key file change and auto reload.
+## For now, gRPC tls config does not support auto reload.
+watch = false
+
 ## MySQL server options.
 [mysql]
 ## Whether to enable.
@@ -65,7 +82,7 @@ addr = "127.0.0.1:4003"
 ## The number of server worker threads.
 runtime_size = 2
 
-## PostgresSQL server TLS options, see `mysql_options.tls` section.
+## PostgresSQL server TLS options, see `mysql.tls` section.
 [postgres.tls]
 ## TLS mode.
 mode = "disable"

src/client/src/client.rs

@@ -19,7 +19,7 @@ use api::v1::prometheus_gateway_client::PrometheusGatewayClient;
 use api::v1::region::region_client::RegionClient as PbRegionClient;
 use api::v1::HealthCheckRequest;
 use arrow_flight::flight_service_client::FlightServiceClient;
-use common_grpc::channel_manager::ChannelManager;
+use common_grpc::channel_manager::{ChannelConfig, ChannelManager, ClientTlsOption};
 use parking_lot::RwLock;
 use snafu::{OptionExt, ResultExt};
 use tonic::codec::CompressionEncoding;
@@ -87,6 +87,17 @@ impl Client {
         Self::with_manager_and_urls(ChannelManager::new(), urls)
     }
 
+    pub fn with_tls_and_urls<U, A>(urls: A, client_tls: ClientTlsOption) -> Result<Self>
+    where
+        U: AsRef<str>,
+        A: AsRef<[U]>,
+    {
+        let channel_config = ChannelConfig::default().client_tls_config(client_tls);
+        let channel_manager = ChannelManager::with_tls_config(channel_config)
+            .context(error::CreateTlsChannelSnafu)?;
+        Ok(Self::with_manager_and_urls(channel_manager, urls))
+    }
+
     pub fn with_manager_and_urls<U, A>(channel_manager: ChannelManager, urls: A) -> Self
     where
         U: AsRef<str>,

src/client/src/error.rs

@@ -82,6 +82,13 @@ pub enum Error {
         source: common_grpc::error::Error,
     },
 
+    #[snafu(display("Failed to create Tls channel manager"))]
+    CreateTlsChannel {
+        #[snafu(implicit)]
+        location: Location,
+        source: common_grpc::error::Error,
+    },
+
     #[snafu(display("Failed to request RegionServer, code: {}", code))]
     RegionServer {
         code: Code,
@@ -129,9 +136,9 @@ impl ErrorExt for Error {
             Error::FlightGet { source, .. }
             | Error::HandleRequest { source, .. }
             | Error::RegionServer { source, .. } => source.status_code(),
-            Error::CreateChannel { source, .. } | Error::ConvertFlightData { source, .. } => {
-                source.status_code()
-            }
+            Error::CreateChannel { source, .. }
+            | Error::ConvertFlightData { source, .. }
+            | Error::CreateTlsChannel { source, .. } => source.status_code(),
             Error::IllegalGrpcClientState { .. } => StatusCode::Unexpected,
         }
     }

src/cmd/src/frontend.rs

@@ -216,6 +216,7 @@ impl StartCommand {
 
         if let Some(addr) = &self.rpc_addr {
             opts.grpc.addr.clone_from(addr);
+            opts.grpc.tls = tls_opts.clone();
         }
 
         if let Some(addr) = &self.mysql_addr {

src/common/grpc/tests/mod.rs

@@ -23,9 +23,9 @@ async fn test_mtls_config() {
 
     // test wrong file
     let config = ChannelConfig::new().client_tls_config(ClientTlsOption {
-        server_ca_cert_path: "tests/tls/wrong_server.cert.pem".to_string(),
-        client_cert_path: "tests/tls/wrong_client.cert.pem".to_string(),
-        client_key_path: "tests/tls/wrong_client.key.pem".to_string(),
+        server_ca_cert_path: "tests/tls/wrong_ca.pem".to_string(),
+        client_cert_path: "tests/tls/wrong_client.pem".to_string(),
+        client_key_path: "tests/tls/wrong_client.key".to_string(),
     });
 
     let re = ChannelManager::with_tls_config(config);
@@ -33,8 +33,8 @@ async fn test_mtls_config() {
 
     // test corrupted file content
     let config = ChannelConfig::new().client_tls_config(ClientTlsOption {
-        server_ca_cert_path: "tests/tls/server.cert.pem".to_string(),
-        client_cert_path: "tests/tls/client.cert.pem".to_string(),
+        server_ca_cert_path: "tests/tls/ca.pem".to_string(),
+        client_cert_path: "tests/tls/client.pem".to_string(),
         client_key_path: "tests/tls/corrupted".to_string(),
     });
 
@@ -44,9 +44,9 @@ async fn test_mtls_config() {
 
     // success
     let config = ChannelConfig::new().client_tls_config(ClientTlsOption {
-        server_ca_cert_path: "tests/tls/server.cert.pem".to_string(),
-        client_cert_path: "tests/tls/client.cert.pem".to_string(),
-        client_key_path: "tests/tls/client.key.pem".to_string(),
+        server_ca_cert_path: "tests/tls/ca.pem".to_string(),
+        client_cert_path: "tests/tls/client.pem".to_string(),
+        client_key_path: "tests/tls/client.key".to_string(),
     });
 
     let re = ChannelManager::with_tls_config(config).unwrap();

src/common/grpc/tests/tls/ca.pem

@@ -0,0 +1,28 @@
+-----BEGIN CERTIFICATE-----
+MIIE3DCCA0SgAwIBAgIRAObeYbJFiVQSGR8yk44dsOYwDQYJKoZIhvcNAQELBQAw
+gYUxHjAcBgNVBAoTFW1rY2VydCBkZXZlbG9wbWVudCBDQTEtMCsGA1UECwwkbHVj
+aW9ATHVjaW9zLVdvcmstTUJQIChMdWNpbyBGcmFuY28pMTQwMgYDVQQDDCtta2Nl
+cnQgbHVjaW9ATHVjaW9zLVdvcmstTUJQIChMdWNpbyBGcmFuY28pMB4XDTE5MDky
+OTIzMzUzM1oXDTI5MDkyOTIzMzUzM1owgYUxHjAcBgNVBAoTFW1rY2VydCBkZXZl
+bG9wbWVudCBDQTEtMCsGA1UECwwkbHVjaW9ATHVjaW9zLVdvcmstTUJQIChMdWNp
+byBGcmFuY28pMTQwMgYDVQQDDCtta2NlcnQgbHVjaW9ATHVjaW9zLVdvcmstTUJQ
+IChMdWNpbyBGcmFuY28pMIIBojANBgkqhkiG9w0BAQEFAAOCAY8AMIIBigKCAYEA
+y/vE61ItbN/1qMYt13LMf+le1svwfkCCOPsygk7nWeRXmomgUpymqn1LnWiuB0+e
+4IdVH2f5E9DknWEpPhKIDMRTCbz4jTwQfHrxCb8EGj3I8oO73pJO5S/xCedM9OrZ
+qWcYWwN0GQ8cO/ogazaoZf1uTrRNHyzRyQsKyb412kDBTNEeldJZ2ljKgXXvh4HO
+2ZIk9K/ZAaAf6VN8K/89rlJ9/KPgRVNsyAapE+Pb8XXKtpzeFiEcUfuXVYWtkoW+
+xyn/Zu8A1L2CXMQ1sARh7P/42BTMKr5pfraYgcBGxKXLrxoySpxCO9KqeVveKy1q
+fPm5FCwFsXDr0koFLrCiR58mcIO/04Q9DKKTV4Z2a+LoqDJRY37KfBSc8sDMPhw5
+k7g3WPoa6QwXRjZTCA5fHWVgLOtcwLsnju5tBE4LDxwF6s+1wPF8NI5yUfufcEjJ
+Z6JBwgoWYosVj27Lx7KBNLU/57PX9ryee691zmtswt0tP0WVBAgalhYWg99RXoa3
+AgMBAAGjRTBDMA4GA1UdDwEB/wQEAwICBDASBgNVHRMBAf8ECDAGAQH/AgEAMB0G
+A1UdDgQWBBQdvlE4Bdcsjc9oaxjDCRu5FiuZkzANBgkqhkiG9w0BAQsFAAOCAYEA
+BP/6o1kPINksMJZSSXgNCPZskDLyGw7auUZBnQ0ocDT3W6gXQvT/27LM1Hxoj9Eh
+qU1TYdEt7ppecLQSGvzQ02MExG7H75art75oLiB+A5agDira937YbK4MCjqW481d
+bDhw6ixJnY1jIvwjEZxyH6g94YyL927aSPch51fys0kSnjkFzC2RmuzDADScc4XH
+5P1+/3dnIm3M5yfpeUzoaOrTXNmhn8p0RDIGrZ5kA5eISIGGD3Mm8FDssUNKndtO
+g4ojHUsxb14icnAYGeye1NOhGiqN6TEFcgr6MPd0XdFNZ5c0HUaBCfN6bc+JxDV5
+MKZVJdNeJsYYwilgJNHAyZgCi30JC20xeYVtTF7CEEsMrFDGJ70Kz7o/FnRiFsA1
+ZSwVVWhhkHG2VkT4vlo0O3fYeZpenYicvy+wZNTbGK83gzHWqxxNC1z3Etg5+HRJ
+F9qeMWPyfA3IHYXygiMcviyLcyNGG/SJ0EhUpYBN/Gg7wI5yFkcsxUDPPzd23O0M
+-----END CERTIFICATE-----

src/common/grpc/tests/tls/server.cert.pem

@@ -1,40 +0,0 @@
------BEGIN CERTIFICATE-----
-MIIG+jCCBOKgAwIBAgIBAjANBgkqhkiG9w0BAQsFADCBhzELMAkGA1UEBhMCSU4x
-EjAQBgNVBAgMCUthcm5hdGFrYTESMBAGA1UEBwwJQkFOR0FMT1JFMRUwEwYDVQQK
-DAxHb0xpbnV4Q2xvdWQxEjAQBgNVBAMMCWNhLXNlcnZlcjElMCMGCSqGSIb3DQEJ
-ARYWYWRtaW5AZ29saW51eGNsb3VkLmNvbTAeFw0yMzAyMTQxMTM5NDBaFw0yNzA4
-MjIxMTM5NDBaMHAxCzAJBgNVBAYTAklOMRIwEAYDVQQIDAlLYXJuYXRha2ExFTAT
-BgNVBAoMDEdvTGludXhDbG91ZDEPMA0GA1UEAwwGc2VydmVyMSUwIwYJKoZIhvcN
-AQkBFhZhZG1pbkBnb2xpbnV4Y2xvdWQuY29tMIICIjANBgkqhkiG9w0BAQEFAAOC
-Ag8AMIICCgKCAgEAvVtxAoRjLRs3Ei4+CgzqJ2+bpc0sBdUm/4LM/D+0KbXxwD7w
-HP6GcKl/9zf9GJg56pVXxXMaerMDLS4Est25+mBgqcePC6utCBYrKA25pKbkFkxZ
-TPh9/R4RHGVJ3KHy9vc4VzqoV7XFMJFFUQ2fQywHZlXh6MNz0WPTIGaH7hvYoHbK
-I3NpPq8TjRuuV61XB0hK+RW0K6/5Yuj74h/mfheX1VIUOjGwKnTPccZQAlrKYjeW
-BZBS4YqahkTIaGLa06SdUSkuhL85rqAxWvhK9GIRlQLNYJOzg+E3jGyqf566xX60
-fxM6alLYf+ZzCwSBuDDj5f+j752gPLYUI82YL4xQ+AEHNR8U1uMvt0EzzFt7mSRe
-fobVr+Y2zpci+mo7kcQGOhenzGclsm+qXwMhYUnJcOYFZWtTJlFaaPreL4M3Dh+2
-pmKj23ZU6zcT3MYtE6phjCLJl0DsFIcOn+tSqMdpwB20EeQjo9bVJuw/HJrlpcnY
-U9aLsnm/4Ls5A0BQutZnxKBIJjpzp8VfK0WU8a4iKok3AS0z1/K+atNrgSUB9DCH
-0MvLqqQmM9TdLcZj7NSEfLyyFVwPRc5dt4CrNDL7JUpMzt36ezU83JU+nfqWDZsL
-+2JOaE4gGLZDcA3cfP83/mYRaAnYW/9W4vEnIpa6subzq1aFOeY/3dKLTx8CAwEA
-AaOCAYUwggGBMAkGA1UdEwQCMAAwEQYJYIZIAYb4QgEBBAQDAgZAMDMGCWCGSAGG
-+EIBDQQmFiRPcGVuU1NMIEdlbmVyYXRlZCBTZXJ2ZXIgQ2VydGlmaWNhdGUwHQYD
-VR0OBBYEFLijeA+RFDQtuVeMUkaXqF7LF50GMIG8BgNVHSMEgbQwgbGAFKVZwpSJ
-CPkNwGXyJX1sl2Pbby4FoYGNpIGKMIGHMQswCQYDVQQGEwJJTjESMBAGA1UECAwJ
-S2FybmF0YWthMRIwEAYDVQQHDAlCQU5HQUxPUkUxFTATBgNVBAoMDEdvTGludXhD
-bG91ZDESMBAGA1UEAwwJY2Etc2VydmVyMSUwIwYJKoZIhvcNAQkBFhZhZG1pbkBn
-b2xpbnV4Y2xvdWQuY29tggkA7NvbvF8jodEwDgYDVR0PAQH/BAQDAgWgMBMGA1Ud
-JQQMMAoGCCsGAQUFBwMBMCkGA1UdEQQiMCCHBMCoAHKHBAoAAg+CEnNlcnZlci5l
-eGFtcGxlLmNvbTANBgkqhkiG9w0BAQsFAAOCAgEAXvaS9+y5g2Kw/4EPsnhjpN1v
-CxXW0+UYSWOaxVJdEAjGQI/1m9LOiF9IHImmiwluJ/Bex1TzuaTCKmpluPwGvd9D
-Zgf0A5SmVqW4WTT4d2nSecxw4OICJ3j6ubKkvMVf9s+ZJwb+fMMUaSt80bWqp1TY
-XbZguv67PkBECPqVe6rgzXnTLwM3lE8EgG8VtM3IOy9a5SIEjm5L8SQ2I2hiytmE
-e4jR1fbZsB5NbBdfA3GFMKQEE2dIymkG3Bz71M3tZi1y4RnHtRKdrFtrIlgclrwd
-nVnQn/NiXUOOzsL2+vwSF32SSbiLvOxu63qO1YDBkKVChog3P/2f6xcJ23wkbHlL
-qaL2jvLo6ylvMPUYHf5ZWat5zayaGUMHYDKcbD4Dw7aY3M0tNgEHdqUqNePmKvmn
-luyXof3KmmLgWlcfBoX96a7hXDtxFyB2N4nzfQBXh+0VAlgqa+ZZhpdEqRQaWkkR
-MDBdsVJ9O3812IaNfMzpS1vb701GFDCM5Hcyw6a/v6Ln08NMhYut4saLi13kHilS
-Wq7wOAfW3rzxuhjOJJxsi0jJNI775q+a/BbbG/CPl826bXPGH43BdPV8mKwsX5HM
-wwDKf3otP/v7bxwJabfhv2EKUy+W1kkFW9FEZ919yTtfhSDrTNcrXtE7RkiAepfm
-95I025URIlhJGLGBUlA=
------END CERTIFICATE-----

src/common/grpc/tests/tls/server.key

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDyptbMyYWztgta
+t1MXLMzIkaQdeeVbs1Y/qCpAdwZe/Y5ZpbzjGIjCxbB6vNRSnEbYKpytKHPzYfM7
+8d8K8bPvpnqXIiTXFT0JQlw1OHLC1fr4e598GJumAmpMYFrtqv0fbmUFTuQGbHxe
+OH2vji0bvr3NKZubMfkEZP3X4sNXXoXIuW2LaS8OMGKoJaeCBvdbszEiSGj/v9Bj
+pM0yLTH89NNMX1T+FtTKnuXag5g7pr6lzJj83+MzAGy4nOjseSuUimuiyG90/C5t
+A5wC0Qh5RbDnkFYhC44Kxof/i6+jnfateIPNiIIwQV+2f6G/aK1hgjekT10m/eoR
+YDTf+e5ZAgMBAAECggEACODt7yRYjhDVLYaTtb9f5t7dYG67Y7WWLFIc6arxQryI
+XuNfm/ej2WyeXn9WTYeGWBaHERbv1zH4UnMxNBdP/C7dQXZwXqZaS2JwOUpNeK+X
+tUvgtAu6dkKUXSMRcKzXAjVp4N3YHhwOGOx8PNY49FDwZPdmyDD16aFAYIvdle6/
+PSMrj38rB1sbQQdmRob2FjJBSDZ44nsr+/nilrcOFNfNnWv7tQIWYVXNcLfdK/WJ
+ZCDFhA8lr/Yon6MEq6ApTj2ZYRRGXPd6UeASJkmTZEUIUbeDcje/MO8cHkREpuRH
+wm3pCjR7OdO4vc+/d/QmEvu5ns6wbTauelYnL616YQKBgQD414gJtpCHauNEUlFB
+v/R3DzPI5NGp9PAqovOD8nCbI49Mw61gP/ExTIPKiR5uUX/5EL04uspaNkuohXk+
+ys0G5At0NfV7W39lzhvALEaSfleybvYxppbBrc20/q8Gvi/i30NY+1LM3RdtMiEw
+hKHjU0SnFhJq0InFg3AO/iCeTQKBgQD5obkbzpOidSsa55aNsUlO2qjiUY9leq9b
+irAohIZ8YnuuixYvkOeSeSz1eIrA4tECeAFSgTZxYe1Iz+USru2Xg/0xNte11dJD
+rBoH/yMn2gDvBK7xQ6uFMPTeYtKG0vfvpXZYSWZzGntyrHTwFk6UV+xdrt9MBdd1
+XdSn7bwOPQKBgC9VQAko8uDvUf+C8PXiv2uONrl13PPJJY3WpR9qFEVOREnDxszS
+HNzVwxPZdTJiykbkCjoqPadfQJDzopZxGQLAifU29lTamKcSx3CMe3gOFDxaovXa
+zD5XAxP0hfJwZsdu1G6uj5dsTrJ0oJ+L+wc0pZBqwGIU/L/XOo9/g1DZAoGAUebL
+kuH98ik7EUK2VJq8EJERI9/ailLsQb6I+WIxtZGiPqwHhWencpkrNQZtj8dbB9JT
+rLwUHrMgZOlAoRafgTyez4zMzS3wJJ/Mkp8U67hM4h7JPwMSvUpIrMYDiJSjIA9L
+er/qSw1/Pypx22uWMHmAZWRAgvLPtAQrB0Wqk4kCgYEAr2H1PvfbwZwkSvlMt5o8
+WLnBbxcM3AKglLRbkShxxgiZYdEP71/uOtRMiL26du5XX8evItITN0DsvmXL/kcd
+h29LK7LM5uLw7efz0Qxs03G6kEyIHVkacowHi5I5Ul1qI61SoV3yMB1TjIU+bXZt
+0ZjC07totO0fqPOLQxonjQg=
+-----END PRIVATE KEY-----

src/common/grpc/tests/tls/server.pem

@@ -0,0 +1,27 @@
+-----BEGIN CERTIFICATE-----
+MIIEmDCCAwCgAwIBAgIQVEJFCgU/CZk9JEwTucWPpzANBgkqhkiG9w0BAQsFADCB
+hTEeMBwGA1UEChMVbWtjZXJ0IGRldmVsb3BtZW50IENBMS0wKwYDVQQLDCRsdWNp
+b0BMdWNpb3MtV29yay1NQlAgKEx1Y2lvIEZyYW5jbykxNDAyBgNVBAMMK21rY2Vy
+dCBsdWNpb0BMdWNpb3MtV29yay1NQlAgKEx1Y2lvIEZyYW5jbykwHhcNMTkwNjAx
+MDAwMDAwWhcNMjkwOTI5MjMzNTM0WjBYMScwJQYDVQQKEx5ta2NlcnQgZGV2ZWxv
+cG1lbnQgY2VydGlmaWNhdGUxLTArBgNVBAsMJGx1Y2lvQEx1Y2lvcy1Xb3JrLU1C
+UCAoTHVjaW8gRnJhbmNvKTCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEB
+APKm1szJhbO2C1q3UxcszMiRpB155VuzVj+oKkB3Bl79jlmlvOMYiMLFsHq81FKc
+RtgqnK0oc/Nh8zvx3wrxs++mepciJNcVPQlCXDU4csLV+vh7n3wYm6YCakxgWu2q
+/R9uZQVO5AZsfF44fa+OLRu+vc0pm5sx+QRk/dfiw1dehci5bYtpLw4wYqglp4IG
+91uzMSJIaP+/0GOkzTItMfz000xfVP4W1Mqe5dqDmDumvqXMmPzf4zMAbLic6Ox5
+K5SKa6LIb3T8Lm0DnALRCHlFsOeQViELjgrGh/+Lr6Od9q14g82IgjBBX7Z/ob9o
+rWGCN6RPXSb96hFgNN/57lkCAwEAAaOBrzCBrDAOBgNVHQ8BAf8EBAMCBaAwEwYD
+VR0lBAwwCgYIKwYBBQUHAwEwDAYDVR0TAQH/BAIwADAfBgNVHSMEGDAWgBQdvlE4
+Bdcsjc9oaxjDCRu5FiuZkzBWBgNVHREETzBNggtleGFtcGxlLmNvbYINKi5leGFt
+cGxlLmNvbYIMZXhhbXBsZS50ZXN0gglsb2NhbGhvc3SHBH8AAAGHEAAAAAAAAAAA
+AAAAAAAAAAEwDQYJKoZIhvcNAQELBQADggGBAKb2TJ8l+e1eraNwZWizLw5fccAf
+y59J1JAWdLxZyAI/bkiTlVO3DQoPZpw7XwLhefCvILkwKAL4TtIGGVC9yTb5Q5eg
+rqGO3FC0yg1fn65Kf1VpVxxUVyoiM5PQ4pFJb4AicAv88rCOLD9FFuE0PKOKU/dm
+Tw0WgPStoh9wsJ1RXUuTJYZs1nd1kMBlfv9NbLilnL+cR2sLktS54X5XagsBYVlf
+oapRb0JtABOoQhX3U8QMq8UF8yzceRHNTN9yfLOUrW26s9nKtlWVniNhw1uPxZw9
+RHM7w9/4+a9LXtEDYg4IP/1mm0ywBoUqy1O6hA73uId+Yi/kFBks/GyYaGjKgYcO
+23B75tkPGYEdGuGZYLzZNHbXg4V0UxFQG3KA1pUiSnD3bN2Rxs+CMpzORnOeK3xi
+EooKgAPYsehItoQOMPpccI2xHdSAMWtwUgOKrefUQujkx2Op+KFlspF0+WJ6AZEe
+2D4hyWaEZsvvILXapwqHDCuN3/jSUlTIqUoE1w==
+-----END CERTIFICATE-----

src/datanode/src/service.rs

@@ -21,6 +21,7 @@ use servers::grpc::{GrpcServer, GrpcServerConfig};
 use servers::http::HttpServerBuilder;
 use servers::metrics_handler::MetricsHandler;
 use servers::server::{ServerHandler, ServerHandlers};
+use servers::tls::TlsOption;
 use snafu::ResultExt;
 
 use crate::config::DatanodeOptions;
@@ -95,6 +96,7 @@ impl<'a> DatanodeServiceBuilder<'a> {
         let config = GrpcServerConfig {
             max_recv_message_size: opts.rpc_max_recv_message_size.as_bytes() as usize,
             max_send_message_size: opts.rpc_max_send_message_size.as_bytes() as usize,
+            tls: TlsOption::default(),
         };
 
         GrpcServerBuilder::new(config, region_server.runtime())

src/frontend/src/error.rs

@@ -355,6 +355,14 @@ pub enum Error {
         location: Location,
         name: String,
     },
+
+    #[snafu(display("Invalid tls config"))]
+    InvalidTlsConfig {
+        #[snafu(source)]
+        error: common_grpc::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -374,7 +382,8 @@ impl ErrorExt for Error {
             | Error::IllegalAuthConfig { .. }
             | Error::EmptyData { .. }
             | Error::ColumnNoneDefaultValue { .. }
-            | Error::IncompleteGrpcRequest { .. } => StatusCode::InvalidArguments,
+            | Error::IncompleteGrpcRequest { .. }
+            | Error::InvalidTlsConfig { .. } => StatusCode::InvalidArguments,
 
             Error::NotSupported { .. } => StatusCode::Unsupported,
 

src/frontend/src/server.rs

@@ -76,9 +76,12 @@ where
         let grpc_config = GrpcServerConfig {
             max_recv_message_size: opts.max_recv_message_size.as_bytes() as usize,
             max_send_message_size: opts.max_send_message_size.as_bytes() as usize,
+            tls: opts.tls.clone(),
         };
-
-        Ok(GrpcServerBuilder::new(grpc_config, grpc_runtime))
+        let builder = GrpcServerBuilder::new(grpc_config, grpc_runtime)
+            .with_tls_config(opts.tls.clone())
+            .context(error::InvalidTlsConfigSnafu)?;
+        Ok(builder)
     }
 
     pub fn http_server_builder(&self, opts: &FrontendOptions) -> HttpServerBuilder {

src/frontend/src/service_config/grpc.rs

@@ -17,6 +17,7 @@ use common_grpc::channel_manager::{
     DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE, DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
 };
 use serde::{Deserialize, Serialize};
+use servers::tls::TlsOption;
 
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
 pub struct GrpcOptions {
@@ -26,6 +27,8 @@ pub struct GrpcOptions {
     pub max_recv_message_size: ReadableSize,
     // Max gRPC sending(encoding) message size
     pub max_send_message_size: ReadableSize,
+    #[serde(default = "Default::default")]
+    pub tls: TlsOption,
 }
 
 impl Default for GrpcOptions {
@@ -35,6 +38,7 @@ impl Default for GrpcOptions {
             runtime_size: 8,
             max_recv_message_size: DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE,
             max_send_message_size: DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
+            tls: TlsOption::default(),
         }
     }
 }

src/servers/src/grpc.rs

@@ -36,6 +36,7 @@ use tokio::net::TcpListener;
 use tokio::sync::oneshot::{self, Receiver, Sender};
 use tokio::sync::Mutex;
 use tonic::transport::server::{Routes, TcpIncoming};
+use tonic::transport::ServerTlsConfig;
 use tonic::{Request, Response, Status};
 use tonic_reflection::server::{ServerReflection, ServerReflectionServer};
 
@@ -44,6 +45,7 @@ use crate::error::{
 };
 use crate::metrics::MetricsMiddlewareLayer;
 use crate::server::Server;
+use crate::tls::TlsOption;
 
 type TonicResult<T> = std::result::Result<T, Status>;
 
@@ -55,6 +57,8 @@ pub struct GrpcServer {
     serve_state: Mutex<Option<Receiver<Result<()>>>>,
     // handlers
     routes: Mutex<Option<Routes>>,
+    // tls config
+    tls_config: Option<ServerTlsConfig>,
 }
 
 /// Grpc Server configuration
@@ -64,6 +68,7 @@ pub struct GrpcServerConfig {
     pub max_recv_message_size: usize,
     // Max gRPC sending(encoding) message size
     pub max_send_message_size: usize,
+    pub tls: TlsOption,
 }
 
 impl Default for GrpcServerConfig {
@@ -71,6 +76,7 @@ impl Default for GrpcServerConfig {
         Self {
             max_recv_message_size: DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE.as_bytes() as usize,
             max_send_message_size: DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE.as_bytes() as usize,
+            tls: TlsOption::default(),
         }
     }
 }
@@ -172,8 +178,11 @@ impl Server for GrpcServer {
             .layer(MetricsMiddlewareLayer)
             .into_inner();
 
-        let builder = tonic::transport::Server::builder()
-            .layer(metrics_layer)
+        let mut builder = tonic::transport::Server::builder().layer(metrics_layer);
+        if let Some(tls_config) = self.tls_config.clone() {
+            builder = builder.tls_config(tls_config).context(StartGrpcSnafu)?;
+        }
+        let builder = builder
             .add_routes(routes)
             .add_service(self.create_healthcheck_service())
             .add_service(self.create_reflection_service());

src/servers/src/grpc/builder.rs

@@ -19,12 +19,15 @@ use api::v1::prometheus_gateway_server::PrometheusGatewayServer;
 use api::v1::region::region_server::RegionServer;
 use arrow_flight::flight_service_server::FlightServiceServer;
 use auth::UserProviderRef;
+use common_grpc::error::{Error, InvalidConfigFilePathSnafu, Result};
 use common_runtime::Runtime;
 use opentelemetry_proto::tonic::collector::metrics::v1::metrics_service_server::MetricsServiceServer;
 use opentelemetry_proto::tonic::collector::trace::v1::trace_service_server::TraceServiceServer;
+use snafu::ResultExt;
 use tokio::sync::Mutex;
 use tonic::codec::CompressionEncoding;
 use tonic::transport::server::RoutesBuilder;
+use tonic::transport::{Identity, ServerTlsConfig};
 use tower::ServiceBuilder;
 
 use super::flight::{FlightCraftRef, FlightCraftWrapper};
@@ -37,6 +40,7 @@ use crate::grpc::otlp::OtlpService;
 use crate::grpc::prom_query_gateway::PrometheusGatewayService;
 use crate::prometheus_handler::PrometheusHandlerRef;
 use crate::query_handler::OpenTelemetryProtocolHandlerRef;
+use crate::tls::TlsOption;
 
 /// Add a gRPC service (`service`) to a `builder`([RoutesBuilder]).
 /// This macro will automatically add some gRPC properties to the service.
@@ -62,6 +66,7 @@ pub struct GrpcServerBuilder {
     config: GrpcServerConfig,
     runtime: Arc<Runtime>,
     routes_builder: RoutesBuilder,
+    tls_config: Option<ServerTlsConfig>,
 }
 
 impl GrpcServerBuilder {
@@ -70,6 +75,7 @@ impl GrpcServerBuilder {
             config,
             runtime,
             routes_builder: RoutesBuilder::default(),
+            tls_config: None,
         }
     }
 
@@ -157,11 +163,34 @@ impl GrpcServerBuilder {
         &mut self.routes_builder
     }
 
+    pub fn with_tls_config(mut self, tls_option: TlsOption) -> Result<Self> {
+        // tonic does not support watching for tls config changes
+        // so we don't support it either for now
+        if tls_option.watch {
+            return Err(Error::NotSupported {
+                feat: "Certificates watch and reloading for gRPC is not supported at the moment"
+                    .to_string(),
+            });
+        }
+        self.tls_config = if tls_option.should_force_tls() {
+            let cert = std::fs::read_to_string(tls_option.cert_path)
+                .context(InvalidConfigFilePathSnafu)?;
+            let key =
+                std::fs::read_to_string(tls_option.key_path).context(InvalidConfigFilePathSnafu)?;
+            let identity = Identity::from_pem(cert, key);
+            Some(ServerTlsConfig::new().identity(identity))
+        } else {
+            None
+        };
+        Ok(self)
+    }
+
     pub fn build(self) -> GrpcServer {
         GrpcServer {
             routes: Mutex::new(Some(self.routes_builder.routes())),
             shutdown_tx: Mutex::new(None),
             serve_state: Mutex::new(None),
+            tls_config: self.tls_config,
         }
     }
 }

tests-integration/src/test_util.rs

@@ -511,13 +511,15 @@ pub async fn setup_grpc_server_with(
 
     let flight_handler = Arc::new(greptime_request_handler.clone());
 
-    let fe_grpc_server = Arc::new(
-        GrpcServerBuilder::new(grpc_config.unwrap_or_default(), runtime)
-            .database_handler(greptime_request_handler)
-            .flight_handler(flight_handler)
-            .prometheus_handler(fe_instance_ref.clone(), user_provider)
-            .build(),
-    );
+    let grpc_config = grpc_config.unwrap_or_default();
+    let grpc_builder = GrpcServerBuilder::new(grpc_config.clone(), runtime)
+        .database_handler(greptime_request_handler)
+        .flight_handler(flight_handler)
+        .prometheus_handler(fe_instance_ref.clone(), user_provider)
+        .with_tls_config(grpc_config.tls)
+        .unwrap();
+
+    let fe_grpc_server = Arc::new(grpc_builder.build());
 
     let fe_grpc_addr = "127.0.0.1:0".parse::<SocketAddr>().unwrap();
     let fe_grpc_addr = fe_grpc_server

tests-integration/tests/grpc.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::sync::Arc;
+
 use api::v1::alter_expr::Kind;
 use api::v1::promql_request::Promql;
 use api::v1::{
@@ -22,14 +24,18 @@ use api::v1::{
 use auth::user_provider_from_option;
 use client::{Client, OutputData, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_catalog::consts::MITO_ENGINE;
+use common_grpc::channel_manager::ClientTlsOption;
 use common_query::Output;
 use common_recordbatch::RecordBatches;
+use common_runtime::Runtime;
+use servers::grpc::builder::GrpcServerBuilder;
 use servers::grpc::GrpcServerConfig;
 use servers::http::prometheus::{
     PromData, PromQueryResult, PromSeriesMatrix, PromSeriesVector, PrometheusJsonResponse,
     PrometheusResponse,
 };
 use servers::server::Server;
+use servers::tls::{TlsMode, TlsOption};
 use tests_integration::database::Database;
 use tests_integration::test_util::{
     setup_grpc_server, setup_grpc_server_with, setup_grpc_server_with_user_provider, StorageType,
@@ -77,6 +83,7 @@ macro_rules! grpc_tests {
                 test_health_check,
                 test_prom_gateway_query,
                 test_grpc_timezone,
+                test_grpc_tls_config,
             );
         )*
     };
@@ -129,6 +136,7 @@ pub async fn test_grpc_message_size_ok(store_type: StorageType) {
     let config = GrpcServerConfig {
         max_recv_message_size: 1024,
         max_send_message_size: 1024,
+        ..Default::default()
     };
     let (addr, mut guard, fe_grpc_server) =
         setup_grpc_server_with(store_type, "auto_create_table", None, Some(config)).await;
@@ -148,6 +156,7 @@ pub async fn test_grpc_zstd_compression(store_type: StorageType) {
     let config = GrpcServerConfig {
         max_recv_message_size: 1024,
         max_send_message_size: 1024,
+        ..Default::default()
     };
     let (addr, mut guard, fe_grpc_server) =
         setup_grpc_server_with(store_type, "auto_create_table", None, Some(config)).await;
@@ -166,6 +175,7 @@ pub async fn test_grpc_message_size_limit_send(store_type: StorageType) {
     let config = GrpcServerConfig {
         max_recv_message_size: 1024,
         max_send_message_size: 50,
+        ..Default::default()
     };
     let (addr, mut guard, fe_grpc_server) =
         setup_grpc_server_with(store_type, "auto_create_table", None, Some(config)).await;
@@ -185,6 +195,7 @@ pub async fn test_grpc_message_size_limit_recv(store_type: StorageType) {
     let config = GrpcServerConfig {
         max_recv_message_size: 10,
         max_send_message_size: 1024,
+        ..Default::default()
     };
     let (addr, mut guard, fe_grpc_server) =
         setup_grpc_server_with(store_type, "auto_create_table", None, Some(config)).await;
@@ -663,6 +674,7 @@ pub async fn test_grpc_timezone(store_type: StorageType) {
     let config = GrpcServerConfig {
         max_recv_message_size: 1024,
         max_send_message_size: 1024,
+        ..Default::default()
     };
     let (addr, mut guard, fe_grpc_server) =
         setup_grpc_server_with(store_type, "auto_create_table", None, Some(config)).await;
@@ -719,3 +731,74 @@ async fn to_batch(output: Output) -> String {
     .pretty_print()
     .unwrap()
 }
+
+pub async fn test_grpc_tls_config(store_type: StorageType) {
+    let comm_dir = std::path::PathBuf::from_iter([
+        std::env!("CARGO_RUSTC_CURRENT_DIR"),
+        "src/common/grpc/tests/tls",
+    ]);
+    let ca_path = comm_dir.join("ca.pem").to_str().unwrap().to_string();
+    let server_cert_path = comm_dir.join("server.pem").to_str().unwrap().to_string();
+    let server_key_path = comm_dir.join("server.key").to_str().unwrap().to_string();
+    let client_cert_path = comm_dir.join("client.pem").to_str().unwrap().to_string();
+    let client_key_path = comm_dir.join("client.key").to_str().unwrap().to_string();
+    let client_corrupted = comm_dir.join("corrupted").to_str().unwrap().to_string();
+
+    let tls = TlsOption::new(
+        Some(TlsMode::Require),
+        Some(server_cert_path),
+        Some(server_key_path),
+    );
+    let config = GrpcServerConfig {
+        max_recv_message_size: 1024,
+        max_send_message_size: 1024,
+        tls,
+    };
+    let (addr, mut guard, fe_grpc_server) =
+        setup_grpc_server_with(store_type, "tls_create_table", None, Some(config)).await;
+
+    let mut client_tls = ClientTlsOption {
+        server_ca_cert_path: ca_path,
+        client_cert_path,
+        client_key_path,
+    };
+    {
+        let grpc_client =
+            Client::with_tls_and_urls(vec![addr.clone()], client_tls.clone()).unwrap();
+        let db = Database::new_with_dbname(
+            format!("{}-{}", DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME),
+            grpc_client,
+        );
+        db.sql("show tables;").await.unwrap();
+    }
+    // test corrupted client key
+    {
+        client_tls.client_key_path = client_corrupted;
+        let grpc_client = Client::with_tls_and_urls(vec![addr], client_tls.clone()).unwrap();
+        let db = Database::new_with_dbname(
+            format!("{}-{}", DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME),
+            grpc_client,
+        );
+        let re = db.sql("show tables;").await;
+        assert!(re.is_err());
+    }
+    // test grpc unsupported tls watch
+    {
+        let tls = TlsOption {
+            watch: true,
+            ..Default::default()
+        };
+        let config = GrpcServerConfig {
+            max_recv_message_size: 1024,
+            max_send_message_size: 1024,
+            tls,
+        };
+        let runtime = Arc::new(Runtime::builder().build().unwrap());
+        let grpc_builder =
+            GrpcServerBuilder::new(config.clone(), runtime).with_tls_config(config.tls);
+        assert!(grpc_builder.is_err());
+    }
+
+    let _ = fe_grpc_server.shutdown().await;
+    guard.remove_all().await;
+}

tests-integration/tests/http.rs

@@ -744,6 +744,12 @@ runtime_size = 8
 max_recv_message_size = "512MiB"
 max_send_message_size = "512MiB"
 
+[grpc.tls]
+mode = "disable"
+cert_path = ""
+key_path = ""
+watch = false
+
 [mysql]
 enable = true
 addr = "127.0.0.1:4002"