clickhouse_test.go

package clickhouse

import (
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/influxdata/telegraf/testutil"
)

func TestClusterIncludeExcludeFilter(t *testing.T) {
	ch := ClickHouse{}
	require.Equal(t, "", ch.clusterIncludeExcludeFilter())
	ch.ClusterExclude = []string{"test_cluster"}
	require.Equal(t, "WHERE cluster NOT IN ('test_cluster')", ch.clusterIncludeExcludeFilter())

	ch.ClusterExclude = []string{"test_cluster"}
	ch.ClusterInclude = []string{"cluster"}
	require.Equal(t, "WHERE cluster IN ('cluster') OR cluster NOT IN ('test_cluster')", ch.clusterIncludeExcludeFilter())

	ch.ClusterExclude = []string{}
	ch.ClusterInclude = []string{"cluster1", "cluster2"}
	require.Equal(t, "WHERE cluster IN ('cluster1', 'cluster2')", ch.clusterIncludeExcludeFilter())

	ch.ClusterExclude = []string{"cluster1", "cluster2"}
	ch.ClusterInclude = []string{}
	require.Equal(t, "WHERE cluster NOT IN ('cluster1', 'cluster2')", ch.clusterIncludeExcludeFilter())
}

func TestChInt64(t *testing.T) {
	assets := map[string]uint64{
		`"1"`:                  1,
		"1":                    1,
		"42":                   42,
		`"42"`:                 42,
		"18446743937525109187": 18446743937525109187,
	}
	for src, expected := range assets {
		var v chUInt64
		err := v.UnmarshalJSON([]byte(src))
		require.NoError(t, err)
		require.Equal(t, expected, uint64(v))
	}
}

