Merge branch 'main' into dependabot/go_modules/ci/it/github.com/ClickHouse/ch-go-0.65.0

Author: trzysiek

docs/public/package-lock.json

@@ -6,7 +6,6 @@ package backend_connectors
 import (
 	"bytes"
 	"context"
-	"crypto/tls"
 	"fmt"
 	"github.com/QuesmaOrg/quesma/platform/config"
 	"github.com/QuesmaOrg/quesma/platform/elasticsearch"
@@ -32,32 +31,29 @@ type ElasticsearchBackendConnector struct {
 
 // NewElasticsearchBackendConnector is a constructor which uses old (v1) configuration object
 func NewElasticsearchBackendConnector(cfg config.ElasticsearchConfiguration) *ElasticsearchBackendConnector {
+	client := elasticsearch.NewHttpsClient(&cfg, esRequestTimeout)
 	conn := &ElasticsearchBackendConnector{
 		config: cfg,
-		client: &http.Client{
-			Transport: &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-			},
-			Timeout: esRequestTimeout,
-		},
+		client: client,
 	}
 	return conn
 }
 
 // NewElasticsearchBackendConnectorFromDbConfig is an alternative constructor which uses the generic database configuration object
 func NewElasticsearchBackendConnectorFromDbConfig(cfg config.RelationalDbConfiguration) *ElasticsearchBackendConnector {
+	esConfig := config.ElasticsearchConfiguration{
+		Url:            cfg.Url,
+		User:           cfg.User,
+		Password:       cfg.Password,
+		ClientCertPath: cfg.ClientCertPath,
+		ClientKeyPath:  cfg.ClientKeyPath,
+		CACertPath:     cfg.CACertPath,
+	}
+
+	client := elasticsearch.NewHttpsClient(&esConfig, esRequestTimeout)
 	conn := &ElasticsearchBackendConnector{
-		config: config.ElasticsearchConfiguration{
-			Url:      cfg.Url,
-			User:     cfg.User,
-			Password: cfg.Password,
-		},
-		client: &http.Client{
-			Transport: &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-			},
-			Timeout: esRequestTimeout,
-		},
+		config: esConfig,
+		client: client,
 	}
 	return conn
 }

platform/backend_connectors/elasticsearch_backend_connector.go

