3 Commits

Author SHA1 Message Date
97e6b8e124 Update module path
Signed-off-by: Nikolaos Karaolidis <nick@karaolidis.com>
2023-07-06 15:39:45 +03:00
7c9a6cc6e0 Optimize NULL handling
Signed-off-by: Nikolaos Karaolidis <nick@karaolidis.com>
2023-07-06 14:35:11 +03:00
318155561b Remove single-table mode
Signed-off-by: Nikolaos Karaolidis <nick@karaolidis.com>
2023-07-06 12:26:35 +03:00
5 changed files with 18 additions and 35 deletions

21
.gitignore vendored
View File

@@ -1,21 +0,0 @@
# If you prefer the allow list template instead of the deny list, see community template:
# https://github.com/github/gitignore/blob/main/community/Golang/Go.AllowList.gitignore
#
# Binaries for programs and plugins
*.exe
*.exe~
*.dll
*.so
*.dylib
# Test binary, built with `go test -c`
*.test
# Output of the go coverage tool, specifically when used with LiteIDE
*.out
# Dependency directories (remove the comment below to include it)
# vendor/
# Go workspace file
go.work

View File

@@ -18,6 +18,8 @@ build:
expire_in: 1d expire_in: 1d
reports: reports:
dotenv: job.env dotenv: job.env
rules:
- if: "$CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH"
release: release:
image: registry.gitlab.com/gitlab-org/release-cli image: registry.gitlab.com/gitlab-org/release-cli
@@ -38,3 +40,5 @@ release:
url: "https://git.karaolidis.com/karaolidis/telegraf-clickhouse-plugin/-/jobs/$JOB_ID/artifacts/raw/telegraf-clickhouse-plugin-linux-amd64" url: "https://git.karaolidis.com/karaolidis/telegraf-clickhouse-plugin/-/jobs/$JOB_ID/artifacts/raw/telegraf-clickhouse-plugin-linux-amd64"
- name: "telegraf-clickhouse-plugin-linux-arm64" - name: "telegraf-clickhouse-plugin-linux-arm64"
url: "https://git.karaolidis.com/karaolidis/telegraf-clickhouse-plugin/-/jobs/$JOB_ID/artifacts/raw/telegraf-clickhouse-plugin-linux-arm64" url: "https://git.karaolidis.com/karaolidis/telegraf-clickhouse-plugin/-/jobs/$JOB_ID/artifacts/raw/telegraf-clickhouse-plugin-linux-arm64"
rules:
- if: "$CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH"

View File

@@ -6,7 +6,7 @@ import (
"os" "os"
"time" "time"
_ "github.com/karaolidis/telegraf-clickhouse-plugin/plugins/outputs/clickhouse" _ "git.karaolidis.com/karaolidis/telegraf-clickhouse-plugin/plugins/outputs/clickhouse"
"github.com/influxdata/telegraf/plugins/common/shim" "github.com/influxdata/telegraf/plugins/common/shim"
) )

2
go.mod
View File

@@ -1,4 +1,4 @@
module github.com/karaolidis/telegraf-clickhouse-plugin module git.karaolidis.com/karaolidis/telegraf-clickhouse-plugin
go 1.20 go 1.20

View File

