<commit_before>// Copyright 2017 The OpenPitrix Authors. All rights reserved.
// Use of this source code is governed by a Apache license
// that can be found in the LICENSE file.

package db_cluster

import (
	"context"
	"database/sql"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/pkg/errors"
	"gopkg.in/gorp.v2"

	"openpitrix.io/openpitrix/pkg/config"
	"openpitrix.io/openpitrix/pkg/logger"
)

const (
	ClusterTableName     = "cluster"
	ClusterNodeTableName = "cluster_node"
)

type Cluster struct {
	Id               string    `db:"id, size:50, primarykey"`
	Name             string    `db:"name, size:50"`
	Description      string    `db:"description, size:1000"`
	AppId            string    `db:"app_id, size:50"`
	AppVersion       string    `db:"app_version, size:50"`
	Status           string    `db:"status, size:50"`
	TransitionStatus string    `db:"transition_status, size:50"`
	Created          time.Time `db:"created"`
	LastModified     time.Time `db:"last_modified"`
}

type ClusterNode struct {
	Id               string    `db:"id, size:50, primarykey"`
	InstanceId       string    `db:"instance_id, size:50"`
	Name             string    `db:"name, size:50"`
	Description      string    `db:"description, size:1000"`
	ClusterId        string    `db:"app_id, size:50"`
	PrivateIp        string    `db:"app_version, size:50"`
	Status           string    `db:"status, size:50"`
	TransitionStatus string    `db:"transition_status, size:50"`
	Created          time.Time `db:"created"`
	LastModified     time.Time `db:"last_modified"`
}

type ClusterDatabase struct {
	cfg               config.Database
	db                *sql.DB
	dbMap             *gorp.DbMap
	createTablesDone  uint32
	createTablesMutex sync.Mutex
}

func OpenClusterDatabase(cfg *config.Database) (p *ClusterDatabase, err error) {
	// https://github.com/go-sql-driver/mysql/issues/9
	db, err := sql.Open(cfg.Type, cfg.GetUrl()+"?parseTime=true")
	if err != nil {
		err = errors.WithStack(err)
		return nil, err
	}

	dialect := gorp.MySQLDialect{
		Encoding: cfg.Encoding,
		Engine:   cfg.Engine,
	}

	dbMap := &gorp.DbMap{Db: db, Dialect: dialect}
	dbMap.AddTableWithName(Cluster{}, ClusterTableName)
	dbMap.AddTableWithName(ClusterNode{}, ClusterNodeTableName)

	p = &ClusterDatabase{
		cfg:   *cfg,
		db:    db,
		dbMap: dbMap,
	}

	p.initTables()
	return
}

func (p *ClusterDatabase) Close() error {
	err := p.db.Close()
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) GetClusters(ctx context.Context, ids string) (clusters []Cluster, err error) {
	p.initTables()
	parts := strings.Split(ids, ",")
	for _, id := range parts {
		if id == "cl-panic000" {
			panic(id)
		}
	}
	in_ids := fmt.Sprintf("'%s'", strings.Join(parts, "','"))
	_, err = p.dbMap.Select(&clusters, fmt.Sprintf("select * from %s where id in (%s)", ClusterTableName, in_ids))
	err = errors.WithStack(err)
	return
}

func (p *ClusterDatabase) GetClusterList(ctx context.Context) (clusters []Cluster, err error) {
	p.initTables()
	_, err = p.dbMap.Select(&clusters, "select * from "+ClusterTableName)
	err = errors.WithStack(err)
	return
}

func (p *ClusterDatabase) CreateCluster(ctx context.Context, cluster *Cluster) error {
	p.initTables()
	err := p.dbMap.Insert(cluster)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) UpdateCluster(ctx context.Context, cluster *Cluster) error {
	p.initTables()
	_, err := p.dbMap.Update(cluster)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) DeleteClusters(ctx context.Context, ids string) error {
	p.initTables()
	parts := strings.Split(ids, ",")
	clusters := make([]interface{}, len(parts))
	for i, id := range parts {
		clusters[i] = &Cluster{Id: id}
	}
	_, err := p.dbMap.Delete(clusters...)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) GetClusterNodes(ctx context.Context, ids string) (clusterNodes []ClusterNode, err error) {
	p.initTables()
	parts := strings.Split(ids, ",")
	in_ids := fmt.Sprintf("'%s'", strings.Join(parts, "','"))
	_, err = p.dbMap.Select(&clusterNodes, fmt.Sprintf("select * from %s where id in (%s)", ClusterNodeTableName, in_ids))
	err = errors.WithStack(err)
	return
}

func (p *ClusterDatabase) GetClusterNodeList(ctx context.Context) (clusterNodes []ClusterNode, err error) {
	p.initTables()
	_, err = p.dbMap.Select(&clusterNodes, "select * from "+ClusterNodeTableName)
	err = errors.WithStack(err)
	return
}

