v3.2.0 ready

.gitignore

@@ -14,6 +14,7 @@ ckmanpasswd
 migrate
 znodefix
 znode_count
+cmd/ckmanctl/ckmanctl
 pkged.go
 static/dist/
 coverage.xml

Dockerfile

@@ -3,10 +3,7 @@ FROM debian:stable-slim
 RUN mkdir -p /etc/ckman && cd /etc/ckman && \
     mkdir bin run logs conf package
 ADD ./ckman /etc/ckman/bin/ckman
-ADD ./migrate /etc/ckman/bin/migrate
-ADD ./ckmanpasswd /etc/ckman/bin/ckmanpasswd
-ADD ./znodefix /etc/ckman/bin/znodefix
-ADD ./znode_count /etc/ckman/bin/znode_count
+ADD ./cmd/ckmanctl/ckmanctl /etc/ckman/bin/ckmanctl
 ADD ./README.md /etc/ckman/package/README.md
 ADD ./resources/ckman.hjson /etc/ckman/conf
 ADD ./resources/migrate.hjson /etc/ckman/conf

Dockerfile.test

@@ -2,8 +2,8 @@
 # You can run command like: "docker build -f Dockerfile.test -t ckman-clickhouse:centos-7 ."
 # the offical image is eoitek/ckman-clickhouse:centos-7, You can pull it from dockerhub.
 
-#FROM centos:7
-FROM ccr.ccs.tencentyun.com/library/centos:7
+FROM centos:7
+#FROM ccr.ccs.tencentyun.com/library/centos:7
 WORKDIR /var/
 
 RUN yum -y update && yum install -y openssh* \

Makefile

@@ -14,6 +14,7 @@ GOARCH?=$(shell go env GOARCH)
 TARNAME=${PKGDIR}-${VERSION}-${DATE}.${OS}.$(GOARCH).tar.gz
 TAG?=$(shell date +%y%m%d)
 LDFLAGS=-ldflags "-X main.BuildTimeStamp=${TIME} -X main.GitCommitHash=${REVISION} -X main.Version=${VERSION}"
+GCFLAGS=-gcflags "all=-N -l"
 PUB_KEY=$(shell cat resources/eoi_public_key.pub 2>/dev/null)
 export GOPROXY=https://goproxy.cn,direct
 
@@ -28,10 +29,13 @@ frontend:
 backend:
 	@rm -rf ${PKGFULLDIR}
 	go build ${LDFLAGS}
-	go build ${LDFLAGS} -o ckmanpasswd cmd/password/password.go
-	go build ${LDFLAGS} -o migrate cmd/migrate/migrate.go
-	go build ${LDFLAGS} -o znodefix cmd/znodefix/znodefix.go
-	go build ${LDFLAGS} -o znode_count cmd/znodecnt/znodecount.go
+	go build ${LDFLAGS} -o cmd/ckmanctl/ckmanctl cmd/ckmanctl/ckmanctl.go
+
+.PHONY: debug
+debug:
+	@rm -rf ${PKGFULLDIR}
+	go build ${GCFLAGS} ${LDFLAGS}
+	go build ${LDFLAGS} -o cmd/ckmanctl/ckmanctl cmd/ckmanctl/ckmanctl.go
 
 .PHONY: pre
 pre:
@@ -66,10 +70,7 @@ package:build
 	@rm -rf ${PKGFULLDIR_TMP}
 	@mkdir -p ${PKGFULLDIR_TMP}/bin ${PKGFULLDIR_TMP}/conf ${PKGFULLDIR_TMP}/run ${PKGFULLDIR_TMP}/logs ${PKGFULLDIR_TMP}/package ${PKGFULLDIR_TMP}/dbscript
 	@mv ${SHDIR}/ckman ${PKGFULLDIR_TMP}/bin