func TestGather(t *testing.T) {
	var (
		ts = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			type result struct {
				Data interface{} `json:"data"`
			}
			enc := json.NewEncoder(w)
			switch query := r.URL.Query().Get("query"); {
			case strings.Contains(query, "system.parts"):
				err := enc.Encode(result{
					Data: []struct {
						Database string   `json:"database"`
						Table    string   `json:"table"`
						Bytes    chUInt64 `json:"bytes"`
						Parts    chUInt64 `json:"parts"`
						Rows     chUInt64 `json:"rows"`
					}{
						{
							Database: "test_database",
							Table:    "test_table",
							Bytes:    1,
							Parts:    10,
							Rows:     100,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "system.events"):
				err := enc.Encode(result{
					Data: []struct {
						Metric string   `json:"metric"`
						Value  chUInt64 `json:"value"`
					}{
						{
							Metric: "TestSystemEvent",
							Value:  1000,
						},
						{
							Metric: "TestSystemEvent2",
							Value:  2000,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "system.metrics"):
				err := enc.Encode(result{
					Data: []struct {
						Metric string   `json:"metric"`
						Value  chUInt64 `json:"value"`
					}{
						{
							Metric: "TestSystemMetric",
							Value:  1000,
						},
						{
							Metric: "TestSystemMetric2",
							Value:  2000,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "system.asynchronous_metrics"):
				err := enc.Encode(result{
					Data: []struct {
						Metric string   `json:"metric"`
						Value  chUInt64 `json:"value"`
					}{
						{
							Metric: "TestSystemAsynchronousMetric",
							Value:  1000,
						},
						{
							Metric: "TestSystemAsynchronousMetric2",
							Value:  2000,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "zk_exists"):
				err := enc.Encode(result{
					Data: []struct {
						ZkExists chUInt64 `json:"zk_exists"`
					}{
						{
							ZkExists: 1,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "zk_root_nodes"):
				err := enc.Encode(result{
					Data: []struct {
						ZkRootNodes chUInt64 `json:"zk_root_nodes"`
					}{
						{
							ZkRootNodes: 2,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "replication_queue_exists"):
				err := enc.Encode(result{
					Data: []struct {
						ReplicationQueueExists chUInt64 `json:"replication_queue_exists"`
					}{
						{
							ReplicationQueueExists: 1,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "replication_too_many_tries_replicas"):
				err := enc.Encode(result{
					Data: []struct {
						TooManyTriesReplicas chUInt64 `json:"replication_too_many_tries_replicas"`
						NumTriesReplicas     chUInt64 `json:"replication_num_tries_replicas"`
					}{
						{
							TooManyTriesReplicas: 10,
							NumTriesReplicas:     100,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "system.detached_parts"):
				err := enc.Encode(result{
					Data: []struct {
						DetachedParts chUInt64 `json:"detached_parts"`
					}{
						{
							DetachedParts: 10,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "system.dictionaries"):
				err := enc.Encode(result{
					Data: []struct {
						Origin         string   `json:"origin"`
						Status         string   `json:"status"`
						BytesAllocated chUInt64 `json:"bytes_allocated"`
					}{
						{
							Origin:         "default.test_dict",
							Status:         "NOT_LOADED",
							BytesAllocated: 100,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "system.mutations"):
				err := enc.Encode(result{
					Data: []struct {
						Failed    chUInt64 `json:"failed"`
						Completed chUInt64 `json:"completed"`
						Running   chUInt64 `json:"running"`
					}{
						{
							Failed:    10,
							Running:   1,
							Completed: 100,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "system.disks"):
				err := enc.Encode(result{
					Data: []struct {
						Name            string   `json:"name"`
						Path            string   `json:"path"`
						FreePercent     chUInt64 `json:"free_space_percent"`
						KeepFreePercent chUInt64 `json:"keep_free_space_percent"`
					}{
						{
							Name:            "default",
							Path:            "/var/lib/clickhouse",
							FreePercent:     1,
							KeepFreePercent: 10,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "system.processes"):
				err := enc.Encode(result{
					Data: []struct {
						QueryType      string  `json:"query_type"`
						Percentile50   float64 `json:"p50"`
						Percentile90   float64 `json:"p90"`
						LongestRunning float64 `json:"longest_running"`
					}{
						{
							QueryType:      "select",
							Percentile50:   0.1,
							Percentile90:   0.5,
							LongestRunning: 10,
						},
						{
							QueryType:      "insert",
							Percentile50:   0.2,
							Percentile90:   1.5,
							LongestRunning: 100,
						},
						{
							QueryType:      "other",
							Percentile50:   0.4,
							Percentile90:   4.5,
							LongestRunning: 1000,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "text_log_exists"):
				err := enc.Encode(result{
					Data: []struct {
						TextLogExists chUInt64 `json:"text_log_exists"`
					}{
						{
							TextLogExists: 1,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "system.text_log"):
				err := enc.Encode(result{
					Data: []struct {
						Level                 string   `json:"level"`
						LastMessagesLast10Min chUInt64 `json:"messages_last_10_min"`
					}{
						{
							Level:                 "Fatal",
							LastMessagesLast10Min: 0,
						},
						{
							Level:                 "Critical",
							LastMessagesLast10Min: 10,
						},
						{
							Level:                 "Error",
							LastMessagesLast10Min: 20,
						},
						{
							Level:                 "Warning",
							LastMessagesLast10Min: 30,
						},
						{
							Level:                 "Notice",
							LastMessagesLast10Min: 40,
						},
					},
				})
				require.NoError(t, err)
			}
		}))
		ch = &ClickHouse{
			Servers: []string{
				ts.URL,
			},
		}
		acc = &testutil.Accumulator{}
	)
	defer ts.Close()
	require.NoError(t, ch.Gather(acc))

	acc.AssertContainsTaggedFields(t, "clickhouse_tables",
		map[string]interface{}{
			"bytes": uint64(1),
			"parts": uint64(10),
			"rows":  uint64(100),
		},
		map[string]string{
			"source":   "127.0.0.1",
			"table":    "test_table",
			"database": "test_database",
		},
	)
	acc.AssertContainsFields(t, "clickhouse_events",
		map[string]interface{}{
			"test_system_event":  uint64(1000),
			"test_system_event2": uint64(2000),
		},
	)
	acc.AssertContainsFields(t, "clickhouse_metrics",
		map[string]interface{}{
			"test_system_metric":  uint64(1000),
			"test_system_metric2": uint64(2000),
		},
	)
	acc.AssertContainsFields(t, "clickhouse_asynchronous_metrics",
		map[string]interface{}{
			"test_system_asynchronous_metric":  float64(1000),
			"test_system_asynchronous_metric2": float64(2000),
		},
	)
	acc.AssertContainsFields(t, "clickhouse_zookeeper",
		map[string]interface{}{
			"root_nodes": uint64(2),
		},
	)
	acc.AssertContainsFields(t, "clickhouse_replication_queue",
		map[string]interface{}{
			"too_many_tries_replicas": uint64(10),
			"num_tries_replicas":      uint64(100),
		},
	)
	acc.AssertContainsFields(t, "clickhouse_detached_parts",
		map[string]interface{}{
			"detached_parts": uint64(10),
		},
	)
	acc.AssertContainsTaggedFields(t, "clickhouse_dictionaries",
		map[string]interface{}{
			"is_loaded":       uint64(0),
			"bytes_allocated": uint64(100),
		},
		map[string]string{
			"source":      "127.0.0.1",
			"dict_origin": "default.test_dict",
		},
	)
	acc.AssertContainsFields(t, "clickhouse_mutations",
		map[string]interface{}{
			"running":   uint64(1),
			"failed":    uint64(10),
			"completed": uint64(100),
		},
	)
	acc.AssertContainsTaggedFields(t, "clickhouse_disks",
		map[string]interface{}{
			"free_space_percent":      uint64(1),
			"keep_free_space_percent": uint64(10),
		},
		map[string]string{
			"source": "127.0.0.1",
			"name":   "default",
			"path":   "/var/lib/clickhouse",
		},
	)
	acc.AssertContainsTaggedFields(t, "clickhouse_processes",
		map[string]interface{}{
			"percentile_50":   0.1,
			"percentile_90":   0.5,
			"longest_running": float64(10),
		},
		map[string]string{
			"source":     "127.0.0.1",
			"query_type": "select",
		},
	)

	acc.AssertContainsTaggedFields(t, "clickhouse_processes",
		map[string]interface{}{
			"percentile_50":   0.2,
			"percentile_90":   1.5,
			"longest_running": float64(100),
		},
		map[string]string{
			"source":     "127.0.0.1",
			"query_type": "insert",
		},
	)
	acc.AssertContainsTaggedFields(t, "clickhouse_processes",
		map[string]interface{}{
			"percentile_50":   0.4,
			"percentile_90":   4.5,
			"longest_running": float64(1000),
		},
		map[string]string{
			"source":     "127.0.0.1",
			"query_type": "other",
		},
	)

	for i, level := range []string{"Fatal", "Critical", "Error", "Warning", "Notice"} {
		acc.AssertContainsTaggedFields(t, "clickhouse_text_log",
			map[string]interface{}{
				"messages_last_10_min": uint64(i * 10),
			},
			map[string]string{
				"source": "127.0.0.1",
				"level":  level,
			},
		)
	}
}

func TestGatherWithSomeTablesNotExists(t *testing.T) {
	var (
		ts = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			type result struct {
				Data interface{} `json:"data"`
			}
			enc := json.NewEncoder(w)
			switch query := r.URL.Query().Get("query"); {
			case strings.Contains(query, "zk_exists"):
				err := enc.Encode(result{
					Data: []struct {
						ZkExists chUInt64 `json:"zk_exists"`
					}{
						{
							ZkExists: 0,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "replication_queue_exists"):
				err := enc.Encode(result{
					Data: []struct {
						ReplicationQueueExists chUInt64 `json:"replication_queue_exists"`
					}{
						{
							ReplicationQueueExists: 0,
						},
					},
				})
				require.NoError(t, err)
			case strings.Contains(query, "text_log_exists"):
				err := enc.Encode(result{
					Data: []struct {
						TextLogExists chUInt64 `json:"text_log_exists"`
					}{
						{
							TextLogExists: 0,
						},
					},
				})
				require.NoError(t, err)
			}
		}))
		ch = &ClickHouse{
			Servers: []string{
				ts.URL,
			},
			Username: "default",
		}
		acc = &testutil.Accumulator{}
	)
	defer ts.Close()
	require.NoError(t, ch.Gather(acc))

	acc.AssertDoesNotContainMeasurement(t, "clickhouse_zookeeper")
	acc.AssertDoesNotContainMeasurement(t, "clickhouse_replication_queue")
	acc.AssertDoesNotContainMeasurement(t, "clickhouse_text_log")
}

func TestWrongJSONMarshalling(t *testing.T) {
	var (
		ts = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			type result struct {
				Data interface{} `json:"data"`
			}
			enc := json.NewEncoder(w)
			//wrong data section json
			err := enc.Encode(result{
				Data: []struct{}{},
			})
			require.NoError(t, err)
		}))
		ch = &ClickHouse{
			Servers: []string{
				ts.URL,
			},
			Username: "default",
		}
		acc = &testutil.Accumulator{}
	)
	defer ts.Close()
	require.NoError(t, ch.Gather(acc))

	require.Equal(t, 0, len(acc.Metrics))
	allMeasurements := []string{
		"clickhouse_events",
		"clickhouse_metrics",
		"clickhouse_asynchronous_metrics",
		"clickhouse_tables",
		"clickhouse_zookeeper",
		"clickhouse_replication_queue",
		"clickhouse_detached_parts",
		"clickhouse_dictionaries",
		"clickhouse_mutations",
		"clickhouse_disks",
		"clickhouse_processes",
		"clickhouse_text_log",
	}
	require.GreaterOrEqual(t, len(allMeasurements), len(acc.Errors))
}

func TestOfflineServer(t *testing.T) {
	var (
		acc = &testutil.Accumulator{}
		ch  = &ClickHouse{
			Servers: []string{
				"http://wrong-domain.local:8123",
			},
			Username: "default",
			HTTPClient: http.Client{
				Timeout: 1 * time.Millisecond,
			},
		}
	)
	require.NoError(t, ch.Gather(acc))

	require.Equal(t, 0, len(acc.Metrics))
	allMeasurements := []string{
		"clickhouse_events",
		"clickhouse_metrics",
		"clickhouse_asynchronous_metrics",
		"clickhouse_tables",
		"clickhouse_zookeeper",
		"clickhouse_replication_queue",
		"clickhouse_detached_parts",
		"clickhouse_dictionaries",
		"clickhouse_mutations",
		"clickhouse_disks",
		"clickhouse_processes",
		"clickhouse_text_log",
	}
	require.GreaterOrEqual(t, len(allMeasurements), len(acc.Errors))
}

func TestAutoDiscovery(t *testing.T) {
	var (
		ts = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			type result struct {
				Data interface{} `json:"data"`
			}
			enc := json.NewEncoder(w)
			query := r.URL.Query().Get("query")
			if strings.Contains(query, "system.clusters") {
				err := enc.Encode(result{
					Data: []struct {
						Cluster  string   `json:"test"`
						Hostname string   `json:"localhost"`
						ShardNum chUInt64 `json:"shard_num"`
					}{
						{
							Cluster:  "test_database",
							Hostname: "test_table",
							ShardNum: 1,
						},
					},
				})
				require.NoError(t, err)
			}
		}))
		ch = &ClickHouse{
			Servers: []string{
				ts.URL,
			},
			Username:      "default",
			AutoDiscovery: true,
		}
		acc = &testutil.Accumulator{}
	)
	defer ts.Close()
	require.NoError(t, ch.Gather(acc))
}