feat: use new schema flag

ee/query-service/app/api/api.go

@@ -39,6 +39,8 @@ type APIHandlerOptions struct {
 	Gateway                       *httputil.ReverseProxy
 	// Querier Influx Interval
 	FluxInterval time.Duration
+
+	UseLogsNewSchema bool
 }
 
 type APIHandler struct {
@@ -63,6 +65,7 @@ func NewAPIHandler(opts APIHandlerOptions) (*APIHandler, error) {
 		LogsParsingPipelineController: opts.LogsParsingPipelineController,
 		Cache:                         opts.Cache,
 		FluxInterval:                  opts.FluxInterval,
+		UseLogsNewSchema:              opts.UseLogsNewSchema,
 	})
 
 	if err != nil {

ee/query-service/app/db/reader.go

@@ -25,8 +25,9 @@ func NewDataConnector(
 	maxOpenConns int,
 	dialTimeout time.Duration,
 	cluster string,
+	useLogsNewSchema bool,
 ) *ClickhouseReader {
-	ch := basechr.NewReader(localDB, promConfigPath, lm, maxIdleConns, maxOpenConns, dialTimeout, cluster)
+	ch := basechr.NewReader(localDB, promConfigPath, lm, maxIdleConns, maxOpenConns, dialTimeout, cluster, useLogsNewSchema)
 	return &ClickhouseReader{
 		conn:             ch.GetConn(),
 		appdb:            localDB,

ee/query-service/app/server.go

@@ -77,6 +77,7 @@ type ServerOptions struct {
 	FluxInterval      string
 	Cluster           string
 	GatewayUrl        string
+	UseLogsNewSchema  bool
 }
 
 // Server runs HTTP api service
@@ -154,6 +155,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) {
 			serverOptions.MaxOpenConns,
 			serverOptions.DialTimeout,
 			serverOptions.Cluster,
+			serverOptions.UseLogsNewSchema,
 		)
 		go qb.Start(readerReady)
 		reader = qb
@@ -176,7 +178,9 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) {
 		localDB,
 		reader,
 		serverOptions.DisableRules,
-		lm)
+		lm,
+		serverOptions.UseLogsNewSchema,
+	)
 
 	if err != nil {
 		return nil, err
@@ -265,6 +269,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) {
 		Cache:                         c,
 		FluxInterval:                  fluxInterval,
 		Gateway:                       gatewayProxy,
+		UseLogsNewSchema:              serverOptions.UseLogsNewSchema,
 	}
 
 	apiHandler, err := api.NewAPIHandler(apiOpts)
