chore: add anomaly provider interface (#5856)

ee/query-service/anomaly/daily.go

@@ -0,0 +1,32 @@
+package anomaly
+
+import (
+	"context"
+)
+
+type DailyProvider struct {
+	BaseSeasonalProvider
+}
+
+var _ BaseProvider = (*DailyProvider)(nil)
+
+func (dp *DailyProvider) GetBaseSeasonalProvider() *BaseSeasonalProvider {
+	return &dp.BaseSeasonalProvider
+}
+
+// NewDailyProvider uses the same generic option type
+func NewDailyProvider(opts ...GenericProviderOption[*DailyProvider]) *DailyProvider {
+	dp := &DailyProvider{
+		BaseSeasonalProvider: BaseSeasonalProvider{},
+	}
+
+	for _, opt := range opts {
+		opt(dp)
+	}
+
+	return dp
+}
+
+func (p *DailyProvider) GetAnomalies(ctx context.Context, req *GetAnomaliesRequest) (*GetAnomaliesResponse, error) {
+	return nil, nil
+}

ee/query-service/anomaly/hourly.go

@@ -0,0 +1,32 @@
+package anomaly
+
+import (
+	"context"
+)
+
+type HourlyProvider struct {
+	BaseSeasonalProvider
+}
+
+var _ BaseProvider = (*HourlyProvider)(nil)
+
+func (hp *HourlyProvider) GetBaseSeasonalProvider() *BaseSeasonalProvider {
+	return &hp.BaseSeasonalProvider
+}
+
+// NewHourlyProvider now uses the generic option type
+func NewHourlyProvider(opts ...GenericProviderOption[*HourlyProvider]) *HourlyProvider {
+	hp := &HourlyProvider{
+		BaseSeasonalProvider: BaseSeasonalProvider{},
+	}
+
+	for _, opt := range opts {
+		opt(hp)
+	}
+
+	return hp
+}
+
+func (p *HourlyProvider) GetAnomalies(ctx context.Context, req *GetAnomaliesRequest) (*GetAnomaliesResponse, error) {
+	return nil, nil
+}

ee/query-service/anomaly/params.go

@@ -0,0 +1,188 @@
+package anomaly
+
+import (
+	"math"
+	"time"
+
+	"go.signoz.io/signoz/pkg/query-service/common"
+	v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
+)
+
+type Seasonality string
+
+const (
+	SeasonalityHourly Seasonality = "hourly"
+	SeasonalityDaily  Seasonality = "daily"
+	SeasonalityWeekly Seasonality = "weekly"
+)
+
+func (s Seasonality) IsValid() bool {
+	switch s {
+	case SeasonalityHourly, SeasonalityDaily, SeasonalityWeekly:
+		return true
+	default:
+		return false
+	}
+}
+
+type GetAnomaliesRequest struct {
+	Params      *v3.QueryRangeParamsV3
+	Seasonality Seasonality
+}
+
+type GetAnomaliesResponse struct {
+	Results []*v3.Result
+}
+
+// anomalyParams is the params for anomaly detection
+// prediction = avg(past_period_query) + avg(current_season_query) - avg(past_season_query)
+//
+//	                  ^                                  ^
+//		              |                                  |
+//			(rounded value for past peiod)    +      (seasonal growth)
+//
+// score = abs(value - prediction) / stddev (current_season_query)
+type anomalyQueryParams struct {
+	// CurrentPeriodQuery is the query range params for period user is looking at or eval window
+	// Example: (now-5m, now), (now-30m, now), (now-1h, now)
+	// The results obtained from this query are used to compare with predicted values
+	// and to detect anomalies
+	CurrentPeriodQuery *v3.QueryRangeParamsV3
+	// PastPeriodQuery is the query range params for past seasonal period
+	// Example: For weekly seasonality, (now-1w-4h-5m, now-1w)
+	//        : For daily seasonality, (now-1d-2h-5m, now-1d)
+	//        : For hourly seasonality, (now-1h-30m-5m, now-1h)
+	PastPeriodQuery *v3.QueryRangeParamsV3
+	// CurrentSeasonQuery is the query range params for current period (seasonal)
+	// Example: For weekly seasonality, this is the query range params for the (now-1w-5m, now)
+	//        : For daily seasonality, this is the query range params for the (now-1d-5m, now)
+	//        : For hourly seasonality, this is the query range params for the (now-1h-5m, now)
+	CurrentSeasonQuery *v3.QueryRangeParamsV3
+	// PastSeasonQuery is the query range params for past seasonal period to the current season
+	// Example: For weekly seasonality, this is the query range params for the (now-2w-5m, now-1w)
+	//        : For daily seasonality, this is the query range params for the (now-2d-5m, now-1d)
+	//        : For hourly seasonality, this is the query range params for the (now-2h-5m, now-1h)
+	PastSeasonQuery *v3.QueryRangeParamsV3
+}
+
+func copyCompositeQuery(req *v3.QueryRangeParamsV3) *v3.CompositeQuery {
+	deepCopyCompositeQuery := *req.CompositeQuery
+	deepCopyCompositeQuery.BuilderQueries = make(map[string]*v3.BuilderQuery)
+	for k, v := range req.CompositeQuery.BuilderQueries {
+		query := *v
+		deepCopyCompositeQuery.BuilderQueries[k] = &query
+	}
+	return &deepCopyCompositeQuery
+}
+
+func updateStepInterval(req *v3.QueryRangeParamsV3) {
+	start := req.Start
+	end := req.End
+
+	req.Step = int64(math.Max(float64(common.MinAllowedStepInterval(start, end)), 60))
+	for _, q := range req.CompositeQuery.BuilderQueries {
+		// If the step interval is less than the minimum allowed step interval, set it to the minimum allowed step interval
+		if minStep := common.MinAllowedStepInterval(start, end); q.StepInterval < minStep {
+			q.StepInterval = minStep
+		}
+	}
+}
+
+func prepareAnomalyQueryParams(req *v3.QueryRangeParamsV3, seasonality Seasonality) *anomalyQueryParams {
+	start := req.Start
+	end := req.End
+
+	currentPeriodQuery := &v3.QueryRangeParamsV3{
+		Start:          start,
+		End:            end,
+		CompositeQuery: req.CompositeQuery,
+		Variables:      make(map[string]interface{}, 0),
+		NoCache:        false,
+	}
+	updateStepInterval(currentPeriodQuery)
+
+	var pastPeriodStart, pastPeriodEnd int64
+
+	switch seasonality {
+	// for one week period, we fetch the data from the past week with 4 hours offset
+	case SeasonalityWeekly:
+		pastPeriodStart = start - 166*time.Hour.Milliseconds() - 4*time.Hour.Milliseconds()
+		pastPeriodEnd = end - 166*time.Hour.Milliseconds()
+	// for one day period, we fetch the data from the past day with 2 hours offset
+	case SeasonalityDaily:
+		pastPeriodStart = start - 23*time.Hour.Milliseconds() - 2*time.Hour.Milliseconds()
+		pastPeriodEnd = end - 23*time.Hour.Milliseconds()
+	// for one hour period, we fetch the data from the past hour with 30 minutes offset
+	case SeasonalityHourly:
+		pastPeriodStart = start - 1*time.Hour.Milliseconds() - 30*time.Minute.Milliseconds()
+		pastPeriodEnd = end - 1*time.Hour.Milliseconds()
+	}
+
+	pastPeriodQuery := &v3.QueryRangeParamsV3{
+		Start:          pastPeriodStart,
+		End:            pastPeriodEnd,
+		CompositeQuery: copyCompositeQuery(req),
+		Variables:      make(map[string]interface{}, 0),
+		NoCache:        false,
+	}
+	updateStepInterval(pastPeriodQuery)
+
+	// seasonality growth trend
+	var currentGrowthPeriodStart, currentGrowthPeriodEnd int64
+	switch seasonality {
+	case SeasonalityWeekly:
+		currentGrowthPeriodStart = start - 7*24*time.Hour.Milliseconds()
+		currentGrowthPeriodEnd = end
+	case SeasonalityDaily:
+		currentGrowthPeriodStart = start - 23*time.Hour.Milliseconds()
+		currentGrowthPeriodEnd = end
+	case SeasonalityHourly:
+		currentGrowthPeriodStart = start - 1*time.Hour.Milliseconds()
+		currentGrowthPeriodEnd = end
+	}
+
+	currentGrowthQuery := &v3.QueryRangeParamsV3{
+		Start:          currentGrowthPeriodStart,
+		End:            currentGrowthPeriodEnd,
+		CompositeQuery: copyCompositeQuery(req),
+		Variables:      make(map[string]interface{}, 0),
+		NoCache:        false,
+	}
+	updateStepInterval(currentGrowthQuery)
+
+	var pastGrowthPeriodStart, pastGrowthPeriodEnd int64
+	switch seasonality {
+	case SeasonalityWeekly:
+		pastGrowthPeriodStart = start - 14*24*time.Hour.Milliseconds()
+		pastGrowthPeriodEnd = start - 7*24*time.Hour.Milliseconds()
+	case SeasonalityDaily:
+		pastGrowthPeriodStart = start - 2*time.Hour.Milliseconds()
+		pastGrowthPeriodEnd = start - 1*time.Hour.Milliseconds()
+	case SeasonalityHourly:
+		pastGrowthPeriodStart = start - 2*time.Hour.Milliseconds()
+		pastGrowthPeriodEnd = start - 1*time.Hour.Milliseconds()
+	}
+
+	pastGrowthQuery := &v3.QueryRangeParamsV3{
+		Start:          pastGrowthPeriodStart,
+		End:            pastGrowthPeriodEnd,
+		CompositeQuery: copyCompositeQuery(req),
+		Variables:      make(map[string]interface{}, 0),
+		NoCache:        false,
+	}
+	updateStepInterval(pastGrowthQuery)
+
+	return &anomalyQueryParams{
+		CurrentPeriodQuery: currentPeriodQuery,
+		PastPeriodQuery:    pastPeriodQuery,
+		CurrentSeasonQuery: currentGrowthQuery,
+		PastSeasonQuery:    pastGrowthQuery,
+	}
+}
+
+type anomalyQueryResults struct {
+	CurrentPeriodResults []*v3.Result
+	PastPeriodResults    []*v3.Result
+	CurrentSeasonResults []*v3.Result
+	PastSeasonResults    []*v3.Result
+}

ee/query-service/anomaly/provider.go

@@ -0,0 +1,9 @@
+package anomaly
+
+import (
+	"context"
+)
+
+type Provider interface {
+	GetAnomalies(ctx context.Context, req *GetAnomaliesRequest) (*GetAnomaliesResponse, error)
+}

ee/query-service/anomaly/seasonal.go

@@ -0,0 +1,229 @@
+package anomaly
+
+import (
+	"context"
+	"math"
+
+	"go.signoz.io/signoz/pkg/query-service/cache"
+	"go.signoz.io/signoz/pkg/query-service/interfaces"
+	v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
+	"go.signoz.io/signoz/pkg/query-service/utils/labels"
+	"go.uber.org/zap"
+)
+
+// BaseProvider is an interface that includes common methods for all provider types
+type BaseProvider interface {
+	GetBaseSeasonalProvider() *BaseSeasonalProvider
+}
+
+// GenericProviderOption is a generic type for provider options
+type GenericProviderOption[T BaseProvider] func(T)
+
+func WithCache[T BaseProvider](cache cache.Cache) GenericProviderOption[T] {
+	return func(p T) {
+		p.GetBaseSeasonalProvider().cache = cache
+	}
+}
+
+func WithKeyGenerator[T BaseProvider](keyGenerator cache.KeyGenerator) GenericProviderOption[T] {
+	return func(p T) {
+		p.GetBaseSeasonalProvider().keyGenerator = keyGenerator
+	}
+}
+
+func WithFeatureLookup[T BaseProvider](ff interfaces.FeatureLookup) GenericProviderOption[T] {
+	return func(p T) {
+		p.GetBaseSeasonalProvider().ff = ff
+	}
+}
+
+func WithReader[T BaseProvider](reader interfaces.Reader) GenericProviderOption[T] {
+	return func(p T) {
+		p.GetBaseSeasonalProvider().reader = reader
+	}
+}
+
+type BaseSeasonalProvider struct {
+	querierV2    interfaces.Querier
+	reader       interfaces.Reader
+	cache        cache.Cache
+	keyGenerator cache.KeyGenerator
+	ff           interfaces.FeatureLookup
+}
+
+func (p *BaseSeasonalProvider) getQueryParams(req *GetAnomaliesRequest) *anomalyQueryParams {
+	if !req.Seasonality.IsValid() {
+		req.Seasonality = SeasonalityWeekly
+	}
+	return prepareAnomalyQueryParams(req.Params, req.Seasonality)
+}
+
+func (p *BaseSeasonalProvider) getResults(ctx context.Context, params *anomalyQueryParams) (*anomalyQueryResults, error) {
+	currentPeriodResults, _, err := p.querierV2.QueryRange(ctx, params.CurrentPeriodQuery, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	pastPeriodResults, _, err := p.querierV2.QueryRange(ctx, params.PastPeriodQuery, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	currentSeasonResults, _, err := p.querierV2.QueryRange(ctx, params.CurrentSeasonQuery, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	pastSeasonResults, _, err := p.querierV2.QueryRange(ctx, params.PastSeasonQuery, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	return &anomalyQueryResults{
+		CurrentPeriodResults: currentPeriodResults,
+		PastPeriodResults:    pastPeriodResults,
+		CurrentSeasonResults: currentSeasonResults,
+		PastSeasonResults:    pastSeasonResults,
+	}, nil
+}
+
+func (p *BaseSeasonalProvider) getMatchingSeries(queryResult *v3.Result, series *v3.Series) *v3.Series {
+	for _, curr := range queryResult.Series {
+		currLabels := labels.FromMap(curr.Labels)
+		seriesLabels := labels.FromMap(series.Labels)
+		if currLabels.Hash() == seriesLabels.Hash() {
+			return curr
+		}
+	}
+	return nil
+}
+
+func (p *BaseSeasonalProvider) getAvg(series *v3.Series) float64 {
+	var sum float64
+	for _, smpl := range series.Points {
+		sum += smpl.Value
+	}
+	return sum / float64(len(series.Points))
+}
+
+func (p *BaseSeasonalProvider) getStdDev(series *v3.Series) float64 {
+	avg := p.getAvg(series)
+	var sum float64
+	for _, smpl := range series.Points {
+		sum += math.Pow(smpl.Value-avg, 2)
+	}
+	return math.Sqrt(sum / float64(len(series.Points)))
+}
+
+func (p *BaseSeasonalProvider) getPredictedSeries(series, prevSeries, currentSeasonSeries, pastSeasonSeries *v3.Series) *v3.Series {
+	predictedSeries := &v3.Series{
+		Labels:      series.Labels,
+		LabelsArray: series.LabelsArray,
+		Points:      []v3.Point{},
+	}
+
+	for _, curr := range series.Points {
+		predictedValue := p.getAvg(prevSeries) + p.getAvg(currentSeasonSeries) - p.getAvg(pastSeasonSeries)
+		predictedSeries.Points = append(predictedSeries.Points, v3.Point{
+			Timestamp: curr.Timestamp,
+			Value:     predictedValue,
+		})
+	}
+
+	return predictedSeries
+}
+
+func (p *BaseSeasonalProvider) getExpectedValue(_, prevSeries, currentSeasonSeries, pastSeasonSeries *v3.Series) float64 {
+	prevSeriesAvg := p.getAvg(prevSeries)
+	currentSeasonSeriesAvg := p.getAvg(currentSeasonSeries)
+	pastSeasonSeriesAvg := p.getAvg(pastSeasonSeries)
+	zap.L().Debug("getExpectedValue",
+		zap.Float64("prevSeriesAvg", prevSeriesAvg),
+		zap.Float64("currentSeasonSeriesAvg", currentSeasonSeriesAvg),
+		zap.Float64("pastSeasonSeriesAvg", pastSeasonSeriesAvg),
+		zap.Float64("expectedValue", prevSeriesAvg+currentSeasonSeriesAvg-pastSeasonSeriesAvg),
+	)
+	return prevSeriesAvg + currentSeasonSeriesAvg - pastSeasonSeriesAvg
+}
+
+func (p *BaseSeasonalProvider) getScore(series, prevSeries, weekSeries, weekPrevSeries *v3.Series, value float64) float64 {
+	expectedValue := p.getExpectedValue(series, prevSeries, weekSeries, weekPrevSeries)
+	return (value - expectedValue) / p.getStdDev(weekSeries)
+}
+
+func (p *BaseSeasonalProvider) getAnomalyScores(series, prevSeries, currentSeasonSeries, pastSeasonSeries *v3.Series) *v3.Series {
+	anomalyScoreSeries := &v3.Series{
+		Labels:      series.Labels,
+		LabelsArray: series.LabelsArray,
+		Points:      []v3.Point{},
+	}
+
+	for _, curr := range series.Points {
+		anomalyScore := p.getScore(series, prevSeries, currentSeasonSeries, pastSeasonSeries, curr.Value)
+		anomalyScoreSeries.Points = append(anomalyScoreSeries.Points, v3.Point{
+			Timestamp: curr.Timestamp,
+			Value:     anomalyScore,
+		})
+	}
+
+	return anomalyScoreSeries
+}
+
+func (p *BaseSeasonalProvider) GetAnomalies(ctx context.Context, req *GetAnomaliesRequest) (*GetAnomaliesResponse, error) {
+	anomalyParams := p.getQueryParams(req)
+	anomalyQueryResults, err := p.getResults(ctx, anomalyParams)
+	if err != nil {
+		return nil, err
+	}
+
+	currentPeriodResultsMap := make(map[string]*v3.Result)
+	for _, result := range anomalyQueryResults.CurrentPeriodResults {
+		currentPeriodResultsMap[result.QueryName] = result
+	}
+
+	pastPeriodResultsMap := make(map[string]*v3.Result)
+	for _, result := range anomalyQueryResults.PastPeriodResults {
+		pastPeriodResultsMap[result.QueryName] = result
+	}
+
+	currentSeasonResultsMap := make(map[string]*v3.Result)
+	for _, result := range anomalyQueryResults.CurrentSeasonResults {
+		currentSeasonResultsMap[result.QueryName] = result
+	}
+
+	pastSeasonResultsMap := make(map[string]*v3.Result)
+	for _, result := range anomalyQueryResults.PastSeasonResults {
+		pastSeasonResultsMap[result.QueryName] = result
+	}
+
+	for _, result := range currentPeriodResultsMap {
+		pastPeriodResult, ok := pastPeriodResultsMap[result.QueryName]
+		if !ok {
+			continue
+		}
+		currentSeasonResult, ok := currentSeasonResultsMap[result.QueryName]
+		if !ok {
+			continue
+		}
+		pastSeasonResult, ok := pastSeasonResultsMap[result.QueryName]
+		if !ok {
+			continue
+		}
+
+		for _, series := range result.Series {
+			pastPeriodSeries := p.getMatchingSeries(pastPeriodResult, series)
+			currentSeasonSeries := p.getMatchingSeries(currentSeasonResult, series)
+			pastSeasonSeries := p.getMatchingSeries(pastSeasonResult, series)
+
+			predictedSeries := p.getPredictedSeries(series, pastPeriodSeries, currentSeasonSeries, pastSeasonSeries)
+			result.PredictedSeries = append(result.PredictedSeries, predictedSeries)
+
+			anomalyScoreSeries := p.getAnomalyScores(series, pastPeriodSeries, currentSeasonSeries, pastSeasonSeries)
+			result.AnomalyScores = append(result.AnomalyScores, anomalyScoreSeries)
+		}
+	}
+
+	return &GetAnomaliesResponse{
+		Results: anomalyQueryResults.CurrentPeriodResults,
+	}, nil
+}

ee/query-service/anomaly/weekly.go

@@ -0,0 +1,31 @@
+package anomaly
+
+import (
+	"context"
+)
+
+type WeeklyProvider struct {
+	BaseSeasonalProvider
+}
+
+var _ BaseProvider = (*WeeklyProvider)(nil)
+
+func (wp *WeeklyProvider) GetBaseSeasonalProvider() *BaseSeasonalProvider {
+	return &wp.BaseSeasonalProvider
+}
+
+func NewWeeklyProvider(opts ...GenericProviderOption[*WeeklyProvider]) *WeeklyProvider {
+	wp := &WeeklyProvider{
+		BaseSeasonalProvider: BaseSeasonalProvider{},
+	}
+
+	for _, opt := range opts {
+		opt(wp)
+	}
+
+	return wp
+}
+
+func (p *WeeklyProvider) GetAnomalies(ctx context.Context, req *GetAnomaliesRequest) (*GetAnomaliesResponse, error) {
+	return nil, nil
+}

pkg/query-service/model/v3/v3.go

@@ -1029,10 +1029,12 @@ type Table struct {
 }
 
 type Result struct {
-	QueryName string    `json:"queryName,omitempty"`
+	QueryName       string    `json:"queryName,omitempty"`
-	Series    []*Series `json:"series,omitempty"`
+	Series          []*Series `json:"series,omitempty"`
-	List      []*Row    `json:"list,omitempty"`
+	PredictedSeries []*Series `json:"predictedSeries,omitempty"`
-	Table     *Table    `json:"table,omitempty"`
+	AnomalyScores   []*Series `json:"anomalyScores,omitempty"`
+	List            []*Row    `json:"list,omitempty"`
+	Table           *Table    `json:"table,omitempty"`
 }
 
 type LogsLiveTailClientV2 struct {