feat: traces v3 exporter integration (#425)

Changes include 
* drop durationsort materialized view. ( please review
#459 first) , no
down operations for this as we don't want this MV anyway
* added the flag `use_new_schema`, when enabled will only use the new
exporters, and usage will be from the new schema.
* Add resources to usage collection by uncommenting the json tag in the
struct


FOR SigNoz/signoz#5713

---------

Co-authored-by: Srikanth Chekuri <[email protected]>

cmd/signozschemamigrator/schema_migrator/traces_migrations.go

@@ -1,9 +1,7 @@
 package schemamigrator
 
-var TracesMigrations = []SchemaMigrationRecord{}
-
 // move them to TracesMigrations once it's ready to deploy
-var TracesMigrationsStaging = []SchemaMigrationRecord{
+var TracesMigrations = []SchemaMigrationRecord{
 	{
 		MigrationID: 1000,
 		UpItems: []Operation{
@@ -294,9 +292,9 @@ var TracesMigrationsStaging = []SchemaMigrationRecord{
 				},
 				Engine: AggregatingMergeTree{
 					MergeTree: MergeTree{
-						PartitionBy: "toDate(start)",
+						PartitionBy: "toDate(end)",
 						OrderBy:     "(trace_id)",
-						TTL:         "toDateTime(start) + toIntervalSecond(1296000)",
+						TTL:         "toDateTime(end) + toIntervalSecond(1296000)",
 						Settings: TableSettings{
 							{Name: "index_granularity", Value: "8192"},
 							{Name: "ttl_only_drop_parts", Value: "1"},
@@ -414,4 +412,26 @@ var TracesMigrationsStaging = []SchemaMigrationRecord{
 			},
 		},
 	},
+	{
+		MigrationID: 1001,
+		UpItems: []Operation{
+			DropTableOperation{
+				Database: "signoz_traces",
+				Table:    "durationSortMV",
+			},
+			DropTableOperation{
+				Database: "signoz_traces",
+				Table:    "distributed_durationSort",
+			},
+			DropTableOperation{
+				Database: "signoz_traces",
+				Table:    "durationSort",
+				// this is added so that we can avoid the following error
+				//1. Size (453.51 GB) is greater than max_[table/partition]_size_to_drop (50.00 GB)
+				// https://stackoverflow.com/questions/78162269/cannot-drop-large-materialized-view-in-clickhouse
+				Settings: TableSettings{{Name: "max_table_size_to_drop", Value: "0"}},
+			},
+		},
+		DownItems: []Operation{},
+	},
 }

exporter/clickhousetracesexporter/clickhouse_exporter.go

@@ -84,8 +84,9 @@ func newExporter(cfg component.Config, logger *zap.Logger) (*storage, error) {
 		config: storageConfig{
 			lowCardinalExceptionGrouping: configClickHouse.LowCardinalExceptionGrouping,
 		},
-		wg:        new(sync.WaitGroup),
-		closeChan: make(chan struct{}),
+		wg:           new(sync.WaitGroup),
+		closeChan:    make(chan struct{}),
+		useNewSchema: configClickHouse.UseNewSchema,
 	}
 
 	return &storage, nil
@@ -188,6 +189,7 @@ func populateOtherDimensions(attributes pcommon.Map, span *Span) {
 		} else if (k == "http.method" || k == "http.request.method") && span.Kind == 3 {
 			span.ExternalHttpMethod = v.Str()
 			span.HttpMethod = v.Str()
+
 		} else if (k == "http.url" || k == "url.full") && span.Kind != 3 {
 			span.HttpUrl = v.Str()
 		} else if (k == "http.method" || k == "http.request.method") && span.Kind != 3 {
@@ -403,6 +405,16 @@ func newStructuredSpan(otelSpan ptrace.Span, ServiceName string, resource pcommo
 
 // traceDataPusher implements OTEL exporterhelper.traceDataPusher
 func (s *storage) pushTraceData(ctx context.Context, td ptrace.Traces) error {
+	// if the new schema is enabled don't write to the old tables
+	err := s.pushTraceDataV3(ctx, td)
+	if err != nil {
+		return err
+	}
+	// if new schema is forced exit here else write to the old tables as well.
+	if s.useNewSchema {
+		return nil
+	}
+
 	s.wg.Add(1)
 	defer s.wg.Done()
 

exporter/clickhousetracesexporter/clickhouse_factory.go

@@ -65,7 +65,7 @@ func ClickHouseNewFactory(exporterId uuid.UUID, config Config) *Factory {
 
 	view.Register(writeLatencyView)
 	return &Factory{
-		Options: NewOptions(exporterId, config, primaryNamespace, archiveNamespace),
+		Options: NewOptions(exporterId, config, primaryNamespace, config.UseNewSchema, archiveNamespace),
 		// makeReader: func(db *clickhouse.Conn, operationsTable, indexTable, spansTable string) (spanstore.Reader, error) {
 		// 	return store.NewTraceReader(db, operationsTable, indexTable, spansTable), nil
 		// },
@@ -120,6 +120,10 @@ func (f *Factory) CreateSpanWriter() (Writer, error) {
 		attributeKeyTable: cfg.AttributeKeyTable,
 		encoding:          cfg.Encoding,
 		exporterId:        cfg.ExporterId,
+
+		useNewSchema:    cfg.UseNewSchema,
+		indexTableV3:    cfg.IndexTableV3,
+		resourceTableV3: cfg.ResourceTableV3,
 	})
 }
 
@@ -140,6 +144,10 @@ func (f *Factory) CreateArchiveSpanWriter() (Writer, error) {
 		attributeKeyTable: cfg.AttributeKeyTable,
 		encoding:          cfg.Encoding,
 		exporterId:        cfg.ExporterId,
+
+		useNewSchema:    cfg.UseNewSchema,
+		indexTableV3:    cfg.IndexTableV3,
+		resourceTableV3: cfg.ResourceTableV3,
 	})
 }
 

exporter/clickhousetracesexporter/config.go

@@ -31,6 +31,7 @@ type Config struct {
 	exporterhelper.TimeoutConfig `mapstructure:",squash"`
 	configretry.BackOffConfig    `mapstructure:"retry_on_failure"`
 	exporterhelper.QueueConfig   `mapstructure:"sending_queue"`
+	UseNewSchema                 bool `mapstructure:"use_new_schema" default:"false"`
 }
 
 var _ component.Config = (*Config)(nil)

exporter/clickhousetracesexporter/options.go

@@ -39,6 +39,8 @@ const (
 	defaultDependencyGraphDbMV      string   = "dependency_graph_minutes_db_calls_mv"
 	DependencyGraphMessagingMV      string   = "dependency_graph_minutes_messaging_calls_mv"
 	defaultEncoding                 Encoding = EncodingJSON
+	defaultIndexTableV3             string   = "distributed_signoz_index_v3"
+	defaultResourceTableV3          string   = "distributed_traces_v3_resource"
 )
 
 // NamespaceConfig is Clickhouse's internal configuration data
@@ -64,6 +66,9 @@ type namespaceConfig struct {
 	Encoding                   Encoding
 	Connector                  Connector
 	ExporterId                 uuid.UUID
+	UseNewSchema               bool
+	IndexTableV3               string
+	ResourceTableV3            string
 }
 
 // Connecto defines how to connect to the database
@@ -105,7 +110,7 @@ type Options struct {
 }
 
 // NewOptions creates a new Options struct.
-func NewOptions(exporterId uuid.UUID, config Config, primaryNamespace string, otherNamespaces ...string) *Options {
+func NewOptions(exporterId uuid.UUID, config Config, primaryNamespace string, useNewSchema bool, otherNamespaces ...string) *Options {
 
 	datasource := config.Datasource
 	if datasource == "" {
@@ -135,6 +140,9 @@ func NewOptions(exporterId uuid.UUID, config Config, primaryNamespace string, ot
 			Encoding:                   defaultEncoding,
 			Connector:                  defaultConnector,
 			ExporterId:                 exporterId,
+			UseNewSchema:               useNewSchema,
+			IndexTableV3:               defaultIndexTableV3,
+			ResourceTableV3:            defaultResourceTableV3,
 		},
 		others: make(map[string]*namespaceConfig, len(otherNamespaces)),
 	}
@@ -150,6 +158,9 @@ func NewOptions(exporterId uuid.UUID, config Config, primaryNamespace string, ot
 				Encoding:        defaultEncoding,
 				Connector:       defaultConnector,
 				ExporterId:      exporterId,
+				UseNewSchema:    useNewSchema,
+				IndexTableV3:    defaultIndexTableV3,
+				ResourceTableV3: defaultResourceTableV3,
 			}
 		} else {
 			options.others[namespace] = &namespaceConfig{namespace: namespace}

exporter/clickhousetracesexporter/schema-signoz.go

@@ -164,7 +164,7 @@ type SpanV3 struct {
 	AttributesNumber map[string]float64 `json:"attributes_number,omitempty"`
 	AttributesBool   map[string]bool    `json:"attributes_bool,omitempty"`
 
-	ResourcesString map[string]string `json:"-"`
+	ResourcesString map[string]string `json:"resources_string,omitempty"`
 
 	// for events
 	Events      []string     `json:"event,omitempty"`

exporter/clickhousetracesexporter/testdata/config.yaml

@@ -7,6 +7,9 @@ processors:
 exporters:
   clickhousetraces:
     datasource: tcp://127.0.0.1:9000/?database=signoz_traces&username=admin&password=password
+  clickhousetraces/new_schema:
+    datasource: tcp://127.0.0.1:9000/?database=signoz_traces&username=admin&password=password
+    use_new_schema: true
   clickhousetraces/2:
     datasource: tcp://127.0.0.1:9000/?database=signoz_traces&username=admin&password=password
     timeout: 5s