[ENH] go/coordinator: grpcserver supports mTLS (#1362)

## Description of changes

*Summarize the changes made by this PR.*
 - New functionality
- Adding mTLS support for coordinator grpcserver, because we need to
make sure the connection & data transport between different component is
secure when these components deployed in public cloud env.

## Test plan
*How are these changes tested?*

- [x] Tests pass locally with `pytest` for python, `yarn test` for js,
`make test` for golang

## Documentation Changes
*Are all docstrings for user-facing APIs updated if required? Do we need
to make documentation changes in the [docs
repository](https://github.com/chroma-core/docs)?*

---------

Signed-off-by: zhangjinpeng1987 <[email protected]>
Co-authored-by: Ben Eggers <[email protected]>

go/coordinator/cmd/grpccoordinator/cmd.go

@@ -22,8 +22,9 @@ var (
 )
 
 func init() {
+
 	// GRPC
-	flag.GRPCAddr(Cmd, &conf.BindAddress)
+	flag.GRPCAddr(Cmd, &conf.GrpcConfig.BindAddress)
 
 	// System Catalog
 	Cmd.Flags().StringVar(&conf.SystemCatalogProvider, "system-catalog-provider", "memory", "System catalog provider")

go/coordinator/internal/grpccoordinator/grpcutils/config.go

@@ -0,0 +1,15 @@
+package grpcutils
+
+type GrpcConfig struct {
+	// BindAddress is the address to bind the GRPC server to.
+	BindAddress string
+
+	// GRPC mTLS config
+	CertPath string
+	KeyPath  string
+	CAPath   string
+}
+
+func (c *GrpcConfig) MTLSEnabled() bool {
+	return c.CertPath != "" && c.KeyPath != "" && c.CAPath != ""
+}

go/coordinator/internal/grpccoordinator/grpcutils/config_test.go

@@ -0,0 +1,37 @@
+package grpcutils
+
+import "testing"
+
+func TestGrpcConfig_TLSEnabled(t *testing.T) {
+	// Create a list of configs and expected check result (true/false)
+	cfgs := []*GrpcConfig{
+		{
+			CertPath: "cert",
+			KeyPath:  "key",
+			CAPath:   "ca",
+		},
+		{
+			CertPath: "",
+			KeyPath:  "",
+			CAPath:   "",
+		},
+		{
+			CertPath: "cert",
+			KeyPath:  "",
+			CAPath:   "ca",
+		},
+		{
+			CertPath: "",
+			KeyPath:  "key",
+			CAPath:   "ca",
+		},
+	}
+	expected := []bool{true, false, false, false}
+
+	// Iterate through the list of configs and check if the result matches the expected result
+	for i, cfg := range cfgs {
+		if cfg.MTLSEnabled() != expected[i] {
+			t.Errorf("Expected %v, got %v", expected[i], cfg.MTLSEnabled())
+		}
+	}
+}

go/coordinator/internal/grpccoordinator/grpcutils/service.go

@@ -1,12 +1,16 @@
 package grpcutils
 
 import (
+	"crypto/tls"
+	"crypto/x509"
 	"io"
 	"net"
+	"os"
 
 	"github.com/pingcap/log"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 )
 
 const (
@@ -22,16 +26,16 @@ type GrpcServer interface {
 }
 
 type GrpcProvider interface {
-	StartGrpcServer(name, bindAddress string, registerFunc func(grpc.ServiceRegistrar)) (GrpcServer, error)
+	StartGrpcServer(name string, grpcConfig *GrpcConfig, registerFunc func(grpc.ServiceRegistrar)) (GrpcServer, error)
 }
 
 var Default = &defaultProvider{}
 
 type defaultProvider struct {
 }
 
-func (d *defaultProvider) StartGrpcServer(name, bindAddress string, registerFunc func(grpc.ServiceRegistrar)) (GrpcServer, error) {
-	return newDefaultGrpcProvider(name, bindAddress, registerFunc)
+func (d *defaultProvider) StartGrpcServer(name string, grpcConfig *GrpcConfig, registerFunc func(grpc.ServiceRegistrar)) (GrpcServer, error) {
+	return newDefaultGrpcProvider(name, grpcConfig, registerFunc)
 }
 
 type defaultGrpcServer struct {
@@ -40,15 +44,39 @@ type defaultGrpcServer struct {
 	port   int
 }
 
-func newDefaultGrpcProvider(name, bindAddress string, registerFunc func(grpc.ServiceRegistrar)) (GrpcServer, error) {
+func newDefaultGrpcProvider(name string, grpcConfig *GrpcConfig, registerFunc func(grpc.ServiceRegistrar)) (GrpcServer, error) {
+	var opts []grpc.ServerOption
+	opts = append(opts, grpc.MaxRecvMsgSize(maxGrpcFrameSize))
+	if grpcConfig.MTLSEnabled() {
+		cert, err := tls.LoadX509KeyPair(grpcConfig.CertPath, grpcConfig.KeyPath)
+		if err != nil {
+			return nil, err
+		}
+
+		ca := x509.NewCertPool()
+		caBytes, err := os.ReadFile(grpcConfig.CAPath)
+		if err != nil {
+			return nil, err
+		}
+		if !ca.AppendCertsFromPEM(caBytes) {
+			return nil, err
+		}
+
+		tlsConfig := &tls.Config{
+			Certificates: []tls.Certificate{cert},
+			ClientCAs:    ca,
+			ClientAuth:   tls.RequireAndVerifyClientCert,
+		}
+
+		opts = append(opts, grpc.Creds(credentials.NewTLS(tlsConfig)))
+	}
+
 	c := &defaultGrpcServer{
-		server: grpc.NewServer(
-			grpc.MaxRecvMsgSize(maxGrpcFrameSize),
-		),
+		server: grpc.NewServer(opts...),
 	}
 	registerFunc(c.server)
 
-	listener, err := net.Listen("tcp", bindAddress)
+	listener, err := net.Listen("tcp", grpcConfig.BindAddress)
 	if err != nil {
 		return nil, err
 	}

go/coordinator/internal/grpccoordinator/server.go

@@ -22,8 +22,8 @@ import (
 )
 
 type Config struct {
-	// GRPC config
-	BindAddress string
+	// GrpcConfig config
+	GrpcConfig *grpcutils.GrpcConfig
 
 	// System catalog provider
 	SystemCatalogProvider string
@@ -175,7 +175,7 @@ func NewWithGrpcProvider(config Config, provider grpcutils.GrpcProvider, db *gor
 			return nil, err
 		}
 
-		s.grpcServer, err = provider.StartGrpcServer("coordinator", config.BindAddress, func(registrar grpc.ServiceRegistrar) {
+    s.grpcServer, err = provider.StartGrpcServer("coordinator", config.GrpcConfig, func(registrar grpc.ServiceRegistrar) {
 			coordinatorpb.RegisterSysDBServer(registrar, s)
 		})
 		if err != nil {