-	@mv ${SHDIR}/ckmanpasswd ${PKGFULLDIR_TMP}/bin
-	@mv ${SHDIR}/migrate ${PKGFULLDIR_TMP}/bin
-	@mv ${SHDIR}/znodefix ${PKGFULLDIR_TMP}/bin
-	@mv ${SHDIR}/znode_count ${PKGFULLDIR_TMP}/bin
+	@mv ${SHDIR}/cmd/ckmanctl/ckmanctl ${PKGFULLDIR_TMP}/binn
 	@cp ${SHDIR}/resources/start ${PKGFULLDIR_TMP}/bin
 	@cp ${SHDIR}/resources/stop ${PKGFULLDIR_TMP}/bin
 	@cp ${SHDIR}/resources/yaml2json.${GOARCH} ${PKGFULLDIR_TMP}/bin/yaml2json

ckconfig/custom.go

@@ -19,6 +19,7 @@ func root(conf *model.CKManClickHouseConfig, ipv6Enable bool) map[string]interfa
 	output["default_replica_path"] = "/clickhouse/tables/{cluster}/{database}/{table}/{shard}"
 	output["default_replica_name"] = "{replica}"
 	output["tcp_port"] = conf.Port
+	output["http_port"] = conf.HttpPort
 	if ipv6Enable {
 		output["listen_host"] = "::"
 	} else {
@@ -173,6 +174,25 @@ func expert(exp map[string]string) map[string]interface{} {
 	return common.ConvertMapping(output)
 }
 
+func query_cache() map[string]interface{} {
+	output := make(map[string]interface{})
+	output["query_cache"] = map[string]interface{}{
+		"max_size_in_bytes":       1073741824,
+		"max_entries":             1024,
+		"max_entry_size_in_bytes": 1048576,
+		"max_entry_size_in_rows":  30000000,
+	}
+	return output
+}
+func merge_tree_metadata_cache() map[string]interface{} {
+	output := make(map[string]interface{})
+	output["merge_tree_metadata_cache"] = map[string]interface{}{
+		"lru_cache_size":        1073741824,
+		"continue_if_corrupted": true,
+	}
+	return output
+}
+
 func GenerateCustomXML(filename string, conf *model.CKManClickHouseConfig, ipv6Enable bool) (string, error) {
 	rootTag := "yandex"
 	if common.CompareClickHouseVersion(conf.Version, "22.x") >= 0 {
@@ -185,6 +205,12 @@ func GenerateCustomXML(filename string, conf *model.CKManClickHouseConfig, ipv6E
 	mergo.Merge(&custom, system_log())
 	mergo.Merge(&custom, distributed_ddl(conf.Cluster))
 	mergo.Merge(&custom, prometheus())
+	if common.CompareClickHouseVersion(conf.Version, "22.4.x") >= 0 {
+		mergo.Merge(&custom, merge_tree_metadata_cache())
+	}
+	if common.CompareClickHouseVersion(conf.Version, "23.4.x") >= 0 {
+		mergo.Merge(&custom, query_cache())
+	}
 	storage_configuration, backups := storage(conf.Storage)
 	mergo.Merge(&custom, storage_configuration)
 	mergo.Merge(&custom, backups)

ckconfig/custom_fake.xml

@@ -50,6 +50,10 @@
         <replicated_fetches_http_receive_timeout>0</replicated_fetches_http_receive_timeout>
         <replicated_fetches_http_send_timeout>0</replicated_fetches_http_send_timeout>
     </merge_tree>
+    <merge_tree_metadata_cache>
+        <continue_if_corrupted>true</continue_if_corrupted>
+        <lru_cache_size>1073741824</lru_cache_size>
+    </merge_tree_metadata_cache>
     <metric_log>
         <flush_interval_milliseconds>30000</flush_interval_milliseconds>
         <partition_by>toYYYYMMDD(event_date)</partition_by>

ckconfig/custom_test.go

@@ -111,7 +111,7 @@ func TestGenerateCustomXML(t *testing.T) {
 		Cwd:       "/home/eoi/clickhouse",
 		NeedSudo:  false,
 		Path:      "/data01/",
-		Version:   "22.3.3.44",
+		Version:   "23.3.3.44",
 	}
 	_, err := GenerateCustomXML("custom_fake.xml", conf, true)
 	assert.Nil(t, err)

ckconfig/keeper.go

@@ -0,0 +1,93 @@
+package ckconfig
+
+import (
+	"path"
+
+	"github.com/housepower/ckman/common"
+	"github.com/housepower/ckman/model"
+	"github.com/imdario/mergo"
+)
+
+func keeper_root(ipv6Enable bool) map[string]interface{} {
+	output := make(map[string]interface{})
+	if ipv6Enable {
+		output["listen_host"] = "::"
+	} else {
+		output["listen_host"] = "0.0.0.0"
+	}
+	output["max_connections"] = 4096
+	return output
+}
+
+func keeper_server(conf *model.CKManClickHouseConfig, ipv6Enable bool, idx int) map[string]interface{} {
+	output := keeper_root(ipv6Enable)
+	output["logger"] = keeper_logger(conf)
+	keeper_server := make(map[string]interface{})
+	mergo.Merge(&keeper_server, conf.KeeperConf.Expert)
+	keeper_server["tcp_port"] = conf.KeeperConf.TcpPort
+	keeper_server["server_id"] = idx
+	keeper_server["log_storage_path"] = conf.KeeperConf.LogPath + "clickhouse/coordination/logs"
+	keeper_server["snapshot_storage_path"] = conf.KeeperConf.SnapshotPath + "clickhouse/coordination/snapshots"
+	keeper_server["coordination_settings"] = coordination_settings(conf.KeeperConf.Coordination)
+	keeper_server["raft_configuration"] = raft_configuration(conf.KeeperConf)
+	output["keeper_server"] = keeper_server
+	return output
+}
+
+// https://github.com/ClickHouse/ClickHouse/blob/master/src/Coordination/CoordinationSettings.h
+func coordination_settings(coordination model.Coordination) map[string]interface{} {
+	output := make(map[string]interface{})
+	mergo.Merge(&output, coordination.Expert)
+	output["operation_timeout_ms"] = coordination.OperationTimeoutMs
+	output["session_timeout_ms"] = coordination.SessionTimeoutMs
+	if coordination.ForceSync {
+		output["force_sync"] = "true"
+	} else {
+		output["force_sync"] = "false"
+	}
+	if coordination.AutoForwarding {
+		output["auto_forwarding"] = "true"
+	} else {
+		output["auto_forwarding"] = "false"
+	}
+	return output
+}
+
+func keeper_logger(conf *model.CKManClickHouseConfig) map[string]interface{} {
+	output := make(map[string]interface{})
+	output["level"] = "debug"
+	output["size"] = "1000M"
+	output["count"] = 10
+	if !conf.NeedSudo {
+		output["log"] = path.Join(conf.Cwd, "log", "clickhouse-keeper", "clickhouse-keeper.log")
+		output["errorlog"] = path.Join(conf.Cwd, "log", "clickhouse-keeper", "clickhouse-keeper.err.log")
+	}
+	return output
+
+}
+
+func raft_configuration(conf *model.KeeperConf) []map[string]interface{} {
+	var outputs []map[string]interface{}
+	for idx, node := range conf.KeeperNodes {
+		output := make(map[string]interface{})
+		output["server"] = map[string]interface{}{
+			"id":       idx + 1,
+			"hostname": node,
+			"port":     conf.RaftPort,
+		}
+		outputs = append(outputs, output)
+	}
+	return outputs
+}
+
+func GenerateKeeperXML(filename string, conf *model.CKManClickHouseConfig, ipv6Enable bool, idx int) (string, error) {
+	xml := common.NewXmlFile(filename)
+	rootTag := "clickhouse"
+	xml.Begin(rootTag)
+	xml.Merge(keeper_server(conf, ipv6Enable, idx))
+	xml.End(rootTag)
+	if err := xml.Dump(); err != nil {
+		return filename, err
+	}
+	return filename, nil
+}

ckconfig/keeper_fake.xml

@@ -0,0 +1,37 @@
+<clickhouse>
+    <keeper_server>
+        <coordination_settings>
+            <auto_forwarding>false</auto_forwarding>
+            <force_sync>true</force_sync>
+            <operation_timeout_ms>10000</operation_timeout_ms>
+            <raft_log_level>Information</raft_log_level>
+            <session_timeout_ms>30000</session_timeout_ms>
+        </coordination_settings>
+        <log_storage_path>/var/lib/clickhouse/coordination/log</log_storage_path>
+        <raft_configuration>
+            <server>
+                <hostname>192.168.101.102</hostname>
+                <id>1</id>
+                <port>9181</port>
+            </server>
+            <server>
+                <hostname>192.168.101.105</hostname>
+                <id>2</id>
+                <port>9181</port>
+            </server>
+            <server>
+                <hostname>192.168.101.107</hostname>
+                <id>3</id>
+                <port>9181</port>
+            </server>
+        </raft_configuration>
+        <server_id>2</server_id>
+        <snapshot_storage_path>/var/lib/clickhouse/coordination/snapshots</snapshot_storage_path>
+        <tcp_port>9181</tcp_port>
+    </keeper_server>
+    <listen_host>::</listen_host>
+    <logger>
+        <level>debug</level>
+    </logger>
+    <max_connections>4096</max_connections>
+</clickhouse>

ckconfig/keeper_test.go

@@ -0,0 +1,27 @@
+package ckconfig
+
+import (
+	"testing"
+
+	"github.com/housepower/ckman/model"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGenerateKeeperXML(t *testing.T) {
+	conf := model.CKManClickHouseConfig{
+		KeeperConf: &model.KeeperConf{
+			KeeperNodes:  []string{"192.168.101.102", "192.168.101.105", "192.168.101.107"},
+			TcpPort:      9181,
+			RaftPort:     9234,
+			LogPath:      "/var/lib/clickhouse/coordination/log",
+			SnapshotPath: "/var/lib/clickhouse/coordination/snapshots",
+			Coordination: model.Coordination{
+				OperationTimeoutMs: 10000,
+				SessionTimeoutMs:   30000,
+				ForceSync:          true,
+			},
+		},
+	}
+	_, err := GenerateKeeperXML("keeper_fake.xml", &conf, true, 2)
+	assert.Nil(t, err)
+}

ckconfig/metrika.go

@@ -3,6 +3,7 @@ package ckconfig
 import (
 	"github.com/housepower/ckman/common"
 	"github.com/housepower/ckman/model"
+	"github.com/housepower/ckman/service/zookeeper"
 )
 
 func GenerateMetrikaXML(filename string, conf *model.CKManClickHouseConfig) (string, error) {
@@ -40,10 +41,11 @@ func GenZookeeperMetrika(indent int, conf *model.CKManClickHouseConfig) string {
 	xml := common.NewXmlFile("")
 	xml.SetIndent(indent)
 	xml.Begin("zookeeper")
-	for index, zk := range conf.ZkNodes {
+	nodes, port := zookeeper.GetZkInfo(conf)
+	for index, zk := range nodes {
 		xml.BeginwithAttr("node", []common.XMLAttr{{Key: "index", Value: index + 1}})
 		xml.Write("host", zk)
-		xml.Write("port", conf.ZkPort)
+		xml.Write("port", port)
 		xml.End("node")
 	}
 	xml.End("zookeeper")
@@ -59,25 +61,6 @@ func GenLocalMetrika(indent int, conf *model.CKManClickHouseConfig) string {
 		secret = false
 	}
 	if secret {
-		xml.Comment(`Inter-server per-cluster secret for Distributed queries
-                 default: no secret (no authentication will be performed)
-
-                 If set, then Distributed queries will be validated on shards, so at least:
-                 - such cluster should exist on the shard,
-                 - such cluster should have the same secret.
-
-                 And also (and which is more important), the initial_user will
-                 be used as current user for the query.
-
-                 Right now the protocol is pretty simple and it only takes into account:
-                 - cluster name
-                 - query
-
-                 Also it will be nice if the following will be implemented:
-                 - source hostname (see interserver_http_host), but then it will depends from DNS,
-                   it can use IP address instead, but then the you need to get correct on the initiator node.
-                 - target hostname / ip address (same notes as for source hostname)
-                 - time-based security tokens`)
 		xml.Write("secret", "foo")
 	}
 	for _, shard := range conf.Shards {