@@ -158,11 +158,12 @@ func (ch *ClickHouse) toDatatype(value interface{}) string {
return datatype return datatype
} }
func (ch *ClickHouse) toNullable(pair *orderedmap.Pair[string, string]) string { func (ch *ClickHouse) toNullableDatatype(key string, value interface{}) string {
if pair.Key != "host" && pair.Key != ch.TimestampColumn && pair.Key != "measurement" { datatype := ch.toDatatype(value)
return fmt.Sprintf("Nullable(%s)", pair.Value) if key != "host" && key != ch.TimestampColumn && key != "measurement" {
return fmt.Sprintf("Nullable(%s)", datatype)
} }
return pair.Value return datatype
} }
func (ch *ClickHouse) pepareMetrics(metrics []telegraf.Metric, metricsData map[string][]map[string]interface{}, columns map[string]*orderedmap.OrderedMap[string, string]) { func (ch *ClickHouse) pepareMetrics(metrics []telegraf.Metric, metricsData map[string][]map[string]interface{}, columns map[string]*orderedmap.OrderedMap[string, string]) {
@@ -188,12 +189,12 @@ func (ch *ClickHouse) pepareMetrics(metrics []telegraf.Metric, metricsData map[s
for _, tag := range metric.TagList() { for _, tag := range metric.TagList() {
metricEntry[tag.Key] = tag.Value metricEntry[tag.Key] = tag.Value
columns[tablename].Set(tag.Key, ch.toDatatype(tag.Value)) columns[tablename].Set(tag.Key, ch.toNullableDatatype(tag.Key, tag.Value))
} }
for _, field := range metric.FieldList() { for _, field := range metric.FieldList() {
metricEntry[field.Key] = field.Value metricEntry[field.Key] = field.Value
columns[tablename].Set(field.Key, ch.toDatatype(field.Value)) columns[tablename].Set(field.Key, ch.toNullableDatatype(field.Key, field.Value))
} }
metricsData[tablename] = append(metricsData[tablename], metricEntry) metricsData[tablename] = append(metricsData[tablename], metricEntry)
@@ -203,10 +204,10 @@ func (ch *ClickHouse) pepareMetrics(metrics []telegraf.Metric, metricsData map[s
func (ch *ClickHouse) generateCreateTable(tablename string, columns *orderedmap.OrderedMap[string, string]) string { func (ch *ClickHouse) generateCreateTable(tablename string, columns *orderedmap.OrderedMap[string, string]) string {
columnDefs := make([]string, 0, columns.Len()) columnDefs := make([]string, 0, columns.Len())
for pair := columns.Oldest(); pair != nil; pair = pair.Next() { for pair := columns.Oldest(); pair != nil; pair = pair.Next() {
columnDefs = append(columnDefs, fmt.Sprintf("%s %s", quoteIdent(pair.Key), ch.toNullable(pair))) columnDefs = append(columnDefs, fmt.Sprintf("%s %s", quoteIdent(pair.Key), pair.Value))
} }
orderBy := make([]string, 0, 3) orderBy := make([]string, 0, 2)
if _, ok := columns.Get("host"); ok { if _, ok := columns.Get("host"); ok {
orderBy = append(orderBy, "host") orderBy = append(orderBy, "host")
} }
@@ -230,15 +231,14 @@ func (ch *ClickHouse) generateAlterTable(tablename string, columns *orderedmap.O
modifyDefs := make([]string, 0, columns.Len()) modifyDefs := make([]string, 0, columns.Len())
for pair := columns.Oldest(); pair != nil; pair = pair.Next() { for pair := columns.Oldest(); pair != nil; pair = pair.Next() {
columnType := ch.toNullable(pair)
alterDefs = append(alterDefs, fmt.Sprintf("ADD COLUMN IF NOT EXISTS %s %s", alterDefs = append(alterDefs, fmt.Sprintf("ADD COLUMN IF NOT EXISTS %s %s",
quoteIdent(pair.Key), quoteIdent(pair.Key),
columnType)) pair.Value))
modifyDefs = append(modifyDefs, fmt.Sprintf("MODIFY COLUMN IF EXISTS %s %s", modifyDefs = append(modifyDefs, fmt.Sprintf("MODIFY COLUMN IF EXISTS %s %s",
quoteIdent(pair.Key), quoteIdent(pair.Key),
columnType)) pair.Value))
} }
return fmt.Sprintf("ALTER TABLE %s %s, %s", return fmt.Sprintf("ALTER TABLE %s %s, %s",
@@ -286,7 +286,7 @@ func (ch *ClickHouse) ensureTable(tablename string, columns *orderedmap.OrderedM
} }
for pair := columns.Oldest(); pair != nil; pair = pair.Next() { for pair := columns.Oldest(); pair != nil; pair = pair.Next() {
if _, ok := tableColumns[pair.Key]; !ok || tableColumns[pair.Key] != ch.toNullable(pair) { if _, ok := tableColumns[pair.Key]; !ok || tableColumns[pair.Key] != pair.Value {
_, err = ch.db.Exec(ch.generateAlterTable(tablename, columns)) _, err = ch.db.Exec(ch.generateAlterTable(tablename, columns))
if err != nil { if err != nil {
return err return err