@@ -728,7 +733,8 @@ func makeRulesManager(
 	db *sqlx.DB,
 	ch baseint.Reader,
 	disableRules bool,
-	fm baseint.FeatureLookup) (*baserules.Manager, error) {
+	fm baseint.FeatureLookup,
+	useLogsNewSchema bool) (*baserules.Manager, error) {
 
 	// create engine
 	pqle, err := pqle.FromConfigPath(promConfigPath)
@@ -756,7 +762,8 @@ func makeRulesManager(
 		Reader:       ch,
 		EvalDelay:    baseconst.GetEvalDelay(),
 
-		PrepareTaskFunc: rules.PrepareTaskFunc,
+		PrepareTaskFunc:  rules.PrepareTaskFunc,
+		UseLogsNewSchema: useLogsNewSchema,
 	}
 
 	// create Manager

ee/query-service/main.go

@@ -87,6 +87,7 @@ func main() {
 	var ruleRepoURL string
 	var cluster string
 
+	var useLogsNewSchema bool
 	var cacheConfigPath, fluxInterval string
 	var enableQueryServiceLogOTLPExport bool
 	var preferSpanMetrics bool
@@ -96,6 +97,7 @@ func main() {
 	var dialTimeout time.Duration
 	var gatewayUrl string
 
+	flag.BoolVar(&useLogsNewSchema, "use-logs-new-schema", false, "use logs_v2 schema for logs")
 	flag.StringVar(&promConfigPath, "config", "./config/prometheus.yml", "(prometheus config to read metrics)")
 	flag.StringVar(&skipTopLvlOpsPath, "skip-top-level-ops", "", "(config file to skip top level operations)")
 	flag.BoolVar(&disableRules, "rules.disable", false, "(disable rule evaluation)")
@@ -134,6 +136,7 @@ func main() {
 		FluxInterval:      fluxInterval,
 		Cluster:           cluster,
 		GatewayUrl:        gatewayUrl,
+		UseLogsNewSchema:  useLogsNewSchema,
 	}
 
 	// Read the jwt secret key

ee/query-service/rules/manager.go

@@ -20,6 +20,7 @@ func PrepareTaskFunc(opts baserules.PrepareTaskOptions) (baserules.Task, error)
 			opts.Rule,
 			opts.FF,
 			opts.Reader,
+			opts.UseLogsNewSchema,
 			baserules.WithEvalDelay(opts.ManagerOpts.EvalDelay),
 		)
 

pkg/query-service/app/clickhouseReader/reader.go

@@ -132,6 +132,8 @@ type ClickHouseReader struct {
 
 	liveTailRefreshSeconds int
 	cluster                string
+
+	useLogsNewSchema bool
 }
 
 // NewTraceReader returns a TraceReader for the database
@@ -143,6 +145,7 @@ func NewReader(
 	maxOpenConns int,
 	dialTimeout time.Duration,
 	cluster string,
+	useLogsNewSchema bool,
 ) *ClickHouseReader {
 
 	datasource := os.Getenv("ClickHouseUrl")
@@ -153,7 +156,7 @@ func NewReader(
 		zap.L().Fatal("failed to initialize ClickHouse", zap.Error(err))
 	}
 
-	return NewReaderFromClickhouseConnection(db, options, localDB, configFile, featureFlag, cluster)
+	return NewReaderFromClickhouseConnection(db, options, localDB, configFile, featureFlag, cluster, useLogsNewSchema)
 }
 
 func NewReaderFromClickhouseConnection(
@@ -163,6 +166,7 @@ func NewReaderFromClickhouseConnection(
 	configFile string,
 	featureFlag interfaces.FeatureLookup,
 	cluster string,
+	useLogsNewSchema bool,
 ) *ClickHouseReader {
 	alertManager, err := am.New("")
 	if err != nil {
@@ -219,6 +223,7 @@ func NewReaderFromClickhouseConnection(
 		featureFlags:            featureFlag,
 		cluster:                 cluster,
 		queryProgressTracker:    queryprogress.NewQueryProgressTracker(),
+		useLogsNewSchema:        useLogsNewSchema,
 	}
 }
 

pkg/query-service/app/http_handler.go

@@ -105,6 +105,8 @@ type APIHandler struct {
 
 	// Websocket connection upgrader
 	Upgrader *websocket.Upgrader
+
+	UseLogsNewSchema bool
 }
 
 type APIHandlerOpts struct {
@@ -140,6 +142,9 @@ type APIHandlerOpts struct {
 
 	// Querier Influx Interval
 	FluxInterval time.Duration
+
+	// Use new schema
+	UseLogsNewSchema bool
 }
 
 // NewAPIHandler returns an APIHandler
@@ -151,19 +156,21 @@ func NewAPIHandler(opts APIHandlerOpts) (*APIHandler, error) {
 	}
 
 	querierOpts := querier.QuerierOptions{
-		Reader:        opts.Reader,
-		Cache:         opts.Cache,
-		KeyGenerator:  queryBuilder.NewKeyGenerator(),
-		FluxInterval:  opts.FluxInterval,
-		FeatureLookup: opts.FeatureFlags,
+		Reader:           opts.Reader,
+		Cache:            opts.Cache,
+		KeyGenerator:     queryBuilder.NewKeyGenerator(),
+		FluxInterval:     opts.FluxInterval,
+		FeatureLookup:    opts.FeatureFlags,
+		UseLogsNewSchema: opts.UseLogsNewSchema,
 	}
 
 	querierOptsV2 := querierV2.QuerierOptions{
-		Reader:        opts.Reader,
-		Cache:         opts.Cache,
-		KeyGenerator:  queryBuilder.NewKeyGenerator(),
-		FluxInterval:  opts.FluxInterval,
-		FeatureLookup: opts.FeatureFlags,
+		Reader:           opts.Reader,
+		Cache:            opts.Cache,
+		KeyGenerator:     queryBuilder.NewKeyGenerator(),
+		FluxInterval:     opts.FluxInterval,
+		FeatureLookup:    opts.FeatureFlags,
+		UseLogsNewSchema: opts.UseLogsNewSchema,
 	}
 
 	querier := querier.NewQuerier(querierOpts)
@@ -185,6 +192,7 @@ func NewAPIHandler(opts APIHandlerOpts) (*APIHandler, error) {
 		LogsParsingPipelineController: opts.LogsParsingPipelineController,
 		querier:                       querier,
 		querierV2:                     querierv2,
+		UseLogsNewSchema:              opts.UseLogsNewSchema,
 	}
 
 	builderOpts := queryBuilder.QueryBuilderOptions{

pkg/query-service/app/querier/querier.go

@@ -54,6 +54,8 @@ type querier struct {
 	timeRanges     [][]int
 	returnedSeries []*v3.Series
 	returnedErr    error
+
+	UseLogsNewSchema bool
 }
 
 type QuerierOptions struct {
@@ -64,9 +66,10 @@ type QuerierOptions struct {
 	FeatureLookup interfaces.FeatureLookup
 
 	// used for testing
-	TestingMode    bool
-	ReturnedSeries []*v3.Series
-	ReturnedErr    error
+	TestingMode      bool
+	ReturnedSeries   []*v3.Series
+	ReturnedErr      error
+	UseLogsNewSchema bool
 }
 
 func NewQuerier(opts QuerierOptions) interfaces.Querier {

pkg/query-service/app/querier/v2/querier.go

@@ -54,6 +54,8 @@ type querier struct {
 	timeRanges     [][]int
 	returnedSeries []*v3.Series
 	returnedErr    error
+
+	UseLogsNewSchema bool
 }
 
 type QuerierOptions struct {
@@ -64,9 +66,10 @@ type QuerierOptions struct {
 	FeatureLookup interfaces.FeatureLookup
 
 	// used for testing
-	TestingMode    bool
-	ReturnedSeries []*v3.Series
-	ReturnedErr    error
+	TestingMode      bool
+	ReturnedSeries   []*v3.Series
+	ReturnedErr      error
+	UseLogsNewSchema bool
 }
 
 func NewQuerier(opts QuerierOptions) interfaces.Querier {

pkg/query-service/app/server.go

@@ -66,6 +66,7 @@ type ServerOptions struct {
 	CacheConfigPath   string
 	FluxInterval      string
 	Cluster           string
+	UseLogsNewSchema  bool
 }
 
 // Server runs HTTP, Mux and a grpc server
@@ -128,6 +129,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) {
 			serverOptions.MaxOpenConns,
 			serverOptions.DialTimeout,
 			serverOptions.Cluster,
+			serverOptions.UseLogsNewSchema,
 		)
 		go clickhouseReader.Start(readerReady)
 		reader = clickhouseReader
@@ -144,7 +146,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) {
 	}
 
 	<-readerReady
-	rm, err := makeRulesManager(serverOptions.PromConfigPath, constants.GetAlertManagerApiPrefix(), serverOptions.RuleRepoURL, localDB, reader, serverOptions.DisableRules, fm)
+	rm, err := makeRulesManager(serverOptions.PromConfigPath, constants.GetAlertManagerApiPrefix(), serverOptions.RuleRepoURL, localDB, reader, serverOptions.DisableRules, fm, serverOptions.UseLogsNewSchema)
 	if err != nil {
 		return nil, err
 	}
@@ -197,6 +199,7 @@ func NewServer(serverOptions *ServerOptions) (*Server, error) {
 		LogsParsingPipelineController: logParsingPipelineController,
 		Cache:                         c,
 		FluxInterval:                  fluxInterval,
+		UseLogsNewSchema:              serverOptions.UseLogsNewSchema,
 	})
 	if err != nil {
 		return nil, err
@@ -713,7 +716,8 @@ func makeRulesManager(
 	db *sqlx.DB,
 	ch interfaces.Reader,
 	disableRules bool,
-	fm interfaces.FeatureLookup) (*rules.Manager, error) {
+	fm interfaces.FeatureLookup,
+	useLogsNewSchema bool) (*rules.Manager, error) {
 
 	// create engine
 	pqle, err := pqle.FromReader(ch)
@@ -730,16 +734,17 @@ func makeRulesManager(
 
 	// create manager opts
 	managerOpts := &rules.ManagerOptions{
-		NotifierOpts: notifierOpts,
-		PqlEngine:    pqle,
-		RepoURL:      ruleRepoURL,
-		DBConn:       db,
-		Context:      context.Background(),
-		Logger:       nil,
-		DisableRules: disableRules,
-		FeatureFlags: fm,
-		Reader:       ch,
-		EvalDelay:    constants.GetEvalDelay(),
+		NotifierOpts:     notifierOpts,
+		PqlEngine:        pqle,
+		RepoURL:          ruleRepoURL,
+		DBConn:           db,
+		Context:          context.Background(),
+		Logger:           nil,
+		DisableRules:     disableRules,
+		FeatureFlags:     fm,
+		Reader:           ch,
+		EvalDelay:        constants.GetEvalDelay(),
+		UseLogsNewSchema: useLogsNewSchema,
 	}
 
 	// create Manager

pkg/query-service/main.go

@@ -33,6 +33,7 @@ func main() {
 	// disables rule execution but allows change to the rule definition
 	var disableRules bool
 
+	var useLogsNewSchema bool
 	// the url used to build link in the alert messages in slack and other systems
 	var ruleRepoURL, cacheConfigPath, fluxInterval string
 	var cluster string
@@ -43,6 +44,7 @@ func main() {
 	var maxOpenConns int
 	var dialTimeout time.Duration
 
+	flag.BoolVar(&useLogsNewSchema, "use-logs-new-schema", false, "use logs_v2 schema for logs")
 	flag.StringVar(&promConfigPath, "config", "./config/prometheus.yml", "(prometheus config to read metrics)")
 	flag.StringVar(&skipTopLvlOpsPath, "skip-top-level-ops", "", "(config file to skip top level operations)")
 	flag.BoolVar(&disableRules, "rules.disable", false, "(disable rule evaluation)")
@@ -79,6 +81,7 @@ func main() {
 		CacheConfigPath:   cacheConfigPath,
 		FluxInterval:      fluxInterval,
 		Cluster:           cluster,
+		UseLogsNewSchema:  useLogsNewSchema,
 	}
 
 	// Read the jwt secret key