mirror of
https://git.mirrors.martin98.com/https://github.com/SigNoz/signoz
synced 2025-08-15 10:15:57 +08:00
chore: remove old data migrations (#5802)
This commit is contained in:
Srikanth Chekuri
GitHub
parent
2f0d98ae51
commit
4295a2756a
@ -1,153 +0,0 @@
|
||||
package alertstov4
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/jmoiron/sqlx"
|
||||
v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
|
||||
"go.signoz.io/signoz/pkg/query-service/rules"
|
||||
"go.uber.org/multierr"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
var Version = "0.45-alerts-to-v4"
|
||||
|
||||
var mapTimeAggregation = map[v3.AggregateOperator]v3.TimeAggregation{
|
||||
v3.AggregateOperatorSum: v3.TimeAggregationSum,
|
||||
v3.AggregateOperatorMin: v3.TimeAggregationMin,
|
||||
v3.AggregateOperatorMax: v3.TimeAggregationMax,
|
||||
v3.AggregateOperatorSumRate: v3.TimeAggregationRate,
|
||||
v3.AggregateOperatorAvgRate: v3.TimeAggregationRate,
|
||||
v3.AggregateOperatorMinRate: v3.TimeAggregationRate,
|
||||
v3.AggregateOperatorMaxRate: v3.TimeAggregationRate,
|
||||
v3.AggregateOperatorHistQuant50: v3.TimeAggregationUnspecified,
|
||||
v3.AggregateOperatorHistQuant75: v3.TimeAggregationUnspecified,
|
||||
v3.AggregateOperatorHistQuant90: v3.TimeAggregationUnspecified,
|
||||
v3.AggregateOperatorHistQuant95: v3.TimeAggregationUnspecified,
|
||||
v3.AggregateOperatorHistQuant99: v3.TimeAggregationUnspecified,
|
||||
}
|
||||
|
||||
var mapSpaceAggregation = map[v3.AggregateOperator]v3.SpaceAggregation{
|
||||
v3.AggregateOperatorSum: v3.SpaceAggregationSum,
|
||||
v3.AggregateOperatorMin: v3.SpaceAggregationMin,
|
||||
v3.AggregateOperatorMax: v3.SpaceAggregationMax,
|
||||
v3.AggregateOperatorSumRate: v3.SpaceAggregationSum,
|
||||
v3.AggregateOperatorAvgRate: v3.SpaceAggregationAvg,
|
||||
v3.AggregateOperatorMinRate: v3.SpaceAggregationMin,
|
||||
v3.AggregateOperatorMaxRate: v3.SpaceAggregationMax,
|
||||
v3.AggregateOperatorHistQuant50: v3.SpaceAggregationPercentile50,
|
||||
v3.AggregateOperatorHistQuant75: v3.SpaceAggregationPercentile75,
|
||||
v3.AggregateOperatorHistQuant90: v3.SpaceAggregationPercentile90,
|
||||
v3.AggregateOperatorHistQuant95: v3.SpaceAggregationPercentile95,
|
||||
v3.AggregateOperatorHistQuant99: v3.SpaceAggregationPercentile99,
|
||||
}
|
||||
|
||||
func canMigrateOperator(operator v3.AggregateOperator) bool {
|
||||
switch operator {
|
||||
case v3.AggregateOperatorSum,
|
||||
v3.AggregateOperatorMin,
|
||||
v3.AggregateOperatorMax,
|
||||
v3.AggregateOperatorSumRate,
|
||||
v3.AggregateOperatorAvgRate,
|
||||
v3.AggregateOperatorMinRate,
|
||||
v3.AggregateOperatorMaxRate,
|
||||
v3.AggregateOperatorHistQuant50,
|
||||
v3.AggregateOperatorHistQuant75,
|
||||
v3.AggregateOperatorHistQuant90,
|
||||
v3.AggregateOperatorHistQuant95,
|
||||
v3.AggregateOperatorHistQuant99:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func Migrate(conn *sqlx.DB) error {
|
||||
ruleDB := rules.NewRuleDB(conn)
|
||||
storedRules, err := ruleDB.GetStoredRules(context.Background())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, storedRule := range storedRules {
|
||||
parsedRule, errs := rules.ParsePostableRule([]byte(storedRule.Data))
|
||||
if len(errs) > 0 {
|
||||
// this should not happen but if it does, we should not stop the migration
|
||||
zap.L().Error("Error parsing rule", zap.Error(multierr.Combine(errs...)), zap.Int("rule", storedRule.Id))
|
||||
continue
|
||||
}
|
||||
zap.L().Info("Rule parsed", zap.Int("rule", storedRule.Id))
|
||||
updated := false
|
||||
if parsedRule.RuleCondition != nil && parsedRule.Version == "" {
|
||||
if parsedRule.RuleCondition.QueryType() == v3.QueryTypeBuilder {
|
||||
// check if all the queries can be converted to v4
|
||||
canMigrate := true
|
||||
for _, query := range parsedRule.RuleCondition.CompositeQuery.BuilderQueries {
|
||||
if query.DataSource == v3.DataSourceMetrics && query.Expression == query.QueryName {
|
||||
if !canMigrateOperator(query.AggregateOperator) {
|
||||
canMigrate = false
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if canMigrate {
|
||||
parsedRule.Version = "v4"
|
||||
for _, query := range parsedRule.RuleCondition.CompositeQuery.BuilderQueries {
|
||||
if query.DataSource == v3.DataSourceMetrics && query.Expression == query.QueryName {
|
||||
// update aggregate attribute
|
||||
if query.AggregateOperator == v3.AggregateOperatorSum ||
|
||||
query.AggregateOperator == v3.AggregateOperatorMin ||
|
||||
query.AggregateOperator == v3.AggregateOperatorMax {
|
||||
query.AggregateAttribute.Type = "Gauge"
|
||||
}
|
||||
if query.AggregateOperator == v3.AggregateOperatorSumRate ||
|
||||
query.AggregateOperator == v3.AggregateOperatorAvgRate ||
|
||||
query.AggregateOperator == v3.AggregateOperatorMinRate ||
|
||||
query.AggregateOperator == v3.AggregateOperatorMaxRate {
|
||||
query.AggregateAttribute.Type = "Sum"
|
||||
}
|
||||
|
||||
if query.AggregateOperator == v3.AggregateOperatorHistQuant50 ||
|
||||
query.AggregateOperator == v3.AggregateOperatorHistQuant75 ||
|
||||
query.AggregateOperator == v3.AggregateOperatorHistQuant90 ||
|
||||
query.AggregateOperator == v3.AggregateOperatorHistQuant95 ||
|
||||
query.AggregateOperator == v3.AggregateOperatorHistQuant99 {
|
||||
query.AggregateAttribute.Type = "Histogram"
|
||||
}
|
||||
query.AggregateAttribute.DataType = v3.AttributeKeyDataTypeFloat64
|
||||
query.AggregateAttribute.IsColumn = true
|
||||
query.TimeAggregation = mapTimeAggregation[query.AggregateOperator]
|
||||
query.SpaceAggregation = mapSpaceAggregation[query.AggregateOperator]
|
||||
query.AggregateOperator = v3.AggregateOperator(query.TimeAggregation)
|
||||
updated = true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !updated {
|
||||
zap.L().Info("Rule not updated", zap.Int("rule", storedRule.Id))
|
||||
continue
|
||||
}
|
||||
|
||||
ruleJSON, jsonErr := json.Marshal(parsedRule)
|
||||
if jsonErr != nil {
|
||||
zap.L().Error("Error marshalling rule; skipping rule migration", zap.Error(jsonErr), zap.Int("rule", storedRule.Id))
|
||||
continue
|
||||
}
|
||||
|
||||
stmt, prepareError := conn.PrepareContext(context.Background(), `UPDATE rules SET data=$3 WHERE id=$4;`)
|
||||
if prepareError != nil {
|
||||
zap.L().Error("Error in preparing statement for UPDATE to rules", zap.Error(prepareError))
|
||||
continue
|
||||
}
|
||||
defer stmt.Close()
|
||||
|
||||
if _, err := stmt.Exec(ruleJSON, storedRule.Id); err != nil {
|
||||
zap.L().Error("Error in Executing prepared statement for UPDATE to rules", zap.Error(err))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@ -1,70 +0,0 @@
|
||||
package alertscustomstep
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"github.com/jmoiron/sqlx"
|
||||
v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
|
||||
"go.signoz.io/signoz/pkg/query-service/rules"
|
||||
"go.uber.org/multierr"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
var Version = "0.47-alerts-custom-step"
|
||||
|
||||
func Migrate(conn *sqlx.DB) error {
|
||||
ruleDB := rules.NewRuleDB(conn)
|
||||
storedRules, err := ruleDB.GetStoredRules(context.Background())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, storedRule := range storedRules {
|
||||
parsedRule, errs := rules.ParsePostableRule([]byte(storedRule.Data))
|
||||
if len(errs) > 0 {
|
||||
// this should not happen but if it does, we should not stop the migration
|
||||
zap.L().Error("Error parsing rule", zap.Error(multierr.Combine(errs...)), zap.Int("rule", storedRule.Id))
|
||||
continue
|
||||
}
|
||||
zap.L().Info("Rule parsed", zap.Int("rule", storedRule.Id))
|
||||
updated := false
|
||||
if parsedRule.RuleCondition != nil {
|
||||
if parsedRule.RuleCondition.QueryType() == v3.QueryTypeBuilder {
|
||||
if parsedRule.EvalWindow <= rules.Duration(6*time.Hour) {
|
||||
for _, query := range parsedRule.RuleCondition.CompositeQuery.BuilderQueries {
|
||||
if query.StepInterval > 60 {
|
||||
updated = true
|
||||
zap.L().Info("Updating step interval", zap.Int("rule", storedRule.Id), zap.Int64("old", query.StepInterval), zap.Int64("new", 60))
|
||||
query.StepInterval = 60
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !updated {
|
||||
zap.L().Info("Rule not updated", zap.Int("rule", storedRule.Id))
|
||||
continue
|
||||
}
|
||||
|
||||
ruleJSON, jsonErr := json.Marshal(parsedRule)
|
||||
if jsonErr != nil {
|
||||
zap.L().Error("Error marshalling rule; skipping rule migration", zap.Error(jsonErr), zap.Int("rule", storedRule.Id))
|
||||
continue
|
||||
}
|
||||
|
||||
stmt, prepareError := conn.PrepareContext(context.Background(), `UPDATE rules SET data=$3 WHERE id=$4;`)
|
||||
if prepareError != nil {
|
||||
zap.L().Error("Error in preparing statement for UPDATE to rules", zap.Error(prepareError))
|
||||
continue
|
||||
}
|
||||
defer stmt.Close()
|
||||
|
||||
if _, err := stmt.Exec(ruleJSON, storedRule.Id); err != nil {
|
||||
zap.L().Error("Error in Executing prepared statement for UPDATE to rules", zap.Error(err))
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@ -7,9 +7,6 @@ import (
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/driver"
|
||||
"github.com/jmoiron/sqlx"
|
||||
alertstov4 "go.signoz.io/signoz/pkg/query-service/migrate/0_45_alerts_to_v4"
|
||||
alertscustomstep "go.signoz.io/signoz/pkg/query-service/migrate/0_47_alerts_custom_step"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
type DataMigration struct {
|
||||
@ -56,28 +53,6 @@ func Migrate(dsn string) error {
|
||||
return err
|
||||
}
|
||||
|
||||
if m, err := getMigrationVersion(conn, "0.45_alerts_to_v4"); err == nil && m == nil {
|
||||
if err := alertstov4.Migrate(conn); err != nil {
|
||||
zap.L().Error("failed to migrate 0.45_alerts_to_v4", zap.Error(err))
|
||||
} else {
|
||||
_, err := conn.Exec("INSERT INTO data_migrations (version, succeeded) VALUES ('0.45_alerts_to_v4', true)")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if m, err := getMigrationVersion(conn, "0.47_alerts_custom_step"); err == nil && m == nil {
|
||||
if err := alertscustomstep.Migrate(conn); err != nil {
|
||||
zap.L().Error("failed to migrate 0.47_alerts_custom_step", zap.Error(err))
|
||||
} else {
|
||||
_, err := conn.Exec("INSERT INTO data_migrations (version, succeeded) VALUES ('0.47_alerts_custom_step', true)")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user