style: modify pkg about database connection (oceanbase#52)

modify database connection

pkg/database/connector.go

@@ -0,0 +1,80 @@
+/*
+Copyright (c) 2023 OceanBase
+ob-operator is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details.
+*/
+
+package database
+
+import (
+	// register mysql driver
+	_ "github.com/go-sql-driver/mysql"
+	"github.com/jmoiron/sqlx"
+)
+
+// Connector represents a connection pool.
+type Connector struct {
+	ds     DataSource
+	client *Client
+}
+
+// DataSource represents a data source.
+type DataSource interface {
+	ID() string
+	DriverName() string
+	DataSourceName() string
+	String() string
+}
+
+// Client represents a wrapper around sqlx.DB.
+type Client struct {
+	*sqlx.DB
+}
+
+func (c *Client) configure() {
+	c.SetMaxOpenConns(DefaultConnMaxOpenCount)
+	c.SetMaxIdleConns(DefaultConnMaxIdleCount)
+	c.SetConnMaxLifetime(DefaultConnMaxLifetime)
+	c.SetConnMaxIdleTime(DefaultConnMaxIdleTime)
+}
+
+func NewConnector(dataSource DataSource) *Connector {
+	return &Connector{
+		ds: dataSource,
+	}
+}
+
+func (c *Connector) Init() error {
+	db, err := sqlx.Connect(c.ds.DriverName(), c.ds.DataSourceName())
+	if err != nil {
+		return err
+	}
+	c.client = &Client{db}
+	c.client.configure()
+	return nil
+}
+
+func (c *Connector) IsAlive() bool {
+	if c.client == nil {
+		return false
+	}
+	err := c.client.Ping()
+	return err == nil
+}
+
+func (c *Connector) GetClient() *Client {
+	return c.client
+}
+
+func (c *Connector) Close() error {
+	if c.client.DB != nil {
+		return c.client.DB.Close()
+	}
+	return nil
+}

pkg/database/const.go

@@ -0,0 +1,8 @@
+package database
+
+const (
+	DefaultConnMaxOpenCount = 20
+	DefaultConnMaxIdleCount = 1
+	DefaultConnMaxLifetime  = 0
+	DefaultConnMaxIdleTime  = 0
+)

pkg/database/pool.go

@@ -0,0 +1,41 @@
+/*
+Copyright (c) 2023 OceanBase
+ob-operator is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details.
+*/
+
+package database
+
+import (
+	"sync"
+
+	"github.com/pkg/errors"
+)
+
+var p = &pool{}
+
+type pool struct {
+	// TODO: maintain cache size and data expiration, maybe change to a cache library.
+	Cache sync.Map
+}
+
+func GetConnector(dataSource DataSource) (*Connector, error) {
+	c, ok := p.Cache.Load(dataSource.ID())
+	if ok && c.(*Connector).IsAlive() {
+		return c.(*Connector), nil
+	}
+	connector := NewConnector(dataSource)
+	err := connector.Init()
+	if err != nil {
+		err = errors.Wrap(err, "init connector failed with datasource: "+dataSource.String())
+		return nil, err
+	}
+	p.Cache.Store(dataSource.ID(), connector)
+	return connector, nil
+}

pkg/oceanbase/connector/datasource.go

@@ -0,0 +1,69 @@
+/*
+Copyright (c) 2023 OceanBase
+ob-operator is licensed under Mulan PSL v2.
+You can use this software according to the terms and conditions of the Mulan PSL v2.
+You may obtain a copy of Mulan PSL v2 at:
+         http://license.coscl.org.cn/MulanPSL2
+THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
+EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
+MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
+See the Mulan PSL v2 for more details.
+*/
+
+package connector
+
+import (
+	"crypto/md5"
+	"encoding/hex"
+	"fmt"
+)
+
+// OceanBaseDataSource implements the database.DataSource interface for OceanBase.
+type OceanBaseDataSource struct {
+	Address  string
+	Port     int64
+	User     string
+	Tenant   string
+	Password string
+	Database string
+}
+
+func NewOceanBaseDataSource(address string, port int64, user, tenant, password, database string) *OceanBaseDataSource {
+	return &OceanBaseDataSource{
+		Address:  address,
+		Port:     port,
+		User:     user,
+		Tenant:   tenant,
+		Password: password,
+		Database: database,
+	}
+}
+
+func (*OceanBaseDataSource) DriverName() string {
+	return "mysql"
+}
+
+func (s *OceanBaseDataSource) DataSourceName() string {
+	passwordPart := ""
+	if s.Password != "" {
+		passwordPart = fmt.Sprintf(":%s", s.Password)
+	}
+	if s.Database != "" {
+		return fmt.Sprintf("%s@%s%s@tcp(%s:%d)/%s?multiStatements=true&interpolateParams=true", s.User, s.Tenant, passwordPart, s.Address, s.Port, s.Database)
+	}
+	return fmt.Sprintf("%s@%s@tcp(%s:%d)/", s.User, s.Tenant, s.Address, s.Port)
+}
+
+func (s *OceanBaseDataSource) ID() string {
+	h := md5.New()
+	key := fmt.Sprintf("%s@%s@%s:%d/%s", s.User, s.Tenant, s.Address, s.Port, s.Database)
+	_, err := h.Write([]byte(key))
+	if err != nil {
+		return key
+	}
+	return hex.EncodeToString(h.Sum(nil))
+}
+
+func (s *OceanBaseDataSource) String() string {
+	return fmt.Sprintf("address: %s, port: %d, user: %s, tenant: %s, database: %s", s.Address, s.Port, s.User, s.Tenant, s.Database)
+}