@@ -137,6 +137,7 @@ type discoveredTable struct {
 	createTableQuery   string
 	timestampFieldName string
 	virtualTable       bool
+	existsOnAllNodes   bool // => table can be queried using `FROM cluster(<clusterName>,<tableName>)` syntax
 }
 
 func (lm *LogManager) ReloadTables() {

platform/clickhouse/clickhouse.go

@@ -28,7 +28,8 @@ type Table struct {
 
 	DiscoveredTimestampFieldName *string
 
-	VirtualTable bool
+	VirtualTable     bool
+	ExistsOnAllNodes bool
 }
 
 func (t *Table) createTableOurFieldsString() []string {

platform/clickhouse/table.go

@@ -217,6 +217,17 @@ func (td *tableDiscovery) ReloadTableDefinitions() {
 			configuredTables = td.configureTables(tables, databaseName)
 		}
 	}
+	var tablePresence map[string][]TablePresence
+	if td.cfg.ClusterName != "" {
+		var err error
+		tablePresence, err = td.getTablePresenceAcrossClusters(databaseName)
+		if err != nil {
+			logger.Warn().Msgf("could not get table presence across clusters: %v", err)
+		}
+		logger.Info().Msgf("Table presence across clusters: %v", tablePresence)
+		configuredTables = td.populateClusterNodes(configuredTables, databaseName, tablePresence)
+	}
+
 	configuredTables = td.readVirtualTables(configuredTables)
 
 	td.ReloadTablesError = nil
@@ -328,7 +339,7 @@ func (td *tableDiscovery) configureTables(tables map[string]map[string]columnMet
 				comment := td.tableComment(databaseName, table)
 				createTableQuery := td.createTableQuery(databaseName, table)
 				// we assume here that @timestamp field is always present in the table, or it's explicitly configured
-				configuredTables[table] = discoveredTable{table, databaseName, columns, indexConfig, comment, createTableQuery, "", false}
+				configuredTables[table] = discoveredTable{table, databaseName, columns, indexConfig, comment, createTableQuery, "", false, false}
 			}
 		} else {
 			notConfiguredTables = append(notConfiguredTables, table)
@@ -358,7 +369,7 @@ func (td *tableDiscovery) autoConfigureTables(tables map[string]map[string]colum
 			maybeTimestampField = td.tableTimestampField(databaseName, table, ClickHouse)
 		}
 		const isVirtualTable = false
-		configuredTables[table] = discoveredTable{table, databaseName, columns, config.IndexConfiguration{}, comment, createTableQuery, maybeTimestampField, isVirtualTable}
+		configuredTables[table] = discoveredTable{table, databaseName, columns, config.IndexConfiguration{}, comment, createTableQuery, maybeTimestampField, isVirtualTable, false}
 
 	}
 	for tableName, table := range configuredTables {
@@ -415,6 +426,7 @@ func (td *tableDiscovery) populateTableDefinitions(configuredTables map[string]d
 				CreateTableQuery:             resTable.createTableQuery,
 				DiscoveredTimestampFieldName: timestampFieldName,
 				VirtualTable:                 resTable.virtualTable,
+				ExistsOnAllNodes:             resTable.existsOnAllNodes,
 			}
 			if containsAttributes(resTable.columnTypes) {
 				table.Config.Attributes = []Attribute{NewDefaultStringAttribute()}
@@ -699,6 +711,90 @@ func (td *tableDiscovery) enrichTableWithMapFields(inputTable map[string]map[str
 	return outputTable
 }
 
+type TablePresence struct {
+	Database         string
+	Table            string
+	FoundNodes       int
+	TotalNodes       int
+	ExistsOnAllNodes bool
+}
+
+func (td *tableDiscovery) getTablePresenceAcrossClusters(database string) (map[string][]TablePresence, error) {
+	// Step 1: Get all cluster names
+	clusterQuery := `SELECT DISTINCT cluster FROM system.clusters ORDER BY cluster`
+	rows, err := td.dbConnPool.Query(context.Background(), clusterQuery)
+	if err != nil {
+		return nil, fmt.Errorf("failed to query cluster list: %w", err)
+	}
+	defer rows.Close()
+
+	var clusters []string
+	for rows.Next() {
+		var cluster string
+		if err := rows.Scan(&cluster); err != nil {
+			return nil, fmt.Errorf("failed to scan cluster name: %w", err)
+		}
+		clusters = append(clusters, cluster)
+	}
+
+	// Step 2: For each cluster, safely query for the given database
+	presenceData := make(map[string][]TablePresence)
+
+	for _, cluster := range clusters {
+		query := `
+            WITH (
+                SELECT count(DISTINCT host_name)
+                FROM system.clusters
+                WHERE cluster = ?
+            ) AS total_nodes
+
+            SELECT
+                database,
+                name AS table_name,
+                count(DISTINCT hostName()) AS found_nodes,
+                total_nodes,
+                count(DISTINCT hostName()) = total_nodes AS exists_on_all_nodes
+            FROM cluster(?, system.tables)
+            WHERE database = ?
+            GROUP BY database, name, total_nodes
+        `
+
+		rows, err := td.dbConnPool.Query(context.Background(), query, cluster, cluster, database)
+		if err != nil {
+			return nil, fmt.Errorf("failed to query tables for cluster %s: %w", cluster, err)
+		}
+		defer rows.Close()
+
+		var tables []TablePresence
+		for rows.Next() {
+			var tp TablePresence
+			if err := rows.Scan(&tp.Database, &tp.Table, &tp.FoundNodes, &tp.TotalNodes, &tp.ExistsOnAllNodes); err != nil {
+				return nil, fmt.Errorf("failed to scan table row: %w", err)
+			}
+			tables = append(tables, tp)
+		}
+
+		if len(tables) > 0 {
+			presenceData[cluster] = tables
+		}
+	}
+
+	return presenceData, nil
+}
+
+func (td *tableDiscovery) populateClusterNodes(configuredTables map[string]discoveredTable, databaseName string, tablePresence map[string][]TablePresence) map[string]discoveredTable {
+	for _, tables := range tablePresence {
+		for _, table := range tables {
+			if table.Database == databaseName {
+				if discoTable, ok := configuredTables[table.Table]; ok {
+					discoTable.existsOnAllNodes = table.ExistsOnAllNodes
+				}
+			}
+		}
+	}
+	return configuredTables
+}
+
 func (td *tableDiscovery) readTables(database string) (map[string]map[string]columnMetadata, error) {
 	logger.Debug().Msgf("describing tables: %s", database)
 

platform/clickhouse/table_discovery.go

@@ -94,6 +94,11 @@ type RelationalDbConfiguration struct {
 	ClusterName   string `koanf:"clusterName"` // When creating tables by Quesma - they'll use `ON CLUSTER ClusterName` clause
 	AdminUrl      *Url   `koanf:"adminUrl"`
 	DisableTLS    bool   `koanf:"disableTLS"`
+
+	// This supports es backend only.
+	ClientCertPath string `koanf:"clientCertPath"`
+	ClientKeyPath  string `koanf:"clientKeyPath"`
+	CACertPath     string `koanf:"caCertPath"`
 }
 
 func (c *RelationalDbConfiguration) IsEmpty() bool {
@@ -1069,9 +1074,12 @@ func (c *QuesmaNewConfiguration) getRelationalDBBackendConnector() (*BackendConn
 func (c *QuesmaNewConfiguration) getElasticsearchConfig() (ElasticsearchConfiguration, error) {
 	if esBackendConn := c.getElasticsearchBackendConnector(); esBackendConn != nil {
 		return ElasticsearchConfiguration{
-			Url:      esBackendConn.Config.Url,
-			User:     esBackendConn.Config.User,
-			Password: esBackendConn.Config.Password,
+			Url:            esBackendConn.Config.Url,
+			User:           esBackendConn.Config.User,
+			Password:       esBackendConn.Config.Password,
+			ClientCertPath: esBackendConn.Config.ClientCertPath,
+			ClientKeyPath:  esBackendConn.Config.ClientKeyPath,
+			CACertPath:     esBackendConn.Config.CACertPath,
 		}, nil
 	}
 	return ElasticsearchConfiguration{}, fmt.Errorf("elasticsearch backend connector must be configured")

platform/config/config_v2.go

@@ -7,4 +7,8 @@ type ElasticsearchConfiguration struct {
 	User     string `koanf:"user"`
 	Password string `koanf:"password"`
 	AdminUrl *Url   `koanf:"adminUrl"`
+
+	ClientCertPath string `koanf:"clientCertPath"`
+	ClientKeyPath  string `koanf:"clientKeyPath"`
+	CACertPath     string `koanf:"caCertPath"`
 }

platform/config/elasticsearch_config.go

@@ -6,10 +6,12 @@ import (
 	"bytes"
 	"context"
 	"crypto/tls"
+	"crypto/x509"
 	"fmt"
 	"github.com/QuesmaOrg/quesma/platform/config"
 	"github.com/QuesmaOrg/quesma/platform/logger"
 	"net/http"
+	"os"
 	"time"
 )
 
@@ -24,18 +26,53 @@ type SimpleClient struct {
 	config *config.ElasticsearchConfiguration
 }
 
-func NewSimpleClient(configuration *config.ElasticsearchConfiguration) *SimpleClient {
-	client := &http.Client{
+// NewHttpsClient should be merged with NewSimpleClient at some point -> TODO!
+func NewHttpsClient(configuration *config.ElasticsearchConfiguration, timeout time.Duration) *http.Client {
+	tlsConfig := &tls.Config{
+		MinVersion:         tls.VersionTLS12,
+		InsecureSkipVerify: true,
+	}
+
+	if configuration.CACertPath != "" {
+		caCert, err := os.ReadFile(configuration.CACertPath)
+		if err != nil {
+			logger.Warn().Msgf("failed to read CA certificate: %v. Fallback to skipping tls.", err)
+		} else {
+			caCertPool := x509.NewCertPool()
+			if !caCertPool.AppendCertsFromPEM(caCert) {
+				logger.Warn().Msgf("failed to append CA certificate: %v. Fallback to skipping tls.", err)
+			} else {
+				tlsConfig.RootCAs = caCertPool
+				tlsConfig.InsecureSkipVerify = false
+			}
+		}
+	}
+
+	if configuration.ClientCertPath != "" && configuration.ClientKeyPath != "" {
+		cert, err := tls.LoadX509KeyPair(configuration.ClientCertPath, configuration.ClientKeyPath)
+		if err != nil {
+			logger.Warn().Msgf("failed to load client certificate/key: %v. Fallback to certificate-less client.", err)
+		} else {
+			tlsConfig.Certificates = []tls.Certificate{cert}
+		}
+	}
+
+	return &http.Client{
 		Transport: &http.Transport{
-			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+			TLSClientConfig: tlsConfig,
 		},
-		Timeout: esRequestTimeout,
+		Timeout: timeout,
 	}
+}
+
+func NewSimpleClient(configuration *config.ElasticsearchConfiguration) *SimpleClient {
+	client := NewHttpsClient(configuration, esRequestTimeout)
 	return &SimpleClient{
 		client: client,
 		config: configuration,
 	}
 }
+
 func (es *SimpleClient) Request(ctx context.Context, method, endpoint string, body []byte) (*http.Response, error) {
 	return es.doRequest(ctx, method, endpoint, body, nil)
 }