Add the "system_replicas_is_session_expired" metric to the exporter.

pkg/apis/metrics/clickhouse_fetcher.go

@@ -21,6 +21,13 @@ import (
 )
 
 const (
+	querySystemReplicasSQL = `
+	SELECT
+		database,
+		table,
+		toString(is_session_expired) AS is_session_expired
+	FROM system.replicas`
+
 	queryMetricsSQL = `
     SELECT
         concat('metric.', metric) AS metric,
@@ -110,16 +117,14 @@ func (f *ClickHouseFetcher) newConn() *clickhouse.Conn {
 func (f *ClickHouseFetcher) clickHouseQueryMetrics() ([][]string, error) {
 	data := make([][]string, 0)
 	conn := f.newConn()
-	if rows, err := conn.Query(heredoc.Doc(queryMetricsSQL)); err != nil {
+	rows, err := conn.Query(heredoc.Doc(queryMetricsSQL))
+	if err != nil {
 		return nil, err
-	} else {
-		for rows.Next() {
-			var metric, value, description, _type string
-			if err := rows.Scan(&metric, &value, &description, &_type); err == nil {
-				data = append(data, []string{metric, value, description, _type})
-			} else {
-				// Skip erroneous line
-			}
+	}
+	for rows.Next() {
+		var metric, value, description, _type string
+		if err := rows.Scan(&metric, &value, &description, &_type); err == nil {
+			data = append(data, []string{metric, value, description, _type})
 		}
 	}
 	return data, nil
@@ -130,16 +135,32 @@ func (f *ClickHouseFetcher) clickHouseQueryMetrics() ([][]string, error) {
 func (f *ClickHouseFetcher) clickHouseQueryTableSizes() ([][]string, error) {
 	data := make([][]string, 0)
 	conn := f.newConn()
-	if rows, err := conn.Query(heredoc.Doc(queryTableSizesSQL)); err != nil {
+	rows, err := conn.Query(heredoc.Doc(queryTableSizesSQL))
+	if err != nil {
 		return nil, err
-	} else {
-		for rows.Next() {
-			var database, table, partitions, parts, bytes, uncompressed, _rows string
-			if err := rows.Scan(&database, &table, &partitions, &parts, &bytes, &uncompressed, &_rows); err == nil {
-				data = append(data, []string{database, table, partitions, parts, bytes, uncompressed, _rows})
-			} else {
-				// Skip erroneous line
-			}
+	}
+	for rows.Next() {
+		var database, table, partitions, parts, bytes, uncompressed, _rows string
+		if err := rows.Scan(&database, &table, &partitions, &parts, &bytes, &uncompressed, &_rows); err == nil {
+			data = append(data, []string{database, table, partitions, parts, bytes, uncompressed, _rows})
+		}
+	}
+	return data, nil
+}
+
+// clickHouseQuerySystemReplicas requests replica information from the ClickHouse database using REST interface
+// data is a concealed output
+func (f *ClickHouseFetcher) clickHouseQuerySystemReplicas() ([][]string, error) {
+	data := make([][]string, 0)
+	conn := f.newConn()
+	rows, err := conn.Query(heredoc.Doc(querySystemReplicasSQL))
+	if err != nil {
+		return nil, err
+	}
+	for rows.Next() {
+		var database, table, isSessionExpired string
+		if err := rows.Scan(&database, &table, &isSessionExpired); err == nil {
+			data = append(data, []string{database, table, isSessionExpired})
 		}
 	}
 	return data, nil

pkg/apis/metrics/exporter.go

@@ -226,4 +226,15 @@ func (e *Exporter) collectFromHost(chi *WatchedChi, hostname string, c chan<- pr
 		e.enqueueToRemoveFromWatched(chi)
 		return
 	}
+
+	glog.Infof("Querying system replicas for %s\n", hostname)
+	if systemReplicas, err := fetcher.clickHouseQuerySystemReplicas(); err == nil {
+		glog.Infof("Extracted %d system replicas for %s\n", len(systemReplicas), hostname)
+		writer.WriteSystemReplicas(systemReplicas)
+	} else {
+		// In case of an error fetching data from clickhouse store CHI name in e.cleanup
+		glog.Infof("Error querying system replicas for %s: %s\n", hostname, err)
+		e.enqueueToRemoveFromWatched(chi)
+		return
+	}
 }

pkg/apis/metrics/prometheus_writer.go

@@ -15,10 +15,11 @@
 package metrics
 
 import (
-	"github.com/golang/glog"
-	"github.com/prometheus/client_golang/prometheus"
 	"strconv"
 	"strings"
+
+	"github.com/golang/glog"
+	"github.com/prometheus/client_golang/prometheus"
 )
 
 const (
@@ -44,7 +45,7 @@ func NewPrometheusWriter(
 	}
 }
 
-// writeMetricsDataToPrometheus pushes set of prometheus.Metric objects created from the ClickHouse system data
+// WriteMetrics pushes set of prometheus.Metric objects created from the ClickHouse system data
 // Expected data structure: metric, value, description, type (gauge|counter)
 // TODO add namespace handling. It is just skipped for now
 func (w *PrometheusWriter) WriteMetrics(data [][]string) {
@@ -70,7 +71,7 @@ func (w *PrometheusWriter) WriteMetrics(data [][]string) {
 	}
 }
 
-// writeTableSizesDataToPrometheus pushes set of prometheus.Metric objects created from the ClickHouse system data
+// WriteTableSizes pushes set of prometheus.Metric objects created from the ClickHouse system data
 // Expected data structure: database, table, partitions, parts, bytes, uncompressed_bytes, rows
 // TODO add namespace handling. It is just skipped for now
 func (w *PrometheusWriter) WriteTableSizes(data [][]string) {
@@ -96,6 +97,14 @@ func (w *PrometheusWriter) WriteTableSizes(data [][]string) {
 	}
 }
 
+func (w *PrometheusWriter) WriteSystemReplicas(data [][]string) {
+	for _, metric := range data {
+		writeSingleMetricToPrometheus(w.out, "system_replicas_is_session_expired", "Number of expired Zookeeper sessions of the table", metric[2], prometheus.GaugeValue,
+			[]string{"chi", "hostname", "database", "table"},
+			w.chi.Name, w.hostname, metric[0], metric[1])
+	}
+}
+
 func writeSingleMetricToPrometheus(out chan<- prometheus.Metric, name string, desc string, value string, metricType prometheus.ValueType, labels []string, labelValues ...string) {
 	floatValue, _ := strconv.ParseFloat(value, 64)
 	m, err := prometheus.NewConstMetric(

pkg/apis/metrics/rest.go

@@ -26,7 +26,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
-// startMetricsExporter start Prometheus metrics exporter in background
+// StartMetricsREST start Prometheus metrics exporter in background
 func StartMetricsREST(
 	username, password string, port int,
 	metricsAddress, metricsPath string,