feat - changes for c3 tool.

cli/cli.go

@@ -4,6 +4,7 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"os"
 	"strings"
 )
 
@@ -81,6 +82,21 @@ func (c *CLI) Lookup(name string) Arg {
 	return Arg{}
 }
 
+func (c *CLI) IsSet(name string) bool {
+	return len(c.Lookup(name).Value) > 0
+}
+
+func (c *CLI) IsPath(name string) bool {
+	_, err := os.Stat(c.Lookup(name).Value)
+	if err == nil {
+		return true
+	}
+	if os.IsNotExist(err) {
+		return false
+	}
+	return false
+}
+
 func (c *CLI) Parser() {
 	f := flag.NewFlagSet("default", flag.ContinueOnError)
 

clickhouse/main.go

@@ -0,0 +1,159 @@
+package clickhouse
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"math"
+	"math/big"
+	"reflect"
+	"time"
+
+	"github.com/debeando/go-common/log"
+
+	"github.com/ClickHouse/clickhouse-go/v2"
+	"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
+	"github.com/google/uuid"
+)
+
+type Column struct {
+	Name     string
+	Kind     reflect.Type
+	Nullable bool
+}
+
+type Connection struct {
+	Instance driver.Conn
+	Name     string
+}
+
+var instance = make(map[string]*Connection)
+
+func Instance(name string) *Connection {
+	if instance[name] == nil {
+		instance[name] = &Connection{}
+		instance[name].Name = name
+	}
+	return instance[name]
+}
+
+func (c *Connection) Connect(host, port, database, username, password string) error {
+	var err error
+
+	if c.Instance != nil {
+		return errors.New("ClickHouse can't connect because instance is empty.")
+	}
+
+	log.DebugWithFields("ClickHouse connection settings.", log.Fields{
+		"Host":     host,
+		"Port":     port,
+		"Database": database,
+		"Username": username,
+		"Password": password,
+	})
+
+	c.Instance, err = clickhouse.Open(&clickhouse.Options{
+		Addr: []string{fmt.Sprintf("%s:%s", host, port)},
+		Auth: clickhouse.Auth{
+			Database: database,
+			Username: username,
+			Password: password,
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	if err = c.Instance.Ping(context.Background()); err != nil {
+		return err
+	}
+
+	log.Debug("ClickHouse connected!")
+
+	return nil
+}
+
+func (c *Connection) Query(query string) (map[int]map[string]any, error) {
+	log.DebugWithFields("ClickHouse execute.", log.Fields{
+		"Query": query,
+	})
+
+	rows, err := c.Instance.Query(context.Background(), query)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	row_id := 0
+	columns := []Column{}
+	types := rows.ColumnTypes()
+	dataset := make(map[int]map[string]any)
+
+	for _, c := range types {
+		columns = append(columns, Column{
+			Name:     c.Name(),
+			Kind:     c.ScanType(),
+			Nullable: c.Nullable(),
+		})
+	}
+
+	values := make([]any, len(types))
+
+	for i := range types {
+		values[i] = reflect.New(types[i].ScanType()).Interface()
+	}
+
+	for rows.Next() {
+		err = rows.Scan(values...)
+		if err != nil {
+			return nil, err
+		}
+
+		row := make(map[string]any)
+
+		for i, v := range values {
+			switch v := v.(type) {
+			case *bool:
+				row[columns[i].Name] = *v
+			case *string:
+				row[columns[i].Name] = *v
+			case *int8:
+				row[columns[i].Name] = *v
+			case *int16:
+				row[columns[i].Name] = *v
+			case *int32:
+				row[columns[i].Name] = *v
+			case *int64:
+				row[columns[i].Name] = *v
+			case *uint8:
+				row[columns[i].Name] = *v
+			case **uint8:
+				row[columns[i].Name] = *v
+			case *uint16:
+				row[columns[i].Name] = *v
+			case *uint32:
+				row[columns[i].Name] = *v
+			case *uint64:
+				row[columns[i].Name] = *v
+			case **big.Int:
+				row[columns[i].Name] = *v
+			case *float32, *float64:
+				if value := *(v.(*float64)); !math.IsNaN(value) {
+					row[columns[i].Name] = value
+				}
+			case *time.Time:
+				row[columns[i].Name] = v.Format("2006-01-02 15:04:05")
+			case *uuid.UUID:
+				row[columns[i].Name] = *v
+			default:
+				row[columns[i].Name] = nil
+			}
+		}
+
+		dataset[row_id] = row
+
+		row_id++
+	}
+
+	return dataset, nil
+}

file/main.go

@@ -0,0 +1,23 @@
+package file
+
+import (
+	"io/ioutil"
+	"path/filepath"
+)
+
+func Read(path string) []byte {
+	c, _ := ioutil.ReadFile(path)
+	return c
+}
+
+func ReadAsString(path string) string {
+	return string(Read(path))
+}
+
+func Name(n string) string {
+	return n[:len(n)-len(filepath.Ext(n))]
+}
+
+func Dir(path string) string {
+	return filepath.Dir(path)
+}

log/main.go

@@ -6,6 +6,22 @@ import (
 
 type Fields map[string]interface{}
 
+type Level uint32
+
+const (
+	PanicLevel Level = iota
+	FatalLevel
+	ErrorLevel
+	WarnLevel
+	InfoLevel
+	DebugLevel
+	TraceLevel
+)
+
+func SetLevel(level Level) {
+	logrus.SetLevel(logrus.Level(level))
+}
+
 func Info(m string) {
 	logrus.Info(m)
 }

mysql/main.go

@@ -35,9 +35,9 @@ func Instance(name string) *Connection {
 	return instance[name]
 }
 
-func (s *Connection) Connect() error {
-	if s.Instance == nil {
-		conn, err := sql.Open("mysql", s.DSN)
+func (c *Connection) Connect() error {
+	if c.Instance == nil {
+		conn, err := sql.Open("mysql", c.DSN)
 		if err != nil {
 			return err
 		}
@@ -46,22 +46,22 @@ func (s *Connection) Connect() error {
 			return err
 		}
 
-		s.Instance = conn
+		c.Instance = conn
 	}
 	return nil
 }
 
-func (s *Connection) Query(query string) (map[int]map[string]string, error) {
+func (c *Connection) Query(query string) (map[int]map[string]string, error) {
 	log.DebugWithFields("MySQL execute.", log.Fields{
 		"Query": query,
 	})
 
-	if err := s.Instance.Ping(); err != nil {
+	if err := c.Instance.Ping(); err != nil {
 		return nil, err
 	}
 
 	// Execute the query
-	rows, err := s.Instance.Query(query)
+	rows, err := c.Instance.Query(query)
 	if err != nil {
 		return nil, err
 	}
@@ -73,8 +73,8 @@ func (s *Connection) Query(query string) (map[int]map[string]string, error) {
 		return nil, err
 	}
 
-	dataset := make(map[int]map[string]string)
 	row_id := 0
+	dataset := make(map[int]map[string]string)
 	columns := make([]sql.RawBytes, len(cols))
 	columnPointers := make([]interface{}, len(cols))
 
@@ -101,9 +101,9 @@ func (s *Connection) Query(query string) (map[int]map[string]string, error) {
 	return dataset, nil
 }
 
-func (s *Connection) Close() {
-	if s.Instance != nil {
-		s.Instance.Close()
+func (c *Connection) Close() {
+	if c.Instance != nil {
+		c.Instance.Close()
 	}
 }
 

proxysql/main.go

@@ -1,10 +1,12 @@
 package proxysql
 
 import (
+	"errors"
 	"fmt"
 
 	"github.com/debeando/go-common/cast"
 	"github.com/debeando/go-common/mysql"
+	"github.com/debeando/go-common/time"
 )
 
 const (
@@ -113,6 +115,31 @@ func (p *ProxySQL) EnableServer(index int) {
 	p.SaveServers()
 }
 
+func (p *ProxySQL) DisableServer(index int) error {
+	var cntQueries int
+	var stats map[string]string
+
+	p.SetStatusServer(0, OFFLINE_SOFT)
+	p.LoadServers()
+	p.SaveServers()
+
+	time.Sleep(300000)
+
+	p.StatConnectionPoolReset()
+	stats = p.StatConnectionPool(0)
+	cntQueries += cast.StringToInt(stats["Queries"])
+
+	time.Sleep(60000)
+
+	stats = p.StatConnectionPool(0)
+	cntQueries += cast.StringToInt(stats["Queries"])
+
+	if cntQueries > 0 {
+		return errors.New("Active connections on MySQL replica.")
+	}
+	return nil
+}
+
 func (p *ProxySQL) LoadServers() {
 	p.Connection.Query("LOAD MYSQL SERVERS TO RUNTIME;")
 }
@@ -125,10 +152,14 @@ func (p *ProxySQL) StatConnectionPoolReset() {
 	p.Connection.Query("SELECT * FROM stats_mysql_connection_pool_reset;")
 }
 
-func (p *ProxySQL) StatConnectionPool(index int) {
+func (p *ProxySQL) StatConnectionPool(index int) map[string]string {
 	sql := fmt.Sprintf(
 		"SELECT hostgroup, substr(srv_host, 0, instr(srv_host, '.')) AS host, status, ConnUsed, ConnOK, ConnERR, Queries FROM stats_mysql_connection_pool WHERE srv_host = '%s';",
 		p.Servers[index].Hostname,
 	)
-	p.Connection.Query(sql)
+	result, _ := p.Connection.Query(sql)
+	if len(result) == 1 {
+		return result[0]
+	}
+	return nil
 }

template/functions.go

@@ -0,0 +1,12 @@
+package template
+
+func Separator(s string) func() string {
+	i := -1
+	return func() string {
+		i++
+		if i == 0 {
+			return ""
+		}
+		return s
+	}
+}