Licensing module (#396)

Licensing service is up and running. I guess we'll add a fancy domain
name once @jakozaur is back.

## Notes:
* licensing module has to initialize **before** the logger. That's
because the licensing module establishes the client id which we later on
used as an identifier for telemetry data being sent. Therefore, there's
no fancy logging, just printing stuff to stdout there. OTOH we'll have
information about failed license API calls in the Licensing Service
logs, so I guess it's not really a deal breaker.
* I left "is ClickHouse Cloud" check empty for now, we only check
Hydrolix.
* There's a breaking change in the configuration !! We introduce the
`connectors` section

.gitignore

@@ -21,4 +21,5 @@ docker/security/es.local
 docker/security/certificate-bundle.zip
 docker/security/clickhouse
 quesma/config.yaml
+quesma/.installation_id
 examples/kibana-sample-data/quesma/logs/*

docker/ci.yml

@@ -9,6 +9,7 @@ services:
       - QUESMA_port=8080
       - QUESMA_logging_path=/var/quesma/logs
       - QUESMA_clickhouse_url=clickhouse://clickhouse:9000
+      - QUESMA_connectors_my-CH-connector_type=clickhouse-os
       - QUESMA_logging_fileLogging=true
     depends_on:
       clickhouse:

docker/quesma/config/ci-config.yaml

@@ -1,5 +1,5 @@
 mode: "dual-write-query-clickhouse"
-licenseKey: "empty-license-key-that-is-just-ci"
+installationId: "just-Quesma-smoke-test-instance"
 port: 8080
 indexes:
   kibana_sample_data_ecommerce:

quesma/buildinfo/build.go

@@ -2,11 +2,6 @@
 // SPDX-License-Identifier: Elastic-2.0
 package buildinfo
 
-const (
-	DevelopmentLicenseKey = "cdd749a3-e777-11ee-bcf8-0242ac150004"
-)
-
 var Version = "development"
 var BuildHash = ""
 var BuildDate = ""
-var LicenseKey = DevelopmentLicenseKey

quesma/clickhouse/clickhouse.go

@@ -32,6 +32,7 @@ const (
 )
 
 type (
+	// LogManager should be renamed to Connector  -> TODO !!!
 	LogManager struct {
 		ctx            context.Context
 		cancel         context.CancelFunc
@@ -262,6 +263,26 @@ func (lm *LogManager) sendCreateTableQuery(ctx context.Context, query string) er
 	return nil
 }
 
+func (lm *LogManager) executeRawQuery(query string) (*sql.Rows, error) {
+	if res, err := lm.chDb.Query(query); err != nil {
+		return nil, fmt.Errorf("error in executeRawQuery: query: %s\nerr:%v", query, err)
+	} else {
+		return res, nil
+	}
+}
+
+func (lm *LogManager) CheckIfConnectedToHydrolix() error {
+	if rows, err := lm.executeRawQuery(`SELECT concat(database,'.', table) FROM system.tables WHERE engine = 'TurbineStorage';`); err != nil {
+		return fmt.Errorf("error executing HDX identifying query: %v", err)
+	} else {
+		defer rows.Close()
+		if rows.Next() {
+			return fmt.Errorf("detected Hydrolix-specific table engine, which is not allowed")
+		}
+		return nil
+	}
+}
+
 func (lm *LogManager) ProcessCreateTableQuery(ctx context.Context, query string, config *ChTableConfig) error {
 	table, err := NewTable(query, config)
 	if err != nil {

quesma/config.yaml.template

@@ -9,6 +9,9 @@ port: 8080  # public tcp port to listen for incoming traffic
 elasticsearch:
   url: "http://localhost:9200"
   call: false
+connectors:
+  - name: "clickhouse-conn"
+    type: "clickhouse" # one of [clickhouse, clickhouse-os, hydrolix]
 #clickhouse:   # this config is going to be removed, but for now let's just comment out
 #  url: "clickhouse://localhost:9000"
 ingestStatistics: true

quesma/connectors/clickhouse.go

@@ -0,0 +1,25 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package connectors
+
+import (
+	"quesma/clickhouse"
+)
+
+type ClickHouseConnector struct {
+	Connector *clickhouse.LogManager
+}
+
+const clickHouseConnectorTypeName = "clickhouse"
+
+func (c *ClickHouseConnector) LicensingCheck() (err error) {
+	return c.Connector.CheckIfConnectedToHydrolix()
+}
+
+func (c *ClickHouseConnector) Type() string {
+	return clickHouseConnectorTypeName
+}
+
+func (c *ClickHouseConnector) GetConnector() *clickhouse.LogManager {
+	return c.Connector
+}

quesma/connectors/clickhouse_os.go

@@ -0,0 +1,24 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package connectors
+
+import "quesma/clickhouse"
+
+type ClickHouseOSConnector struct {
+	Connector *clickhouse.LogManager
+}
+
+const clickHouseOSConnectorTypeName = "clickhouse-os"
+
+func (c *ClickHouseOSConnector) LicensingCheck() error {
+	// TODO: Check if you're connected to ClickHouse Cloud OR Hydrolix and fail if so
+	return c.Connector.CheckIfConnectedToHydrolix()
+}
+
+func (c *ClickHouseOSConnector) Type() string {
+	return clickHouseOSConnectorTypeName
+}
+
+func (c *ClickHouseOSConnector) GetConnector() *clickhouse.LogManager {
+	return c.Connector
+}

quesma/connectors/connector.go

@@ -0,0 +1,64 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package connectors
+
+import (
+	"database/sql"
+	"fmt"
+	"quesma/clickhouse"
+	"quesma/licensing"
+	"quesma/logger"
+	"quesma/quesma/config"
+	"quesma/telemetry"
+)
+
+type Connector interface {
+	LicensingCheck() error
+	Type() string
+	GetConnector() *clickhouse.LogManager // enforce contract for having connector instance ... maybe unnecessary
+}
+
+type ConnectorManager struct {
+	connectors []Connector
+}
+
+// GetConnector - TODO this is just bypassing the fact that we support only 1 connector at a time today :>
+func (c *ConnectorManager) GetConnector() *clickhouse.LogManager {
+	if len(c.connectors) == 0 {
+		panic("No connectors found")
+	}
+	conn := c.connectors[0]
+	if err := conn.LicensingCheck(); err != nil {
+		licensing.PanicWithLicenseViolation(fmt.Errorf("connector [%s] reported licensing issue: [%v]", conn.Type(), err))
+	}
+	return c.connectors[0].GetConnector()
+}
+
+func NewConnectorManager(cfg config.QuesmaConfiguration, chDb *sql.DB, phoneHomeAgent telemetry.PhoneHomeAgent, loader clickhouse.TableDiscovery) *ConnectorManager {
+	return &ConnectorManager{
+		connectors: registerConnectors(cfg, chDb, phoneHomeAgent, loader),
+	}
+}
+
+func registerConnectors(cfg config.QuesmaConfiguration, chDb *sql.DB, phoneHomeAgent telemetry.PhoneHomeAgent, loader clickhouse.TableDiscovery) (conns []Connector) {
+	for connName, conn := range cfg.Connectors {
+		logger.Info().Msgf("Registering connector named [%s] of type [%s]", connName, conn.ConnectorType)
+		switch conn.ConnectorType {
+		case clickHouseConnectorTypeName:
+			conns = append(conns, &ClickHouseConnector{
+				Connector: clickhouse.NewEmptyLogManager(cfg, chDb, phoneHomeAgent, loader),
+			})
+		case clickHouseOSConnectorTypeName:
+			conns = append(conns, &ClickHouseOSConnector{
+				Connector: clickhouse.NewEmptyLogManager(cfg, chDb, phoneHomeAgent, loader),
+			})
+		case hydrolixConnectorTypeName:
+			conns = append(conns, &HydrolixConnector{
+				Connector: clickhouse.NewEmptyLogManager(cfg, chDb, phoneHomeAgent, loader),
+			})
+		default:
+			logger.Error().Msgf("Unknown connector type [%s]", conn.ConnectorType)
+		}
+	}
+	return conns
+}

quesma/connectors/hydrolix.go

@@ -0,0 +1,27 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package connectors
+
+import (
+	"quesma/clickhouse"
+	"quesma/logger"
+)
+
+type HydrolixConnector struct {
+	Connector *clickhouse.LogManager
+}
+
+const hydrolixConnectorTypeName = "hydrolix"
+
+func (h *HydrolixConnector) LicensingCheck() error {
+	logger.Debug().Msg("Runtime checks for Hydrolix connector is not required, as static configuration disables it.")
+	return nil
+}
+
+func (h *HydrolixConnector) Type() string {
+	return hydrolixConnectorTypeName
+}
+
+func (h *HydrolixConnector) GetConnector() *clickhouse.LogManager {
+	return h.Connector
+}

quesma/licensing/error.go

@@ -0,0 +1,14 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package licensing
+
+import "fmt"
+
+const (
+	errorMessage = `There's been license violation detected. Please contact us at:
+		[email protected]`
+)
+
+func PanicWithLicenseViolation(initialErr error) {
+	panic(fmt.Sprintf("Error thrown: %v\n%s", initialErr, errorMessage))
+}

quesma/licensing/license_manager.go

@@ -0,0 +1,88 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package licensing
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+)
+
+const (
+	obtainLicenseEndpoint = "https://quesma-licensing-service-gd46dsvxda-uc.a.run.app/api/license/obtain"
+	verifyLicenseEndpoint = "https://quesma-licensing-service-gd46dsvxda-uc.a.run.app/api/license/verify"
+)
+
+type InstallationIDPayload struct {
+	InstallationID string `json:"installation_id"`
+}
+
+type LicensePayload struct {
+	LicenseKey []byte `json:"license_key"`
+}
+
+// obtainLicenseKey presents an InstallationId to the license server and receives a LicenseKey in return
+func (l *LicenseModule) obtainLicenseKey() (err error) {
+	fmt.Printf("Obtaining license key for installation ID [%s]\n", l.InstallationID)
+	var payloadBytes []byte
+	if payloadBytes, err = json.Marshal(InstallationIDPayload{InstallationID: l.InstallationID}); err != nil {
+		return err
+	}
+	resp, err := http.Post(obtainLicenseEndpoint, "application/json", bytes.NewReader(payloadBytes))
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return err
+	}
+
+	var licenseResponse LicensePayload
+	if err = json.Unmarshal(body, &licenseResponse); err != nil {
+		return err
+	}
+	l.LicenseKey = licenseResponse.LicenseKey
+	fmt.Printf("License key obtained and set successfully, key=[%s]\n", l.LicenseKey)
+	return nil
+}
+
+// processLicense presents the license to the license server and receives an AllowList in return
+func (l *LicenseModule) processLicense() error {
+	if fetchedLicense, err := l.fetchLicense(); err != nil {
+		return fmt.Errorf("failed processing license by the license server: %v", err)
+	} else {
+		l.License = fetchedLicense
+		fmt.Printf("Allowlist loaded successfully\n%s\n", fetchedLicense.String())
+	}
+	if l.License.ExpirationDate.Before(time.Now()) {
+		return fmt.Errorf("license expired on %s", l.License.ExpirationDate)
+	}
+	return nil
+}
+
+func (l *LicenseModule) fetchLicense() (a *License, err error) {
+	var payloadBytes []byte
+	if payloadBytes, err = json.Marshal(LicensePayload{LicenseKey: l.LicenseKey}); err != nil {
+		return nil, err
+	}
+	resp, err := http.Post(verifyLicenseEndpoint, "application/json", bytes.NewReader(payloadBytes))
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	if err = json.Unmarshal(body, &a); err != nil {
+		return nil, err
+	} else {
+		return a, nil
+	}
+}

quesma/licensing/model.go

@@ -0,0 +1,23 @@
+// Copyright Quesma, licensed under the Elastic License 2.0.
+// SPDX-License-Identifier: Elastic-2.0
+package licensing
+
+import (
+	"fmt"
+	"strings"
+	"time"
+)
+
+// License is an object returned by the license server based on the provided (and positively verified) license key
+type License struct {
+	InstallationID string    `json:"installation_id"`
+	ClientID       string    `json:"client_id"`
+	Connectors     []string  `json:"connectors"`
+	Processors     []string  `json:"processors"`
+	ExpirationDate time.Time `json:"expiration_date"`
+}
+
+func (a *License) String() string {
+	return fmt.Sprintf("[Quesma License]\n\tInstallation ID: %s\n\tClient Name: %s\n\tConnectors: [%v]\n\tProcessors: [%v]\n\tExpires: %s\n",
+		a.InstallationID, a.ClientID, strings.Join(a.Connectors, ", "), strings.Join(a.Processors, ", "), a.ExpirationDate)
+}