func (p *ClusterDatabase) CreateClusterNodes(ctx context.Context, clusterNodes []*ClusterNode) error {
	p.initTables()
	dst := make([]interface{}, len(clusterNodes))
	for i, v := range clusterNodes {
		dst[i] = v
	}
	err := p.dbMap.Insert(dst...)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) UpdateClusterNode(ctx context.Context, clusterNode *ClusterNode) error {
	p.initTables()
	_, err := p.dbMap.Update(clusterNode)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) DeleteClusterNodes(ctx context.Context, ids string) error {
	p.initTables()
	parts := strings.Split(ids, ",")
	clusterNodes := make([]interface{}, len(parts))
	for i, id := range parts {
		clusterNodes[i] = &ClusterNode{Id: id}
	}
	_, err := p.dbMap.Delete(clusterNodes...)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) TruncateTables() error {
	p.initTables()
	err := p.dbMap.TruncateTables()
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) initTables() {
	if atomic.LoadUint32(&p.createTablesDone) == 1 {
		return
	}

	// Slow-path.
	p.createTablesMutex.Lock()
	defer p.createTablesMutex.Unlock()

	if atomic.LoadUint32(&p.createTablesDone) == 1 {
		return
	}

	db, err := sql.Open(p.cfg.Type, p.cfg.GetServerAddr())
	if err != nil {
		logger.Warningf("sql.Open failed: %+v", err)
		return
	}
	defer db.Close()

	sqlCreateDbIfNotExists := fmt.Sprintf(
		"CREATE DATABASE IF NOT EXISTS %s DEFAULT CHARACTER SET utf8;",
		p.cfg.DbName,
	)
	if _, err = db.Exec(sqlCreateDbIfNotExists); err != nil {
		logger.Warningf("CREATE DATABASE failed: %+v", err)
		return
	}

	if err := p.dbMap.CreateTablesIfNotExists(); err != nil {
		logger.Warningf("CreateTablesIfNotExists: %+v", err)
		return
	}

	atomic.StoreUint32(&p.createTablesDone, 1)
}
<commit_msg>pkg/db/cluster: fix orm tag string<commit_after>// Copyright 2017 The OpenPitrix Authors. All rights reserved.
// Use of this source code is governed by a Apache license
// that can be found in the LICENSE file.

package db_cluster

import (
	"context"
	"database/sql"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	_ "github.com/go-sql-driver/mysql"
	"github.com/pkg/errors"
	"gopkg.in/gorp.v2"

	"openpitrix.io/openpitrix/pkg/config"
	"openpitrix.io/openpitrix/pkg/logger"
)

const (
	ClusterTableName     = "cluster"
	ClusterNodeTableName = "cluster_node"
)

type Cluster struct {
	Id               string    `db:"id, size:50, primarykey"`
	Name             string    `db:"name, size:50"`
	Description      string    `db:"description, size:1000"`
	AppId            string    `db:"app_id, size:50"`
	AppVersion       string    `db:"app_version, size:50"`
	Status           string    `db:"status, size:50"`
	TransitionStatus string    `db:"transition_status, size:50"`
	Created          time.Time `db:"created"`
	LastModified     time.Time `db:"last_modified"`
}

type ClusterNode struct {
	Id               string    `db:"id, size:50, primarykey"`
	InstanceId       string    `db:"instance_id, size:50"`
	Name             string    `db:"name, size:50"`
	Description      string    `db:"description, size:1000"`
	ClusterId        string    `db:"cluster_id, size:50"`
	PrivateIp        string    `db:"private_ip, size:50"`
	Status           string    `db:"status, size:50"`
	TransitionStatus string    `db:"transition_status, size:50"`
	Created          time.Time `db:"created"`
	LastModified     time.Time `db:"last_modified"`
}

type ClusterDatabase struct {
	cfg               config.Database
	db                *sql.DB
	dbMap             *gorp.DbMap
	createTablesDone  uint32
	createTablesMutex sync.Mutex
}

func OpenClusterDatabase(cfg *config.Database) (p *ClusterDatabase, err error) {
	// https://github.com/go-sql-driver/mysql/issues/9
	db, err := sql.Open(cfg.Type, cfg.GetUrl()+"?parseTime=true")
	if err != nil {
		err = errors.WithStack(err)
		return nil, err
	}

	dialect := gorp.MySQLDialect{
		Encoding: cfg.Encoding,
		Engine:   cfg.Engine,
	}

	dbMap := &gorp.DbMap{Db: db, Dialect: dialect}
	dbMap.AddTableWithName(Cluster{}, ClusterTableName)
	dbMap.AddTableWithName(ClusterNode{}, ClusterNodeTableName)

	p = &ClusterDatabase{
		cfg:   *cfg,
		db:    db,
		dbMap: dbMap,
	}

	p.initTables()
	return
}

