V2 - Utility dependencies clean up, part 1 (#1107)

This is the first PR  to clean utility dependencies.
It contains:
1. Split `PhoneHomeAgent` interface into a client and statistic provider
2. Extract `DebugInfoCollector` interface. It is passed instead of the
whole console management instance. More work is required in that area.
The management console is quite bulky.
3. Add a `Diagnostic` interface that will provide components with
telemetry/debug/... functionality. We should discuss component
granularity here.
4. Introduce a dependency injection framework. 

Building Quesma infrastructure (pipelines, processors, etc) is a
two-phase process:
1.  initialize components and connect them
2. inject utilities


@pdelewski please review this PR

Author: nablaone

quesma/clickhouse/clickhouse.go

@@ -12,8 +12,8 @@ import (
 	"quesma/quesma/config"
 	"quesma/quesma/recovery"
 	"quesma/schema"
-	"quesma/telemetry"
 	"quesma/util"
+	"quesma_v2/core/diag"
 	"slices"
 	"strings"
 	"sync/atomic"
@@ -32,7 +32,7 @@ type (
 		chDb           *sql.DB
 		tableDiscovery TableDiscovery
 		cfg            *config.QuesmaConfiguration
-		phoneHomeAgent telemetry.PhoneHomeAgent
+		phoneHomeAgent diag.PhoneHomeClient
 	}
 	TableMap  = util.SyncMap[string, *Table]
 	SchemaMap = map[string]interface{} // TODO remove
@@ -317,7 +317,7 @@ func (lm *LogManager) Ping() error {
 	return lm.chDb.Ping()
 }
 
-func NewEmptyLogManager(cfg *config.QuesmaConfiguration, chDb *sql.DB, phoneHomeAgent telemetry.PhoneHomeAgent, loader TableDiscovery) *LogManager {
+func NewEmptyLogManager(cfg *config.QuesmaConfiguration, chDb *sql.DB, phoneHomeAgent diag.PhoneHomeClient, loader TableDiscovery) *LogManager {
 	ctx, cancel := context.WithCancel(context.Background())
 	return &LogManager{ctx: ctx, cancel: cancel, chDb: chDb, tableDiscovery: loader, cfg: cfg, phoneHomeAgent: phoneHomeAgent}
 }
@@ -326,22 +326,22 @@ func NewLogManager(tables *TableMap, cfg *config.QuesmaConfiguration) *LogManage
 	var tableDefinitions = atomic.Pointer[TableMap]{}
 	tableDefinitions.Store(tables)
 	return &LogManager{chDb: nil, tableDiscovery: NewTableDiscoveryWith(cfg, nil, *tables),
-		cfg: cfg, phoneHomeAgent: telemetry.NewPhoneHomeEmptyAgent(),
+		cfg: cfg, phoneHomeAgent: diag.NewPhoneHomeEmptyAgent(),
 	}
 }
 
 // right now only for tests purposes
 func NewLogManagerWithConnection(db *sql.DB, tables *TableMap) *LogManager {
 	return &LogManager{chDb: db, tableDiscovery: NewTableDiscoveryWith(&config.QuesmaConfiguration{}, db, *tables),
-		phoneHomeAgent: telemetry.NewPhoneHomeEmptyAgent()}
+		phoneHomeAgent: diag.NewPhoneHomeEmptyAgent()}
 }
 
 func NewLogManagerEmpty() *LogManager {
 	var tableDefinitions = atomic.Pointer[TableMap]{}
 	tableDefinitions.Store(NewTableMap())
 	cfg := &config.QuesmaConfiguration{}
 	return &LogManager{tableDiscovery: NewTableDiscovery(cfg, nil, persistence.NewStaticJSONDatabase()), cfg: cfg,
-		phoneHomeAgent: telemetry.NewPhoneHomeEmptyAgent()}
+		phoneHomeAgent: diag.NewPhoneHomeEmptyAgent()}
 }
 
 func NewDefaultCHConfig() *ChTableConfig {

quesma/frontend_connectors/router_v2.go

@@ -22,9 +22,9 @@ import (
 	"quesma/quesma/types"
 	"quesma/quesma/ui"
 	"quesma/schema"
-	"quesma/telemetry"
 	"quesma/util"
 	quesma_api "quesma_v2/core"
+	"quesma_v2/core/diag"
 	"quesma_v2/core/routes"
 	"quesma_v2/core/tracing"
 	"strings"
@@ -72,15 +72,20 @@ func responseFromQuesmaV2(ctx context.Context, unzipped []byte, w http.ResponseW
 }
 
 type RouterV2 struct {
-	Config                  *config.QuesmaConfiguration
-	RequestPreprocessors    quesma_api.ProcessorChain
-	QuesmaManagementConsole *ui.QuesmaManagementConsole
-	PhoneHomeAgent          telemetry.PhoneHomeAgent
-	HttpClient              *http.Client
-	FailedRequests          atomic.Int64
+	Config               *config.QuesmaConfiguration
+	RequestPreprocessors quesma_api.ProcessorChain
+
+	HttpClient     *http.Client
+	FailedRequests atomic.Int64
+
+	diagnostic diag.Diagnostic
 }
 
-func NewRouterV2(config *config.QuesmaConfiguration, qmc *ui.QuesmaManagementConsole, agent telemetry.PhoneHomeAgent) *RouterV2 {
+func (r *RouterV2) InjectDiagnostic(s diag.Diagnostic) {
+	r.diagnostic = s
+}
+
+func NewRouterV2(config *config.QuesmaConfiguration) *RouterV2 {
 	tr := &http.Transport{
 		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 	}
@@ -91,11 +96,9 @@ func NewRouterV2(config *config.QuesmaConfiguration, qmc *ui.QuesmaManagementCon
 	requestProcessors := quesma_api.ProcessorChain{}
 	requestProcessors = append(requestProcessors, quesma_api.NewTraceIdPreprocessor())
 	return &RouterV2{
-		Config:                  config,
-		RequestPreprocessors:    requestProcessors,
-		QuesmaManagementConsole: qmc,
-		PhoneHomeAgent:          agent,
-		HttpClient:              client,
+		Config:               config,
+		RequestPreprocessors: requestProcessors,
+		HttpClient:           client,
 	}
 }
 
@@ -258,7 +261,7 @@ func (r *RouterV2) Reroute(ctx context.Context, w http.ResponseWriter, req *http
 	}
 	dispatcher := &quesma_api.Dispatcher{}
 	if handlersPipe != nil {
-		quesmaResponse, err := recordRequestToClickhouseV2(req.URL.Path, r.QuesmaManagementConsole, func() (*quesma_api.Result, error) {
+		quesmaResponse, err := recordRequestToClickhouseV2(req.URL.Path, r.diagnostic.DebugInfoCollector(), func() (*quesma_api.Result, error) {
 			var result *quesma_api.Result
 			result, err = handlersPipe.Handler(ctx, quesmaRequest)
 
@@ -338,22 +341,24 @@ func (r *RouterV2) sendHttpRequestToElastic(ctx context.Context, req *http.Reque
 	}
 
 	go func() {
-		elkResponseChan <- recordRequestToElasticV2(req.URL.Path, r.QuesmaManagementConsole, func() elasticResultV2 {
+		elkResponseChan <- recordRequestToElasticV2(req.URL.Path, r.diagnostic.DebugInfoCollector(), func() elasticResultV2 {
 
 			isWrite := elasticsearch.IsWriteRequest(req)
 
-			var span telemetry.Span
+			phoneHome := r.diagnostic.PhoneHomeAgent()
+
+			var span diag.Span
 			if isManagement {
 				if isWrite {
-					span = r.PhoneHomeAgent.ElasticBypassedWriteRequestsDuration().Begin()
+					span = phoneHome.ElasticBypassedWriteRequestsDuration().Begin()
 				} else {
-					span = r.PhoneHomeAgent.ElasticBypassedReadRequestsDuration().Begin()
+					span = phoneHome.ElasticBypassedReadRequestsDuration().Begin()
 				}
 			} else {
 				if isWrite {
-					span = r.PhoneHomeAgent.ElasticWriteRequestsDuration().Begin()
+					span = phoneHome.ElasticWriteRequestsDuration().Begin()
 				} else {
-					span = r.PhoneHomeAgent.ElasticReadRequestsDuration().Begin()
+					span = phoneHome.ElasticReadRequestsDuration().Begin()
 				}
 			}
 
@@ -373,7 +378,7 @@ func isIngestV2(path string) bool {
 	return strings.HasSuffix(path, routes.BulkPath) // We may add more methods in future such as `_put` or `_create`
 }
 
-func recordRequestToClickhouseV2(path string, qmc *ui.QuesmaManagementConsole, requestFunc func() (*quesma_api.Result, error)) (*quesma_api.Result, error) {
+func recordRequestToClickhouseV2(path string, qmc diag.DebugInfoCollector, requestFunc func() (*quesma_api.Result, error)) (*quesma_api.Result, error) {
 	statName := ui.RequestStatisticKibana2Clickhouse
 	if isIngestV2(path) {
 		statName = ui.RequestStatisticIngest2Clickhouse
@@ -384,7 +389,7 @@ func recordRequestToClickhouseV2(path string, qmc *ui.QuesmaManagementConsole, r
 	return response, err
 }
 
-func recordRequestToElasticV2(path string, qmc *ui.QuesmaManagementConsole, requestFunc func() elasticResultV2) elasticResultV2 {
+func recordRequestToElasticV2(path string, qmc diag.DebugInfoCollector, requestFunc func() elasticResultV2) elasticResultV2 {
 	statName := ui.RequestStatisticKibana2Elasticsearch
 	if isIngestV2(path) {
 		statName = ui.RequestStatisticIngest2Elasticsearch

quesma/ingest/processor.go

@@ -25,6 +25,7 @@ import (
 	"quesma/telemetry"
 	"quesma/util"
 	"quesma_v2/core"
+	"quesma_v2/core/diag"
 	"slices"
 	"sort"
 	"strings"
@@ -61,7 +62,7 @@ type (
 		chDb                      *sql.DB
 		tableDiscovery            chLib.TableDiscovery
 		cfg                       *config.QuesmaConfiguration
-		phoneHomeAgent            telemetry.PhoneHomeAgent
+		phoneHomeAgent            diag.PhoneHomeClient
 		schemaRegistry            schema.Registry
 		ingestCounter             int64
 		ingestFieldStatistics     IngestFieldStatistics

quesma/ingest/processor_test.go

@@ -9,8 +9,8 @@ import (
 	"quesma/quesma/config"
 	"quesma/quesma/types"
 	"quesma/schema"
-	"quesma/telemetry"
 	"quesma/util"
+	"quesma_v2/core/diag"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -22,7 +22,7 @@ func newIngestProcessorWithEmptyTableMap(tables *TableMap, cfg *config.QuesmaCon
 	var tableDefinitions = atomic.Pointer[TableMap]{}
 	tableDefinitions.Store(tables)
 	return &IngestProcessor{chDb: nil, tableDiscovery: clickhouse.NewTableDiscoveryWith(cfg, nil, *tables),
-		cfg: cfg, phoneHomeAgent: telemetry.NewPhoneHomeEmptyAgent(),
+		cfg: cfg, phoneHomeAgent: diag.NewPhoneHomeEmptyAgent(),
 		ingestFieldStatistics: make(IngestFieldStatistics),
 		virtualTableStorage:   persistence.NewStaticJSONDatabase(),
 	}
@@ -33,7 +33,7 @@ func newIngestProcessorEmpty() *IngestProcessor {
 	tableDefinitions.Store(NewTableMap())
 	cfg := &config.QuesmaConfiguration{}
 	return &IngestProcessor{tableDiscovery: clickhouse.NewTableDiscovery(cfg, nil, persistence.NewStaticJSONDatabase()), cfg: cfg,
-		phoneHomeAgent: telemetry.NewPhoneHomeEmptyAgent(), ingestFieldStatistics: make(IngestFieldStatistics)}
+		phoneHomeAgent: diag.NewPhoneHomeEmptyAgent(), ingestFieldStatistics: make(IngestFieldStatistics)}
 }
 
 var hasOthersConfig = &clickhouse.ChTableConfig{

quesma/main.go

@@ -51,6 +51,8 @@ const EnableConcurrencyProfiling = false
 // buildIngestOnlyQuesma is for now a helper function to help establishing the way of v2 module api import
 func buildIngestOnlyQuesma() quesma_api.QuesmaBuilder {
 	var quesmaBuilder quesma_api.QuesmaBuilder = quesma_api.NewQuesma()
+	quesmaBuilder.SetDependencies(quesma_api.EmptyDependencies())
+
 	ingestFrontendConnector := frontend_connectors.NewElasticsearchIngestFrontendConnector(":8080")
 
 	var ingestPipeline quesma_api.PipelineBuilder = quesma_api.NewPipeline()
@@ -206,7 +208,7 @@ func main() {
 
 func constructQuesma(cfg *config.QuesmaConfiguration, sl clickhouse.TableDiscovery, lm *clickhouse.LogManager, ip *ingest.IngestProcessor, im elasticsearch.IndexManagement, schemaRegistry schema.Registry, phoneHomeAgent telemetry.PhoneHomeAgent, quesmaManagementConsole *ui.QuesmaManagementConsole, logChan <-chan logger.LogWithLevel, abResultsrepository ab_testing.Sender, indexRegistry table_resolver.TableResolver) *quesma.Quesma {
 	if cfg.TransparentProxy {
-		return quesma.NewQuesmaTcpProxy(phoneHomeAgent, cfg, quesmaManagementConsole, logChan, false)
+		return quesma.NewQuesmaTcpProxy(cfg, quesmaManagementConsole, logChan, false)
 	} else {
 		const quesma_v2 = false
 		return quesma.NewHttpProxy(phoneHomeAgent, lm, ip, sl, im, schemaRegistry, cfg, quesmaManagementConsole, abResultsrepository, indexRegistry, quesma_v2)

quesma/main_test.go

@@ -41,6 +41,7 @@ func Test_backendConnectorValidation(t *testing.T) {
 	var postgressPipeline quesma_api.PipelineBuilder = quesma_api.NewPipeline()
 	postgressPipeline.AddProcessor(tcpProcessor)
 	var quesmaBuilder quesma_api.QuesmaBuilder = quesma_api.NewQuesma()
+	quesmaBuilder.SetDependencies(quesma_api.EmptyDependencies())
 	const endpoint = "root:password@tcp(127.0.0.1:3306)/test"
 	var mySqlBackendConnector quesma_api.BackendConnector = &backend_connectors.MySqlBackendConnector{
 		Endpoint: endpoint,
@@ -62,6 +63,7 @@ func fallback(_ context.Context, _ *quesma_api.Request) (*quesma_api.Result, err
 
 func ab_testing_scenario() quesma_api.QuesmaBuilder {
 	var quesmaBuilder quesma_api.QuesmaBuilder = quesma_api.NewQuesma()
+	quesmaBuilder.SetDependencies(quesma_api.EmptyDependencies())
 
 	ingestFrontendConnector := frontend_connectors.NewBasicHTTPFrontendConnector(":8888")
 	ingestHTTPRouter := quesma_api.NewPathRouter()
@@ -105,6 +107,7 @@ func ab_testing_scenario() quesma_api.QuesmaBuilder {
 
 func fallbackScenario() quesma_api.QuesmaBuilder {
 	var quesmaBuilder quesma_api.QuesmaBuilder = quesma_api.NewQuesma()
+	quesmaBuilder.SetDependencies(quesma_api.EmptyDependencies())
 
 	ingestFrontendConnector := frontend_connectors.NewBasicHTTPFrontendConnector(":8888")
 	ingestHTTPRouter := quesma_api.NewPathRouter()

quesma/queryparser/query_parser_test.go

@@ -12,9 +12,9 @@ import (
 	"quesma/quesma/config"
 	"quesma/quesma/types"
 	"quesma/schema"
-	"quesma/telemetry"
 	"quesma/testdata"
 	"quesma/util"
+	"quesma_v2/core/diag"
 	"strconv"
 	"strings"
 	"testing"
@@ -40,7 +40,7 @@ func TestQueryParserStringAttrConfig(t *testing.T) {
 
 	cfg.IndexConfig["logs-generic-default"] = config.IndexConfiguration{}
 
-	lm := clickhouse.NewEmptyLogManager(&cfg, nil, telemetry.NewPhoneHomeEmptyAgent(), clickhouse.NewTableDiscovery(&config.QuesmaConfiguration{}, nil, persistence.NewStaticJSONDatabase()))
+	lm := clickhouse.NewEmptyLogManager(&cfg, nil, diag.NewPhoneHomeEmptyAgent(), clickhouse.NewTableDiscovery(&config.QuesmaConfiguration{}, nil, persistence.NewStaticJSONDatabase()))
 	lm.AddTableIfDoesntExist(table)
 	s := schema.StaticRegistry{
 		Tables: map[schema.IndexName]schema.Schema{
@@ -99,7 +99,7 @@ func TestQueryParserNoFullTextFields(t *testing.T) {
 		},
 		Created: true,
 	}
-	lm := clickhouse.NewEmptyLogManager(&config.QuesmaConfiguration{}, nil, telemetry.NewPhoneHomeEmptyAgent(), clickhouse.NewTableDiscovery(&config.QuesmaConfiguration{}, nil, persistence.NewStaticJSONDatabase()))
+	lm := clickhouse.NewEmptyLogManager(&config.QuesmaConfiguration{}, nil, diag.NewPhoneHomeEmptyAgent(), clickhouse.NewTableDiscovery(&config.QuesmaConfiguration{}, nil, persistence.NewStaticJSONDatabase()))
 	lm.AddTableIfDoesntExist(&table)
 	cfg := config.QuesmaConfiguration{IndexConfig: map[string]config.IndexConfiguration{}}
 

quesma/quesma/dual_write_proxy.go

@@ -30,6 +30,7 @@ import (
 	"quesma/telemetry"
 	"quesma/util"
 	quesma_api "quesma_v2/core"
+	"quesma_v2/core/diag"
 	"quesma_v2/core/routes"
 	tracing "quesma_v2/core/tracing"
 	"strconv"
@@ -447,7 +448,7 @@ func (r *router) sendHttpRequestToElastic(ctx context.Context, req *http.Request
 
 			isWrite := elasticsearch.IsWriteRequest(req)
 
-			var span telemetry.Span
+			var span diag.Span
 			if isManagement {
 				if isWrite {
 					span = r.phoneHomeAgent.ElasticBypassedWriteRequestsDuration().Begin()

quesma/quesma/dual_write_proxy_v2.go

@@ -15,10 +15,8 @@ import (
 	"quesma/queryparser"
 	"quesma/quesma/async_search_storage"
 	"quesma/quesma/config"
-	"quesma/quesma/ui"
 	"quesma/schema"
 	"quesma/table_resolver"
-	"quesma/telemetry"
 	"quesma/util"
 	quesma_api "quesma_v2/core"
 	"strconv"
@@ -70,8 +68,9 @@ func (q *dualWriteHttpProxyV2) Stop(ctx context.Context) {
 	q.Close(ctx)
 }
 
-func newDualWriteProxyV2(schemaLoader clickhouse.TableDiscovery, logManager *clickhouse.LogManager, indexManager elasticsearch.IndexManagement, registry schema.Registry, config *config.QuesmaConfiguration, quesmaManagementConsole *ui.QuesmaManagementConsole, agent telemetry.PhoneHomeAgent, ingestProcessor *ingest.IngestProcessor, resolver table_resolver.TableResolver, abResultsRepository ab_testing.Sender) *dualWriteHttpProxyV2 {
-	queryProcessor := NewQueryRunner(logManager, config, indexManager, quesmaManagementConsole, registry, abResultsRepository, resolver, schemaLoader)
+func newDualWriteProxyV2(dependencies *quesma_api.Dependencies, schemaLoader clickhouse.TableDiscovery, logManager *clickhouse.LogManager, indexManager elasticsearch.IndexManagement, registry schema.Registry, config *config.QuesmaConfiguration, ingestProcessor *ingest.IngestProcessor, resolver table_resolver.TableResolver, abResultsRepository ab_testing.Sender) *dualWriteHttpProxyV2 {
+
+	queryProcessor := NewQueryRunner(logManager, config, indexManager, dependencies.Diagnostic.DebugInfoCollector(), registry, abResultsRepository, resolver, schemaLoader)
 
 	// not sure how we should configure our query translator ???
 	// is this a config option??
@@ -81,22 +80,23 @@ func newDualWriteProxyV2(schemaLoader clickhouse.TableDiscovery, logManager *cli
 	// tests should not be run with optimization enabled by default
 	queryProcessor.EnableQueryOptimization(config)
 
-	routerInstance := frontend_connectors.NewRouterV2(config,
-		quesmaManagementConsole, agent)
+	routerInstance := frontend_connectors.NewRouterV2(config)
+
+	dependencies.Diagnostic.PhoneHomeAgent().FailedRequestsCollector(func() int64 {
 
-	agent.FailedRequestsCollector(func() int64 {
 		return routerInstance.FailedRequests.Load()
 	})
 
-	ingestRouter := ConfigureIngestRouterV2(config, ingestProcessor, agent, resolver)
-	searchRouter := ConfigureSearchRouterV2(config, registry, logManager, quesmaManagementConsole, queryProcessor, resolver)
+	ingestRouter := ConfigureIngestRouterV2(config, dependencies, ingestProcessor, resolver)
+	searchRouter := ConfigureSearchRouterV2(config, dependencies, registry, logManager, queryProcessor, resolver)
 
 	elasticHttpIngestFrontendConnector := NewElasticHttpIngestFrontendConnector(":"+strconv.Itoa(int(config.PublicTcpPort)),
-		routerInstance, logManager, registry, agent)
+
+		routerInstance, logManager, registry)
 	elasticHttpIngestFrontendConnector.AddRouter(ingestRouter)
 
 	elasticHttpQueryFrontendConnector := NewElasticHttpQueryFrontendConnector(":"+strconv.Itoa(int(config.PublicTcpPort)),
-		routerInstance, logManager, registry, agent)
+		routerInstance, logManager, registry)
 	elasticHttpQueryFrontendConnector.AddRouter(searchRouter)
 
 	quesmaBuilder := quesma_api.NewQuesma()
@@ -107,6 +107,7 @@ func newDualWriteProxyV2(schemaLoader clickhouse.TableDiscovery, logManager *cli
 	queryPipeline.AddFrontendConnector(elasticHttpQueryFrontendConnector)
 	quesmaBuilder.AddPipeline(ingestPipeline)
 	quesmaBuilder.AddPipeline(queryPipeline)
+	quesmaBuilder.SetDependencies(dependencies)
 	_, err := quesmaBuilder.Build()
 	if err != nil {
 		logger.Fatal().Msgf("Error building Quesma: %v", err)