func (p *ClusterDatabase) Close() error {
	err := p.db.Close()
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) GetClusters(ctx context.Context, ids string) (clusters []Cluster, err error) {
	p.initTables()
	parts := strings.Split(ids, ",")
	for _, id := range parts {
		if id == "cl-panic000" {
			panic(id)
		}
	}
	in_ids := fmt.Sprintf("'%s'", strings.Join(parts, "','"))
	_, err = p.dbMap.Select(&clusters, fmt.Sprintf("select * from %s where id in (%s)", ClusterTableName, in_ids))
	err = errors.WithStack(err)
	return
}

func (p *ClusterDatabase) GetClusterList(ctx context.Context) (clusters []Cluster, err error) {
	p.initTables()
	_, err = p.dbMap.Select(&clusters, "select * from "+ClusterTableName)
	err = errors.WithStack(err)
	return
}

func (p *ClusterDatabase) CreateCluster(ctx context.Context, cluster *Cluster) error {
	p.initTables()
	err := p.dbMap.Insert(cluster)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) UpdateCluster(ctx context.Context, cluster *Cluster) error {
	p.initTables()
	_, err := p.dbMap.Update(cluster)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) DeleteClusters(ctx context.Context, ids string) error {
	p.initTables()
	parts := strings.Split(ids, ",")
	clusters := make([]interface{}, len(parts))
	for i, id := range parts {
		clusters[i] = &Cluster{Id: id}
	}
	_, err := p.dbMap.Delete(clusters...)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) GetClusterNodes(ctx context.Context, ids string) (clusterNodes []ClusterNode, err error) {
	p.initTables()
	parts := strings.Split(ids, ",")
	in_ids := fmt.Sprintf("'%s'", strings.Join(parts, "','"))
	_, err = p.dbMap.Select(&clusterNodes, fmt.Sprintf("select * from %s where id in (%s)", ClusterNodeTableName, in_ids))
	err = errors.WithStack(err)
	return
}

func (p *ClusterDatabase) GetClusterNodeList(ctx context.Context) (clusterNodes []ClusterNode, err error) {
	p.initTables()
	_, err = p.dbMap.Select(&clusterNodes, "select * from "+ClusterNodeTableName)
	err = errors.WithStack(err)
	return
}

func (p *ClusterDatabase) CreateClusterNodes(ctx context.Context, clusterNodes []*ClusterNode) error {
	p.initTables()
	dst := make([]interface{}, len(clusterNodes))
	for i, v := range clusterNodes {
		dst[i] = v
	}
	err := p.dbMap.Insert(dst...)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) UpdateClusterNode(ctx context.Context, clusterNode *ClusterNode) error {
	p.initTables()
	_, err := p.dbMap.Update(clusterNode)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) DeleteClusterNodes(ctx context.Context, ids string) error {
	p.initTables()
	parts := strings.Split(ids, ",")
	clusterNodes := make([]interface{}, len(parts))
	for i, id := range parts {
		clusterNodes[i] = &ClusterNode{Id: id}
	}
	_, err := p.dbMap.Delete(clusterNodes...)
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) TruncateTables() error {
	p.initTables()
	err := p.dbMap.TruncateTables()
	err = errors.WithStack(err)
	return err
}

func (p *ClusterDatabase) initTables() {
	if atomic.LoadUint32(&p.createTablesDone) == 1 {
		return
	}

	// Slow-path.
	p.createTablesMutex.Lock()
	defer p.createTablesMutex.Unlock()

	if atomic.LoadUint32(&p.createTablesDone) == 1 {
		return
	}

	db, err := sql.Open(p.cfg.Type, p.cfg.GetServerAddr())
	if err != nil {
		logger.Warningf("sql.Open failed: %+v", err)
		return
	}
	defer db.Close()

	sqlCreateDbIfNotExists := fmt.Sprintf(
		"CREATE DATABASE IF NOT EXISTS %s DEFAULT CHARACTER SET utf8;",
		p.cfg.DbName,
	)
	if _, err = db.Exec(sqlCreateDbIfNotExists); err != nil {
		logger.Warningf("CREATE DATABASE failed: %+v", err)
		return
	}

	if err := p.dbMap.CreateTablesIfNotExists(); err != nil {
		logger.Warningf("CreateTablesIfNotExists: %+v", err)
		return
	}

	atomic.StoreUint32(&p.createTablesDone, 1)
}

{'fraction_non_alphanumeric': 0.14579365079365078, 'fraction_numerical': 0.00873015873015873, 'mean_word_length': 3.9280406726632773, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2068268', 'n_tokens_mistral': 4503, 'n_tokens_neox': 4352, 'n_words': 1016}